Add initial implementation of electrical component types and related utilities

This commit introduces a comprehensive set of classes and functions for managing electrical components within the microgrid framework. Key additions include:

- Definitions for various electrical components such as batteries, inverters, chargers, and more, each with specific attributes and categories.
- Protobuf message loading functions to convert messages into corresponding Python objects.
- JSON encoding utilities for serializing component data for API interactions.

These changes lay the groundwork for enhanced functionality and integration within the microgrid system.

Signed-off-by: eduardiazf <[email protected]>

Author: eduardiazf

src/frequenz/client/assets/_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.v1alpha8.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 "<NO CODE>"
+        code_type = (
+            f"type={self.code_type}"
+            if isinstance(self.code_type, int)
+            else self.code_type.name
+        )
+        return f"{code}[{code_type}]"

src/frequenz/client/assets/_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.v1alpha8.grid import delivery_area_pb2
+from frequenz.client.common import enum_proto
+
+from ._delivery_area import DeliveryArea, EnergyMarketCodeType
+
+_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_proto.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)

src/frequenz/client/assets/_lifetime.py

@@ -0,0 +1,54 @@
+# 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(
+                f"Start ({self.start}) must be before or equal to end ({self.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))

src/frequenz/client/assets/_lifetime_proto.py

@@ -0,0 +1,26 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Loading of Lifetime objects from protobuf messages."""
+
+from frequenz.api.common.v1alpha8.microgrid import lifetime_pb2
+from frequenz.client.base.conversion import to_datetime
+
+from ._lifetime import Lifetime
+
+
+def lifetime_from_proto(
+    message: lifetime_pb2.Lifetime,
+) -> Lifetime:
+    """Create a [`Lifetime`][frequenz.client.microgrid.Lifetime] from a protobuf message."""
+    start = (
+        to_datetime(message.start_timestamp)
+        if message.HasField("start_timestamp")
+        else None
+    )
+    end = (
+        to_datetime(message.end_timestamp)
+        if message.HasField("end_timestamp")
+        else None
+    )
+    return Lifetime(start=start, end=end)

src/frequenz/client/assets/_location.py

@@ -0,0 +1,31 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Location information for a microgrid."""
+
+
+from dataclasses import dataclass
+
+
+@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."""
+
+    def __str__(self) -> str:
+        """Return the short string representation of this instance."""
+        country = self.country_code or "<NO COUNTRY CODE>"
+        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}"

src/frequenz/client/assets/_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.v1alpha8.types 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)

src/frequenz/client/assets/_microgrid.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.v1alpha8.microgrid import microgrid_pb2
+from frequenz.client.common.microgrid import EnterpriseId, MicrogridId
+
+from ._delivery_area import DeliveryArea
+from ._location import Location
+
+_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 Microgrid:
+    """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}"