feat(assets): List electrical component connections

Add a new client method to retrieve microgrid electrical components connections.

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

Author: eduardiazf

src/frequenz/client/assets/_client.py

@@ -13,12 +13,11 @@
 from frequenz.client.base import channel
 from frequenz.client.base.client import BaseApiClient, call_stub_method
 
-from frequenz.client.assets.electrical_component._electrical_component import (
-    ElectricalComponent,
-)
-
 from ._microgrid import Microgrid
 from ._microgrid_proto import microgrid_from_proto
+from .electrical_component._connection import ComponentConnection
+from .electrical_component._connection_proto import component_connection_from_proto
+from .electrical_component._electrical_component import ElectricalComponent
 from .electrical_component._electrical_component_proto import electrical_component_proto
 from .exceptions import ClientNotConnected
 
@@ -138,3 +137,47 @@ async def list_microgrid_electrical_components(
         return [
             electrical_component_proto(component) for component in response.components
         ]
+
+    async def list_microgrid_electrical_component_connections(
+        self,
+        microgrid_id: int,
+        source_component_ids: list[int] | None = None,
+        destination_component_ids: list[int] | None = None,
+    ) -> list[ComponentConnection | None]:
+        """
+        Get the electrical component connections of a microgrid.
+
+        Args:
+            microgrid_id: The ID of the microgrid to get the electrical
+                component connections of.
+            source_component_ids: Only return connections that originate from
+                these component IDs. If None or empty, no filtering is applied.
+            destination_component_ids: Only return connections that terminate at
+                these component IDs. If None or empty, no filtering is applied.
+
+        Returns:
+            The electrical component connections of the microgrid.
+        """
+        request = assets_pb2.ListMicrogridElectricalComponentConnectionsRequest(
+            microgrid_id=microgrid_id,
+        )
+
+        if source_component_ids:
+            request.source_component_ids.extend(source_component_ids)
+
+        if destination_component_ids:
+            request.destination_component_ids.extend(destination_component_ids)
+
+        response = await call_stub_method(
+            self,
+            lambda: self.stub.ListMicrogridElectricalComponentConnections(
+                request,
+                timeout=DEFAULT_GRPC_CALL_TIMEOUT,
+            ),
+            method_name="ListMicrogridElectricalComponentConnections",
+        )
+
+        return [
+            component_connection_from_proto(connection)
+            for connection in response.connections
+        ]

src/frequenz/client/assets/cli/__main__.py

@@ -40,7 +40,11 @@
 from frequenz.client.assets._client import AssetsApiClient
 from frequenz.client.assets.exceptions import ApiClientError
 
-from ._utils import print_electrical_components, print_microgrid_details
+from ._utils import (
+    print_component_connections,
+    print_electrical_components,
+    print_microgrid_details,
+)
 
 
 @click.group(invoke_without_command=True)
@@ -201,6 +205,97 @@ async def list_microgrid_electrical_components(
         raise click.Abort()
 
 
+@cli.command("component-connections")
+@click.pass_context
+@click.argument("microgrid-id", required=True, type=int)
+@click.option(
+    "--source",
+    "source_component_ids",
+    help="Filter connections by source component ID(s). Can be specified multiple times.",
+    type=int,
+    multiple=True,
+    required=False,
+)
+@click.option(
+    "--destination",
+    "destination_component_ids",
+    help="Filter connections by destination component ID(s). Can be specified multiple times.",
+    type=int,
+    multiple=True,
+    required=False,
+)
+async def list_microgrid_electrical_component_connections(
+    ctx: click.Context,
+    microgrid_id: int,
+    source_component_ids: tuple[int, ...],
+    destination_component_ids: tuple[int, ...],
+) -> None:
+    """
+    Get and display electrical component connections by microgrid ID.
+
+    This command fetches detailed information about all electrical component connections
+    in a specific microgrid from the Assets API and displays it in JSON format.
+    The output can be piped to other tools for further processing.
+
+    Args:
+        ctx: Click context object containing the initialized API client.
+        microgrid_id: The unique identifier of the microgrid to retrieve.
+        source_component_ids: Optional filter for connections from specific
+            source component IDs.
+        destination_component_ids: Optional filter for connections to specific
+            destination component IDs.
+
+    Raises:
+        click.Abort: If there is an error printing the electrical component connections.
+
+    Example:
+        ```bash
+        # Get all connections for microgrid with ID 123
+        assets-cli component-connections 123
+
+        # Filter by source component
+        assets-cli component-connections 123 --source 5
+
+        # Filter by destination component
+        assets-cli component-connections 123 --destination 10
+
+        # Filter by both source and destination
+        assets-cli component-connections 123 --source 5 --destination 10
+
+        # Filter by multiple source components
+        assets-cli component-connections 123 --source 5 --source 6 --source 7
+
+        # Pipe output to jq for filtering
+        assets-cli component-connections 123 | jq ".[]"
+        ```
+    """
+    try:
+        client = ctx.obj["client"]
+        component_connections = (
+            await client.list_microgrid_electrical_component_connections(
+                microgrid_id,
+                source_component_ids=(
+                    list(source_component_ids) if source_component_ids else None
+                ),
+                destination_component_ids=(
+                    list(destination_component_ids)
+                    if destination_component_ids
+                    else None
+                ),
+            )
+        )
+        print_component_connections(component_connections)
+    except ApiClientError as e:
+        error_dict = {
+            "error_type": type(e).__name__,
+            "server_url": e.server_url,
+            "operation": e.operation,
+            "description": e.description,
+        }
+        click.echo(json.dumps(error_dict, indent=2))
+        raise click.Abort()
+
+
 def main() -> None:
     """
     Initialize and run the CLI application.

src/frequenz/client/assets/cli/_utils.py

@@ -4,6 +4,10 @@
 
 from frequenz.client.assets._microgrid import Microgrid
 from frequenz.client.assets._microgrid_json import microgrid_to_json
+from frequenz.client.assets.electrical_component._connection import ComponentConnection
+from frequenz.client.assets.electrical_component._connection_json import (
+    component_connections_to_json,
+)
 from frequenz.client.assets.electrical_component._electrical_component import (
     ElectricalComponent,
 )
@@ -42,3 +46,17 @@ def print_electrical_components(
         electrical_components: The list of ElectricalComponent instances to print to console.
     """
     click.echo(electrical_components_to_json(electrical_components))
+
+
+def print_component_connections(
+    component_connections: list[ComponentConnection],
+) -> None:
+    """
+    Print electrical component connections to console in JSON format using custom encoder.
+
+    This function converts the ComponentConnection instances to JSON using a custom
+    encoder and outputs it as formatted JSON to the console. The output is
+    designed to be machine-readable and can be piped to tools like jq for
+    further processing.
+    """
+    click.echo(component_connections_to_json(component_connections))

src/frequenz/client/assets/electrical_component/_connection.py

@@ -0,0 +1,72 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Component connection."""
+
+import dataclasses
+from datetime import datetime, timezone
+
+from frequenz.client.common.microgrid.components 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 object 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 allow 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")
+
+    def is_operational_at(self, timestamp: datetime) -> bool:
+        """Check whether this connection is operational at a specific timestamp."""
+        return self.operational_lifetime.is_operational_at(timestamp)
+
+    def is_operational_now(self) -> bool:
+        """Whether this connection is currently operational."""
+        return self.is_operational_at(datetime.now(timezone.utc))
+
+    def __str__(self) -> str:
+        """Return a human-readable string representation of this instance."""
+        return f"{self.source}->{self.destination}"

src/frequenz/client/assets/electrical_component/_connection_json.py

@@ -0,0 +1,21 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""JSON encoder for ComponentConnection objects."""
+
+import json
+from dataclasses import asdict
+
+from .._utils import AssetsJSONEncoder
+from ._connection import ComponentConnection
+
+
+def component_connections_to_json(
+    component_connections: list[ComponentConnection],
+) -> str:
+    """Convert a list of ElectricalComponent objects to a JSON string."""
+    return json.dumps(
+        [asdict(connection) for connection in component_connections],
+        cls=AssetsJSONEncoder,
+        indent=2,
+    )