Implement ListConnections

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

Author: llucax

src/frequenz/client/microgrid/_client.py

@@ -22,6 +22,8 @@
 from .component._category import ComponentCategory
 from .component._component import ComponentTypes
 from .component._component_proto import component_from_proto
+from .component._connection import ComponentConnection
+from .component._connection_proto import component_connection_from_proto
 
 DEFAULT_GRPC_CALL_TIMEOUT = 60.0
 """The default timeout for gRPC calls made by this client (in seconds)."""
@@ -170,6 +172,60 @@ async def list_components(  # noqa: DOC502 (raises ApiClientError indirectly)
 
         return map(component_from_proto, component_list.components)
 
+    async def list_connections(  # noqa: DOC502 (raises ApiClientError indirectly)
+        self,
+        *,
+        sources: Iterable[ComponentId | Component] = (),
+        destinations: Iterable[ComponentId | Component] = (),
+    ) -> Iterable[ComponentConnection]:
+        """Fetch all the connections 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.
+
+        The direction of a connection is always away from the grid endpoint, i.e.
+        aligned with the direction of positive current according to the passive sign
+        convention: https://en.wikipedia.org/wiki/Passive_sign_convention
+
+        The request may be filtered by `source`/`destination` component(s) of individual
+        connections.  If provided, the `sources` and `destinations` filters have an
+        `AND` relationship between each other, meaning that they are applied serially,
+        but an `OR` relationship with other elements in the same list.
+
+        Example:
+            If `sources = {1, 2, 3}`, and `destinations = {4,
+            5, 6}`, then the result should have all the connections where:
+
+            * Each `source` component ID is either `1`, `2`, OR `3`; **AND**
+            * Each `destination` component ID is either `4`, `5`, OR `6`.
+
+        Args:
+            sources: The component from which the connections originate.
+            destinations: The component at which the connections terminate.
+
+        Returns:
+            Iterator whose elements are all the connections 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].
+        """
+        connection_list = await client.call_stub_method(
+            self,
+            lambda: self.stub.ListConnections(
+                microgrid_pb2.ListConnectionsRequest(
+                    starts=map(_get_component_id, sources),
+                    ends=map(_get_component_id, destinations),
+                ),
+                timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
+            ),
+            method_name="ListConnections",
+        )
+
+        return map(component_connection_from_proto, connection_list.connections)
+
 
 def _get_component_id(component: ComponentId | Component) -> int:
     """Get the component ID from a component or component ID."""

src/frequenz/client/microgrid/component/__init__.py

@@ -21,6 +21,7 @@
     UnrecognizedComponentTypes,
     UnspecifiedComponentTypes,
 )
+from ._connection import ComponentConnection
 from ._converter import Converter
 from ._crypto_miner import CryptoMiner
 from ._electrolyzer import Electrolyzer
@@ -67,6 +68,7 @@
     "Chp",
     "Component",
     "ComponentCategory",
+    "ComponentConnection",
     "ComponentStatus",
     "ComponentTypes",
     "Converter",

src/frequenz/client/microgrid/component/_connection.py

@@ -0,0 +1,68 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Component connection."""
+
+import dataclasses
+from functools import cached_property
+
+from .._id import ComponentId
+from .._lifetime import Lifetime
+
+
+@dataclasses.dataclass(frozen=True, kw_only=True)
+class ComponentConnection:
+    """A single electrical link between two components within a microgrid.
+
+    A component connection represents the physical wiring as viewed from the grid
+    connection point, if one exists, or from the islanding point, in case of an islanded
+    microgrids.
+
+    Note: Physical Representation
+        This message is not about data flow but rather about the physical
+        electrical connections between components. Therefore, the IDs for the
+        source and destination components correspond to the actual setup within
+        the microgrid.
+
+    Note: Direction
+        The direction of the connection follows the flow of current away from the
+        grid connection point, or in case of islands, away from the islanding
+        point. This direction is aligned with positive current according to the
+        [Passive Sign Convention]
+        (https://en.wikipedia.org/wiki/Passive_sign_convention).
+
+    Note: Historical Data
+        The timestamps of when a connection was created and terminated allows for
+        tracking the changes over time to a microgrid, providing insights into
+        when and how the microgrid infrastructure has been modified.
+    """
+
+    source: ComponentId
+    """The unique identifier of the component where the connection originates.
+
+    This is aligned with the direction of current flow away from the grid connection
+    point, or in case of islands, away from the islanding point.
+    """
+
+    destination: ComponentId
+    """The unique ID of the component where the connection terminates.
+
+    This is the component towards which the current flows.
+    """
+
+    operational_lifetime: Lifetime = dataclasses.field(default_factory=Lifetime)
+    """The operational lifetime of the connection."""
+
+    def __post_init__(self) -> None:
+        """Ensure that the source and destination components are different."""
+        if self.source == self.destination:
+            raise ValueError("Source and destination components must be different")
+
+    @cached_property
+    def active(self) -> bool:
+        """Whether this connection is currently active."""
+        return self.operational_lifetime.active
+
+    def __str__(self) -> str:
+        """Return a human-readable string representation of this instance."""
+        return f"{self.source}->{self.destination}"

src/frequenz/client/microgrid/component/_connection_proto.py

@@ -0,0 +1,75 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Loading of ComponentConnection objects from protobuf messages."""
+
+import logging
+
+from frequenz.api.common.v1.microgrid.components import components_pb2
+
+from .._id import ComponentId
+from .._lifetime import Lifetime
+from .._lifetime_proto import lifetime_from_proto
+from ._connection import ComponentConnection
+
+_logger = logging.getLogger(__name__)
+
+
+def component_connection_from_proto(
+    message: components_pb2.ComponentConnection,
+) -> ComponentConnection:
+    """Create a `ComponentConnection` from a protobuf message."""
+    major_issues: list[str] = []
+    minor_issues: list[str] = []
+
+    source_component_id = ComponentId(message.source_component_id)
+    destination_component_id = ComponentId(message.destination_component_id)
+    if source_component_id == destination_component_id:
+        major_issues.append(
+            f"source and destination are the same: {source_component_id}",
+        )
+
+    lifetime = _get_operational_lifetime_from_proto(
+        message, major_issues=major_issues, minor_issues=minor_issues
+    )
+
+    if major_issues:
+        _logger.warning(
+            "Found issues in component connection: %s | Protobuf message:\n%s",
+            ", ".join(major_issues),
+            message,
+        )
+    if minor_issues:
+        _logger.debug(
+            "Found minor issues in component connection: %s | Protobuf message:\n%s",
+            ", ".join(minor_issues),
+            message,
+        )
+
+    return ComponentConnection(
+        source=source_component_id,
+        destination=destination_component_id,
+        operational_lifetime=lifetime,
+    )
+
+
+def _get_operational_lifetime_from_proto(
+    message: components_pb2.ComponentConnection,
+    *,
+    major_issues: list[str],
+    minor_issues: list[str],
+) -> Lifetime:
+    """Get the operational lifetime from a protobuf message."""
+    if message.HasField("operational_lifetime"):
+        try:
+            return lifetime_from_proto(message.operational_lifetime)
+        except ValueError as exc:
+            major_issues.append(
+                f"invalid operational lifetime ({exc}), considering it as missing "
+                "(i.e. always operational)",
+            )
+    else:
+        minor_issues.append(
+            "missing operational lifetime, considering it always operational",
+        )
+    return Lifetime()