Add CLI tool to print formulas

RELEASE_NOTES.md

@@ -10,6 +10,8 @@
 
 ## New Features
 
+* Add CLI tool to print formulas from assets API component graph.
+
 <!-- Here goes the main new features and examples or instructions on how to use them -->
 
 ## Bug Fixes

pyproject.toml

@@ -15,7 +15,6 @@ description = "High-level interface to grid pools for the Frequenz platform."
 readme = "README.md"
 license = { text = "MIT" }
 keywords = ["frequenz", "python", "lib", "library", "gridpool"]
-# TODO(cookiecutter): Remove and add more classifiers if appropriate
 classifiers = [
   "Development Status :: 3 - Alpha",
   "Intended Audience :: Developers",
@@ -26,17 +25,21 @@ classifiers = [
   "Typing :: Typed",
 ]
 requires-python = ">= 3.11, < 4"
-# TODO(cookiecutter): Remove and add more dependencies if appropriate
 dependencies = [
+  "asyncclick >= 8.3.0.4, < 9",
   "typing-extensions >= 4.14.1, < 5",
+  "frequenz-microgrid-component-graph @ git+https://github.com/shsms/frequenz-microgrid-component-graph-python.git@c4a458e06e541846de0a25142d5b821dd7934f47",
+  "frequenz-client-assets @ git+https://github.com/shsms/frequenz-client-assets-python@bbf09104df24ddfac497e2a3dd66fe68cfdacd25"
 ]
 dynamic = ["version"]
 
+[project.scripts]
+gridpool-cli = "frequenz.gridpool.cli.__main__:main"
+
 [[project.authors]]
 name = "Frequenz Energy-as-a-Service GmbH"
 email = "floss@frequenz.com"
 
-# TODO(cookiecutter): Remove and add more optional dependencies if appropriate
 [project.optional-dependencies]
 dev-flake8 = [
   "flake8 == 7.3.0",

src/frequenz/gridpool/__init__.py

@@ -1,25 +1,8 @@
 # License: MIT
 # Copyright © 2025 Frequenz Energy-as-a-Service GmbH
 
-"""High-level interface to grid pools for the Frequenz platform..
+"""High-level interface to grid pools for the Frequenz platform."""
 
-TODO(cookiecutter): Add a more descriptive module description.
-"""
+from ._graph_generator import ComponentGraphGenerator
 
-
-# TODO(cookiecutter): Remove this function
-def delete_me(*, blow_up: bool = False) -> bool:
-    """Do stuff for demonstration purposes.
-
-    Args:
-        blow_up: If True, raise an exception.
-
-    Returns:
-        True if no exception was raised.
-
-    Raises:
-        RuntimeError: if blow_up is True.
-    """
-    if blow_up:
-        raise RuntimeError("This function should be removed!")
-    return True
+__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

src/frequenz/gridpool/cli/__init__.py

@@ -0,0 +1 @@
+"""Package for CLI tool for gridpool functionality."""

src/frequenz/gridpool/cli/__main__.py

@@ -0,0 +1,65 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""CLI tool for gridpool functionality."""
+
+import os
+
+import asyncclick as click
+from frequenz.client.common.microgrid import MicrogridId
+
+from frequenz.gridpool import ComponentGraphGenerator
+
+
+@click.group()
+async def cli() -> None:
+    """CLI tool for gridpool functionality."""
+
+
+@cli.command()
+@click.argument("microgrid_id", type=int)
+@click.option(
+    "--prefix",
+    type=str,
+    default="{component}",
+    help="Prefix format for the output (Supports wildcards {microgrid_id} and {component}).",
+)
+async def print_formulas(
+    microgrid_id: int,
+    prefix: str,
+) -> None:
+    """Fetch and print component graph formulas for a microgrid."""
+    url = os.environ.get("ASSETS_API_URL")
+    key = os.environ.get("ASSETS_API_KEY")
+    secret = os.environ.get("ASSETS_API_SECRET")
+    if not url or not key or not secret:
+        raise click.ClickException(
+            "ASSETS_API_URL, ASSETS_API_KEY, ASSETS_API_SECRET must be set."
+        )
+    cgg = ComponentGraphGenerator(url, auth_key=key, sign_secret=secret)
+
+    graph = await cgg.get_component_graph(MicrogridId(microgrid_id))
+    power_formulas = {
+        "consumption": graph.consumer_formula(),
+        "generation": graph.producer_formula(),
+        "grid": graph.grid_formula(),
+        "pv": graph.pv_formula(None),
+        "battery": graph.battery_formula(None),
+        "chp": graph.chp_formula(None),
+        "ev": graph.ev_charger_formula(None),
+    }
+
+    for component, formula in power_formulas.items():
+        print(
+            prefix.format(component=component, microgrid_id=microgrid_id)
+            + f' = "{formula}"'
+        )
+
+
+def main() -> None:
+    """Run the CLI tool."""
+    cli()
+
+
+if __name__ == "__main__":
+    main()

tests/test_gridpool.py

@@ -2,17 +2,66 @@
 # Copyright © 2025 Frequenz Energy-as-a-Service GmbH
 
 """Tests for the frequenz.gridpool package."""
-import pytest
 
-from frequenz.gridpool import delete_me
+from unittest.mock import AsyncMock, MagicMock, patch
 
+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
 
-def test_gridpool_succeeds() -> None:  # TODO(cookiecutter): Remove
-    """Test that the delete_me function succeeds."""
-    assert delete_me() is True
+from frequenz.gridpool._graph_generator import ComponentGraphGenerator
 
 
-def test_gridpool_fails() -> None:  # TODO(cookiecutter): Remove
-    """Test that the delete_me function fails."""
-    with pytest.raises(RuntimeError, match="This function should be removed!"):
-        delete_me(blow_up=True)
+async def test_formula_generation() -> None:
+    """Test that the frequenz.gridpool package loads correctly."""
+    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("grpc://never.where:-55", connect=False)
+    with patch.object(g, "_client", 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)"