Index

frequenz.gridpool

High-level interface to grid pools for the Frequenz platform.

Classes

frequenz.gridpool.ComponentGraphGenerator

Generates component graphs for microgrids using the Assets API.

Source code in src/frequenz/gridpool/_graph_generator.py

class ComponentGraphGenerator:
    """Generates component graphs for microgrids using the Assets API."""

    def __init__(
        self,
        client: AssetsApiClient,
    ) -> None:
        """Initialize this instance.

        Args:
            client: The Assets API client to use for fetching components and
                connections.
        """
        self._client: AssetsApiClient = client

    async def get_component_graph(
        self, microgrid_id: MicrogridId
    ) -> MicrogridComponentGraph:
        """Generate a component graph for the given microgrid ID.

        Args:
            microgrid_id: The ID of the microgrid to generate the graph for.

        Returns:
            The component graph representing the microgrid's electrical
                components and their connections.

        Raises:
            ValueError: If any component connections could not be loaded.
        """
        components = await self._client.list_microgrid_electrical_components(
            microgrid_id
        )
        connections = (
            await self._client.list_microgrid_electrical_component_connections(
                microgrid_id
            )
        )

        if any(c is None for c in connections):
            raise ValueError("Failed to load all electrical component connections.")

        breakers = [c for c in components if isinstance(c, Breaker)]
        connected_breakers = [
            b
            for b in breakers
            if any(
                b.id in (c.source, c.destination) for c in connections if c is not None
            )
        ]

        if connected_breakers:
            _logger.warning(
                "The following breakers are connected to other components, "
                + "which is not supported by the component graph generator and may "
                + "lead to graph traversal issues: %s",
                [b.id for b in connected_breakers],
            )
        elif breakers:
            _logger.debug("Dropping unconnected breakers: %s", [b.id for b in breakers])
            components = [c for c in components if not isinstance(c, Breaker)]

        graph = ComponentGraph[
            ElectricalComponent, ComponentConnection, ElectricalComponentId
        ](components, connections)

        return graph

Methods:

__init__

__init__(client: AssetsApiClient) -> None

Initialize this instance.

PARAMETER	DESCRIPTION
client	The Assets API client to use for fetching components and connections. TYPE: AssetsApiClient

Source code in src/frequenz/gridpool/_graph_generator.py

def __init__(
    self,
    client: AssetsApiClient,
) -> None:
    """Initialize this instance.

    Args:
        client: The Assets API client to use for fetching components and
            connections.
    """
    self._client: AssetsApiClient = client

get_component_graph async

get_component_graph(
    microgrid_id: MicrogridId,
) -> MicrogridComponentGraph

Generate a component graph for the given microgrid ID.

PARAMETER	DESCRIPTION
microgrid_id	The ID of the microgrid to generate the graph for. TYPE: MicrogridId

RETURNS	DESCRIPTION
MicrogridComponentGraph	The component graph representing the microgrid's electrical components and their connections.

RAISES	DESCRIPTION
ValueError	If any component connections could not be loaded.

Source code in src/frequenz/gridpool/_graph_generator.py

async def get_component_graph(
    self, microgrid_id: MicrogridId
) -> MicrogridComponentGraph:
    """Generate a component graph for the given microgrid ID.

    Args:
        microgrid_id: The ID of the microgrid to generate the graph for.

    Returns:
        The component graph representing the microgrid's electrical
            components and their connections.

    Raises:
        ValueError: If any component connections could not be loaded.
    """
    components = await self._client.list_microgrid_electrical_components(
        microgrid_id
    )
    connections = (
        await self._client.list_microgrid_electrical_component_connections(
            microgrid_id
        )
    )

    if any(c is None for c in connections):
        raise ValueError("Failed to load all electrical component connections.")

    breakers = [c for c in components if isinstance(c, Breaker)]
    connected_breakers = [
        b
        for b in breakers
        if any(
            b.id in (c.source, c.destination) for c in connections if c is not None
        )
    ]

    if connected_breakers:
        _logger.warning(
            "The following breakers are connected to other components, "
            + "which is not supported by the component graph generator and may "
            + "lead to graph traversal issues: %s",
            [b.id for b in connected_breakers],
        )
    elif breakers:
        _logger.debug("Dropping unconnected breakers: %s", [b.id for b in breakers])
        components = [c for c in components if not isinstance(c, Breaker)]

    graph = ComponentGraph[
        ElectricalComponent, ComponentConnection, ElectricalComponentId
    ](components, connections)

    return graph

frequenz.gridpool.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.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)

Functions:

frequenz.gridpool.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.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.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.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.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