Add CLI command to generate configs

RELEASE_NOTES.md

@@ -2,23 +2,25 @@
 
 ## Summary
 
-This release reworks the microgrid configuration loading API: the loaders are now module-level functions exported from `frequenz.gridpool` for loading configs from files, from the Assets API, or from both layered together.
+This release reworks the microgrid configuration loading API: the loaders are now module-level functions exported from `frequenz.gridpool` for loading configs from files, from the Assets API, or from both layered together. It also adds a `generate-config` CLI command.
 
 ## Upgrading
 
 The configuration loaders are no longer `MicrogridConfig` static methods — they are module-level functions exported from `frequenz.gridpool`, and their signatures have changed:
 
 * `MicrogridConfig.load_configs(files, dir)` → `load_configs_from_files(files)`. The `microgrid_config_dir` argument has been removed; pass an explicit list of files instead, e.g. `load_configs_from_files(list(Path(my_dir).glob("*.toml")))`. Note that the name `load_configs` now refers to the new merge wrapper (see New Features), not the file loader.
 * `MicrogridConfig.load_configs_with_formulas(assets_url, assets_auth_key, assets_sign_secret, files, dir)` → `load_configs(default_files=files, assets_client=client, override_files=...)`. It now takes an `AssetsApiClient` you construct instead of raw credentials, and layers its sources by explicit precedence (default files < Assets API < override files) rather than the old "load files, then fill in missing formulas" behavior.
+* The CLI dependencies (`asyncclick` and the new `tomlkit`) are no longer core dependencies; they have moved to a `cli` optional extra. Install the CLI with `pip install frequenz-gridpool[cli]` (the `render-graph` extra now pulls this in automatically).
 
 ## New Features
 
 * Added config loading functions, exported from `frequenz.gridpool`:
    * `load_configs_from_files(...)` loads microgrid configs from one or more TOML files.
-   * `load_configs_from_api(...)` loads microgrid metadata (latitude/longitude) from the Assets API and optionally populates formulas from the component graph. Formulas are derived for all supported component types by default, and a microgrid that cannot be loaded is logged and skipped rather than aborting the whole batch.
+   * `load_configs_from_api(...)` loads microgrid metadata (latitude/longitude) from the Assets API and derives the per-type formulas and meter/inverter/component IDs directly from the component graph. The two steps fail independently: a microgrid whose metadata cannot be fetched is skipped, while one whose component graph cannot be derived is still returned with metadata only. Both are logged rather than aborting the whole batch.
    * `load_configs(...)` loads from up to three layered sources — default files, the Assets API, and override files — and merges them, with higher layers winning on conflicts. The microgrid IDs fetched from the Assets API can be given explicitly or are otherwise taken from the files.
 * Added `merge_microgrid_configs(...)` for deep-merging two `MicrogridConfig` objects where override values take precedence and `None` does not overwrite base values.
 * Added `merge_config_maps(...)` for merging two dictionaries of microgrid configs by microgrid ID.
+* Added a `generate-config` CLI command that derives a microgrid config from the Assets API and prints it as TOML, optionally layering `--default` and `--override` config files by precedence (`--default` < Assets API < `--override`).
 
 ## Bug Fixes
 

pyproject.toml

@@ -5,6 +5,7 @@
 requires = [
   "setuptools == 82.0.1",
   "setuptools_scm[toml] == 10.0.5",
+  "vcs_versioning < 2",  # setuptools_scm 10.0.5 needs parse_version, dropped in 2.x
   "frequenz-repo-config[lib] == 0.17.0",
 ]
 build-backend = "setuptools.build_meta"
@@ -27,7 +28,6 @@ classifiers = [
 requires-python = ">= 3.11, < 4"
 dependencies = [
   "marshmallow-dataclass >= 8.7.1, < 9",
-  "asyncclick >= 8.3.0.4, < 9",
   "typing-extensions >= 4.14.1, < 5",
   "frequenz-microgrid-component-graph >= 0.4.0, < 0.5",
   "frequenz-client-assets >= 0.3.1, < 0.4",
@@ -42,7 +42,12 @@ name = "Frequenz Energy-as-a-Service GmbH"
 email = "floss@frequenz.com"
 
 [project.optional-dependencies]
+cli = [
+  "asyncclick >= 8.3.0.4, < 9",
+  "tomlkit >= 0.13.0, < 1",
+]
 render-graph = [
+  "frequenz-gridpool[cli]",
   "matplotlib >= 3.7.0, < 4",
   "networkx >= 3.0, < 4",
 ]
@@ -91,6 +96,7 @@ dev-pytest = [
   "pytest-mock == 3.15.1",
   "pytest-asyncio == 1.4.0",
   "async-solipsism == 0.9",
+  "frequenz-gridpool[cli]",  # The tests cover the cli package
 ]
 dev = [
   "frequenz-gridpool[dev-mkdocs,dev-flake8,dev-formatting,dev-mkdocs,dev-mypy,dev-noxfile,dev-pylint,dev-pytest,render-graph]",

src/frequenz/gridpool/_graph_generator.py

@@ -1,22 +1,40 @@
 # License: MIT
 # Copyright © 2025 Frequenz Energy-as-a-Service GmbH
 
-"""Formula generation from assets API component/connection configurations."""
+"""Component graph generation and per-type queries over the Assets API.
+
+The `ComponentGraphGenerator` builds a `MicrogridComponentGraph` from the
+Platform Assets API. The module-level query functions (e.g. `pv_meter_ids`)
+return component IDs read back out of such a graph.
+"""
 
 import logging
+from collections.abc import Callable
 
 from frequenz.client.assets import AssetsApiClient
 from frequenz.client.assets.electrical_component import (
+    Battery,
+    BatteryInverter,
     Breaker,
+    Chp,
     ComponentConnection,
     ElectricalComponent,
+    EvCharger,
+    GridConnectionPoint,
+    Meter,
+    SolarInverter,
 )
 from frequenz.client.common.microgrid import MicrogridId
 from frequenz.client.common.microgrid.electrical_components import ElectricalComponentId
 from frequenz.microgrid_component_graph import ComponentGraph
 
 _logger = logging.getLogger(__name__)
 
+MicrogridComponentGraph = ComponentGraph[
+    ElectricalComponent, ComponentConnection, ElectricalComponentId
+]
+"""Component graph specialized for a microgrid's electrical components."""
+
 
 class ComponentGraphGenerator:
     """Generates component graphs for microgrids using the Assets API."""
@@ -35,9 +53,7 @@ def __init__(
 
     async def get_component_graph(
         self, microgrid_id: MicrogridId
-    ) -> ComponentGraph[
-        ElectricalComponent, ComponentConnection, ElectricalComponentId
-    ]:
+    ) -> MicrogridComponentGraph:
         """Generate a component graph for the given microgrid ID.
 
         Args:
@@ -87,3 +103,79 @@ async def get_component_graph(
         ](components, connections)
 
         return graph
+
+
+def _ids_of(
+    graph: MicrogridComponentGraph, component_class: type[ElectricalComponent]
+) -> list[int]:
+    """Return the sorted IDs of all components of the given class."""
+    return sorted(int(c.id) for c in graph.components(matching_types=component_class))
+
+
+def _meter_ids_where(
+    graph: MicrogridComponentGraph,
+    is_meter: Callable[[ElectricalComponentId], bool],
+) -> list[int]:
+    """Return the sorted IDs of all meters matching the given classifier."""
+    return sorted(
+        int(m.id) for m in graph.components(matching_types=Meter) if is_meter(m.id)
+    )
+
+
+def grid_meter_ids(graph: MicrogridComponentGraph) -> list[int]:
+    """Return the meters directly downstream of a grid connection point.
+
+    Grid meters have no dedicated classifier, so they are derived from the graph
+    topology.
+    """
+    grid_ids = {int(c.id) for c in graph.components(matching_types=GridConnectionPoint)}
+    return sorted(
+        int(m.id)
+        for m in graph.components(matching_types=Meter)
+        if any(int(p.id) in grid_ids for p in graph.predecessors(m.id))
+    )
+
+
+def pv_meter_ids(graph: MicrogridComponentGraph) -> list[int]:
+    """Return the sorted IDs of all PV meters."""
+    return _meter_ids_where(graph, graph.is_pv_meter)
+
+
+def pv_inverter_ids(graph: MicrogridComponentGraph) -> list[int]:
+    """Return the sorted IDs of all solar inverters."""
+    return _ids_of(graph, SolarInverter)
+
+
+def battery_meter_ids(graph: MicrogridComponentGraph) -> list[int]:
+    """Return the sorted IDs of all battery meters."""
+    return _meter_ids_where(graph, graph.is_battery_meter)
+
+
+def battery_inverter_ids(graph: MicrogridComponentGraph) -> list[int]:
+    """Return the sorted IDs of all battery inverters."""
+    return _ids_of(graph, BatteryInverter)
+
+
+def battery_ids(graph: MicrogridComponentGraph) -> list[int]:
+    """Return the sorted IDs of all batteries."""
+    return _ids_of(graph, Battery)
+
+
+def chp_meter_ids(graph: MicrogridComponentGraph) -> list[int]:
+    """Return the sorted IDs of all CHP meters."""
+    return _meter_ids_where(graph, graph.is_chp_meter)
+
+
+def chp_ids(graph: MicrogridComponentGraph) -> list[int]:
+    """Return the sorted IDs of all CHPs."""
+    return _ids_of(graph, Chp)
+
+
+def ev_charger_meter_ids(graph: MicrogridComponentGraph) -> list[int]:
+    """Return the sorted IDs of all EV charger meters."""
+    return _meter_ids_where(graph, graph.is_ev_charger_meter)
+
+
+def ev_charger_ids(graph: MicrogridComponentGraph) -> list[int]:
+    """Return the sorted IDs of all EV chargers."""
+    return _ids_of(graph, EvCharger)

src/frequenz/gridpool/cli/__main__.py

@@ -4,12 +4,14 @@
 """CLI tool for gridpool functionality."""
 
 import os
+from pathlib import Path
 
 import asyncclick as click
 from frequenz.client.assets import AssetsApiClient
 from frequenz.client.common.microgrid import MicrogridId
 
-from frequenz.gridpool import ComponentGraphGenerator
+from frequenz.gridpool import ComponentGraphGenerator, load_configs
+from frequenz.gridpool.cli._dump_config import dump_map
 from frequenz.gridpool.cli._render_graph import ComponentGraphRenderer, RenderOptions
 
 
@@ -105,6 +107,61 @@ async def render_graph(microgrid_id: int, output: str, show: bool) -> None:
         raise click.ClickException(str(exc)) from exc
 
 
+@cli.command("generate-config")
+@click.argument("microgrid_ids", type=int, nargs=-1)
+@click.option(
+    "--default",
+    "default_file",
+    type=click.Path(exists=True, dir_okay=False, path_type=Path),
+    default=None,
+    help="Config file whose values the Assets API overrides (lowest precedence).",
+)
+@click.option(
+    "--override",
+    "override_file",
+    type=click.Path(exists=True, dir_okay=False, path_type=Path),
+    default=None,
+    help="Config file whose values override the Assets API (highest precedence).",
+)
+async def generate_config(
+    microgrid_ids: tuple[int, ...],
+    default_file: Path | None,
+    override_file: Path | None,
+) -> None:
+    """Generate microgrid config from the Assets API as TOML.
+
+    Derives metadata, formulas and component IDs for the given microgrid IDs and
+    prints the result as dotted-key TOML to stdout.
+
+    `--default` and `--override` each take a config file and are layered with the
+    Assets API by precedence: `--default` < Assets API < `--override`. So a file
+    passed as `--override` keeps its values where it has them (the API only fills
+    gaps), while a file passed as `--default` is overridden by the API. If no
+    microgrid IDs are given, they are taken from the supplied files. Files are
+    only read; redirect stdout to save the result.
+    """
+    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."
+        )
+
+    async with AssetsApiClient(url, auth_key=key, sign_secret=secret) as client:
+        configs = await load_configs(
+            default_files=default_file,
+            assets_client=client,
+            override_files=override_file,
+            microgrid_ids=list(dict.fromkeys(microgrid_ids)) or None,
+        )
+
+    if not configs:
+        raise click.ClickException("No microgrids could be loaded; nothing to write.")
+
+    click.echo(dump_map(configs), nl=False)
+
+
 def main() -> None:
     """Run the CLI tool."""
     cli(_anyio_backend="asyncio")

src/frequenz/gridpool/cli/_dump_config.py

@@ -0,0 +1,77 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Serialize microgrid configurations to dotted-key TOML.
+
+Renders a `{microgrid_id: MicrogridConfig}` mapping as dotted-key TOML, e.g.:
+
+    115.meta.microgrid_id = 115
+    115.meta.name = "Demo Grid"
+    115.meta.latitude = 52.52
+
+This is the inverse of `load_configs_from_files`. Value rendering (quoting,
+escaping, key-quoting, list/number/datetime formatting) is delegated to
+`tomlkit`; only the flattening to dotted keys and the dropping of empty/`None`
+fields are done here, since TOML has no null and `tomlkit` rejects it.
+"""
+
+from typing import Any
+
+import tomlkit
+
+from frequenz.gridpool import MicrogridConfig
+
+
+def _is_empty(value: Any) -> bool:
+    """Whether a dumped value should be omitted from the output."""
+    return value is None or value == {} or value == []
+
+
+def _iter_leaves(
+    prefix: list[str], data: dict[str, Any]
+) -> list[tuple[list[str], Any]]:
+    """Flatten a nested dict into `(key_path, value)` pairs, dropping empties.
+
+    Args:
+        prefix: The key-path prefix accumulated so far.
+        data: The nested mapping to flatten.
+
+    Returns:
+        A list of `(key_path, leaf_value)` pairs, where `key_path` is the list
+        of nested keys leading to a non-empty scalar or list value.
+    """
+    leaves: list[tuple[list[str], Any]] = []
+    for key, value in data.items():
+        if _is_empty(value):
+            continue
+        path = prefix + [key]
+        if isinstance(value, dict):
+            leaves.extend(_iter_leaves(path, value))
+        else:
+            leaves.append((path, value))
+    return leaves
+
+
+def dump_map(configs: dict[str, MicrogridConfig]) -> str:
+    """Serialize a mapping of microgrid configs to dotted-key TOML.
+
+    Args:
+        configs: Mapping from microgrid ID (as string) to `MicrogridConfig`.
+
+    Returns:
+        The TOML representation as a string, with one blank line between
+        microgrids and entries sorted by numeric microgrid ID.
+    """
+    schema = MicrogridConfig.Schema()
+    doc = tomlkit.document()
+    for mid in sorted(configs, key=int):
+        dumped = schema.dump(configs[mid])
+        assert isinstance(dumped, dict)
+        leaves = _iter_leaves([mid], dumped)
+        if not leaves:
+            continue
+        if doc.body:
+            doc.add(tomlkit.nl())
+        for path, value in leaves:
+            doc.append(tomlkit.key(path), value)
+    return tomlkit.dumps(doc)