From 9c9cd757734eaa1032f84d3b7dc2cea47575acd5 Mon Sep 17 00:00:00 2001 From: Flora Date: Thu, 29 May 2025 10:26:08 +0200 Subject: [PATCH] Add wrappers for microgrid and electrical components Signed-off-by: Flora --- pyproject.toml | 2 + .../client/common/microgrid/_delivery_area.py | 89 +++++ .../common/microgrid/_delivery_area_proto.py | 44 +++ .../client/common/microgrid/_lifetime.py | 52 +++ .../client/common/microgrid/_location.py | 50 +++ .../common/microgrid/_location_proto.py | 47 +++ .../common/microgrid/_microgrid_info.py | 87 +++++ .../common/microgrid/_microgrid_info_proto.py | 79 ++++ src/frequenz/client/common/microgrid/_util.py | 47 +++ .../electrical_components/__init__.py | 90 +---- .../electrical_components/_connection.py | 35 ++ .../_electrical_component.py | 345 ++++++++++++++++++ src/frequenz/client/common/microgrid/id.py | 223 +++++++++++ 13 files changed, 1101 insertions(+), 89 deletions(-) create mode 100644 src/frequenz/client/common/microgrid/_delivery_area.py create mode 100644 src/frequenz/client/common/microgrid/_delivery_area_proto.py create mode 100644 src/frequenz/client/common/microgrid/_lifetime.py create mode 100644 src/frequenz/client/common/microgrid/_location.py create mode 100644 src/frequenz/client/common/microgrid/_location_proto.py create mode 100644 src/frequenz/client/common/microgrid/_microgrid_info.py create mode 100644 src/frequenz/client/common/microgrid/_microgrid_info_proto.py create mode 100644 src/frequenz/client/common/microgrid/_util.py create mode 100644 src/frequenz/client/common/microgrid/electrical_components/_connection.py create mode 100644 src/frequenz/client/common/microgrid/electrical_components/_electrical_component.py create mode 100644 src/frequenz/client/common/microgrid/id.py diff --git a/pyproject.toml b/pyproject.toml index 8be5a8e..451707f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,6 +27,8 @@ classifiers = [ requires-python = ">= 3.11, < 4" dependencies = [ "typing-extensions >= 4.6.0, < 5", + "timezonefinder >= 6.2.0, < 7", + "frequenz-client-base >= 0.8.0, < 0.12.0", "frequenz-api-common @ git+https://github.com/frequenz-floss/frequenz-api-common.git@2e89add6a16d42b23612f0f791a499919f3738ed", ] dynamic = ["version"] diff --git a/src/frequenz/client/common/microgrid/_delivery_area.py b/src/frequenz/client/common/microgrid/_delivery_area.py new file mode 100644 index 0000000..1746f60 --- /dev/null +++ b/src/frequenz/client/common/microgrid/_delivery_area.py @@ -0,0 +1,89 @@ +# License: MIT +# Copyright © 2025 Frequenz Energy-as-a-Service GmbH + +"""Delivery area information for the energy market.""" + +import enum +from dataclasses import dataclass + +from frequenz.api.common.v1.grid import delivery_area_pb2 + + +@enum.unique +class EnergyMarketCodeType(enum.Enum): + """The identification code types used in the energy market. + + CodeType specifies the type of identification code used for uniquely + identifying various entities such as delivery areas, market participants, + and grid components within the energy market. + + This enumeration aims to + offer compatibility across different jurisdictional standards. + + Note: Understanding Code Types + Different regions or countries may have their own standards for uniquely + identifying various entities within the energy market. For example, in + Europe, the Energy Identification Code (EIC) is commonly used for this + purpose. + + Note: Extensibility + New code types can be added to this enum to accommodate additional regional + standards, enhancing the API's adaptability. + + Danger: Validation Required + The chosen code type should correspond correctly with the `code` field in + the relevant message objects, such as `DeliveryArea` or `Counterparty`. + Failure to match the code type with the correct code could lead to + processing errors. + """ + + UNSPECIFIED = delivery_area_pb2.ENERGY_MARKET_CODE_TYPE_UNSPECIFIED + """Unspecified type. This value is a placeholder and should not be used.""" + + EUROPE_EIC = delivery_area_pb2.ENERGY_MARKET_CODE_TYPE_EUROPE_EIC + """European Energy Identification Code Standard.""" + + US_NERC = delivery_area_pb2.ENERGY_MARKET_CODE_TYPE_US_NERC + """North American Electric Reliability Corporation identifiers.""" + + +@dataclass(frozen=True, kw_only=True) +class DeliveryArea: + """A geographical or administrative region where electricity deliveries occur. + + DeliveryArea represents the geographical or administrative region, usually defined + and maintained by a Transmission System Operator (TSO), where electricity deliveries + for a contract occur. + + The concept is important to energy trading as it delineates the agreed-upon delivery + location. Delivery areas can have different codes based on the jurisdiction in + which they operate. + + Note: Jurisdictional Differences + This is typically represented by specific codes according to local jurisdiction. + + In Europe, this is represented by an + [EIC](https://en.wikipedia.org/wiki/Energy_Identification_Code) (Energy + Identification Code). [List of + EICs](https://www.entsoe.eu/data/energy-identification-codes-eic/eic-approved-codes/). + """ + + code: str | None + """The code representing the unique identifier for the delivery area.""" + + code_type: EnergyMarketCodeType | int + """Type of code used for identifying the delivery area itself. + + This code could be extended in the future, in case an unknown code type is + encountered, a plain integer value is used to represent it. + """ + + def __str__(self) -> str: + """Return a human-readable string representation of this instance.""" + code = self.code or "" + code_type = ( + f"type={self.code_type}" + if isinstance(self.code_type, int) + else self.code_type.name + ) + return f"{code}[{code_type}]" diff --git a/src/frequenz/client/common/microgrid/_delivery_area_proto.py b/src/frequenz/client/common/microgrid/_delivery_area_proto.py new file mode 100644 index 0000000..7824757 --- /dev/null +++ b/src/frequenz/client/common/microgrid/_delivery_area_proto.py @@ -0,0 +1,44 @@ +# License: MIT +# Copyright © 2025 Frequenz Energy-as-a-Service GmbH + +"""Loading of DeliveryArea objects from protobuf messages.""" + +import logging + +from frequenz.api.common.v1.grid import delivery_area_pb2 + +from ._delivery_area import DeliveryArea, EnergyMarketCodeType +from ._util import enum_from_proto + +_logger = logging.getLogger(__name__) + + +def delivery_area_from_proto(message: delivery_area_pb2.DeliveryArea) -> DeliveryArea: + """Convert a protobuf delivery area message to a delivery area object. + + Args: + message: The protobuf message to convert. + + Returns: + The resulting delivery area object. + """ + issues: list[str] = [] + + code = message.code or None + if code is None: + issues.append("code is empty") + + code_type = enum_from_proto(message.code_type, EnergyMarketCodeType) + if code_type is EnergyMarketCodeType.UNSPECIFIED: + issues.append("code_type is unspecified") + elif isinstance(code_type, int): + issues.append("code_type is unrecognized") + + if issues: + _logger.warning( + "Found issues in delivery area: %s | Protobuf message:\n%s", + ", ".join(issues), + message, + ) + + return DeliveryArea(code=code, code_type=code_type) diff --git a/src/frequenz/client/common/microgrid/_lifetime.py b/src/frequenz/client/common/microgrid/_lifetime.py new file mode 100644 index 0000000..fa1ab04 --- /dev/null +++ b/src/frequenz/client/common/microgrid/_lifetime.py @@ -0,0 +1,52 @@ +# License: MIT +# Copyright © 2025 Frequenz Energy-as-a-Service GmbH + +"""Lifetime of a microgrid asset.""" + + +from dataclasses import dataclass +from datetime import datetime, timezone + + +@dataclass(frozen=True, kw_only=True) +class Lifetime: + """An active operational period of a microgrid asset. + + Warning: + The [`end`][frequenz.client.microgrid.Lifetime.end] timestamp indicates that the + asset has been permanently removed from the system. + """ + + start: datetime | None = None + """The moment when the asset became operationally active. + + If `None`, the asset is considered to be active in any past moment previous to the + [`end`][frequenz.client.microgrid.Lifetime.end]. + """ + + end: datetime | None = None + """The moment when the asset's operational activity ceased. + + If `None`, the asset is considered to be active with no plans to be deactivated. + """ + + def __post_init__(self) -> None: + """Validate this lifetime.""" + if self.start is not None and self.end is not None and self.start > self.end: + raise ValueError("Start must be before or equal to end.") + + def is_operational_at(self, timestamp: datetime) -> bool: + """Check whether this lifetime is active at a specific timestamp.""" + # Handle start time - it's not active if start is in the future + if self.start is not None and self.start > timestamp: + return False + # Handle end time - active up to and including end time + if self.end is not None: + return self.end >= timestamp + # self.end is None, and either self.start is None or self.start <= timestamp, + # so it is active at this timestamp + return True + + def is_operational_now(self) -> bool: + """Whether this lifetime is currently active.""" + return self.is_operational_at(datetime.now(timezone.utc)) diff --git a/src/frequenz/client/common/microgrid/_location.py b/src/frequenz/client/common/microgrid/_location.py new file mode 100644 index 0000000..af565e0 --- /dev/null +++ b/src/frequenz/client/common/microgrid/_location.py @@ -0,0 +1,50 @@ +# License: MIT +# Copyright © 2025 Frequenz Energy-as-a-Service GmbH + +"""Location information for a microgrid.""" + + +import logging +from dataclasses import dataclass +from functools import cached_property +from zoneinfo import ZoneInfo + +import timezonefinder + +_timezone_finder = timezonefinder.TimezoneFinder() +_logger = logging.getLogger(__name__) + + +@dataclass(frozen=True, kw_only=True) +class Location: + """A location of a microgrid.""" + + latitude: float | None + """The latitude of the microgrid in degree.""" + + longitude: float | None + """The longitude of the microgrid in degree.""" + + country_code: str | None + """The country code of the microgrid in ISO 3166-1 Alpha 2 format.""" + + @cached_property + def timezone(self) -> ZoneInfo | None: + """The timezone of the microgrid, or `None` if it could not be determined.""" + if self.latitude is None or self.longitude is None: + _logger.warning( + "Latitude (%s) or longitude (%s) missing, cannot determine timezone" + ) + return None + timezone = _timezone_finder.timezone_at(lat=self.latitude, lng=self.longitude) + return ZoneInfo(key=timezone) if timezone else None + + def __str__(self) -> str: + """Return the short string representation of this instance.""" + country = self.country_code or "" + lat = f"{self.latitude:.2f}" if self.latitude is not None else "?" + lon = f"{self.longitude:.2f}" if self.longitude is not None else "?" + coordinates = "" + if self.latitude is not None or self.longitude is not None: + coordinates = f":({lat}, {lon})" + return f"{country}{coordinates}" diff --git a/src/frequenz/client/common/microgrid/_location_proto.py b/src/frequenz/client/common/microgrid/_location_proto.py new file mode 100644 index 0000000..301fada --- /dev/null +++ b/src/frequenz/client/common/microgrid/_location_proto.py @@ -0,0 +1,47 @@ +# License: MIT +# Copyright © 2025 Frequenz Energy-as-a-Service GmbH + +"""Loading of Location objects from protobuf messages.""" + +import logging + +from frequenz.api.common.v1 import location_pb2 + +from ._location import Location + +_logger = logging.getLogger(__name__) + + +def location_from_proto(message: location_pb2.Location) -> Location: + """Convert a protobuf location message to a location object. + + Args: + message: The protobuf message to convert. + + Returns: + The resulting location object. + """ + issues: list[str] = [] + + latitude: float | None = message.latitude if -90 <= message.latitude <= 90 else None + if latitude is None: + issues.append("latitude out of range [-90, 90]") + + longitude: float | None = ( + message.longitude if -180 <= message.longitude <= 180 else None + ) + if longitude is None: + issues.append("longitude out of range [-180, 180]") + + country_code = message.country_code or None + if country_code is None: + issues.append("country code is empty") + + if issues: + _logger.warning( + "Found issues in location: %s | Protobuf message:\n%s", + ", ".join(issues), + message, + ) + + return Location(latitude=latitude, longitude=longitude, country_code=country_code) diff --git a/src/frequenz/client/common/microgrid/_microgrid_info.py b/src/frequenz/client/common/microgrid/_microgrid_info.py new file mode 100644 index 0000000..e73c421 --- /dev/null +++ b/src/frequenz/client/common/microgrid/_microgrid_info.py @@ -0,0 +1,87 @@ +# License: MIT +# Copyright © 2025 Frequenz Energy-as-a-Service GmbH + +"""Definition of a microgrid.""" + +import datetime +import enum +import logging +from dataclasses import dataclass +from functools import cached_property + +from frequenz.api.common.v1.microgrid import microgrid_pb2 + +from ._delivery_area import DeliveryArea +from ._location import Location +from .id import EnterpriseId, MicrogridId + +_logger = logging.getLogger(__name__) + + +@enum.unique +class MicrogridStatus(enum.Enum): + """The possible statuses for a microgrid.""" + + UNSPECIFIED = microgrid_pb2.MICROGRID_STATUS_UNSPECIFIED + """The status is unspecified. This should not be used.""" + + ACTIVE = microgrid_pb2.MICROGRID_STATUS_ACTIVE + """The microgrid is active.""" + + INACTIVE = microgrid_pb2.MICROGRID_STATUS_INACTIVE + """The microgrid is inactive.""" + + +@dataclass(frozen=True, kw_only=True) +class MicrogridInfo: + """A localized grouping of electricity generation, energy storage, and loads. + + A microgrid is a localized grouping of electricity generation, energy storage, and + loads that normally operates connected to a traditional centralized grid. + + Each microgrid has a unique identifier and is associated with an enterprise account. + + A key feature is that it has a physical location and is situated in a delivery area. + + Note: Key Concepts + - Physical Location: Geographical coordinates specify the exact physical + location of the microgrid. + - Delivery Area: Each microgrid is part of a broader delivery area, which is + crucial for energy trading and compliance. + """ + + id: MicrogridId + """The unique identifier of the microgrid.""" + + enterprise_id: EnterpriseId + """The unique identifier linking this microgrid to its parent enterprise account.""" + + name: str | None + """Name of the microgrid.""" + + delivery_area: DeliveryArea | None + """The delivery area where the microgrid is located, as identified by a specific code.""" + + location: Location | None + """Physical location of the microgrid, in geographical co-ordinates.""" + + status: MicrogridStatus | int + """The current status of the microgrid.""" + + create_timestamp: datetime.datetime + """The UTC timestamp indicating when the microgrid was initially created.""" + + @cached_property + def is_active(self) -> bool: + """Whether the microgrid is active.""" + if self.status is MicrogridStatus.UNSPECIFIED: + # Because this is a cached property, the warning will only be logged once. + _logger.warning( + "Microgrid %s has an unspecified status. Assuming it is active.", self + ) + return self.status in (MicrogridStatus.ACTIVE, MicrogridStatus.UNSPECIFIED) + + def __str__(self) -> str: + """Return the ID of this microgrid as a string.""" + name = f":{self.name}" if self.name else "" + return f"{self.id}{name}" diff --git a/src/frequenz/client/common/microgrid/_microgrid_info_proto.py b/src/frequenz/client/common/microgrid/_microgrid_info_proto.py new file mode 100644 index 0000000..53211b3 --- /dev/null +++ b/src/frequenz/client/common/microgrid/_microgrid_info_proto.py @@ -0,0 +1,79 @@ +# License: MIT +# Copyright © 2025 Frequenz Energy-as-a-Service GmbH + +"""Loading of MicrogridInfo objects from protobuf messages.""" + + +import logging + +from frequenz.api.common.v1.microgrid import microgrid_pb2 +from frequenz.client.base import conversion + +from ._delivery_area import DeliveryArea +from ._delivery_area_proto import delivery_area_from_proto +from ._location import Location +from ._location_proto import location_from_proto +from ._microgrid_info import MicrogridInfo, MicrogridStatus +from ._util import enum_from_proto +from .id import EnterpriseId, MicrogridId + +_logger = logging.getLogger(__name__) + + +def microgrid_info_from_proto(message: microgrid_pb2.Microgrid) -> MicrogridInfo: + """Convert a protobuf microgrid message to a microgrid object. + + Args: + message: The protobuf message to convert. + + Returns: + The resulting microgrid object. + """ + major_issues: list[str] = [] + minor_issues: list[str] = [] + + delivery_area: DeliveryArea | None = None + if message.HasField("delivery_area"): + delivery_area = delivery_area_from_proto(message.delivery_area) + else: + major_issues.append("delivery_area is missing") + + location: Location | None = None + if message.HasField("location"): + location = location_from_proto(message.location) + else: + major_issues.append("location is missing") + + name = message.name or None + if name is None: + minor_issues.append("name is empty") + + status = enum_from_proto(message.status, MicrogridStatus) + if status is MicrogridStatus.UNSPECIFIED: + major_issues.append("status is unspecified") + elif isinstance(status, int): + major_issues.append("status is unrecognized") + + if major_issues: + _logger.warning( + "Found issues in microgrid: %s | Protobuf message:\n%s", + ", ".join(major_issues), + message, + ) + + if minor_issues: + _logger.debug( + "Found minor issues in microgrid: %s | Protobuf message:\n%s", + ", ".join(minor_issues), + message, + ) + + return MicrogridInfo( + id=MicrogridId(message.id), + enterprise_id=EnterpriseId(message.enterprise_id), + name=message.name or None, + delivery_area=delivery_area, + location=location, + status=status, + create_timestamp=conversion.to_datetime(message.create_timestamp), + ) diff --git a/src/frequenz/client/common/microgrid/_util.py b/src/frequenz/client/common/microgrid/_util.py new file mode 100644 index 0000000..e543914 --- /dev/null +++ b/src/frequenz/client/common/microgrid/_util.py @@ -0,0 +1,47 @@ +# License: MIT +# Copyright © 2025 Frequenz Energy-as-a-Service GmbH + +"""Utility functions.""" + +import enum +from typing import TypeVar + +EnumT = TypeVar("EnumT", bound=enum.Enum) +"""A type variable that is bound to an enum.""" + + +def enum_from_proto(value: int, enum_type: type[EnumT]) -> EnumT | int: + """Convert a protobuf int enum value to a python enum. + + Example: + ```python + import enum + + from proto import proto_pb2 # Just an example. pylint: disable=import-error + + @enum.unique + class SomeEnum(enum.Enum): + # These values should match the protobuf enum values. + UNSPECIFIED = 0 + SOME_VALUE = 1 + + enum_value = enum_from_proto(proto_pb2.SomeEnum.SOME_ENUM_SOME_VALUE, SomeEnum) + # -> SomeEnum.SOME_VALUE + + enum_value = enum_from_proto(42, SomeEnum) + # -> 42 + ``` + + Args: + value: The protobuf int enum value. + enum_type: The python enum type to convert to, + typically an enum class. + + Returns: + The resulting python enum value if the protobuf value is known, otherwise + the input value converted to a plain `int`. + """ + try: + return enum_type(value) + except ValueError: + return value diff --git a/src/frequenz/client/common/microgrid/electrical_components/__init__.py b/src/frequenz/client/common/microgrid/electrical_components/__init__.py index 4110a0d..4b8a2de 100644 --- a/src/frequenz/client/common/microgrid/electrical_components/__init__.py +++ b/src/frequenz/client/common/microgrid/electrical_components/__init__.py @@ -6,10 +6,7 @@ from enum import Enum -# pylint: disable=no-name-in-module -from frequenz.api.common.v1.microgrid.electrical_components.electrical_components_pb2 import ( - ElectricalComponentCategory as PBElectricalComponentCategory, -) +# # pylint: disable=no-name-in-module from frequenz.api.common.v1.microgrid.electrical_components.electrical_components_pb2 import ( ElectricalComponentDiagnosticCode as PBElectricalComponentDiagnosticCode, ) @@ -20,91 +17,6 @@ # pylint: enable=no-name-in-module -class ElectricalComponentCategory(Enum): - """Possible types of microgrid electrical component.""" - - UNSPECIFIED = ( - PBElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_UNSPECIFIED - ) - """An unknown component category. - - Useful for error handling, and marking unknown components in - a list of components with otherwise known categories. - """ - - GRID = PBElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_GRID - """The point where the local microgrid is connected to the grid.""" - - METER = PBElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_METER - """A meter, for measuring electrical metrics, e.g., current, voltage, etc.""" - - INVERTER = PBElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_INVERTER - """An electricity generator, with batteries or solar energy.""" - - CONVERTER = PBElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_CONVERTER - """An electricity converter, e.g., a DC-DC converter.""" - - BATTERY = PBElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_BATTERY - """A storage system for electrical energy, used by inverters.""" - - EV_CHARGER = PBElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_EV_CHARGER - """A station for charging electrical vehicles.""" - - CRYPTO_MINER = ( - PBElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_CRYPTO_MINER - ) - """A device for mining cryptocurrencies.""" - - ELECTROLYZER = ( - PBElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_ELECTROLYZER - ) - """A device for splitting water into hydrogen and oxygen using electricity.""" - - CHP = PBElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_CHP - """A heat and power combustion plant (CHP stands for combined heat and power).""" - - RELAY = PBElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_RELAY - """A relay, used for switching electrical circuits on and off.""" - - PRECHARGER = PBElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_PRECHARGER - """A precharger, used for preparing electrical circuits for switching on.""" - - FUSE = PBElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_FUSE - """A fuse, used for protecting electrical circuits from overcurrent.""" - - TRANSFORMER = ( - PBElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_VOLTAGE_TRANSFORMER - ) - """A transformer, used for changing the voltage of electrical circuits.""" - - HVAC = PBElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_HVAC - """A heating, ventilation, and air conditioning (HVAC) system.""" - - @classmethod - def from_proto( - cls, component_category: PBElectricalComponentCategory.ValueType - ) -> ElectricalComponentCategory: - """Convert a protobuf ElectricalComponentCategory message to enum. - - Args: - component_category: protobuf enum to convert - - Returns: - Enum value corresponding to the protobuf message. - """ - if not any(t.value == component_category for t in ElectricalComponentCategory): - return ElectricalComponentCategory.UNSPECIFIED - return cls(component_category) - - def to_proto(self) -> PBElectricalComponentCategory.ValueType: - """Convert a ElectricalComponentCategory enum to protobuf message. - - Returns: - Enum value corresponding to the protobuf message. - """ - return self.value - - class ElectricalComponentStateCode(Enum): """All possible states of a microgrid electrical component.""" diff --git a/src/frequenz/client/common/microgrid/electrical_components/_connection.py b/src/frequenz/client/common/microgrid/electrical_components/_connection.py new file mode 100644 index 0000000..ef71e7a --- /dev/null +++ b/src/frequenz/client/common/microgrid/electrical_components/_connection.py @@ -0,0 +1,35 @@ +# License: MIT +# Copyright © 2022 Frequenz Energy-as-a-Service GmbH + +"""Defines the connections between microgrid components.""" + + +from dataclasses import dataclass + +from ..id import ComponentId + + +@dataclass(frozen=True) +class Connection: + """Metadata for a connection between microgrid components.""" + + source_component_id: int + """The component ID that represents the source component of the connection.""" + + destination_component_id: int + """The component ID that represents the destination component of the connection.""" + + start: ComponentId + """The component ID that represents the start component of the connection.""" + + end: ComponentId + """The component ID that represents the end component of the connection.""" + + def is_valid(self) -> bool: + """Check if this instance contains valid data. + + Returns: + `True` if `start >= 0`, `end > 0`, and `start != end`, `False` + otherwise. + """ + return int(self.start) >= 0 and int(self.end) > 0 and self.start != self.end diff --git a/src/frequenz/client/common/microgrid/electrical_components/_electrical_component.py b/src/frequenz/client/common/microgrid/electrical_components/_electrical_component.py new file mode 100644 index 0000000..edcf501 --- /dev/null +++ b/src/frequenz/client/common/microgrid/electrical_components/_electrical_component.py @@ -0,0 +1,345 @@ +# License: MIT +# Copyright © 2025 Frequenz Energy-as-a-Service GmbH + +"""Defines the electrical components that can be used in a microgrid.""" + +from dataclasses import dataclass +from datetime import datetime +from enum import Enum + +from frequenz.api.common.v1.microgrid.electrical_components import ( + electrical_components_pb2, + grid_pb2, + inverter_pb2, +) + +from ..id import ComponentId, MicrogridId + + +class ComponentType(Enum): + """A base class from which individual component types are derived.""" + + +class InverterType(ComponentType): + """Enum representing inverter types.""" + + NONE = inverter_pb2.InverterType.TYPE_UNSPECIFIED + """Unspecified inverter type.""" + + BATTERY = inverter_pb2.InverterType.TYPE_BATTERY + """Battery inverter.""" + + SOLAR = inverter_pb2.InverterType.TYPE_SOLAR + """Solar inverter.""" + + HYBRID = inverter_pb2.InverterType.TYPE_HYBRID + """Hybrid inverter.""" + + +def component_type_from_protobuf( + component_category: electrical_components_pb2.ElectricalComponentCategory.ValueType, + component_metadata: inverter_pb2.Inverter, +) -> ComponentType | None: + """Convert a protobuf InverterType message to Component enum. + + For internal-only use by the `microgrid` package. + + Args: + component_category: category the type belongs to. + component_metadata: protobuf metadata to fetch type from. + + Returns: + Enum value corresponding to the protobuf message. + """ + # ComponentType values in the protobuf definition are not unique across categories + # as of v0.11.0, so we need to check the component category first, before doing any + # component type checks. + if ( + component_category + == electrical_components_pb2.ElectricalComponentCategory.COMPONENT_CATEGORY_INVERTER + ): + if not any(int(t.value) == int(component_metadata.type) for t in InverterType): + return None + + return InverterType(component_metadata.type) + + return None + + +class ElectricalComponentCategory(Enum): + """Possible types of microgrid electrical component.""" + + UNSPECIFIED = ( + electrical_components_pb2.ElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_UNSPECIFIED + ) + """An unknown component category. + + Useful for error handling, and marking unknown components in + a list of components with otherwise known categories. + """ + + GRID = ( + electrical_components_pb2.ElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_GRID + ) + """The point where the local microgrid is connected to the grid.""" + + METER = ( + electrical_components_pb2.ElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_METER + ) + """A meter, for measuring electrical metrics, e.g., current, voltage, etc.""" + + INVERTER = ( + electrical_components_pb2.ElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_INVERTER + ) + """An electricity generator, with batteries or solar energy.""" + + CONVERTER = ( + electrical_components_pb2.ElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_CONVERTER + ) + """An electricity converter, e.g., a DC-DC converter.""" + + BATTERY = ( + electrical_components_pb2.ElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_BATTERY + ) + """A storage system for electrical energy, used by inverters.""" + + EV_CHARGER = ( + electrical_components_pb2.ElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_EV_CHARGER + ) + """A station for charging electrical vehicles.""" + + CRYPTO_MINER = ( + electrical_components_pb2.ElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_CRYPTO_MINER + ) + """A device for mining cryptocurrencies.""" + + ELECTROLYZER = ( + electrical_components_pb2.ElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_ELECTROLYZER + ) + """A device for splitting water into hydrogen and oxygen using electricity.""" + + CHP = ( + electrical_components_pb2.ElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_CHP + ) + """A heat and power combustion plant (CHP stands for combined heat and power).""" + + RELAY = ( + electrical_components_pb2.ElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_RELAY + ) + """A relay, used for switching electrical circuits on and off.""" + + PRECHARGER = ( + electrical_components_pb2.ElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_PRECHARGER + ) + """A precharger, used for preparing electrical circuits for switching on.""" + + FUSE = ( + electrical_components_pb2.ElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_FUSE + ) + """A fuse, used for protecting electrical circuits from overcurrent.""" + + TRANSFORMER = ( + electrical_components_pb2.ElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_VOLTAGE_TRANSFORMER + ) + """A transformer, used for changing the voltage of electrical circuits.""" + + HVAC = ( + electrical_components_pb2.ElectricalComponentCategory.ELECTRICAL_COMPONENT_CATEGORY_HVAC + ) + """A heating, ventilation, and air conditioning (HVAC) system.""" + + +def component_category_from_protobuf( + component_category: electrical_components_pb2.ElectricalComponentCategory.ValueType, +) -> ElectricalComponentCategory: + """Convert a protobuf ElectricalComponentCategory message to ElectricalComponentCategory enum. + + For internal-only use by the `microgrid` package. + + Args: + component_category: protobuf enum to convert + + Returns: + Enum value corresponding to the protobuf message. + """ + if not any(t.value == component_category for t in ElectricalComponentCategory): + return ElectricalComponentCategory.UNSPECIFIED + + return ElectricalComponentCategory(component_category) + + +@dataclass(frozen=True) +class Fuse: + """Fuse data class.""" + + max_current: float + """Rated current of the fuse.""" + + +@dataclass(frozen=True) +class ComponentMetadata: + """Base class for component metadata classes.""" + + fuse: Fuse | None = None + """The fuse at the grid connection point.""" + + +@dataclass(frozen=True) +class GridMetadata(ComponentMetadata): + """Metadata for a grid connection point.""" + + +def component_metadata_from_protobuf( + component_category: electrical_components_pb2.ElectricalComponentCategory.ValueType, + component_metadata: grid_pb2.GridConnectionPoint, +) -> GridMetadata | None: + """Convert a protobuf GridMetadata message to GridMetadata class. + + Args: + component_category: category the type belongs to. + component_metadata: protobuf metadata to fetch type from. + + Returns: + GridMetadata instance corresponding to the protobuf message. + """ + if ( + component_category + == electrical_components_pb2.ElectricalComponentCategory.COMPONENT_CATEGORY_GRID + ): + max_current = component_metadata.rated_fuse_current + fuse = Fuse(max_current) + return GridMetadata(fuse) + + return None + + +class ComponentMetricId(Enum): + """An enum representing the various metrics available in the microgrid.""" + + ACTIVE_POWER = "active_power" + """Active power.""" + + ACTIVE_POWER_PHASE_1 = "active_power_phase_1" + """Active power in phase 1.""" + ACTIVE_POWER_PHASE_2 = "active_power_phase_2" + """Active power in phase 2.""" + ACTIVE_POWER_PHASE_3 = "active_power_phase_3" + """Active power in phase 3.""" + + REACTIVE_POWER = "reactive_power" + """Reactive power.""" + + REACTIVE_POWER_PHASE_1 = "reactive_power_phase_1" + """Reactive power in phase 1.""" + REACTIVE_POWER_PHASE_2 = "reactive_power_phase_2" + """Reactive power in phase 2.""" + REACTIVE_POWER_PHASE_3 = "reactive_power_phase_3" + """Reactive power in phase 3.""" + + CURRENT_PHASE_1 = "current_phase_1" + """Current in phase 1.""" + CURRENT_PHASE_2 = "current_phase_2" + """Current in phase 2.""" + CURRENT_PHASE_3 = "current_phase_3" + """Current in phase 3.""" + + VOLTAGE_PHASE_1 = "voltage_phase_1" + """Voltage in phase 1.""" + VOLTAGE_PHASE_2 = "voltage_phase_2" + """Voltage in phase 2.""" + VOLTAGE_PHASE_3 = "voltage_phase_3" + """Voltage in phase 3.""" + + FREQUENCY = "frequency" + + SOC = "soc" + """State of charge.""" + SOC_LOWER_BOUND = "soc_lower_bound" + """Lower bound of state of charge.""" + SOC_UPPER_BOUND = "soc_upper_bound" + """Upper bound of state of charge.""" + CAPACITY = "capacity" + """Capacity.""" + + POWER_INCLUSION_LOWER_BOUND = "power_inclusion_lower_bound" + """Power inclusion lower bound.""" + POWER_EXCLUSION_LOWER_BOUND = "power_exclusion_lower_bound" + """Power exclusion lower bound.""" + POWER_EXCLUSION_UPPER_BOUND = "power_exclusion_upper_bound" + """Power exclusion upper bound.""" + POWER_INCLUSION_UPPER_BOUND = "power_inclusion_upper_bound" + """Power inclusion upper bound.""" + + ACTIVE_POWER_INCLUSION_LOWER_BOUND = "active_power_inclusion_lower_bound" + """Active power inclusion lower bound.""" + ACTIVE_POWER_EXCLUSION_LOWER_BOUND = "active_power_exclusion_lower_bound" + """Active power exclusion lower bound.""" + ACTIVE_POWER_EXCLUSION_UPPER_BOUND = "active_power_exclusion_upper_bound" + """Active power exclusion upper bound.""" + ACTIVE_POWER_INCLUSION_UPPER_BOUND = "active_power_inclusion_upper_bound" + """Active power inclusion upper bound.""" + + TEMPERATURE = "temperature" + """Temperature.""" + + +# pylint: disable=too-many-instance-attributes +@dataclass(frozen=True) +class ElectricalComponent: + """Metadata for a single microgrid electrical component.""" + + id: ComponentId + """The ID of this component.""" + + microgrid_id: MicrogridId + """The unique identifier of the microgrid.""" + + name: str | None + """Name of the microgrid.""" + + category: ElectricalComponentCategory + """The category of this component.""" + + metadata: ComponentMetadata | None = None + """The metadata of this component.""" + + manufacturer: str | None = None + """The manufacturer of this component.""" + + model: str | None = None + """The model of this component.""" + + status: str | None = None + """The status of this component, e.g., "active", "inactive", etc.""" + + start: datetime | None = None + """The moment when the component became operationally active.""" + + end: datetime | None = None + """The moment when the component's operational activity ceased.""" + + metric_config_bounds: ComponentMetricId | None = None + """Configuration bounds for the metrics of this component.""" + + def is_valid(self) -> bool: + """Check if this instance contains valid data. + + Returns: + `True` if `id > 0` and `type` is a valid `ComponentCategory`, or if `id + == 0` and `type` is `GRID`, `False` otherwise + """ + return ( + int(self.id) > 0 + and any(t == self.category for t in ElectricalComponentCategory) + ) or ( + int(self.id) == 0 + and self.category == ElectricalComponentCategory.GRID + ) + + def __hash__(self) -> int: + """Compute a hash of this instance, obtained by hashing the `component_id` field. + + Returns: + Hash of this instance. + """ + return hash(self.id) diff --git a/src/frequenz/client/common/microgrid/id.py b/src/frequenz/client/common/microgrid/id.py new file mode 100644 index 0000000..cfa2aa6 --- /dev/null +++ b/src/frequenz/client/common/microgrid/id.py @@ -0,0 +1,223 @@ +# License: MIT +# Copyright © 2025 Frequenz Energy-as-a-Service GmbH + +r'''Provides strongly-typed unique identifiers for entities. + +This module offers a base class, +[`BaseId`][frequenz.client.microgrid.id.BaseId], which can be subclassed to +create distinct ID types for different components or concepts within a system. +These IDs ensure type safety, meaning that an ID for one type of entity (e.g., a +sensor) cannot be mistakenly used where an ID for another type (e.g., a +microgrid) is expected. + +# Creating Custom ID Types + +To define a new ID type, create a class that inherits from +[`BaseId`][frequenz.client.microgrid.id.BaseId] and provide a unique +`str_prefix` as a keyword argument in the class definition. This prefix is used +in the string representation of the ID and must be unique across all ID types. + +Note: + The `str_prefix` must be unique across all ID types. If you try to use a + prefix that is already registered, a `ValueError` will be raised when defining + the class. + +To encourage consistency, the class name must end with the suffix "Id" (e.g., +`MyNewId`). This check can be bypassed by passing `allow_custom_name=True` when +defining the class (e.g., `class MyCustomName(BaseId, str_prefix="MCN", +allow_custom_name=True):`). + +Tip: + Use the [`@typing.final`][typing.final] decorator to prevent subclassing of + ID classes. + +Example: Creating a standard ID type + ```python + from typing import final + from frequenz.client.microgrid.id import BaseId + + @final + class InverterId(BaseId, str_prefix="INV"): + """A unique identifier for an inverter.""" + + inv_id = InverterId(123) + print(inv_id) # Output: INV123 + print(int(inv_id)) # Output: 123 + ``` + +Example: Creating an ID type with a non-standard name + ```python + from typing import final + from frequenz.client.microgrid.id import BaseId + + @final + class CustomNameForId(BaseId, str_prefix="CST", allow_custom_name=True): + """An ID with a custom name, not ending in 'Id'.""" + + custom_id = CustomNameForId(456) + print(custom_id) # Output: CST456 + print(int(custom_id)) # Output: 456 + ``` + +# Predefined ID Types + +This module predefines the following ID types: + +- [`ComponentId`][frequenz.client.microgrid.id.ComponentId]: For identifying + generic components. +- [`MicrogridId`][frequenz.client.microgrid.id.MicrogridId]: For identifying + microgrids. +- [`SensorId`][frequenz.client.microgrid.id.SensorId]: For identifying sensors. +''' + + +from typing import Any, ClassVar, Self, cast, final + + +class BaseId: + """A base class for unique identifiers. + + Subclasses must provide a unique `str_prefix` keyword argument during + definition, which is used in the string representation of the ID. + + By default, subclass names must end with "Id". This can be overridden by + passing `allow_custom_name=True` during class definition. + + For more information and examples, see the [module's + documentation][frequenz.client.microgrid.id]. + """ + + _id: int + _str_prefix: ClassVar[str] + _registered_prefixes: ClassVar[set[str]] = set() + + def __new__(cls, *_: Any, **__: Any) -> Self: + """Create a new instance of the ID class, only if it is a subclass of BaseId.""" + if cls is BaseId: + raise TypeError("BaseId cannot be instantiated directly. Use a subclass.") + return super().__new__(cls) + + def __init_subclass__( + cls, + *, + str_prefix: str, + allow_custom_name: bool = False, + **kwargs: Any, + ) -> None: + """Initialize a subclass, set its string prefix, and perform checks. + + Args: + str_prefix: The string prefix for the ID type (e.g., "MID"). + Must be unique across all ID types. + allow_custom_name: If True, bypasses the check that the class name + must end with "Id". Defaults to False. + **kwargs: Forwarded to the parent's __init_subclass__. + + Raises: + ValueError: If the `str_prefix` is already registered by another + ID type. + TypeError: If `allow_custom_name` is False and the class name + does not end with "Id". + """ + super().__init_subclass__(**kwargs) + + if str_prefix in BaseId._registered_prefixes: + raise ValueError( + f"Prefix '{str_prefix}' is already registered. " + "ID prefixes must be unique." + ) + BaseId._registered_prefixes.add(str_prefix) + + if not allow_custom_name and not cls.__name__.endswith("Id"): + raise TypeError( + f"Class name '{cls.__name__}' for an ID class must end with 'Id' " + "(e.g., 'SomeId'), or use `allow_custom_name=True`." + ) + + cls._str_prefix = str_prefix + + def __init__(self, id_: int, /) -> None: + """Initialize this instance. + + Args: + id_: The numeric unique identifier. + + Raises: + ValueError: If the ID is negative. + """ + if id_ < 0: + raise ValueError(f"{type(self).__name__} can't be negative.") + self._id = id_ + + @property + def str_prefix(self) -> str: + """The prefix used for the string representation of this ID.""" + return self._str_prefix + + def __int__(self) -> int: + """Return the numeric ID of this instance.""" + return self._id + + def __eq__(self, other: object) -> bool: + """Check if this instance is equal to another object. + + Equality is defined as being of the exact same type and having the same + underlying ID. + """ + # pylint thinks this is not an unidiomatic typecheck, but in this case + # it is not. isinstance() returns True for subclasses, which is not + # what we want here, as different ID types should never be equal. + # pylint: disable-next=unidiomatic-typecheck + if type(other) is not type(self): + return NotImplemented + # We already checked type(other) is type(self), but mypy doesn't + # understand that, so we need to cast it to Self. + other_id = cast(Self, other) + return self._id == other_id._id + + def __lt__(self, other: object) -> bool: + """Check if this instance is less than another object. + + Comparison is only defined between instances of the exact same type. + """ + # pylint: disable-next=unidiomatic-typecheck + if type(other) is not type(self): + return NotImplemented + other_id = cast(Self, other) + return self._id < other_id._id + + def __hash__(self) -> int: + """Return the hash of this instance. + + The hash is based on the exact type and the underlying ID to ensure + that IDs of different types but with the same numeric value have different hashes. + """ + return hash((type(self), self._id)) + + def __repr__(self) -> str: + """Return the string representation of this instance.""" + return f"{type(self).__name__}({self._id!r})" + + def __str__(self) -> str: + """Return the short string representation of this instance.""" + return f"{type(self)._str_prefix}{self._id}" + + +@final +class EnterpriseId(BaseId, str_prefix="EID"): + """A unique identifier for an enterprise account.""" + + +@final +class MicrogridId(BaseId, str_prefix="MID"): + """A unique identifier for a microgrid.""" + + +@final +class ComponentId(BaseId, str_prefix="CID"): + """A unique identifier for a microgrid component.""" + + +@final +class SensorId(BaseId, str_prefix="SID"): + """A unique identifier for a microgrid sensor."""