Implement ListComponents

Implement the `ListComponents` 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,
and translate most enums that encode types of components to concrete
Python classes. This way, we can use the type system to ensure we are
using the right type of component in the right place, and enables using
match statements to handle different types of components in an
exhaustive way.

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

@@ -31,13 +31,15 @@
     UnknownError,
     UnrecognizedGrpcStatus,
 )
-from ._id import EnterpriseId, MicrogridId
+from ._id import ComponentId, EnterpriseId, MicrogridId
+from ._lifetime import Lifetime
 from ._location import Location
 from ._microgrid_info import MicrogridInfo, MicrogridStatus
 
 __all__ = [
     "ApiClientError",
     "ClientNotConnected",
+    "ComponentId",
     "DataLoss",
     "DeliveryArea",
     "EnergyMarketCodeType",
@@ -47,6 +49,7 @@
     "GrpcError",
     "InternalError",
     "InvalidArgument",
+    "Lifetime",
     "Location",
     "MicrogridApiClient",
     "MicrogridId",

src/frequenz/client/microgrid/_client.py

@@ -3,20 +3,26 @@
 
 """Client for requests to the Microgrid API."""
 
+from collections.abc import Iterable
 from dataclasses import replace
-from typing import Any
+from typing import Any, assert_never
 
-from frequenz.api.microgrid.v1 import microgrid_pb2_grpc
+from frequenz.api.common.v1.microgrid.components import components_pb2
+from frequenz.api.microgrid.v1 import microgrid_pb2, microgrid_pb2_grpc
 from frequenz.client.base import channel, client, retry, streaming
 from google.protobuf.empty_pb2 import Empty
 
+from ._id import ComponentId
 from ._microgrid_info import MicrogridInfo
 from ._microgrid_info_proto import microgrid_info_from_proto
+from .component._base import Component
+from .component._category import ComponentCategory
+from .component._component import ComponentTypes
+from .component._component_proto import component_from_proto
 
 DEFAULT_GRPC_CALL_TIMEOUT = 60.0
 """The default timeout for gRPC calls made by this client (in seconds)."""
 
-
 DEFAULT_CHANNEL_OPTIONS = replace(
     channel.ChannelOptions(), ssl=channel.SslOptions(enabled=False)
 )
@@ -93,3 +99,82 @@ async def get_microgrid_info(  # noqa: DOC502 (raises ApiClientError indirectly)
         )
 
         return microgrid_info_from_proto(microgrid.microgrid)
+
+    async def list_components(  # noqa: DOC502 (raises ApiClientError indirectly)
+        self,
+        *,
+        components: Iterable[ComponentId | Component] = (),
+        categories: Iterable[ComponentCategory | int] = (),
+    ) -> Iterable[ComponentTypes]:
+        """Fetch all the components present in the local microgrid.
+
+        Electrical components are a part of a microgrid's electrical infrastructure
+        are can be connected to each other to form an electrical circuit, which can
+        then be represented as a graph.
+
+        If provided, the filters for component and categories have an `AND`
+        relationship with one another, meaning that they are applied serially,
+        but the elements within a single filter list have an `OR` relationship with
+        each other.
+
+        Example:
+            If `ids = {1, 2, 3}`, and `categories = {ComponentCategory.INVERTER,
+            ComponentCategory.BATTERY}`, then the results will consist of elements that
+            have:
+
+            * The IDs 1, `OR` 2, `OR` 3; `AND`
+            * Are of the categories `ComponentCategory.INVERTER` `OR`
+              `ComponentCategory.BATTERY`.
+
+        If a filter list is empty, then that filter is not applied.
+
+        Args:
+            components: The components to fetch. See the method description for details.
+            categories: The categories of the components to fetch. See the method
+                description for details.
+
+        Returns:
+            Iterator whose elements are all the components in 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].
+        """
+        component_list = await client.call_stub_method(
+            self,
+            lambda: self._async_stub.ListComponents(
+                microgrid_pb2.ListComponentsRequest(
+                    component_ids=map(_get_component_id, components),
+                    categories=map(_get_category_value, categories),
+                ),
+                timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
+            ),
+            method_name="ListComponents",
+        )
+
+        return map(component_from_proto, component_list.components)
+
+
+def _get_component_id(component: ComponentId | Component) -> int:
+    """Get the component ID from a component or component ID."""
+    match component:
+        case ComponentId():
+            return int(component)
+        case Component():
+            return int(component.id)
+        case unexpected:
+            assert_never(unexpected)
+
+
+def _get_category_value(
+    category: ComponentCategory | int,
+) -> components_pb2.ComponentCategory.ValueType:
+    """Get the category value from a component or component category."""
+    match category:
+        case ComponentCategory():
+            return components_pb2.ComponentCategory.ValueType(category.value)
+        case int():
+            return components_pb2.ComponentCategory.ValueType(category)
+        case unexpected:
+            assert_never(unexpected)

src/frequenz/client/microgrid/_id.py

@@ -86,3 +86,45 @@ def __repr__(self) -> str:
     def __str__(self) -> str:
         """Return the short string representation of this instance."""
         return f"MID{self._id}"
+
+
+class ComponentId:
+    """A unique identifier for a microgrid component."""
+
+    def __init__(self, id_: int, /) -> None:
+        """Initialize this instance.
+
+        Args:
+            id_: The numeric unique identifier of the microgrid component.
+
+        Raises:
+            ValueError: If the ID is negative.
+        """
+        if id_ < 0:
+            raise ValueError("Component 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 ComponentId and self._id == other._id
+
+    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((ComponentId, 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"CID{self._id}"

src/frequenz/client/microgrid/_lifetime.py

@@ -0,0 +1,45 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Lifetime of a microgrid asset."""
+
+
+from dataclasses import dataclass
+from datetime import datetime
+from functools import cached_property
+
+
+@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.")
+
+    @cached_property
+    def active(self) -> bool:
+        """Whether this lifetime is currently active."""
+        now = datetime.now()
+        if self.start is not None and self.start > now:
+            return False
+        return self.end is None or self.end >= now

src/frequenz/client/microgrid/_lifetime_proto.py

@@ -0,0 +1,26 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Loading of Lifetime objects from protobuf messages."""
+
+from frequenz.api.common.v1.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/microgrid/component/__init__.py

@@ -0,0 +1,92 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""All classes and functions related to microgrid components."""
+
+from ._base import Component
+from ._battery import (
+    Battery,
+    BatteryType,
+    LiIonBattery,
+    NaIonBattery,
+    UnrecognizedBattery,
+    UnspecifiedBattery,
+)
+from ._category import ComponentCategory
+from ._chp import Chp
+from ._component import ComponentTypes
+from ._converter import Converter
+from ._crypto_miner import CryptoMiner
+from ._electrolyzer import Electrolyzer
+from ._ev_charger import (
+    AcEvCharger,
+    DcEvCharger,
+    EvCharger,
+    EvChargerType,
+    UnrecognizedEvCharger,
+    UnspecifiedEvCharger,
+)
+from ._fuse import Fuse
+from ._grid_connection_point import GridConnectionPoint
+from ._hvac import Hvac
+from ._inverter import (
+    BatteryInverter,
+    HybridInverter,
+    Inverter,
+    InverterType,
+    SolarInverter,
+    UnrecognizedInverter,
+    UnspecifiedInverter,
+)
+from ._meter import Meter
+from ._precharger import Precharger
+from ._problematic import (
+    MismatchedCategoryComponent,
+    ProblematicComponentTypes,
+    UnrecognizedComponent,
+    UnspecifiedComponent,
+)
+from ._relay import Relay
+from ._status import ComponentStatus
+from ._voltage_transformer import VoltageTransformer
+
+__all__ = [
+    "AcEvCharger",
+    "Battery",
+    "BatteryInverter",
+    "BatteryType",
+    "Chp",
+    "Component",
+    "ComponentCategory",
+    "ComponentStatus",
+    "ComponentTypes",
+    "Converter",
+    "CryptoMiner",
+    "DcEvCharger",
+    "Electrolyzer",
+    "EvCharger",
+    "EvChargerType",
+    "Fuse",
+    "GridConnectionPoint",
+    "Hvac",
+    "HybridInverter",
+    "Inverter",
+    "InverterType",
+    "LiIonBattery",
+    "Meter",
+    "MismatchedCategoryComponent",
+    "NaIonBattery",
+    "Precharger",
+    "ProblematicComponentTypes",
+    "Relay",
+    "SolarInverter",
+    "UnrecognizedBattery",
+    "UnrecognizedComponent",
+    "UnrecognizedEvCharger",
+    "UnrecognizedInverter",
+    "UnspecifiedBattery",
+    "UnspecifiedComponent",
+    "UnspecifiedEvCharger",
+    "UnspecifiedInverter",
+    "VoltageTransformer",
+]