Implement a ComponentGraphGenerator from the assets API

Signed-off-by: Sahas Subramanian <[email protected]>

Author: shsms

src/frequenz/gridpool/__init__.py

@@ -2,3 +2,7 @@
 # Copyright © 2025 Frequenz Energy-as-a-Service GmbH
 
 """High-level interface to grid pools for the Frequenz platform."""
+
+from ._graph_generator import ComponentGraphGenerator
+
+__all__ = ["ComponentGraphGenerator"]

src/frequenz/gridpool/_graph_generator.py

@@ -0,0 +1,90 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Formula generation from assets API component/connection configurations."""
+
+from typing import cast
+
+from frequenz.client.assets import AssetsApiClient
+from frequenz.client.assets.electrical_component import (
+    ComponentConnection,
+    ElectricalComponent,
+)
+from frequenz.client.base import channel
+from frequenz.client.common.microgrid import MicrogridId
+from frequenz.client.common.microgrid.electrical_components import ElectricalComponentId
+from frequenz.microgrid_component_graph import ComponentGraph
+
+
+class ComponentGraphGenerator:
+    """Generates component graphs for microgrids using the Assets API."""
+
+    def __init__(  # pylint: disable=too-many-arguments
+        self,
+        server_url: str,
+        *,
+        auth_key: str | None = None,
+        sign_secret: str | None = None,
+        channel_defaults: channel.ChannelOptions = channel.ChannelOptions(),
+        connect: bool = True,
+    ) -> None:
+        """Initialize this instance.
+
+        Args:
+            server_url: The location of the microgrid API server in the form of a URL.
+                The following format is expected:
+                "grpc://hostname{:`port`}{?ssl=`ssl`}",
+                where the `port` should be an int between 0 and 65535 (defaulting to
+                9090) and `ssl` should be a boolean (defaulting to `true`).
+                For example: `grpc://localhost:1090?ssl=true`.
+            auth_key: The authentication key to use for the connection.
+            sign_secret: The secret to use for signing requests.
+            channel_defaults: The default options use to create the channel when not
+                specified in the URL.
+            connect: Whether to connect to the server as soon as a client instance is
+                created. If `False`, the client will not connect to the server until
+                [connect()][frequenz.client.base.client.BaseApiClient.connect] is
+                called.
+        """
+        self._client: AssetsApiClient = AssetsApiClient(
+            server_url,
+            auth_key=auth_key,
+            sign_secret=sign_secret,
+            channel_defaults=channel_defaults,
+            connect=connect,
+        )
+
+    async def get_component_graph(
+        self, microgrid_id: MicrogridId
+    ) -> ComponentGraph[
+        ElectricalComponent, ComponentConnection, ElectricalComponentId
+    ]:
+        """Generate a component graph for the given microgrid ID.
+
+        Args:
+            microgrid_id: The ID of the microgrid to generate the graph for.
+
+        Returns:
+            The component graph representing the microgrid's electrical
+                components and their connections.
+
+        Raises:
+            ValueError: If any component connections could not be loaded.
+        """
+        components = await self._client.list_microgrid_electrical_components(
+            microgrid_id
+        )
+        connections = (
+            await self._client.list_microgrid_electrical_component_connections(
+                microgrid_id
+            )
+        )
+
+        if any(c is None for c in connections):
+            raise ValueError("Failed to load all electrical component connections.")
+
+        graph = ComponentGraph[
+            ElectricalComponent, ComponentConnection, ElectricalComponentId
+        ](components, cast(list[ComponentConnection], connections))
+
+        return graph