Add config load assets api

RELEASE_NOTES.md

@@ -2,15 +2,17 @@
 
 ## Summary
 
-<!-- Here goes a general summary of what this release is about -->
+This release adds a new Assets API based configuration loader, introduces helpers to merge microgrid configs, and updates PV curtailability behavior to support unspecified values.
 
 ## Upgrading
 
 <!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with -->
 
 ## New Features
 
-* Add start and end time to microgrid configs.
+* Added `MicrogridConfig.load_configs_from_assets_api(...)` to load microgrid metadata (latitude/longitude) from the Assets API and optionally populate formulas from the component graph.
+* 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.
 
 ## Bug Fixes
 

src/frequenz/gridpool/__init__.py

@@ -4,6 +4,17 @@
 """High-level interface to grid pools for the Frequenz platform."""
 
 from ._graph_generator import ComponentGraphGenerator
-from ._microgrid_config import Metadata, MicrogridConfig
+from ._microgrid_config import (
+    Metadata,
+    MicrogridConfig,
+    merge_config_maps,
+    merge_microgrid_configs,
+)
 
-__all__ = ["ComponentGraphGenerator", "Metadata", "MicrogridConfig"]
+__all__ = [
+    "ComponentGraphGenerator",
+    "Metadata",
+    "MicrogridConfig",
+    "merge_config_maps",
+    "merge_microgrid_configs",
+]

src/frequenz/gridpool/_microgrid_config.py

@@ -6,6 +6,7 @@
 import logging
 import re
 import tomllib
+from copy import deepcopy
 from dataclasses import field
 from datetime import datetime
 from pathlib import Path
@@ -420,6 +421,77 @@ def load_configs(
 
         return microgrid_configs
 
+    @staticmethod
+    async def load_configs_from_assets_api(
+        assets_url: str,
+        assets_auth_key: str,
+        assets_sign_secret: str,
+        microgrid_ids: list[int],
+        populate_formulas: bool = True,
+    ) -> dict[str, "MicrogridConfig"]:
+        """Load microgrid configs with location metadata from the Assets API.
+
+        Fetches each microgrid's location (latitude, longitude) and optionally
+        populates formulas from the component graph.  This is the canonical
+        single-source loader for both metadata and formulas so that callers
+        (e.g. the forecast pipeline) do not have to re-implement this logic.
+
+        Args:
+            assets_url:
+                Base URL of the Assets API.
+            assets_auth_key:
+                Authentication key used to access the Assets API.
+            assets_sign_secret:
+                Signing secret used for authenticated API requests.
+            microgrid_ids:
+                List of microgrid IDs to load configurations for.
+            populate_formulas:
+                When `True` (default), formulas are derived from the component
+                graph and written into each config via
+                `populate_missing_formulas`.  Set to `False` to load
+                metadata only.
+
+        Returns:
+            dict[str, MicrogridConfig]:
+                Mapping from microgrid ID (as string) to the populated
+                `MicrogridConfig` instance. Microgrids that could not be loaded
+                are logged as warnings and omitted, so the returned mapping may
+                cover fewer microgrids than were requested.
+        """
+        async with AssetsApiClient(
+            assets_url,
+            auth_key=assets_auth_key,
+            sign_secret=assets_sign_secret,
+        ) as client:
+            configs: dict[str, MicrogridConfig] = {}
+            for microgrid_id in microgrid_ids:
+                try:
+                    mgrid = await client.get_microgrid(MicrogridId(microgrid_id))
+                    location = mgrid.location if mgrid.location else None
+                    cfg = MicrogridConfig(
+                        meta=Metadata(
+                            microgrid_id=microgrid_id,
+                            latitude=location.latitude if location else None,
+                            longitude=location.longitude if location else None,
+                        )
+                    )
+                    if populate_formulas:
+                        await populate_missing_formulas(
+                            microgrid_id=microgrid_id,
+                            config=cfg,
+                            assets_client=client,
+                        )
+                except Exception as exc:  # pylint: disable=broad-except
+                    _logger.warning(
+                        "Failed to load microgrid %s from the Assets API: %s",
+                        microgrid_id,
+                        exc,
+                    )
+                    continue
+                configs[str(microgrid_id)] = cfg
+
+        return configs
+
     @staticmethod
     async def load_configs_with_formulas(
         assets_url: str,
@@ -451,10 +523,10 @@ async def load_configs_with_formulas(
         Returns:
             dict[str, MicrogridConfig]:
                 Mapping from microgrid ID (as string) to the corresponding populated
-                ``MicrogridConfig`` instance.
+                `MicrogridConfig` instance.
 
         Notes:
-            - Configuration files are first loaded via ``MicrogridConfig.load_configs``.
+            - Configuration files are first loaded via `MicrogridConfig.load_configs`.
             - Any missing formulas are populated by querying the Assets API and
             generating formulas from the microgrid component graph.
         """
@@ -479,6 +551,84 @@ async def load_configs_with_formulas(
         return microgrid_configs
 
 
+def merge_microgrid_configs(
+    base: MicrogridConfig,
+    override: MicrogridConfig,
+) -> MicrogridConfig:
+    """Merge two `MicrogridConfig` objects.
+
+    The *override* config takes precedence over *base*.  Nested dictionaries
+    are merged recursively.  If a field in *override* is `None` the value
+    from *base* is retained, so partial overrides never nullify existing data.
+
+    Args:
+        base: The base MicrogridConfig.
+        override: The overriding MicrogridConfig.
+
+    Returns:
+        A new MicrogridConfig representing the merged result.
+    """
+    schema = MicrogridConfig.Schema()
+    base_dict = schema.dump(base)
+    override_dict = schema.dump(override)
+
+    def _deep_merge(a: dict[Any, Any], b: dict[Any, Any]) -> dict[Any, Any]:
+        result = deepcopy(a)
+        for k, v in b.items():
+            if v is None:
+                continue
+            if isinstance(v, dict) and isinstance(result.get(k), dict):
+                result[k] = _deep_merge(result[k], v)
+            else:
+                result[k] = v
+        return result
+
+    merged = schema.load(_deep_merge(base_dict, override_dict))
+    assert isinstance(merged, MicrogridConfig)
+    return merged
+
+
+def merge_config_maps(
+    base: dict[str, MicrogridConfig],
+    override: dict[str, MicrogridConfig],
+) -> dict[str, MicrogridConfig]:
+    """Merge two dictionaries of `MicrogridConfig` objects.
+
+    For microgrid IDs present in both maps the configs are merged via
+    `merge_microgrid_configs`.  IDs that exist only in one map are
+    included unchanged.
+
+    Args:
+        base: The base dictionary of MicrogridConfig objects.
+        override: The overriding dictionary of MicrogridConfig objects.
+
+    Returns:
+        A new dictionary representing the merged result.
+    """
+    merged = dict(base)
+    for mid, cfg in override.items():
+        if mid in merged:
+            merged[mid] = merge_microgrid_configs(merged[mid], cfg)
+        else:
+            merged[mid] = cfg
+    return merged
+
+
+def _is_zero_formula(formula: str) -> bool:
+    """Return whether a derived formula is empty or a constant zero.
+
+    Component types absent from a microgrid yield an empty formula or one that
+    is just a zero constant (e.g. `0.0`), which is not worth storing.
+    """
+    stripped = formula.strip()
+    if not stripped:
+        return True
+    try:
+        return float(stripped) == 0.0
+    except ValueError:
+        return False
+
+
 async def populate_missing_formulas(
     microgrid_id: int,
     config: "MicrogridConfig",
@@ -488,9 +638,9 @@ async def populate_missing_formulas(
     """Populate missing component formulas from the assets API graph.
 
     Builds a component graph for the given microgrid and derives default formulas
-    for common component types such as consumption, generation, grid, PV, battery,
-    CHP, and EV charging. Existing formulas already present in the configuration
-    are preserved; only missing component-type entries or missing metric formulas
+    for common component types such as consumption, grid, PV, battery, CHP, and
+    EV charging. Existing formulas already present in the configuration are
+    preserved; only missing component-type entries or missing metric formulas
     are filled in.
 
     Args:
@@ -502,52 +652,47 @@ async def populate_missing_formulas(
         assets_client:
             Assets API client used to fetch the component graph.
         component_types:
-            Optional set of component types to consider when populating formulas.
+            Set of component types to consider when populating formulas. When
+            `None` (the default), every component type a formula can be derived
+            for is considered.
 
     Returns:
         None. The configuration is modified in place.
 
     Notes:
-        - Existing formulas in ``config`` are never overwritten.
-        - For missing component types, a new ``ComponentTypeConfig`` is created.
-        - The same derived formula is assigned to all supported metric keys for a
-        given component type when missing.
+        - Existing formulas in `config` are never overwritten.
+        - For missing component types, a new `ComponentTypeConfig` is created.
+        - The derived formula is assigned to the `AC_POWER_ACTIVE` metric key
+        for a given component type when missing.
     """
     cgg = ComponentGraphGenerator(assets_client)
     graph = await cgg.get_component_graph(MicrogridId(microgrid_id))
 
     auto_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),
     }
 
-    metrics = (
-        "AC_POWER_ACTIVE",
-        "AC_ACTIVE_POWER",
-        "AC_ENERGY_ACTIVE",
-        "AC_ENERGY_ACTIVE_CONSUMED",
-        "AC_ENERGY_ACTIVE_DELIVERED",
-    )
-
-    # Default: only populate component types already present in the config.
-    allowed_ctypes = component_types or set(config.ctype.keys())
-
     for ctype, formula in auto_formulas.items():
-        if ctype not in allowed_ctypes:
+        if component_types is not None and ctype not in component_types:
+            continue
+
+        # Skip component types, whose derived formula
+        # is empty or evaluates to a constant zero.
+        if _is_zero_formula(formula):
             continue
 
         cfg = config.ctype.get(ctype)
         if cfg is None:
-            continue
+            cfg = ComponentTypeConfig()
+            config.ctype[ctype] = cfg
 
         if cfg.formula is None:
             cfg.formula = {}
 
-        for metric in metrics:
-            if metric not in cfg.formula:
-                cfg.formula[metric] = formula
+        if "AC_POWER_ACTIVE" not in cfg.formula:
+            cfg.formula["AC_POWER_ACTIVE"] = formula