diff --git a/README.md b/README.md index 7986dab..ef01a42 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,50 @@ The following platforms are officially supported (tested): - **Operating System:** Ubuntu Linux 20.04 - **Architectures:** amd64, arm64 +## CLI + +This package ships the `gridpool-cli` command with two subcommands. + +### Setup + +Set the Assets API credentials before running the CLI: + +```bash +export ASSETS_API_URL="grpc://..." +export ASSETS_API_AUTH_KEY="..." +export ASSETS_API_SIGN_SECRET="..." +``` + +### Print component formulas + +```bash +gridpool-cli print-formulas +``` + +Optional prefix formatting: + +```bash +gridpool-cli print-formulas --prefix "{microgrid_id}.{component}" +``` + +### Render component graph + +Rendering requires optional dependencies. Install with: + +```bash +pip install frequenz-gridpool[render-graph] +``` + +```bash +gridpool-cli render-graph +``` + +To save without opening a window: + +```bash +gridpool-cli render-graph --no-show --output component_graph.png +``` + ## Contributing If you want to know how to build this project and contribute to it, please diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 43a6405..eddb4f9 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -10,7 +10,9 @@ ## New Features - +* Added `gridpool-cli render-graph` to visualize microgrid component graphs using the + Assets API credentials (`ASSETS_API_URL`, `ASSETS_API_AUTH_KEY`, and + `ASSETS_API_SIGN_SECRET`). ## Bug Fixes diff --git a/pyproject.toml b/pyproject.toml index 2adcc54..f0845d2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -42,6 +42,11 @@ name = "Frequenz Energy-as-a-Service GmbH" email = "floss@frequenz.com" [project.optional-dependencies] +render-graph = [ + "matplotlib >= 3.7.0, < 4", + "networkx >= 3.0, < 4", +] + dev-flake8 = [ "flake8 == 7.3.0", "flake8-docstrings == 1.7.0", @@ -64,6 +69,7 @@ dev-mkdocs = [ ] dev-mypy = [ "mypy == 1.19.0", + "types-networkx>=3.6.1.20251220", "types-Markdown == 3.10.0.20251106", # For checking the noxfile, docs/ script, and tests "frequenz-gridpool[dev-mkdocs,dev-noxfile,dev-pytest]", @@ -86,7 +92,7 @@ dev-pytest = [ "async-solipsism == 0.9", ] dev = [ - "frequenz-gridpool[dev-mkdocs,dev-flake8,dev-formatting,dev-mkdocs,dev-mypy,dev-noxfile,dev-pylint,dev-pytest]", + "frequenz-gridpool[dev-mkdocs,dev-flake8,dev-formatting,dev-mkdocs,dev-mypy,dev-noxfile,dev-pylint,dev-pytest,render-graph]", ] [project.urls] diff --git a/src/frequenz/gridpool/cli/__main__.py b/src/frequenz/gridpool/cli/__main__.py index 96cf28a..0445e88 100644 --- a/src/frequenz/gridpool/cli/__main__.py +++ b/src/frequenz/gridpool/cli/__main__.py @@ -10,6 +10,7 @@ from frequenz.client.common.microgrid import MicrogridId from frequenz.gridpool import ComponentGraphGenerator +from frequenz.gridpool.cli._render_graph import ComponentGraphRenderer, RenderOptions @click.group() @@ -63,6 +64,47 @@ async def print_formulas( ) +@cli.command("render-graph") +@click.argument("microgrid_id", type=int) +@click.option( + "--output", + type=click.Path(dir_okay=False, writable=True), + default="component_graph.png", + show_default=True, + help="Output image path.", +) +@click.option( + "--show/--no-show", + default=True, + show_default=True, + help="Display the graph interactively.", +) +async def render_graph(microgrid_id: int, output: str, show: bool) -> None: + """Render and save a component graph visualization for a microgrid.""" + url = os.environ.get("ASSETS_API_URL") + key = os.environ.get("ASSETS_API_AUTH_KEY") + secret = os.environ.get("ASSETS_API_SIGN_SECRET") + if not url or not key or not secret: + raise click.ClickException( + "ASSETS_API_URL, ASSETS_API_AUTH_KEY, ASSETS_API_SIGN_SECRET must be set." + ) + + try: + async with AssetsApiClient( + url, + auth_key=key, + sign_secret=secret, + ) as client: + renderer = ComponentGraphRenderer(client) + graph = await renderer.build_graph(MicrogridId(microgrid_id)) + if not graph.nodes: + raise click.ClickException("No components found for this microgrid.") + pos = renderer.compute_layout(graph) + renderer.render(graph, pos, RenderOptions(output=output, show=show)) + except RuntimeError as exc: + raise click.ClickException(str(exc)) from exc + + def main() -> None: """Run the CLI tool.""" cli(_anyio_backend="asyncio") diff --git a/src/frequenz/gridpool/cli/_render_graph.py b/src/frequenz/gridpool/cli/_render_graph.py new file mode 100644 index 0000000..a47089d --- /dev/null +++ b/src/frequenz/gridpool/cli/_render_graph.py @@ -0,0 +1,244 @@ +# License: MIT +# Copyright © 2025 Frequenz Energy-as-a-Service GmbH +# pylint: disable=import-error + +"""Render component graphs via the Assets API.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from frequenz.client.assets import AssetsApiClient +from frequenz.client.common.microgrid import MicrogridId + + +def _format_category(category: Any) -> str: + """Normalize a component category into a short display label.""" + if category is None: + return "UNKNOWN" + if hasattr(category, "name"): + return str(category.name).replace("COMPONENT_CATEGORY_", "") + return str(category).replace("COMPONENT_CATEGORY_", "") + + +def _require_networkx() -> Any: + try: + # pylint: disable=import-outside-toplevel + import networkx as nx + except ModuleNotFoundError as exc: + raise RuntimeError( + "Rendering requires optional dependencies. Install " + "`frequenz-gridpool[render-graph]`." + ) from exc + return nx + + +def _require_matplotlib() -> Any: + try: + # pylint: disable=import-outside-toplevel + import matplotlib.pyplot as plt + except ModuleNotFoundError as exc: + raise RuntimeError( + "Rendering requires optional dependencies. Install " + "`frequenz-gridpool[render-graph]`." + ) from exc + return plt + + +@dataclass(frozen=True) +class RenderOptions: + """Rendering options for component graphs.""" + + output: str + show: bool = True + figsize: tuple[int, int] = (15, 10) + title: str = "Microgrid Component Graph" + + +class ComponentGraphRenderer: + """Build and render component graphs for a microgrid.""" + + def __init__(self, client: AssetsApiClient) -> None: + """Initialize the renderer. + + Args: + client: API client used to fetch microgrid components and their connections. + """ + self._client = client + + async def build_graph(self, microgrid_id: MicrogridId) -> Any: + """Build a directed graph representing the microgrid electrical topology. + + Fetches electrical components and their connections for the given microgrid and + constructs a directed graph where nodes represent components and edges represent + connections from source to destination. + + Args: + microgrid_id: Identifier of the microgrid to fetch and model. + + Returns: + A directed graph containing component nodes and connection edges. + """ + components = await self._client.list_microgrid_electrical_components( + microgrid_id + ) + connections = ( + await self._client.list_microgrid_electrical_component_connections( + microgrid_id + ) + ) + + nx = _require_networkx() + graph = nx.DiGraph() + + for component in components: + graph.add_node( + component.id, + name=component.name or str(component.id), + category=_format_category(component.category), + orig_id=component.id, + ) + + for connection in connections: + if connection is None: + continue + graph.add_edge(connection.source, connection.destination) + + return graph + + def compute_layout(self, graph: Any) -> dict[Any, tuple[float, float]]: + """Compute a layered node layout with the root on the left. + + Selects a root node, groups nodes by shortest-path distance from the root, and + assigns (x, y) coordinates such that layers are spaced along the x-axis. + + Args: + graph: A graph-like object containing nodes and edges. + + Returns: + A mapping from node identifiers to (x, y) coordinates. + """ + if not graph.nodes: + return {} + + root_node = self._select_root(graph) + layered_nodes = self._group_by_level(graph, root_node) + return self._build_positions(layered_nodes) + + def _select_root(self, graph: Any) -> Any: + """Select a root node for layout. + + Prefers a node with no incoming edges and at least one outgoing edge. If no such + node exists, falls back to an arbitrary node. + + Args: + graph: A graph-like object containing nodes and edges. + + Returns: + The selected root node identifier. + """ + roots = [ + node + for node in graph.nodes + if graph.in_degree(node) == 0 and graph.out_degree(node) > 0 + ] + return roots[0] if roots else list(graph.nodes)[0] + + def _group_by_level(self, graph: Any, root_node: Any) -> dict[int, list[Any]]: + """Group nodes into layers by shortest-path distance from a root node. + + Nodes reachable from the root are assigned to layers based on their shortest-path + distance. Nodes not reachable from the root are placed into a final "orphan" + layer after the deepest reachable layer. + + Args: + graph: A graph-like object containing nodes and edges. + root_node: The node to treat as the root for distance computation. + + Returns: + A mapping from layer index to the list of nodes in that layer. + """ + nx = _require_networkx() + levels = nx.single_source_shortest_path_length(graph, root_node) + layered_nodes: dict[int, list[Any]] = {} + for node, dist in levels.items(): + layered_nodes.setdefault(dist, []).append(node) + + orphans = [node for node in graph.nodes if node not in levels] + if orphans: + orphan_layer = max(layered_nodes.keys()) + 1 if layered_nodes else 0 + layered_nodes[orphan_layer] = orphans + return layered_nodes + + def _build_positions( + self, layered_nodes: dict[int, list[Any]] + ) -> dict[Any, tuple[float, float]]: + """Compute node positions from pre-grouped layers. + + Assigns x-coordinates based on layer index and y-coordinates by distributing + nodes evenly within each layer. + + Args: + layered_nodes: Mapping from layer index to the list of nodes in that layer. + + Returns: + A mapping from node identifiers to (x, y) coordinates. + """ + x_spacing, y_spacing = 2.5, 1.2 + pos: dict[Any, tuple[float, float]] = {} + for level, nodes in layered_nodes.items(): + x_pos = level * x_spacing + sorted_nodes = sorted(nodes, key=str) + y_start = (len(sorted_nodes) - 1) * y_spacing / 2 + for i, node in enumerate(sorted_nodes): + pos[node] = (x_pos, y_start - (i * y_spacing)) + return pos + + def render( + self, graph: Any, pos: dict[Any, tuple[float, float]], options: RenderOptions + ) -> None: + """Render a component graph to an image file and optionally display it. + + This method uses NetworkX and Matplotlib to draw the given graph using the + provided node positions, applies basic styling (labels, colors, arrows), + saves the resulting figure to the path specified in ``options.output``, + and, if configured, opens an interactive window to show the plot. + + Args: + graph: A NetworkX graph-like object (typically a ``DiGraph``) whose + nodes and edges represent the microgrid components and their + electrical connections. + pos: A mapping from node identifiers to ``(x, y)`` coordinates used + to place each node in the rendered figure, usually obtained from + :meth:`compute_layout`. + options: Rendering configuration, including output file path, whether + to show the figure interactively, figure size, and plot title. + """ + nx = _require_networkx() + plt = _require_matplotlib() + + node_labels = { + node: ( + f'{graph.nodes[node].get("name", str(node))}\n' + f'ID: {graph.nodes[node].get("orig_id", node)}' + ) + for node in graph.nodes + } + + plt.figure(figsize=options.figsize) + nx.draw( + graph, + pos, + with_labels=True, + labels=node_labels, + node_color="lightblue", + edge_color="gray", + node_size=800, + font_size=8, + ) + plt.title(options.title) + plt.tight_layout() + plt.savefig(options.output, dpi=300) + if options.show: + plt.show() diff --git a/tests/test_render_graph.py b/tests/test_render_graph.py new file mode 100644 index 0000000..fca10f1 --- /dev/null +++ b/tests/test_render_graph.py @@ -0,0 +1,156 @@ +# License: MIT +# Copyright © 2025 Frequenz Energy-as-a-Service GmbH +# pylint: disable=import-error + +"""Tests for component graph rendering utilities.""" + +from types import SimpleNamespace +from typing import TYPE_CHECKING, Any +from unittest.mock import AsyncMock, MagicMock + +import pytest +from frequenz.client.assets import AssetsApiClient +from frequenz.client.common.microgrid import MicrogridId + +from frequenz.gridpool.cli._render_graph import ( + ComponentGraphRenderer, + RenderOptions, + _format_category, +) + +if TYPE_CHECKING: + import matplotlib.pyplot as plt + import networkx as nx +else: + nx = pytest.importorskip("networkx") + plt = pytest.importorskip("matplotlib.pyplot") + + +class _Category: + def __init__(self, name: str) -> None: + self.name = name + + +def test_format_category_handles_none() -> None: + """It should default to UNKNOWN when no category is set.""" + assert _format_category(None) == "UNKNOWN" + + +def test_format_category_handles_named_enum() -> None: + """It should strip the COMPONENT_CATEGORY_ prefix from enum-like values.""" + assert _format_category(_Category("COMPONENT_CATEGORY_PV")) == "PV" + + +def test_format_category_handles_string() -> None: + """It should strip the prefix from string values.""" + assert _format_category("COMPONENT_CATEGORY_CHP") == "CHP" + + +@pytest.mark.asyncio +async def test_build_graph_populates_nodes_and_edges() -> None: + """It should create nodes with attributes and edges from connections.""" + client = MagicMock(spec=AssetsApiClient) + components = [ + SimpleNamespace( + id=1, name="Meter-1", category=_Category("COMPONENT_CATEGORY_METER") + ), + SimpleNamespace(id=2, name=None, category=None), + ] + connections = [SimpleNamespace(source=1, destination=2), None] + client.list_microgrid_electrical_components = AsyncMock(return_value=components) + client.list_microgrid_electrical_component_connections = AsyncMock( + return_value=connections + ) + + renderer = ComponentGraphRenderer(client) + graph = await renderer.build_graph(MicrogridId(10)) + + assert graph.has_edge(1, 2) + assert graph.nodes[1]["name"] == "Meter-1" + assert graph.nodes[1]["category"] == "METER" + assert graph.nodes[2]["name"] == "2" + assert graph.nodes[2]["category"] == "UNKNOWN" + assert graph.nodes[2]["orig_id"] == 2 + + +def test_compute_layout_empty_graph_returns_empty_mapping() -> None: + """It should return an empty mapping when the graph is empty.""" + renderer = ComponentGraphRenderer(MagicMock(spec=AssetsApiClient)) + graph: nx.DiGraph[Any] = nx.DiGraph() + assert not renderer.compute_layout(graph) + + +def test_compute_layout_positions_nodes_by_layer() -> None: + """It should position nodes by layer with the root on the left.""" + renderer = ComponentGraphRenderer(MagicMock(spec=AssetsApiClient)) + graph: nx.DiGraph[Any] = nx.DiGraph() + graph.add_edge("root", "child") + + pos = renderer.compute_layout(graph) + + assert pos["root"] == (0.0, 0.0) + assert pos["child"] == (2.5, 0.0) + + +def test_select_root_prefers_nodes_with_children() -> None: + """It should select a root node that has outgoing edges.""" + renderer = ComponentGraphRenderer(MagicMock(spec=AssetsApiClient)) + graph: nx.DiGraph[Any] = nx.DiGraph() + graph.add_node(3) + graph.add_node(1) + graph.add_edge(1, 2) + + assert renderer._select_root(graph) == 1 # pylint: disable=protected-access + + +def test_group_by_level_adds_orphans() -> None: + """It should append orphan nodes after the deepest layer.""" + renderer = ComponentGraphRenderer(MagicMock(spec=AssetsApiClient)) + graph: nx.DiGraph[Any] = nx.DiGraph() + graph.add_edge("root", "child") + graph.add_node("orphan") + + layered = renderer._group_by_level( # pylint: disable=protected-access + graph, "root" + ) + + assert layered[0] == ["root"] + assert layered[1] == ["child"] + assert layered[2] == ["orphan"] + + +def test_build_positions_centers_nodes_in_layer() -> None: + """It should center nodes vertically within each layer.""" + renderer = ComponentGraphRenderer(MagicMock(spec=AssetsApiClient)) + layered_nodes = {0: ["b", "a"], 1: ["c"]} + + pos = renderer._build_positions(layered_nodes) # pylint: disable=protected-access + + assert pos["a"] == (0.0, 0.6) + assert pos["b"] == (0.0, -0.6) + assert pos["c"] == (2.5, 0.0) + + +def test_render_writes_file_without_show(monkeypatch: pytest.MonkeyPatch) -> None: + """It should save a figure and avoid showing it when configured.""" + renderer = ComponentGraphRenderer(MagicMock(spec=AssetsApiClient)) + graph: nx.DiGraph[Any] = nx.DiGraph() + graph.add_node(1, name="Node-1", orig_id=1) + pos = {1: (0.0, 0.0)} + + draw_calls: list[dict[str, object]] = [] + saved_paths: list[str] = [] + shown: list[bool] = [] + + monkeypatch.setattr(nx, "draw", lambda *args, **kwargs: draw_calls.append(kwargs)) + monkeypatch.setattr(plt, "figure", lambda *args, **kwargs: None) + monkeypatch.setattr(plt, "title", lambda *args, **kwargs: None) + monkeypatch.setattr(plt, "tight_layout", lambda *args, **kwargs: None) + monkeypatch.setattr(plt, "savefig", lambda path, dpi=None: saved_paths.append(path)) + monkeypatch.setattr(plt, "show", lambda *args, **kwargs: shown.append(True)) + + renderer.render(graph, pos, RenderOptions(output="component_graph.png", show=False)) + + assert saved_paths == ["component_graph.png"] + assert not shown + assert draw_calls