Add MicrogridInfo and MicrogridStatus

This will be used to retrieve the microgrid information.

Signed-off-by: Leandro Lucarella <[email protected]>

Author: llucax

src/frequenz/client/microgrid/__init__.py

@@ -33,6 +33,7 @@
 )
 from ._lifetime import Lifetime
 from ._location import Location
+from ._microgrid_info import MicrogridInfo, MicrogridStatus
 
 __all__ = [
     "ApiClientError",
@@ -48,6 +49,8 @@
     "Lifetime",
     "Location",
     "MicrogridApiClient",
+    "MicrogridInfo",
+    "MicrogridStatus",
     "OperationAborted",
     "OperationCancelled",
     "OperationNotImplemented",

src/frequenz/client/microgrid/_microgrid_info.py

@@ -0,0 +1,85 @@
+# 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 ._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 = 0
+    """The status is unspecified. This should not be used."""
+
+    ACTIVE = 1
+    """The microgrid is active."""
+
+    INACTIVE = 2
+    """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}"

src/frequenz/client/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),
+    )