Index

frequenz.gridpool.config

Microgrid configuration data model and loading.

Attributes

frequenz.gridpool.config.ComponentCategory module-attribute

ComponentCategory = Literal[
    "meter", "inverter", "component"
]

Valid component categories.

frequenz.gridpool.config.ComponentType module-attribute

ComponentType = Literal[
    "grid", "pv", "battery", "consumption", "chp", "ev"
]

Valid component types.

Classes

frequenz.gridpool.config.BatteryConfig

Configuration of a battery in a microgrid.

Source code in src/frequenz/gridpool/config/microgrid.py

@dataclass(frozen=True)
class BatteryConfig:
    """Configuration of a battery in a microgrid."""

    start_time: datetime | None = None
    """Start time of the battery installation."""

    end_time: datetime | None = None
    """End time of the battery installation."""

    capacity: float | None = None
    """Capacity of the battery in Wh."""

Attributes

capacity class-attribute instance-attribute

capacity: float | None = None

Capacity of the battery in Wh.

end_time class-attribute instance-attribute

end_time: datetime | None = None

End time of the battery installation.

start_time class-attribute instance-attribute

start_time: datetime | None = None

Start time of the battery installation.

frequenz.gridpool.config.ComponentTypeConfig

Configuration of a microgrid component type.

Source code in src/frequenz/gridpool/config/microgrid.py

@dataclass
class ComponentTypeConfig:
    """Configuration of a microgrid component type."""

    meter: list[int] | None = None
    """List of meter IDs for this component."""

    inverter: list[int] | None = None
    """List of inverter IDs for this component."""

    component: list[int] | None = None
    """List of component IDs for this component."""

    formula: dict[str, str] | None = None
    """Formula to calculate the power of this component."""

    def __post_init__(self) -> None:
        """Set the default formula if none is provided."""
        self.formula = self.formula or {}
        if "AC_ACTIVE_POWER" in self.formula:
            _logger.warning(
                "ComponentTypeConfig: 'AC_ACTIVE_POWER' formula is deprecated, "
                "please use 'AC_POWER_ACTIVE' instead."
            )

    def cids(self, metric: str = "") -> list[int]:
        """Get component IDs for this component.

        By default, the meter IDs are returned if available, otherwise the inverter IDs.
        For components without meters or inverters, the component IDs are returned.

        If a metric is provided, the component IDs are extracted from the formula.

        Args:
            metric: Metric name of the formula.

        Returns:
            List of component IDs for this component.

        Raises:
            ValueError: If the metric is not supported or improperly set.
        """
        if metric:
            if not isinstance(self.formula, dict):
                raise ValueError("Formula must be a dictionary.")
            formula = self.formula.get(metric)
            if not formula:
                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])]

        return self._default_cids()

    def _default_cids(self) -> list[int]:
        """Get the default component IDs for this component.

        If available, the meter IDs are returned, otherwise the inverter IDs.
        For components without meters or inverters, the component IDs are returned.

        Returns:
            List of component IDs for this component.

        Raises:
            ValueError: If no IDs are available.
        """
        if self.meter:
            return self.meter
        if self.inverter:
            return self.inverter
        if self.component:
            return self.component

        raise ValueError("No IDs available")

    @classmethod
    def is_valid_type(cls, ctype: str) -> bool:
        """Check if `ctype` is a valid enum value."""
        return ctype in get_args(ComponentType)

Attributes

component class-attribute instance-attribute

component: list[int] | None = None

List of component IDs for this component.

formula class-attribute instance-attribute

formula: dict[str, str] | None = None

Formula to calculate the power of this component.

inverter class-attribute instance-attribute

inverter: list[int] | None = None

List of inverter IDs for this component.

meter class-attribute instance-attribute

meter: list[int] | None = None

List of meter IDs for this component.

Methods:

__post_init__

__post_init__() -> None

Set the default formula if none is provided.

Source code in src/frequenz/gridpool/config/microgrid.py

def __post_init__(self) -> None:
    """Set the default formula if none is provided."""
    self.formula = self.formula or {}
    if "AC_ACTIVE_POWER" in self.formula:
        _logger.warning(
            "ComponentTypeConfig: 'AC_ACTIVE_POWER' formula is deprecated, "
            "please use 'AC_POWER_ACTIVE' instead."
        )

cids

cids(metric: str = '') -> list[int]

Get component IDs for this component.

By default, the meter IDs are returned if available, otherwise the inverter IDs. For components without meters or inverters, the component IDs are returned.

If a metric is provided, the component IDs are extracted from the formula.

PARAMETER	DESCRIPTION
metric	Metric name of the formula. TYPE: str DEFAULT: ''

RETURNS	DESCRIPTION
list[int]	List of component IDs for this component.

RAISES	DESCRIPTION
ValueError	If the metric is not supported or improperly set.

Source code in src/frequenz/gridpool/config/microgrid.py

def cids(self, metric: str = "") -> list[int]:
    """Get component IDs for this component.

    By default, the meter IDs are returned if available, otherwise the inverter IDs.
    For components without meters or inverters, the component IDs are returned.

    If a metric is provided, the component IDs are extracted from the formula.

    Args:
        metric: Metric name of the formula.

    Returns:
        List of component IDs for this component.

    Raises:
        ValueError: If the metric is not supported or improperly set.
    """
    if metric:
        if not isinstance(self.formula, dict):
            raise ValueError("Formula must be a dictionary.")
        formula = self.formula.get(metric)
        if not formula:
            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])]

    return self._default_cids()

is_valid_type classmethod

is_valid_type(ctype: str) -> bool

Check if ctype is a valid enum value.

Source code in src/frequenz/gridpool/config/microgrid.py

@classmethod
def is_valid_type(cls, ctype: str) -> bool:
    """Check if `ctype` is a valid enum value."""
    return ctype in get_args(ComponentType)

frequenz.gridpool.config.Metadata

Metadata for a microgrid.

Source code in src/frequenz/gridpool/config/microgrid.py

@dataclass(frozen=True)
class Metadata:
    """Metadata for a microgrid."""

    microgrid_id: int
    """ID of the microgrid."""

    name: str | None = None
    """Name of the microgrid."""

    enterprise_id: int | None = None
    """Enterprise ID of the microgrid."""

    gid: int | None = None
    """Gridpool ID of the microgrid."""

    delivery_area: str | None = None
    """Delivery area of the microgrid."""

    latitude: float | None = None
    """Geographic latitude of the microgrid."""

    longitude: float | None = None
    """Geographic longitude of the microgrid."""

    altitude: float | None = None
    """Geographic altitude of the microgrid."""

    start_time: datetime | None = None
    """Start time of the microgrid operation."""

    end_time: datetime | None = None
    """End time of the microgrid operation."""

Attributes

altitude class-attribute instance-attribute

altitude: float | None = None

Geographic altitude of the microgrid.

delivery_area class-attribute instance-attribute

delivery_area: str | None = None

Delivery area of the microgrid.

end_time class-attribute instance-attribute

end_time: datetime | None = None

End time of the microgrid operation.

enterprise_id class-attribute instance-attribute

enterprise_id: int | None = None

Enterprise ID of the microgrid.

gid class-attribute instance-attribute

gid: int | None = None

Gridpool ID of the microgrid.

latitude class-attribute instance-attribute

latitude: float | None = None

Geographic latitude of the microgrid.

longitude class-attribute instance-attribute

longitude: float | None = None

Geographic longitude of the microgrid.

microgrid_id instance-attribute

microgrid_id: int

ID of the microgrid.

name class-attribute instance-attribute

name: str | None = None

Name of the microgrid.

start_time class-attribute instance-attribute

start_time: datetime | None = None

Start time of the microgrid operation.

frequenz.gridpool.config.MicrogridConfig

Configuration of a microgrid.

Source code in src/frequenz/gridpool/config/microgrid.py

@dataclass
class MicrogridConfig:
    """Configuration of a microgrid."""

    meta: Metadata
    """Metadata of the 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."""

    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.ctype.keys())

    def component_type_ids(
        self,
        component_type: str,
        component_category: str | None = None,
        metric: str = "",
    ) -> list[int]:
        """Get a list of all component IDs for a component type.

        Args:
            component_type: Component type to be aggregated.
            component_category: Specific category of component IDs to retrieve
                (e.g., "meter", "inverter", or "component"). If not provided,
                the default logic is used.
            metric: Metric name of the formula if CIDs should be extracted from the formula.

        Returns:
            List of component IDs for this component type.

        Raises:
            ValueError: If the component type is unknown.
            KeyError: If `component_category` is invalid.
        """
        cfg = self.ctype.get(component_type)
        if not cfg:
            raise ValueError(f"{component_type} not found in config.")

        if component_category:
            valid_categories = get_args(ComponentCategory)
            if component_category not in valid_categories:
                raise KeyError(
                    f"Invalid component category: {component_category}. "
                    f"Valid categories are {valid_categories}"
                )
            category_ids = cast(list[int], getattr(cfg, component_category, []))
            return category_ids

        return cfg.cids(metric)

    def formula(self, component_type: str, metric: str) -> str:
        """Get the formula for a component type.

        Args:
            component_type: Component type to be aggregated.
            metric: Metric to be aggregated.

        Returns:
            Formula to be used for this aggregated component as string.

        Raises:
            ValueError: If the component type is unknown or formula is missing.
        """
        cfg = self.ctype.get(component_type)
        if not cfg:
            raise ValueError(f"{component_type} not found in config.")
        if cfg.formula is None:
            raise ValueError(f"No formula set for {component_type}")
        formula = cfg.formula.get(metric)
        if not formula:
            raise ValueError(f"{component_type} is missing formula for {metric}")

        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[str, 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)

Attributes

battery class-attribute instance-attribute

battery: dict[str, BatteryConfig] | None = None

Configuration of the batteries.

ctype class-attribute instance-attribute

ctype: dict[str, ComponentTypeConfig] = field(
    default_factory=dict
)

Mapping of component category types to ac power component config.

meta instance-attribute

meta: Metadata

Metadata of the microgrid.

pv class-attribute instance-attribute

pv: dict[str, PVConfig] | None = None

Configuration of the PV system.

wind class-attribute instance-attribute

wind: dict[str, WindConfig] | None = None

Configuration of the wind turbines.

Methods:

component_type_ids

component_type_ids(
    component_type: str,
    component_category: str | None = None,
    metric: str = "",
) -> list[int]

Get a list of all component IDs for a component type.

PARAMETER	DESCRIPTION
component_type	Component type to be aggregated. TYPE: str
component_category	Specific category of component IDs to retrieve (e.g., "meter", "inverter", or "component"). If not provided, the default logic is used. TYPE: str | None DEFAULT: None
metric	Metric name of the formula if CIDs should be extracted from the formula. TYPE: str DEFAULT: ''

RETURNS	DESCRIPTION
list[int]	List of component IDs for this component type.

RAISES	DESCRIPTION
ValueError	If the component type is unknown.
KeyError	If component_category is invalid.

Source code in src/frequenz/gridpool/config/microgrid.py

def component_type_ids(
    self,
    component_type: str,
    component_category: str | None = None,
    metric: str = "",
) -> list[int]:
    """Get a list of all component IDs for a component type.

    Args:
        component_type: Component type to be aggregated.
        component_category: Specific category of component IDs to retrieve
            (e.g., "meter", "inverter", or "component"). If not provided,
            the default logic is used.
        metric: Metric name of the formula if CIDs should be extracted from the formula.

    Returns:
        List of component IDs for this component type.

    Raises:
        ValueError: If the component type is unknown.
        KeyError: If `component_category` is invalid.
    """
    cfg = self.ctype.get(component_type)
    if not cfg:
        raise ValueError(f"{component_type} not found in config.")

    if component_category:
        valid_categories = get_args(ComponentCategory)
        if component_category not in valid_categories:
            raise KeyError(
                f"Invalid component category: {component_category}. "
                f"Valid categories are {valid_categories}"
            )
        category_ids = cast(list[int], getattr(cfg, component_category, []))
        return category_ids

    return cfg.cids(metric)

component_types

component_types() -> list[str]

Get a list of all component types in the configuration.

Source code in src/frequenz/gridpool/config/microgrid.py

def component_types(self) -> list[str]:
    """Get a list of all component types in the configuration."""
    return list(self.ctype.keys())

formula

formula(component_type: str, metric: str) -> str

Get the formula for a component type.

PARAMETER	DESCRIPTION
component_type	Component type to be aggregated. TYPE: str
metric	Metric to be aggregated. TYPE: str

RETURNS	DESCRIPTION
str	Formula to be used for this aggregated component as string.

RAISES	DESCRIPTION
ValueError	If the component type is unknown or formula is missing.

Source code in src/frequenz/gridpool/config/microgrid.py

def formula(self, component_type: str, metric: str) -> str:
    """Get the formula for a component type.

    Args:
        component_type: Component type to be aggregated.
        metric: Metric to be aggregated.

    Returns:
        Formula to be used for this aggregated component as string.

    Raises:
        ValueError: If the component type is unknown or formula is missing.
    """
    cfg = self.ctype.get(component_type)
    if not cfg:
        raise ValueError(f"{component_type} not found in config.")
    if cfg.formula is None:
        raise ValueError(f"No formula set for {component_type}")
    formula = cfg.formula.get(metric)
    if not formula:
        raise ValueError(f"{component_type} is missing formula for {metric}")

    return formula

load_from_file classmethod

load_from_file(config_path: Path) -> dict[str, Self]

Load and validate configuration settings from a TOML file.

PARAMETER	DESCRIPTION
config_path	the path to the TOML configuration file. TYPE: Path

RETURNS	DESCRIPTION
dict[str, Self]	A dict mapping microgrid IDs to MicrogridConfig instances.

Source code in src/frequenz/gridpool/config/microgrid.py

@classmethod
def load_from_file(cls, config_path: Path) -> dict[str, 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)

frequenz.gridpool.config.PVConfig

Configuration of a PV system in a microgrid.

Source code in src/frequenz/gridpool/config/microgrid.py

@dataclass(frozen=True)
class PVConfig:
    """Configuration of a PV system in a microgrid."""

    start_time: datetime | None = None
    """Start time of the PV system installation."""

    end_time: datetime | None = None
    """End time of the PV system installation."""

    peak_power: float | None = None
    """Peak power of the PV system in Watt."""

    rated_power: float | None = None
    """Rated power of the inverters in Watt."""

    curtailable: bool | None = None
    """Flag to indicate if PV system can be curtailed."""

Attributes

curtailable class-attribute instance-attribute

curtailable: bool | None = None

Flag to indicate if PV system can be curtailed.

end_time class-attribute instance-attribute

end_time: datetime | None = None

End time of the PV system installation.

peak_power class-attribute instance-attribute

peak_power: float | None = None

Peak power of the PV system in Watt.

rated_power class-attribute instance-attribute

rated_power: float | None = None

Rated power of the inverters in Watt.

start_time class-attribute instance-attribute

start_time: datetime | None = None

Start time of the PV system installation.

frequenz.gridpool.config.WindConfig

Configuration of a wind turbine in a microgrid.

Source code in src/frequenz/gridpool/config/microgrid.py

@dataclass(frozen=True)
class WindConfig:
    # pylint: disable=too-many-instance-attributes
    """Configuration of a wind turbine in a microgrid."""

    start_time: datetime | None = None
    """Start time of the wind turbine installation."""

    end_time: datetime | None = None
    """End time of the wind turbine installation."""

    turbine_model: str | None = None
    """Model name of the wind turbine."""

    rated_power: float | None = None
    """Rated power of the wind turbine in Watt."""

    turbine_height: float | None = None
    """Height of the wind turbine in meters."""

    number_of_turbines: int = 1
    """Number of wind turbines."""

    hellmann_exponent: float | None = None
    """Hellmann exponent for wind speed extrapolation. See: https://w.wiki/FMw9"""

    longitude: float | None = None
    """Geographic longitude of the wind turbine."""

    latitude: float | None = None
    """Geographic latitude of the wind turbine."""

Attributes

end_time class-attribute instance-attribute

end_time: datetime | None = None

End time of the wind turbine installation.

hellmann_exponent class-attribute instance-attribute

hellmann_exponent: float | None = None

Hellmann exponent for wind speed extrapolation. See: https://w.wiki/FMw9

latitude class-attribute instance-attribute

latitude: float | None = None

Geographic latitude of the wind turbine.

longitude class-attribute instance-attribute

longitude: float | None = None

Geographic longitude of the wind turbine.

number_of_turbines class-attribute instance-attribute

number_of_turbines: int = 1

Number of wind turbines.

rated_power class-attribute instance-attribute

rated_power: float | None = None

Rated power of the wind turbine in Watt.

start_time class-attribute instance-attribute

start_time: datetime | None = None

Start time of the wind turbine installation.

turbine_height class-attribute instance-attribute

turbine_height: float | None = None

Height of the wind turbine in meters.

turbine_model class-attribute instance-attribute

turbine_model: str | None = None

Model name of the wind turbine.

Functions:

frequenz.gridpool.config.load_configs async

load_configs(
    default_files: (
        str | Path | list[str | Path] | None
    ) = None,
    assets_client: AssetsApiClient | None = None,
    override_files: (
        str | Path | list[str | Path] | None
    ) = None,
    microgrid_ids: list[int] | None = None,
) -> dict[str, MicrogridConfig]

Load configs from up to three sources and merge them in layers.

Combines up to three sources, listed here from lowest to highest precedence: a default config file layer, the Assets API, and an override config file layer. Higher layers win on conflicts, while lower layers fill in anything the higher ones leave unset. This lets callers pick a strategy by choosing which sources to pass, for example:

• default_files + assets_client: files provide defaults that the Assets API overrides.
• assets_client + override_files: the Assets API provides the base that files override.
• all three: the Assets API sits between a default and an override file layer.

The microgrid IDs fetched from the Assets API are microgrid_ids when given, otherwise the IDs found in the default and override files. This lets the Assets API layer be used even when no files are given.

PARAMETER	DESCRIPTION
default_files	Optional path or list of paths to config files forming the lowest-precedence layer. TYPE: str | Path | list[str | Path] | None DEFAULT: None
assets_client	Optional Assets API client. When given, microgrid metadata and formulas are fetched and layered above the default files. TYPE: AssetsApiClient | None DEFAULT: None
override_files	Optional path or list of paths to config files forming the highest-precedence layer. TYPE: str | Path | list[str | Path] | None DEFAULT: None
microgrid_ids	Optional explicit microgrid IDs to fetch from the Assets API. When given, these replace the IDs derived from the files, so the Assets API layer can be used without any files. TYPE: list[int] | None DEFAULT: None

RETURNS	DESCRIPTION
dict[str, MicrogridConfig]	dict[str, MicrogridConfig]: Mapping from microgrid ID (as string) to the merged MicrogridConfig instance.

RAISES	DESCRIPTION
ValueError	If none of the three sources is provided, or if microgrid_ids is given without an assets_client.

Source code in src/frequenz/gridpool/config/load.py

async def load_configs(
    default_files: str | Path | list[str | Path] | None = None,
    assets_client: AssetsApiClient | None = None,
    override_files: str | Path | list[str | Path] | None = None,
    microgrid_ids: list[int] | None = None,
) -> dict[str, "MicrogridConfig"]:
    """Load configs from up to three sources and merge them in layers.

    Combines up to three sources, listed here from lowest to highest
    precedence: a *default* config file layer, the Assets API, and an
    *override* config file layer.  Higher layers win on conflicts, while
    lower layers fill in anything the higher ones leave unset.  This lets
    callers pick a strategy by choosing which sources to pass, for example:

    - `default_files` + `assets_client`: files provide defaults that the
      Assets API overrides.
    - `assets_client` + `override_files`: the Assets API provides the base
      that files override.
    - all three: the Assets API sits between a default and an override file
      layer.

    The microgrid IDs fetched from the Assets API are `microgrid_ids` when
    given, otherwise the IDs found in the default and override files.  This
    lets the Assets API layer be used even when no files are given.

    Args:
        default_files:
            Optional path or list of paths to config files forming the
            lowest-precedence layer.
        assets_client:
            Optional Assets API client.  When given, microgrid metadata and
            formulas are fetched and layered above the default files.
        override_files:
            Optional path or list of paths to config files forming the
            highest-precedence layer.
        microgrid_ids:
            Optional explicit microgrid IDs to fetch from the Assets API.
            When given, these replace the IDs derived from the files, so the
            Assets API layer can be used without any files.

    Returns:
        dict[str, MicrogridConfig]:
            Mapping from microgrid ID (as string) to the merged
            `MicrogridConfig` instance.

    Raises:
        ValueError: If none of the three sources is provided, or if
            `microgrid_ids` is given without an `assets_client`.
    """
    if default_files is None and assets_client is None and override_files is None:
        raise ValueError("At least one config source must be provided.")

    if microgrid_ids is not None and assets_client is None:
        raise ValueError("microgrid_ids requires an assets_client.")

    configs: dict[str, MicrogridConfig] = {}
    if default_files is not None:
        configs = load_configs_from_files(
            microgrid_config_files=default_files,
        )

    override_configs: dict[str, MicrogridConfig] = {}
    if override_files is not None:
        override_configs = load_configs_from_files(
            microgrid_config_files=override_files,
        )

    if assets_client is not None:
        if microgrid_ids is None:
            microgrid_ids = sorted({int(mid) for mid in (*configs, *override_configs)})
        assets_configs = await load_configs_from_api(
            assets_client=assets_client,
            microgrid_ids=microgrid_ids,
        )
        configs = merge_config_maps(base=configs, override=assets_configs)

    return merge_config_maps(base=configs, override=override_configs)

frequenz.gridpool.config.load_configs_from_api async

load_configs_from_api(
    assets_client: AssetsApiClient, microgrid_ids: list[int]
) -> dict[str, MicrogridConfig]

Load microgrid configs from the Assets API.

For each microgrid, fetches its location metadata (latitude, longitude) and then derives the per-type formulas and meter/inverter/component IDs from its component graph. This is the canonical single-source loader so that callers (e.g. the forecast pipeline) do not have to re-implement this logic.

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 failures are logged as warnings.

PARAMETER	DESCRIPTION
assets_client	Assets API client used to fetch microgrid metadata and the component graph. TYPE: AssetsApiClient
microgrid_ids	List of microgrid IDs to load configurations for. TYPE: list[int]

RETURNS	DESCRIPTION
dict[str, MicrogridConfig]	dict[str, MicrogridConfig]: Mapping from microgrid ID (as string) to the loaded MicrogridConfig instance. Microgrids whose metadata could not be loaded are omitted, so the returned mapping may cover fewer microgrids than were requested.

Source code in src/frequenz/gridpool/config/load.py

async def load_configs_from_api(
    assets_client: AssetsApiClient,
    microgrid_ids: list[int],
) -> dict[str, "MicrogridConfig"]:
    """Load microgrid configs from the Assets API.

    For each microgrid, fetches its location metadata (latitude, longitude) and
    then derives the per-type formulas and meter/inverter/component IDs from its
    component graph. This is the canonical single-source loader so that callers
    (e.g. the forecast pipeline) do not have to re-implement this logic.

    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 failures are logged as warnings.

    Args:
        assets_client:
            Assets API client used to fetch microgrid metadata and the
            component graph.
        microgrid_ids:
            List of microgrid IDs to load configurations for.

    Returns:
        dict[str, MicrogridConfig]:
            Mapping from microgrid ID (as string) to the loaded
            `MicrogridConfig` instance. Microgrids whose metadata could not be
            loaded are omitted, so the returned mapping may cover fewer
            microgrids than were requested.
    """
    configs: dict[str, MicrogridConfig] = {}
    for microgrid_id in microgrid_ids:
        try:
            cfg = await _build_config_from_metadata(assets_client, microgrid_id)
        except Exception as exc:  # pylint: disable=broad-except
            _logger.warning(
                "Failed to load microgrid %s metadata from the Assets API: %s",
                microgrid_id,
                exc,
            )
            continue

        try:
            graph = await ComponentGraphGenerator(assets_client).get_component_graph(
                MicrogridId(microgrid_id)
            )
            cfg.ctype = _derive_component_configs(graph)
        except Exception as exc:  # pylint: disable=broad-except
            _logger.warning(
                "Failed to derive component config for microgrid %s from the "
                "graph: %s",
                microgrid_id,
                exc,
            )

        configs[str(microgrid_id)] = cfg

    return configs

frequenz.gridpool.config.load_configs_from_files

load_configs_from_files(
    microgrid_config_files: (
        str | Path | list[str | Path] | None
    ) = None,
) -> dict[str, MicrogridConfig]

Load multiple microgrid configurations from one or more files.

Configs for a single microgrid are expected to be in a single file. Later files with the same microgrid ID will overwrite the previous configs.

PARAMETER	DESCRIPTION
microgrid_config_files	Path to a single microgrid config file or list of paths. TYPE: str | Path | list[str | Path] | None DEFAULT: None

RETURNS	DESCRIPTION
dict[str, MicrogridConfig]	Dictionary of single microgrid formula configs with microgrid IDs as keys.

RAISES	DESCRIPTION
ValueError	If no config files are provided, or if no config files are found.

Source code in src/frequenz/gridpool/config/load.py

def load_configs_from_files(
    microgrid_config_files: str | Path | list[str | Path] | None = None,
) -> dict[str, "MicrogridConfig"]:
    """Load multiple microgrid configurations from one or more files.

    Configs for a single microgrid are expected to be in a single file.
    Later files with the same microgrid ID will overwrite the previous configs.

    Args:
        microgrid_config_files: Path to a single microgrid config file or list of paths.

    Returns:
        Dictionary of single microgrid formula configs with microgrid IDs as keys.

    Raises:
        ValueError: If no config files are provided, or if no config files are found.
    """
    if microgrid_config_files is None:
        raise ValueError(
            "No microgrid config files provided. Please provide at least one."
        )

    config_files: list[Path] = []

    if microgrid_config_files:
        if isinstance(microgrid_config_files, str):
            config_files = [Path(microgrid_config_files)]
        elif isinstance(microgrid_config_files, Path):
            config_files = [microgrid_config_files]
        elif isinstance(microgrid_config_files, list):
            config_files = [Path(f) for f in microgrid_config_files]

    if len(config_files) == 0:
        raise ValueError(
            "No microgrid config files found. "
            "Please provide at least one valid config file."
        )

    microgrid_configs: dict[str, "MicrogridConfig"] = {}

    for config_path in config_files:
        if not config_path.is_file():
            _logger.warning("Config path %s is not a file, skipping.", config_path)
            continue

        mcfgs = MicrogridConfig.load_from_file(config_path)
        microgrid_configs.update({str(key): value for key, value in mcfgs.items()})

    return microgrid_configs

frequenz.gridpool.config.merge_config_maps

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.

PARAMETER	DESCRIPTION
base	The base dictionary of MicrogridConfig objects. TYPE: dict[str, MicrogridConfig]
override	The overriding dictionary of MicrogridConfig objects. TYPE: dict[str, MicrogridConfig]

RETURNS	DESCRIPTION
dict[str, MicrogridConfig]	A new dictionary representing the merged result.

Source code in src/frequenz/gridpool/config/microgrid.py

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

frequenz.gridpool.config.merge_microgrid_configs

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.

PARAMETER	DESCRIPTION
base	The base MicrogridConfig. TYPE: MicrogridConfig
override	The overriding MicrogridConfig. TYPE: MicrogridConfig

RETURNS	DESCRIPTION
MicrogridConfig	A new MicrogridConfig representing the merged result.

Source code in src/frequenz/gridpool/config/microgrid.py

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