load

frequenz.gridpool.config.load

Loading and merging of microgrid configurations.

Attributes

Classes

Functions:

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