Merge pull request #176 from cwasicki/cfg

Switch to schema-based microgrid config loading

Author: cwasicki

RELEASE_NOTES.md

@@ -8,6 +8,10 @@
 
 <!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with -->
 
+- `MicrogridConfig`: Switch to schema-based loading of microgrid config files and updates to the config class:
+   - Remove unused nested field `assets` and replace by its contents `pv`, `wind`, `battery`.
+   - Make `meta` and `ctype` public fields.
+   - Require `meta.microgrid_id` to be set.
 - The minimum supported version of `matplotlib` is now `v3.9.2`.
 - Add `src/frequenz/lib/notebooks/reporting/schema_mapping.yaml` to your deployment so notebooks can load the canonical column definitions via `ColumnMapper`.
 

pyproject.toml

@@ -35,6 +35,7 @@ dependencies = [
   "pvlib >= 0.13.0, < 0.14.0",
   "python-dotenv >= 0.21.0, < 1.2.0",
   "toml >= 0.10.2, < 0.11.0",
+  "marshmallow_dataclass >= 8.7.1, < 9",
   "plotly >= 6.0.0, < 6.4.0",
   "kaleido >= 0.2.1, < 1.2.0",
   "frequenz-client-reporting >= 0.19.0, < 0.20.0",

src/frequenz/data/microgrid/config.py

@@ -6,9 +6,12 @@
 import logging
 import re
 import tomllib
-from dataclasses import dataclass
+from dataclasses import field
 from pathlib import Path
-from typing import Any, Literal, cast, get_args
+from typing import Any, ClassVar, Literal, Self, Type, cast, get_args
+
+from marshmallow import Schema
+from marshmallow_dataclass import dataclass
 
 _logger = logging.getLogger(__name__)
 
@@ -23,9 +26,6 @@
 class ComponentTypeConfig:
     """Configuration of a microgrid component type."""
 
-    component_type: ComponentType
-    """Type of the component."""
-
     meter: list[int] | None = None
     """List of meter IDs for this component."""
 
@@ -45,12 +45,6 @@ def __post_init__(self) -> None:
             self.formula["AC_ACTIVE_POWER"] = "+".join(
                 [f"#{cid}" for cid in self._default_cids()]
             )
-        if self.component_type == "battery" and "BATTERY_SOC_PCT" not in self.formula:
-            if self.component:
-                cids = self.component
-                form = "+".join([f"#{cid}" for cid in cids])
-                form = f"({form})/({len(cids)})"
-                self.formula["BATTERY_SOC_PCT"] = form
 
     def cids(self, metric: str = "") -> list[int]:
         """Get component IDs for this component.
@@ -74,9 +68,7 @@ def cids(self, metric: str = "") -> list[int]:
                 raise ValueError("Formula must be a dictionary.")
             formula = self.formula.get(metric)
             if not formula:
-                raise ValueError(
-                    f"{metric} does not have a formula for {self.component_type}"
-                )
+                raise ValueError(f"{metric} does not have a formula")
             # Extract component IDs from the formula which are given as e.g. #123
             pattern = r"#(\d+)"
             return [int(e) for e in re.findall(pattern, self.formula[metric])]
@@ -102,7 +94,7 @@ def _default_cids(self) -> list[int]:
         if self.component:
             return self.component
 
-        raise ValueError(f"No IDs available for {self.component_type}")
+        raise ValueError("No IDs available")
 
     @classmethod
     def is_valid_type(cls, ctype: str) -> bool:
@@ -158,31 +150,17 @@ class BatteryConfig:
     """Capacity of the battery in Wh."""
 
 
-@dataclass(frozen=True)
-class AssetsConfig:
-    """Configuration of the assets in a microgrid."""
-
-    pv: dict[str, PVConfig] | None = None
-    """Configuration of the PV system."""
-
-    wind: dict[str, WindConfig] | None = None
-    """Configuration of the wind turbines."""
-
-    battery: dict[str, BatteryConfig] | None = None
-    """Configuration of the batteries."""
-
-
 # pylint: disable=too-many-instance-attributes
 @dataclass(frozen=True)
 class Metadata:
     """Metadata for a microgrid."""
 
+    microgrid_id: int
+    """ID of the microgrid."""
+
     name: str | None = None
     """Name of the microgrid."""
 
-    microgrid_id: int | None = None
-    """ID of the microgrid."""
-
     enterprise_id: int | None = None
     """Enterprise ID of the microgrid."""
 
@@ -206,48 +184,24 @@ class Metadata:
 class MicrogridConfig:
     """Configuration of a microgrid."""
 
-    _metadata: Metadata
+    meta: Metadata
     """Metadata of the microgrid."""
 
-    _assets_cfg: AssetsConfig
-    """Configuration of the assets in the microgrid."""
+    pv: dict[str, PVConfig] | None = None
+    """Configuration of the PV system."""
 
-    _component_types_cfg: dict[str, ComponentTypeConfig]
-    """Mapping of component category types to ac power component config."""
+    wind: dict[str, WindConfig] | None = None
+    """Configuration of the wind turbines."""
 
-    def __init__(self, config_dict: dict[str, Any]) -> None:
-        """Initialize the microgrid configuration.
+    battery: dict[str, BatteryConfig] | None = None
+    """Configuration of the batteries."""
 
-        Args:
-            config_dict: Dictionary with component type as key and config as value.
-        """
-        self._metadata = Metadata(**(config_dict.get("meta") or {}))
-
-        self._assets_cfg = AssetsConfig(
-            pv=config_dict.get("pv") or {},
-            wind=config_dict.get("wind") or {},
-            battery=config_dict.get("battery") or {},
-        )
-
-        self._component_types_cfg = {
-            ctype: ComponentTypeConfig(component_type=cast(ComponentType, ctype), **cfg)
-            for ctype, cfg in config_dict.get("ctype", {}).items()
-            if ComponentTypeConfig.is_valid_type(ctype)
-        }
-
-    @property
-    def meta(self) -> Metadata:
-        """Return the metadata of the microgrid."""
-        return self._metadata
-
-    @property
-    def assets(self) -> AssetsConfig:
-        """Return the assets configuration of the microgrid."""
-        return self._assets_cfg
+    ctype: dict[str, ComponentTypeConfig] = field(default_factory=dict)
+    """Mapping of component category types to ac power component config."""
 
     def component_types(self) -> list[str]:
         """Get a list of all component types in the configuration."""
-        return list(self._component_types_cfg.keys())
+        return list(self.ctype.keys())
 
     def component_type_ids(
         self,
@@ -271,7 +225,7 @@ def component_type_ids(
             ValueError: If the component type is unknown.
             KeyError: If `component_category` is invalid.
         """
-        cfg = self._component_types_cfg.get(component_type)
+        cfg = self.ctype.get(component_type)
         if not cfg:
             raise ValueError(f"{component_type} not found in config.")
 
@@ -300,7 +254,7 @@ def formula(self, component_type: str, metric: str) -> str:
         Raises:
             ValueError: If the component type is unknown or formula is missing.
         """
-        cfg = self._component_types_cfg.get(component_type)
+        cfg = self.ctype.get(component_type)
         if not cfg:
             raise ValueError(f"{component_type} not found in config.")
         if cfg.formula is None:
@@ -311,6 +265,67 @@ def formula(self, component_type: str, metric: str) -> str:
 
         return formula
 
+    Schema: ClassVar[Type[Schema]] = Schema
+
+    @classmethod
+    def _load_table_entries(cls, data: dict[str, Any]) -> dict[str, Self]:
+        """Load microgrid configurations from table entries.
+
+        Args:
+            data: The loaded TOML data.
+
+        Returns:
+            A dict mapping microgrid IDs to MicrogridConfig instances.
+
+        Raises:
+            ValueError: If top-level keys are not numeric microgrid IDs
+                or if there is a microgrid ID mismatch.
+            TypeError: If microgrid data is not a dict.
+        """
+        if not all(str(k).isdigit() for k in data.keys()):
+            raise ValueError("All top-level keys must be numeric microgrid IDs.")
+
+        mgrids = {}
+        for mid, entry in data.items():
+            if not mid.isdigit():
+                raise ValueError(
+                    f"Table reader: Microgrid ID key must be numeric, got {mid}"
+                )
+            if not isinstance(entry, dict):
+                raise TypeError("Table reader: Each microgrid entry must be a dict")
+
+            mgrid = cls.Schema().load(entry)
+            if mgrid.meta is None or mgrid.meta.microgrid_id is None:
+                raise ValueError(
+                    "Table reader: Each microgrid entry must have a meta.microgrid_id"
+                )
+            if int(mgrid.meta.microgrid_id) != int(mid):
+                raise ValueError(
+                    f"Table reader: Microgrid ID mismatch: key {mid} != {mgrid.meta.microgrid_id}"
+                )
+
+            mgrids[mid] = mgrid
+
+        return mgrids
+
+    @classmethod
+    def load_from_file(cls, config_path: Path) -> dict[int, Self]:
+        """
+        Load and validate configuration settings from a TOML file.
+
+        Args:
+            config_path: the path to the TOML configuration file.
+
+        Returns:
+            A dict mapping microgrid IDs to MicrogridConfig instances.
+        """
+        with config_path.open("rb") as f:
+            data = tomllib.load(f)
+
+        assert isinstance(data, dict)
+
+        return cls._load_table_entries(data)
+
     @staticmethod
     def load_configs(
         microgrid_config_files: str | Path | list[str | Path] | None = None,
@@ -369,14 +384,7 @@ def load_configs(
                 _logger.warning("Config path %s is not a file, skipping.", config_path)
                 continue
 
-            with config_path.open("rb") as f:
-                cfg_dict = tomllib.load(f)
-                for microgrid_id, mcfg in cfg_dict.items():
-                    _logger.debug(
-                        "Loading microgrid config for ID %s from %s",
-                        microgrid_id,
-                        config_path,
-                    )
-                    microgrid_configs[microgrid_id] = MicrogridConfig(mcfg)
+            mcfgs = MicrogridConfig.load_from_file(config_path)
+            microgrid_configs.update({str(key): value for key, value in mcfgs.items()})
 
         return microgrid_configs

tests/test_config.py

@@ -4,7 +4,7 @@
 """Tests for the frequenz.lib.notebooks.config module."""
 
 from pathlib import Path
-from typing import Any, cast
+from typing import Any
 
 import pytest
 from pytest_mock import MockerFixture
@@ -14,7 +14,7 @@
 
 VALID_CONFIG: dict[str, dict[str, Any]] = {
     "1": {
-        "meta": {"name": "Test Grid", "gid": 1},
+        "meta": {"name": "Test Grid", "gid": 1, "microgrid_id": 1},
         "ctype": {
             "pv": {"meter": [101, 102], "formula": {"AC_ACTIVE_POWER": "#12+#23"}},
             "battery": {
@@ -41,7 +41,8 @@
 @pytest.fixture
 def valid_microgrid_config() -> MicrogridConfig:
     """Fixture to provide a valid MicrogridConfig instance."""
-    return MicrogridConfig(VALID_CONFIG["1"])
+    # pylint: disable=protected-access
+    return MicrogridConfig._load_table_entries(VALID_CONFIG)["1"]
 
 
 def test_is_valid_type() -> None:
@@ -52,25 +53,22 @@ def test_is_valid_type() -> None:
 
 def test_component_type_config_cids() -> None:
     """Test the retrieval of component IDs for various configurations."""
-    config = ComponentTypeConfig(component_type="pv", meter=[1, 2, 3])
+    config = ComponentTypeConfig(inverter=[1, 2, 3])
     assert config.cids() == [1, 2, 3]
 
-    config = ComponentTypeConfig(component_type="battery", inverter=[4, 5])
+    config = ComponentTypeConfig(meter=[4, 5], inverter=[1, 2, 3])
     assert config.cids() == [4, 5]
 
-    with pytest.raises(ValueError):
-        config = ComponentTypeConfig(component_type="grid")
-        config.cids()
-
 
 def test_microgrid_config_init(valid_microgrid_config: MicrogridConfig) -> None:
     """Test initialisation of MicrogridConfig with valid configuration data."""
+    assert valid_microgrid_config.meta is not None
     assert valid_microgrid_config.meta.name == "Test Grid"
-    pv_config = valid_microgrid_config.assets.pv
-    if pv_config:
-        _assert_optional_field(
-            cast(dict[str, float], pv_config["PV1"]).get("peak_power"), 5000
-        )
+    pv_config = valid_microgrid_config.pv
+    assert pv_config is not None
+    pv_system = pv_config.get("PV1")
+    assert pv_system is not None
+    assert pv_system.peak_power == 5000
 
 
 def test_microgrid_config_component_types(
@@ -115,10 +113,10 @@ def test_microgrid_config_formula(valid_microgrid_config: MicrogridConfig) -> No
 def test_load_configs(mocker: MockerFixture) -> None:
     """Test loading configurations for multiple microgrids from mock TOML files."""
     toml_data = """
+    1.meta.microgrid_id = 1
     1.meta.name = "Test Grid"
     1.meta.gid = 1
     1.ctype.pv.meter = [101, 102]
-    1.ctype.pv.formula = "AC_ACTIVE_POWER"
     1.ctype.battery.inverter = [201, 202, 203]
     1.ctype.battery.component = [301, 302, 303, 304, 305, 306]
     1.pv.PV1.peak_power = 5000
@@ -133,16 +131,20 @@ def test_load_configs(mocker: MockerFixture) -> None:
     configs = MicrogridConfig.load_configs(Path("mock_path.toml"))
 
     assert "1" in configs
+    assert configs["1"].meta is not None
     assert configs["1"].meta.name == "Test Grid"
-    pv_config = configs["1"].assets.pv
-    battery_config = configs["1"].assets.battery
-    if pv_config and battery_config:
-        _assert_optional_field(
-            cast(dict[str, float], pv_config["PV1"]).get("peak_power"), 5000
-        )
-        _assert_optional_field(
-            cast(dict[str, float], battery_config["BAT1"]).get("capacity"), 10000
-        )
+
+    pv_config = configs["1"].pv
+    assert pv_config is not None
+    pv_system = pv_config.get("PV1")
+    assert pv_system is not None
+    assert pv_system.peak_power == 5000
+
+    battery_config = configs["1"].battery
+    assert battery_config is not None
+    battery_system = battery_config.get("BAT1")
+    assert battery_system is not None
+    assert battery_system.capacity == 10000
 
 
 def _assert_optional_field(value: float | None, expected: float) -> None: