feat(assets): Add functionality to list electrical component connections for microgrids

This commit introduces a new command to the CLI for retrieving and displaying electrical component connections within a specified microgrid. The new method `list_microgrid_electrical_component_connections` in the `AssetsApiClient` allows filtering by source and destination component IDs. Additionally, utility functions for printing component connections in JSON format have been added, along with the necessary data structures and protobuf handling for `ComponentConnection` objects.

The CLI command enhances usability by allowing users to filter connections and output results in a machine-readable format.

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,
+    )