Implement a ComponentGraphGenerator from the assets API

RELEASE_NOTES.md

@@ -12,6 +12,8 @@
 
 * Expose microgrid config `Metadata`.
 
+* This introduces the `ComponentGraphGenerator`, that uses the assets API to fetch assets for a specified microgrid, and builds a component graph for it, from which various formulas for the microgrid can be generated.
+
 ## Bug Fixes
 
 <!-- Here goes notable bug fixes that are worth a special mention or explanation -->

pyproject.toml

@@ -30,6 +30,8 @@ requires-python = ">= 3.11, < 4"
 dependencies = [
   "marshmallow-dataclass>=8.7.1,<9",
   "typing-extensions >= 4.14.1, < 5",
+  "frequenz-microgrid-component-graph >= 0.2.0, < 0.3",
+  "frequenz-client-assets >= 0.2.0, < 0.3",
 ]
 dynamic = ["version"]
 

src/frequenz/gridpool/__init__.py

@@ -3,6 +3,7 @@
 
 """High-level interface to grid pools for the Frequenz platform."""
 
+from ._graph_generator import ComponentGraphGenerator
 from ._microgrid_config import Metadata, MicrogridConfig
 
-__all__ = ["Metadata", "MicrogridConfig"]
+__all__ = ["ComponentGraphGenerator", "Metadata", "MicrogridConfig"]

src/frequenz/gridpool/_graph_generator.py

@@ -0,0 +1,66 @@
+# 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.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__(
+        self,
+        client: AssetsApiClient,
+    ) -> None:
+        """Initialize this instance.
+
+        Args:
+            client: The Assets API client to use for fetching components and
+                connections.
+        """
+        self._client: AssetsApiClient = client
+
+    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

tests/test_graph_generator.py

@@ -0,0 +1,66 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Tests for the component graph generator."""
+
+from unittest.mock import AsyncMock, MagicMock
+
+from frequenz.client.assets import AssetsApiClient
+from frequenz.client.assets.electrical_component import (
+    ComponentConnection,
+    GridConnectionPoint,
+    Meter,
+    SolarInverter,
+)
+from frequenz.client.common.microgrid import MicrogridId
+from frequenz.client.common.microgrid.electrical_components import ElectricalComponentId
+
+from frequenz.gridpool._graph_generator import ComponentGraphGenerator
+
+
+async def test_formula_generation() -> None:
+    """Test formula generation from component graph created from Assets API."""
+    assets_client_mock = MagicMock(spec=AssetsApiClient)
+    assets_client_mock.list_microgrid_electrical_components = AsyncMock(
+        return_value=[
+            GridConnectionPoint(
+                id=ElectricalComponentId(1),
+                microgrid_id=MicrogridId(10),
+                rated_fuse_current=100,
+            ),
+            Meter(
+                id=ElectricalComponentId(2),
+                microgrid_id=MicrogridId(10),
+            ),
+            Meter(
+                id=ElectricalComponentId(3),
+                microgrid_id=MicrogridId(10),
+            ),
+            SolarInverter(
+                id=ElectricalComponentId(4),
+                microgrid_id=MicrogridId(10),
+            ),
+        ]
+    )
+    assets_client_mock.list_microgrid_electrical_component_connections = AsyncMock(
+        return_value=[
+            ComponentConnection(
+                source=ElectricalComponentId(1),
+                destination=ElectricalComponentId(2),
+            ),
+            ComponentConnection(
+                source=ElectricalComponentId(1),
+                destination=ElectricalComponentId(3),
+            ),
+            ComponentConnection(
+                source=ElectricalComponentId(2),
+                destination=ElectricalComponentId(4),
+            ),
+        ]
+    )
+
+    g = ComponentGraphGenerator(assets_client_mock)
+    graph = await g.get_component_graph(MicrogridId(10))
+
+    assert graph.grid_formula() == "COALESCE(#2, #4, 0.0) + #3"
+    assert graph.pv_formula(None) == "COALESCE(#4, #2, 0.0)"