Implement GetMicrogridMetadata

Implement the `GetMicrogridMetadata` RPC call in the
`MicrogridApiClient`. To do this we also need to implement the data
structures that will be used to wrap the data returned by the API.

These wrappers use a more Pythonic approach than the protobuf messages.
For example, if some fields don't come in the serialized data, a `None`
will be used instead of using the protobuf defaults. This is to avoid
ending up exposing wrong data (for example, if we don't have location
information, it doesn't make sense to tell the user we are located at
latitude 0 and longitude 0.

Most wrappers should probably be moved to `api-common` in the future.

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

Author: llucax

src/frequenz/client/microgrid/__init__.py

@@ -8,6 +8,7 @@
 
 
 from ._client import MicrogridApiClient
+from ._delivery_area import DeliveryArea, EnergyMarketCodeType
 from ._exception import (
     ApiClientError,
     ClientNotConnected,
@@ -30,20 +31,28 @@
     UnknownError,
     UnrecognizedGrpcStatus,
 )
-from ._id import ComponentId, MicrogridId
+from ._id import ComponentId, EnterpriseId, MicrogridId
+from ._location import Location
+from ._microgrid_info import MicrogridInfo, MicrogridStatus
 
 __all__ = [
     "ApiClientError",
     "ClientNotConnected",
     "ComponentId",
     "DataLoss",
+    "DeliveryArea",
+    "EnergyMarketCodeType",
+    "EnterpriseId",
     "EntityAlreadyExists",
     "EntityNotFound",
     "GrpcError",
     "InternalError",
     "InvalidArgument",
+    "Location",
     "MicrogridApiClient",
     "MicrogridId",
+    "MicrogridInfo",
+    "MicrogridStatus",
     "OperationAborted",
     "OperationCancelled",
     "OperationNotImplemented",

src/frequenz/client/microgrid/_client.py

@@ -10,9 +10,12 @@
 
 from frequenz.api.microgrid.v1 import microgrid_pb2_grpc
 from frequenz.client.base import channel, client, retry, streaming
+from google.protobuf.empty_pb2 import Empty
 
 from ._exception import ClientNotConnected
 from ._id import ComponentId
+from ._microgrid_info import MicrogridInfo
+from ._microgrid_info_proto import microgrid_info_from_proto
 
 DEFAULT_GRPC_CALL_TIMEOUT = 60.0
 """The default timeout for gRPC calls made by this client (in seconds)."""
@@ -79,3 +82,30 @@ def stub(self) -> microgrid_pb2_grpc.MicrogridAsyncStub:
         # actually exists to the eyes of the interpreter, it only exists for the
         # type-checker, so it can only be used for type hints.
         return self._stub  # type: ignore
+
+    async def get_microgrid_info(  # noqa: DOC502 (raises ApiClientError indirectly)
+        self,
+    ) -> MicrogridInfo:
+        """Fetch the information about the local microgrid.
+
+        This consists of information that describes the overall microgrid, as opposed to
+        its electrical components or sensors, e.g., the microgrid ID, location.
+
+        Returns:
+            The information about the local microgrid.
+
+        Raises:
+            ApiClientError: If the are any errors communicating with the Microgrid API,
+                most likely a subclass of
+                [GrpcError][frequenz.client.microgrid.GrpcError].
+        """
+        microgrid = await client.call_stub_method(
+            self,
+            lambda: self.stub.GetMicrogridMetadata(
+                Empty(),
+                timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
+            ),
+            method_name="GetMicrogridMetadata",
+        )
+
+        return microgrid_info_from_proto(microgrid.microgrid)

src/frequenz/client/microgrid/_delivery_area.py

@@ -0,0 +1,87 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Delivery area information for the energy market."""
+
+import enum
+from dataclasses import dataclass
+
+
+@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 = 0
+    """Unspecified type. This value is a placeholder and should not be used."""
+
+    EUROPE_EIC = 1
+    """European Energy Identification Code Standard."""
+
+    US_NERC = 2
+    """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/microgrid/_delivery_area_proto.py

@@ -0,0 +1,46 @@
+# License: MIT
+# Copyright © 2024 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)

src/frequenz/client/microgrid/_id.py

@@ -1,7 +1,56 @@
 # License: MIT
 # Copyright © 2025 Frequenz Energy-as-a-Service GmbH
 
-"""Strongly typed IDs for microgrids and components."""
+"""Strongly typed IDs for enterprises, microgrids and components."""
+
+
+class EnterpriseId:
+    """A unique identifier for an enterprise account."""
+
+    def __init__(self, id_: int, /) -> None:
+        """Initialize this instance.
+
+        Args:
+            id_: The numeric unique identifier of the enterprise account.
+
+        Raises:
+            ValueError: If the ID is negative.
+        """
+        if id_ < 0:
+            raise ValueError("Enterprise ID can't be negative.")
+        self._id = id_
+
+    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."""
+        # This is not an unidiomatic typecheck, that's an odd name for the check.
+        # isinstance() returns True for subclasses, which is not what we want here.
+        # pylint: disable-next=unidiomatic-typecheck
+        return type(other) is EnterpriseId and self._id == other._id
+
+    def __lt__(self, other: object) -> bool:
+        """Check if this instance is less than another object."""
+        # pylint: disable-next=unidiomatic-typecheck
+        if type(other) is EnterpriseId:
+            return self._id < other._id
+        return NotImplemented
+
+    def __hash__(self) -> int:
+        """Return the hash of this instance."""
+        # We include the class because we explicitly want to avoid the same ID to give
+        # the same hash for different classes of IDs
+        return hash((EnterpriseId, 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"EID{self._id}"
 
 
 from typing import final

src/frequenz/client/microgrid/_location.py

@@ -0,0 +1,50 @@
+# License: MIT
+# Copyright © 2024 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 "<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/microgrid/_location_proto.py

@@ -0,0 +1,49 @@
+# License: MIT
+# Copyright © 2024 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)