Switch to schema-based microgrid config loading

Previously, microgrid configs were initialized by unpacking nested
dictionaries. Now the class defines a marshmallow schema and uses it to
deserialize toml data into valudated python objects.

Introduces a new `load_from_file` function to load multiple configs as
dict from a single file. This is also used now in the existing function
to load multiple files or directories.

Signed-off-by: cwasicki <[email protected]>

Author: cwasicki

src/frequenz/data/microgrid/config.py

@@ -8,8 +8,9 @@
 import tomllib
 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__)
@@ -198,24 +199,6 @@ class MicrogridConfig:
     ctype: dict[str, ComponentTypeConfig] = field(default_factory=dict)
     """Mapping of component category types to ac power component config."""
 
-    def __init__(self, config_dict: dict[str, Any]) -> None:
-        """Initialize the microgrid configuration.
-
-        Args:
-            config_dict: Dictionary with component type as key and config as value.
-        """
-        self.meta = Metadata(**(config_dict.get("meta") or {}))
-
-        self.pv = config_dict.get("pv") or {}
-        self.wind = config_dict.get("wind") or {}
-        self.battery = config_dict.get("battery") or {}
-
-        self.ctype = {
-            ctype: ComponentTypeConfig(**cfg)
-            for ctype, cfg in config_dict.get("ctype", {}).items()
-            if ComponentTypeConfig.is_valid_type(ctype)
-        }
-
     def component_types(self) -> list[str]:
         """Get a list of all component types in the configuration."""
         return list(self.ctype.keys())
@@ -282,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,
@@ -340,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
@@ -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:
@@ -64,10 +65,10 @@ def test_microgrid_config_init(valid_microgrid_config: MicrogridConfig) -> None:
     assert valid_microgrid_config.meta is not None
     assert valid_microgrid_config.meta.name == "Test Grid"
     pv_config = valid_microgrid_config.pv
-    if pv_config:
-        _assert_optional_field(
-            cast(dict[str, float], pv_config["PV1"]).get("peak_power"), 5000
-        )
+    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(
@@ -132,15 +133,18 @@ def test_load_configs(mocker: MockerFixture) -> None:
     assert "1" in configs
     assert configs["1"].meta is not None
     assert configs["1"].meta.name == "Test Grid"
+
     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
-    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
-        )
+    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: