diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index e053502..08b143f 100644 --- a/RELEASE_NOTES.md +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index c38873c..a853915 100644 --- a/pyproject.toml +++ b/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"] diff --git a/src/frequenz/gridpool/__init__.py b/src/frequenz/gridpool/__init__.py index 2481eba..78c4629 100644 --- a/src/frequenz/gridpool/__init__.py +++ b/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"] diff --git a/src/frequenz/gridpool/_graph_generator.py b/src/frequenz/gridpool/_graph_generator.py new file mode 100644 index 0000000..69db5f2 --- /dev/null +++ b/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 diff --git a/tests/test_graph_generator.py b/tests/test_graph_generator.py new file mode 100644 index 0000000..019b235 --- /dev/null +++ b/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)"