config

frequenz.gridpool.config

Asset configuration data models 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.AssetsConfig

Entities described by a config document, keyed by their ID.

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

@dataclass
class AssetsConfig:
    """Entities described by a config document, keyed by their ID."""

    version: int = _CURRENT_VERSION
    """Format version of the `assets` namespace, stamped by the migration."""

    microgrids: dict[int, MicrogridConfig] = field(default_factory=dict)
    """Microgrids, keyed by microgrid ID."""

    gridpools: dict[int, GridpoolConfig] = field(default_factory=dict)
    """Gridpools, keyed by gridpool ID."""

    market_locations: dict[str, MarketLocationConfig] = field(default_factory=dict)
    """Market locations, keyed by their identifier."""

    relations: dict[str, RelationConfig] = field(default_factory=dict)
    """Market topology relations, keyed by the composite of the sides they connect."""

    class Meta:
        """Ignore entity tables this version does not know about.

        A reader must keep working against files that already carry entities
        added after it, so unknown tables are skipped rather than rejected.
        `_warn_unknown_entities` reports them, so a mistyped table is still
        visible instead of silently loading as empty.
        """

        unknown = marshmallow.EXCLUDE

    Schema: ClassVar[Type[Schema]] = Schema

    def __post_init__(self) -> None:
        """Check that every microgrid is filed under its own ID.

        Relations are not checked here: an override names only the fields it
        changes, so a single document may hold an incomplete record. `check`
        looks at the merged result.

        Raises:
            ValueError: If a key is not the ID of the entry it holds.
        """
        for mid, cfg in self.microgrids.items():
            if int(cfg.microgrid_id) != mid:
                raise ValueError(
                    f"Microgrid ID mismatch: key {mid} != {cfg.microgrid_id}"
                )
        for gpid, gridpool in self.gridpools.items():
            if int(gridpool.gridpool_id) != gpid:
                raise ValueError(
                    f"Gridpool ID mismatch: key {gpid} != {gridpool.gridpool_id}"
                )

    def check(self) -> None:
        """Check the document as a whole, once every layer has been merged.

        Raises:
            ValueError: If an entry's identifier disagrees with the key it is
                filed under, a relation names fewer than two sides, its key
                disagrees with its fields, a gridpool relation names no delivery
                area, one market location is placed in two of them, a legacy
                microgrid gridpool ID disagrees with its relations, or a
                gridpool's declared and inferred enterprise disagree.
        """
        for key, location in self.market_locations.items():
            if location.id is None:
                raise ValueError(f"Market location {key}: must name its id")
            if location.id != key:
                raise ValueError(
                    f"Market location key mismatch: key {key} != {location.id}"
                )
        self._check_relations()
        self._check_legacy_gridpool_ids()
        self._check_gridpool_enterprises()

    def _check_relations(self) -> None:
        """Check the relations are complete, well-keyed and area-consistent.

        Raises:
            ValueError: If a relation names fewer than two sides, its key
                disagrees with its fields, a gridpool relation names no delivery
                area, one market location is placed in two of them, or one
                delivery-area code appears with two code types.
        """
        for key, relation in self.relations.items():
            if not relation.is_complete:
                raise ValueError(
                    f"Relation {key}: must name at least two of gridpool, microgrid "
                    "and market location"
                )
            if key != relation.key:
                raise ValueError(
                    f"Relation key mismatch: key {key} != {relation.key}, derived "
                    "from the sides the record names"
                )
            if relation.gridpool_id is not None and relation.delivery_area is None:
                raise ValueError(
                    f"Relation {key}: a gridpool relation must name a delivery area"
                )

        zones: dict[str, DeliveryAreaConfig] = {}
        for relation in self.relations.values():
            mlid, zone = relation.market_location_id, relation.delivery_area
            if mlid is None or zone is None:
                continue
            if zones.setdefault(mlid, zone) != zone:
                raise ValueError(
                    f"Market location {mlid} is placed in two delivery areas: "
                    f"{zones[mlid].code} and {zone.code}"
                )

        seen: dict[str, DeliveryAreaConfig] = {}
        for relation in self.relations.values():
            area = relation.delivery_area
            if area is None or area.code is None:
                continue
            if seen.setdefault(area.code, area).code_type != area.code_type:
                raise ValueError(
                    f"Delivery area code {area.code} appears with two code types: "
                    f"{seen[area.code].code_type.name} and {area.code_type.name}"
                )

    def _check_legacy_gridpool_ids(self) -> None:
        """Check legacy microgrid gridpool IDs against the relations."""
        gridpools_by_microgrid: dict[int, set[int]] = {}
        for relation in self.relations.values():
            if relation.microgrid_id is None or relation.gridpool_id is None:
                continue
            gridpools_by_microgrid.setdefault(relation.microgrid_id, set()).add(
                relation.gridpool_id
            )

        for microgrid in self.microgrids.values():
            legacy_gid = microgrid.gid
            if legacy_gid is None:
                continue
            relation_gids = gridpools_by_microgrid.get(microgrid.microgrid_id)
            if relation_gids and relation_gids != {legacy_gid}:
                raise ValueError(
                    f"Microgrid {microgrid.microgrid_id}: legacy gid "
                    f"{legacy_gid} disagrees with relation gridpools "
                    f"{sorted(relation_gids)}; remove gid when several apply"
                )

    def _derive_enterprise(self, gridpool_id: int) -> int | None:
        """Infer a gridpool's enterprise from the microgrids its relations name.

        Args:
            gridpool_id: The gridpool whose enterprise to infer.

        Returns:
            The inferred enterprise ID, or `None` if none can be inferred.

        Raises:
            ValueError: If the related microgrids disagree on the enterprise.
        """
        enterprises: set[int] = set()
        for mid in self.find_microgrids(gridpool_id=gridpool_id):
            microgrid = self.microgrids.get(mid)
            if microgrid is not None and microgrid.enterprise_id is not None:
                enterprises.add(microgrid.enterprise_id)
        if not enterprises:
            return None
        if len(enterprises) > 1:
            raise ValueError(
                f"Gridpool {gridpool_id}: its microgrids disagree on the owning "
                f"enterprise: {sorted(enterprises)}"
            )
        return enterprises.pop()

    def _check_gridpool_enterprises(self) -> None:
        """Check declared and inferable gridpool enterprises agree.

        A gridpool owns one enterprise, so its microgrids must not disagree on
        it, and a declared `gridpools` entry must match what they imply.

        Raises:
            ValueError: If a gridpool's microgrids disagree on the enterprise,
                or a declared enterprise differs from the inferred one.
        """
        gridpool_ids = set(self.gridpools) | {
            relation.gridpool_id
            for relation in self.relations.values()
            if relation.gridpool_id is not None
        }
        for gpid in gridpool_ids:
            inferred = self._derive_enterprise(gpid)
            declared = self.gridpools.get(gpid)
            if (
                declared is not None
                and inferred is not None
                and declared.enterprise_id != inferred
            ):
                raise ValueError(
                    f"Gridpool {gpid}: declared enterprise "
                    f"{declared.enterprise_id} disagrees with its microgrids' "
                    f"enterprise {inferred}"
                )

    def find_relations(
        self,
        *,
        gridpool_id: int | None = None,
        microgrid_id: int | None = None,
        market_location_id: str | None = None,
        delivery_area: str | DeliveryAreaConfig | None = None,
        participation: MarketParticipationType | None = None,
        at: datetime | None = None,
    ) -> list[RelationConfig]:
        """Find the relations naming all of the given sides.

        Args:
            gridpool_id: Gridpool to match, or `None` to ignore.
            microgrid_id: Microgrid to match, or `None` to ignore.
            market_location_id: Market location to match, or `None` to ignore.
            delivery_area: Delivery area to match, a `DeliveryAreaConfig` matched
                on code and code type or a bare code string; `None` to ignore.
            participation: Use case the relation must serve, or `None` to ignore.
            at: Instant the relations, or the given participation, must apply at,
                or `None` to ignore.

        Returns:
            The matching relations, in document order.
        """
        return [
            relation
            for relation in self.relations.values()
            if relation.matches(
                gridpool_id=gridpool_id,
                microgrid_id=microgrid_id,
                market_location_id=market_location_id,
                delivery_area=delivery_area,
                participation=participation,
                at=at,
            )
        ]

    def find_delivery_areas(
        self,
        *,
        gridpool_id: int | None = None,
        microgrid_id: int | None = None,
        market_location_id: str | None = None,
        at: datetime | None = None,
    ) -> list[DeliveryAreaConfig]:
        """List the delivery areas of the matching relations.

        Args:
            gridpool_id: Gridpool to match, or `None` to ignore.
            microgrid_id: Microgrid to match, or `None` to ignore.
            market_location_id: Market location to match, or `None` to ignore.
            at: Instant the relations must apply at, or `None` to ignore.

        Returns:
            The delivery areas, each with its code and code type, deduplicated,
            in document order.
        """
        return list(
            dict.fromkeys(
                relation.delivery_area
                for relation in self.find_relations(
                    gridpool_id=gridpool_id,
                    microgrid_id=microgrid_id,
                    market_location_id=market_location_id,
                    at=at,
                )
                if relation.delivery_area is not None
            )
        )

    def find_market_locations(
        self,
        *,
        gridpool_id: int | None = None,
        microgrid_id: int | None = None,
        delivery_area: str | DeliveryAreaConfig | None = None,
        at: datetime | None = None,
    ) -> list[str]:
        """List the market locations of the matching relations.

        Args:
            gridpool_id: Gridpool to match, or `None` to ignore.
            microgrid_id: Microgrid to match, or `None` to ignore.
            delivery_area: Delivery area to match, a `DeliveryAreaConfig` matched
                on code and code type or a bare code string; `None` to ignore.
            at: Instant the relations must apply at, or `None` to ignore.

        Returns:
            The market locations, deduplicated, in document order.
        """
        return list(
            dict.fromkeys(
                relation.market_location_id
                for relation in self.find_relations(
                    gridpool_id=gridpool_id,
                    microgrid_id=microgrid_id,
                    delivery_area=delivery_area,
                    at=at,
                )
                if relation.market_location_id is not None
            )
        )

    def find_microgrids(
        self,
        *,
        gridpool_id: int | None = None,
        market_location_id: str | None = None,
        delivery_area: str | DeliveryAreaConfig | None = None,
        at: datetime | None = None,
    ) -> list[int]:
        """List the microgrids of the matching relations.

        Args:
            gridpool_id: Gridpool to match, or `None` to ignore.
            market_location_id: Market location to match, or `None` to ignore.
            delivery_area: Delivery area to match, a `DeliveryAreaConfig` matched
                on code and code type or a bare code string; `None` to ignore.
            at: Instant the relations must apply at, or `None` to ignore.

        Returns:
            The microgrids, deduplicated, in document order.
        """
        return list(
            dict.fromkeys(
                relation.microgrid_id
                for relation in self.find_relations(
                    gridpool_id=gridpool_id,
                    market_location_id=market_location_id,
                    delivery_area=delivery_area,
                    at=at,
                )
                if relation.microgrid_id is not None
            )
        )

    def find_enterprise(self, gridpool_id: int) -> int | None:
        """Find the configured enterprise owning `gridpool_id`.

        Args:
            gridpool_id: The gridpool to look up.

        Returns:
            The owning enterprise ID, or `None` when the gridpool is not configured.
        """
        gridpool = self.gridpools.get(gridpool_id)
        return gridpool.enterprise_id if gridpool is not None else None

    @classmethod
    def _warn_unknown_entities(cls, assets: dict[str, Any], source: Path) -> None:
        """Warn about entity tables that this version drops on load."""
        if unknown := sorted(set(assets) - set(cls.Schema().fields)):
            _logger.warning(
                "%s: ignoring unknown entity tables under `assets`: %s",
                source,
                ", ".join(unknown),
            )

    @classmethod
    def _read_assets_table(cls, config_path: Path) -> dict[str, Any]:
        """Read the raw `assets` table from a TOML file.

        The document is migrated to the current format before its `assets`
        table is returned for merging. A file with no `assets` table contributes
        nothing to the merge, so consumers can pass a mixed list of files and
        gridpool reads only the `assets`-bearing ones.

        Args:
            config_path: The path to the TOML configuration file.

        Returns:
            The raw `assets` table, unvalidated, for merging before it is loaded,
            or an empty table if the file has none.

        Raises:
            TypeError: If `assets` is present but not a table.
        """
        with config_path.open("rb") as f:
            data: dict[str, Any] = tomllib.load(f)

        data = migrate(data, config_path)

        assets = data.get("assets")
        if assets is None:
            return {}
        if not isinstance(assets, dict):
            raise TypeError(
                f"{config_path}: `assets` must be a table, got {type(assets)}"
            )

        cls._warn_unknown_entities(assets, config_path)
        return assets

    @classmethod
    def load_from_files(
        cls,
        config_files: str | Path | list[str | Path],
        check: bool = True,
    ) -> Self:
        """Load and validate a config document from one or more TOML files.

        Later files take precedence, entry by entry, so a file can override
        single fields of an entry another defines. The raw tables are merged
        before they are loaded, so a field left unset in an override keeps the
        base value rather than being reset to its default. Paths that are not
        files are skipped with a warning.

        Args:
            config_files: A path or list of paths to TOML config files.
            check: Whether to run the whole-document `check`. It skips only that
                cross-entity pass; the schema and each entry's own validation
                still run. Pass all layers of a stack together rather than
                loading one incomplete override with `check=False`.

        Returns:
            The merged document.
        """
        merged = _merge_file_tables(config_files)
        loaded = cls.Schema().load(merged)
        assert isinstance(loaded, cls)
        if check:
            loaded.check()
        return loaded

Attributes

gridpools class-attribute instance-attribute

gridpools: dict[int, GridpoolConfig] = field(
    default_factory=dict
)

Gridpools, keyed by gridpool ID.

market_locations class-attribute instance-attribute

market_locations: dict[str, MarketLocationConfig] = field(
    default_factory=dict
)

Market locations, keyed by their identifier.

microgrids class-attribute instance-attribute

microgrids: dict[int, MicrogridConfig] = field(
    default_factory=dict
)

Microgrids, keyed by microgrid ID.

relations class-attribute instance-attribute

relations: dict[str, RelationConfig] = field(
    default_factory=dict
)

Market topology relations, keyed by the composite of the sides they connect.

version class-attribute instance-attribute

version: int = _CURRENT_VERSION

Format version of the assets namespace, stamped by the migration.

Classes

Meta

Ignore entity tables this version does not know about.

A reader must keep working against files that already carry entities added after it, so unknown tables are skipped rather than rejected. _warn_unknown_entities reports them, so a mistyped table is still visible instead of silently loading as empty.

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

class Meta:
    """Ignore entity tables this version does not know about.

    A reader must keep working against files that already carry entities
    added after it, so unknown tables are skipped rather than rejected.
    `_warn_unknown_entities` reports them, so a mistyped table is still
    visible instead of silently loading as empty.
    """

    unknown = marshmallow.EXCLUDE

Methods:

__post_init__

__post_init__() -> None

Check that every microgrid is filed under its own ID.

Relations are not checked here: an override names only the fields it changes, so a single document may hold an incomplete record. check looks at the merged result.

RAISES	DESCRIPTION
ValueError	If a key is not the ID of the entry it holds.

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

def __post_init__(self) -> None:
    """Check that every microgrid is filed under its own ID.

    Relations are not checked here: an override names only the fields it
    changes, so a single document may hold an incomplete record. `check`
    looks at the merged result.

    Raises:
        ValueError: If a key is not the ID of the entry it holds.
    """
    for mid, cfg in self.microgrids.items():
        if int(cfg.microgrid_id) != mid:
            raise ValueError(
                f"Microgrid ID mismatch: key {mid} != {cfg.microgrid_id}"
            )
    for gpid, gridpool in self.gridpools.items():
        if int(gridpool.gridpool_id) != gpid:
            raise ValueError(
                f"Gridpool ID mismatch: key {gpid} != {gridpool.gridpool_id}"
            )

check

check() -> None

Check the document as a whole, once every layer has been merged.

RAISES	DESCRIPTION
ValueError	If an entry's identifier disagrees with the key it is filed under, a relation names fewer than two sides, its key disagrees with its fields, a gridpool relation names no delivery area, one market location is placed in two of them, a legacy microgrid gridpool ID disagrees with its relations, or a gridpool's declared and inferred enterprise disagree.

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

def check(self) -> None:
    """Check the document as a whole, once every layer has been merged.

    Raises:
        ValueError: If an entry's identifier disagrees with the key it is
            filed under, a relation names fewer than two sides, its key
            disagrees with its fields, a gridpool relation names no delivery
            area, one market location is placed in two of them, a legacy
            microgrid gridpool ID disagrees with its relations, or a
            gridpool's declared and inferred enterprise disagree.
    """
    for key, location in self.market_locations.items():
        if location.id is None:
            raise ValueError(f"Market location {key}: must name its id")
        if location.id != key:
            raise ValueError(
                f"Market location key mismatch: key {key} != {location.id}"
            )
    self._check_relations()
    self._check_legacy_gridpool_ids()
    self._check_gridpool_enterprises()

find_delivery_areas

find_delivery_areas(
    *,
    gridpool_id: int | None = None,
    microgrid_id: int | None = None,
    market_location_id: str | None = None,
    at: datetime | None = None
) -> list[DeliveryAreaConfig]

List the delivery areas of the matching relations.

PARAMETER	DESCRIPTION
gridpool_id	Gridpool to match, or None to ignore. TYPE: int | None DEFAULT: None
microgrid_id	Microgrid to match, or None to ignore. TYPE: int | None DEFAULT: None
market_location_id	Market location to match, or None to ignore. TYPE: str | None DEFAULT: None
at	Instant the relations must apply at, or None to ignore. TYPE: datetime | None DEFAULT: None

RETURNS	DESCRIPTION
list[DeliveryAreaConfig]	The delivery areas, each with its code and code type, deduplicated,
list[DeliveryAreaConfig]	in document order.

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

def find_delivery_areas(
    self,
    *,
    gridpool_id: int | None = None,
    microgrid_id: int | None = None,
    market_location_id: str | None = None,
    at: datetime | None = None,
) -> list[DeliveryAreaConfig]:
    """List the delivery areas of the matching relations.

    Args:
        gridpool_id: Gridpool to match, or `None` to ignore.
        microgrid_id: Microgrid to match, or `None` to ignore.
        market_location_id: Market location to match, or `None` to ignore.
        at: Instant the relations must apply at, or `None` to ignore.

    Returns:
        The delivery areas, each with its code and code type, deduplicated,
        in document order.
    """
    return list(
        dict.fromkeys(
            relation.delivery_area
            for relation in self.find_relations(
                gridpool_id=gridpool_id,
                microgrid_id=microgrid_id,
                market_location_id=market_location_id,
                at=at,
            )
            if relation.delivery_area is not None
        )
    )

find_enterprise

find_enterprise(gridpool_id: int) -> int | None

Find the configured enterprise owning gridpool_id.

PARAMETER	DESCRIPTION
gridpool_id	The gridpool to look up. TYPE: int

RETURNS	DESCRIPTION
int | None	The owning enterprise ID, or None when the gridpool is not configured.

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

def find_enterprise(self, gridpool_id: int) -> int | None:
    """Find the configured enterprise owning `gridpool_id`.

    Args:
        gridpool_id: The gridpool to look up.

    Returns:
        The owning enterprise ID, or `None` when the gridpool is not configured.
    """
    gridpool = self.gridpools.get(gridpool_id)
    return gridpool.enterprise_id if gridpool is not None else None

find_market_locations

find_market_locations(
    *,
    gridpool_id: int | None = None,
    microgrid_id: int | None = None,
    delivery_area: str | DeliveryAreaConfig | None = None,
    at: datetime | None = None
) -> list[str]

List the market locations of the matching relations.

PARAMETER	DESCRIPTION
gridpool_id	Gridpool to match, or None to ignore. TYPE: int | None DEFAULT: None
microgrid_id	Microgrid to match, or None to ignore. TYPE: int | None DEFAULT: None
delivery_area	Delivery area to match, a DeliveryAreaConfig matched on code and code type or a bare code string; None to ignore. TYPE: str | DeliveryAreaConfig | None DEFAULT: None
at	Instant the relations must apply at, or None to ignore. TYPE: datetime | None DEFAULT: None

RETURNS	DESCRIPTION
list[str]	The market locations, deduplicated, in document order.

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

def find_market_locations(
    self,
    *,
    gridpool_id: int | None = None,
    microgrid_id: int | None = None,
    delivery_area: str | DeliveryAreaConfig | None = None,
    at: datetime | None = None,
) -> list[str]:
    """List the market locations of the matching relations.

    Args:
        gridpool_id: Gridpool to match, or `None` to ignore.
        microgrid_id: Microgrid to match, or `None` to ignore.
        delivery_area: Delivery area to match, a `DeliveryAreaConfig` matched
            on code and code type or a bare code string; `None` to ignore.
        at: Instant the relations must apply at, or `None` to ignore.

    Returns:
        The market locations, deduplicated, in document order.
    """
    return list(
        dict.fromkeys(
            relation.market_location_id
            for relation in self.find_relations(
                gridpool_id=gridpool_id,
                microgrid_id=microgrid_id,
                delivery_area=delivery_area,
                at=at,
            )
            if relation.market_location_id is not None
        )
    )

find_microgrids

find_microgrids(
    *,
    gridpool_id: int | None = None,
    market_location_id: str | None = None,
    delivery_area: str | DeliveryAreaConfig | None = None,
    at: datetime | None = None
) -> list[int]

List the microgrids of the matching relations.

PARAMETER	DESCRIPTION
gridpool_id	Gridpool to match, or None to ignore. TYPE: int | None DEFAULT: None
market_location_id	Market location to match, or None to ignore. TYPE: str | None DEFAULT: None
delivery_area	Delivery area to match, a DeliveryAreaConfig matched on code and code type or a bare code string; None to ignore. TYPE: str | DeliveryAreaConfig | None DEFAULT: None
at	Instant the relations must apply at, or None to ignore. TYPE: datetime | None DEFAULT: None

RETURNS	DESCRIPTION
list[int]	The microgrids, deduplicated, in document order.

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

def find_microgrids(
    self,
    *,
    gridpool_id: int | None = None,
    market_location_id: str | None = None,
    delivery_area: str | DeliveryAreaConfig | None = None,
    at: datetime | None = None,
) -> list[int]:
    """List the microgrids of the matching relations.

    Args:
        gridpool_id: Gridpool to match, or `None` to ignore.
        market_location_id: Market location to match, or `None` to ignore.
        delivery_area: Delivery area to match, a `DeliveryAreaConfig` matched
            on code and code type or a bare code string; `None` to ignore.
        at: Instant the relations must apply at, or `None` to ignore.

    Returns:
        The microgrids, deduplicated, in document order.
    """
    return list(
        dict.fromkeys(
            relation.microgrid_id
            for relation in self.find_relations(
                gridpool_id=gridpool_id,
                market_location_id=market_location_id,
                delivery_area=delivery_area,
                at=at,
            )
            if relation.microgrid_id is not None
        )
    )

find_relations

find_relations(
    *,
    gridpool_id: int | None = None,
    microgrid_id: int | None = None,
    market_location_id: str | None = None,
    delivery_area: str | DeliveryAreaConfig | None = None,
    participation: MarketParticipationType | None = None,
    at: datetime | None = None
) -> list[RelationConfig]

Find the relations naming all of the given sides.

PARAMETER	DESCRIPTION
gridpool_id	Gridpool to match, or None to ignore. TYPE: int | None DEFAULT: None
microgrid_id	Microgrid to match, or None to ignore. TYPE: int | None DEFAULT: None
market_location_id	Market location to match, or None to ignore. TYPE: str | None DEFAULT: None
delivery_area	Delivery area to match, a DeliveryAreaConfig matched on code and code type or a bare code string; None to ignore. TYPE: str | DeliveryAreaConfig | None DEFAULT: None
participation	Use case the relation must serve, or None to ignore. TYPE: MarketParticipationType | None DEFAULT: None
at	Instant the relations, or the given participation, must apply at, or None to ignore. TYPE: datetime | None DEFAULT: None

RETURNS	DESCRIPTION
list[RelationConfig]	The matching relations, in document order.

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

def find_relations(
    self,
    *,
    gridpool_id: int | None = None,
    microgrid_id: int | None = None,
    market_location_id: str | None = None,
    delivery_area: str | DeliveryAreaConfig | None = None,
    participation: MarketParticipationType | None = None,
    at: datetime | None = None,
) -> list[RelationConfig]:
    """Find the relations naming all of the given sides.

    Args:
        gridpool_id: Gridpool to match, or `None` to ignore.
        microgrid_id: Microgrid to match, or `None` to ignore.
        market_location_id: Market location to match, or `None` to ignore.
        delivery_area: Delivery area to match, a `DeliveryAreaConfig` matched
            on code and code type or a bare code string; `None` to ignore.
        participation: Use case the relation must serve, or `None` to ignore.
        at: Instant the relations, or the given participation, must apply at,
            or `None` to ignore.

    Returns:
        The matching relations, in document order.
    """
    return [
        relation
        for relation in self.relations.values()
        if relation.matches(
            gridpool_id=gridpool_id,
            microgrid_id=microgrid_id,
            market_location_id=market_location_id,
            delivery_area=delivery_area,
            participation=participation,
            at=at,
        )
    ]

load_from_files classmethod

load_from_files(
    config_files: str | Path | list[str | Path],
    check: bool = True,
) -> Self

Load and validate a config document from one or more TOML files.

Later files take precedence, entry by entry, so a file can override single fields of an entry another defines. The raw tables are merged before they are loaded, so a field left unset in an override keeps the base value rather than being reset to its default. Paths that are not files are skipped with a warning.

PARAMETER	DESCRIPTION
config_files	A path or list of paths to TOML config files. TYPE: str | Path | list[str | Path]
check	Whether to run the whole-document check. It skips only that cross-entity pass; the schema and each entry's own validation still run. Pass all layers of a stack together rather than loading one incomplete override with check=False. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
Self	The merged document.

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

@classmethod
def load_from_files(
    cls,
    config_files: str | Path | list[str | Path],
    check: bool = True,
) -> Self:
    """Load and validate a config document from one or more TOML files.

    Later files take precedence, entry by entry, so a file can override
    single fields of an entry another defines. The raw tables are merged
    before they are loaded, so a field left unset in an override keeps the
    base value rather than being reset to its default. Paths that are not
    files are skipped with a warning.

    Args:
        config_files: A path or list of paths to TOML config files.
        check: Whether to run the whole-document `check`. It skips only that
            cross-entity pass; the schema and each entry's own validation
            still run. Pass all layers of a stack together rather than
            loading one incomplete override with `check=False`.

    Returns:
        The merged document.
    """
    merged = _merge_file_tables(config_files)
    loaded = cls.Schema().load(merged)
    assert isinstance(loaded, cls)
    if check:
        loaded.check()
    return loaded

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 = field(default=None, metadata=toml_datetime_metadata())
    """Start time of the battery installation."""

    end_time: datetime | None = field(default=None, metadata=toml_datetime_metadata())
    """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 = field(
    default=None, metadata=toml_datetime_metadata()
)

End time of the battery installation.

start_time class-attribute instance-attribute

start_time: datetime | None = field(
    default=None, metadata=toml_datetime_metadata()
)

Start time of the battery installation.

frequenz.gridpool.config.ComponentGraphConfig

Configuration for the component graph.

Source code in frequenz/microgrid_component_graph/__init__.py

    "FormulaGenerationError",
    "FormulaOverrides",
    "InvalidGraphError",
]

Methods:

__init__

__init__(
    *,
    allow_component_validation_failures: bool = False,
    allow_unconnected_components: bool = False,
    allow_unspecified_inverters: bool = False,
    disable_fallback_components: bool = False,
    include_phantom_loads_in_consumer_formula: bool = False,
    prefer_meters_in_component_formulas: bool = False,
    formula_overrides: FormulaOverrides | None = None
) -> None

Initialize this instance.

PARAMETER	DESCRIPTION
allow_component_validation_failures	Whether to allow validation errors on components. When this is True, the graph will be built even if there are validation errors on the components. TYPE: bool DEFAULT: False
allow_unconnected_components	Whether to allow unconnected components in the graph, that are not reachable from the root. TYPE: bool DEFAULT: False
allow_unspecified_inverters	Whether to allow untyped inverters in the graph. When this is True, inverters that have InverterType::Unspecified will be assumed to be Battery inverters. TYPE: bool DEFAULT: False
disable_fallback_components	Whether to disable fallback components in generated formulas. When this is True, the formulas will not include fallback components. TYPE: bool DEFAULT: False
include_phantom_loads_in_consumer_formula	Whether to consider phantom loads in the consumer formula. Meters with successors can still have loads not represented in the component graph. These are called phantom loads. When this is true, phantom loads are included in formulas by excluding the measurements of successor meters from the measurements of their predecessor meters. When false, consumer formula is generated by excluding production and battery components from the grid measurements. The non-consumer components behind one internal meter are excluded as one group: the meter reading first, the component readings as the fallback. While the meter reading is used, a load behind that meter that is not in the component graph is then also excluded together with the group. TYPE: bool DEFAULT: False
prefer_meters_in_component_formulas	Default policy for the per-category formulas. When False (the default), the component measurement is the primary source and the meter measurement is the fallback for battery_formula, chp_formula, pv_formula, wind_turbine_formula, ev_charger_formula, and steam_boiler_formula. When True, the meter is primary and the component is the fallback. Has no effect on grid_formula, consumer_formula, producer_formula, or any of the coalesce formulas. TYPE: bool DEFAULT: False
formula_overrides	Per-formula overrides for the meter/component preference; see FormulaOverrides. Each entry, when set, takes precedence over prefer_meters_in_component_formulas for that formula. TYPE: FormulaOverrides | None DEFAULT: None

Source code in frequenz/microgrid_component_graph/__init__.py

]

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 {}

    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 {}

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.DeliveryAreaConfig

Configuration of a delivery area.

Mirrors the Assets API DeliveryArea: a grid-zone code and the code_type that says how to read it. code_type defaults to EIC, so the common case is just a code. An EIC code is checked for valid syntax and check character; other code types are taken as given.

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

@dataclass(frozen=True)
class DeliveryAreaConfig:
    """Configuration of a delivery area.

    Mirrors the Assets API `DeliveryArea`: a grid-zone `code` and the `code_type`
    that says how to read it. `code_type` defaults to EIC, so the common case is
    just a code. An EIC code is checked for valid syntax and check character;
    other code types are taken as given.
    """

    code: str | None = None
    """The delivery-area code."""

    code_type: EnergyMarketCodeType = EnergyMarketCodeType.EUROPE_EIC
    """Identifier scheme of the code, defaulting to EIC."""

    Schema: ClassVar[Type[Schema]] = Schema

    def __post_init__(self) -> None:
        """Check the code against its type.

        Raises:
            ValueError: If the code type is unspecified, no code is given, or an
                EIC code is malformed.
        """
        if self.code_type is EnergyMarketCodeType.UNSPECIFIED:
            raise ValueError("Delivery area code type must be specified")
        if self.code is None:
            raise ValueError("Delivery area must name a code")
        if self.code_type is EnergyMarketCodeType.EUROPE_EIC:
            _require_eic(self.code)

Attributes

code class-attribute instance-attribute

code: str | None = None

The delivery-area code.

code_type class-attribute instance-attribute

code_type: EnergyMarketCodeType = (
    EnergyMarketCodeType.EUROPE_EIC
)

Identifier scheme of the code, defaulting to EIC.

Methods:

__post_init__

__post_init__() -> None

Check the code against its type.

RAISES	DESCRIPTION
ValueError	If the code type is unspecified, no code is given, or an EIC code is malformed.

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

def __post_init__(self) -> None:
    """Check the code against its type.

    Raises:
        ValueError: If the code type is unspecified, no code is given, or an
            EIC code is malformed.
    """
    if self.code_type is EnergyMarketCodeType.UNSPECIFIED:
        raise ValueError("Delivery area code type must be specified")
    if self.code is None:
        raise ValueError("Delivery area must name a code")
    if self.code_type is EnergyMarketCodeType.EUROPE_EIC:
        _require_eic(self.code)

frequenz.gridpool.config.FormulaOverrides

Per-formula overrides for the meter/component preference.

Each parameter is None by default, meaning the corresponding formula follows the global prefer_meters_in_component_formulas setting on ComponentGraphConfig. Setting True forces the meter as primary for that formula; False forces the component.

Methods:

__init__

__init__(
    *,
    prefer_meters_in_pv_formula: bool | None = None,
    prefer_meters_in_battery_formula: bool | None = None,
    prefer_meters_in_chp_formula: bool | None = None,
    prefer_meters_in_ev_charger_formula: bool | None = None,
    prefer_meters_in_wind_turbine_formula: (
        bool | None
    ) = None,
    prefer_meters_in_steam_boiler_formula: (
        bool | None
    ) = None
) -> None

Initialize this instance.

PARAMETER	DESCRIPTION
prefer_meters_in_pv_formula	Override for pv_formula. TYPE: bool | None DEFAULT: None
prefer_meters_in_battery_formula	Override for battery_formula. TYPE: bool | None DEFAULT: None
prefer_meters_in_chp_formula	Override for chp_formula. TYPE: bool | None DEFAULT: None
prefer_meters_in_ev_charger_formula	Override for ev_charger_formula. TYPE: bool | None DEFAULT: None
prefer_meters_in_wind_turbine_formula	Override for wind_turbine_formula. TYPE: bool | None DEFAULT: None
prefer_meters_in_steam_boiler_formula	Override for steam_boiler_formula. TYPE: bool | None DEFAULT: None

frequenz.gridpool.config.GridpoolConfig

Configuration of a gridpool.

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

@dataclass
class GridpoolConfig:
    """Configuration of a gridpool."""

    gridpool_id: int
    """ID of the gridpool."""

    enterprise_id: int
    """Enterprise that owns the gridpool."""

Attributes

enterprise_id instance-attribute

enterprise_id: int

Enterprise that owns the gridpool.

gridpool_id instance-attribute

gridpool_id: int

ID of the gridpool.

frequenz.gridpool.config.MarketLocationConfig

Configuration of a market location.

Mirrors the Assets API MarketLocationRef: the market area, identifier id and how to read it. id repeats the key it is filed under, so the object is self-describing; AssetsConfig.check verifies the two agree. The grid zone the location sits in is not kept here but on the relations that name it. Raw IDs must be unique within a document, including across market areas.

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

@dataclass(frozen=True)
class MarketLocationConfig:
    """Configuration of a market location.

    Mirrors the Assets API `MarketLocationRef`: the market area, identifier `id`
    and how to read it. `id` repeats the key it is filed under, so the object is
    self-describing; `AssetsConfig.check` verifies the two agree. The grid zone
    the location sits in is not kept here but on the relations that name it. Raw
    IDs must be unique within a document, including across market areas.
    """

    id: str | None = None
    """The market location identifier."""

    type: MarketLocationIdType = MarketLocationIdType.MALO_ID
    """Identifier scheme of the market location."""

    market_area: int = _MARKET_AREA_EU_DE
    """Assets API market area, defaulting to EU_DE (`101`)."""

    Schema: ClassVar[Type[Schema]] = Schema

    def __post_init__(self) -> None:
        """Check that the identifier scheme and market area are specified.

        Raises:
            ValueError: If the identifier scheme or market area is unspecified.
        """
        if self.type is MarketLocationIdType.UNSPECIFIED:
            raise ValueError("Market location type must be specified")
        if self.market_area <= 0:
            raise ValueError("Market area must be specified")

Attributes

id class-attribute instance-attribute

id: str | None = None

The market location identifier.

market_area class-attribute instance-attribute

market_area: int = _MARKET_AREA_EU_DE

Assets API market area, defaulting to EU_DE (101).

type class-attribute instance-attribute

type: MarketLocationIdType = MarketLocationIdType.MALO_ID

Identifier scheme of the market location.

Methods:

__post_init__

__post_init__() -> None

Check that the identifier scheme and market area are specified.

RAISES	DESCRIPTION
ValueError	If the identifier scheme or market area is unspecified.

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

def __post_init__(self) -> None:
    """Check that the identifier scheme and market area are specified.

    Raises:
        ValueError: If the identifier scheme or market area is unspecified.
    """
    if self.type is MarketLocationIdType.UNSPECIFIED:
        raise ValueError("Market location type must be specified")
    if self.market_area <= 0:
        raise ValueError("Market area must be specified")

frequenz.gridpool.config.MicrogridConfig

Configuration of a microgrid.

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

@dataclass
class MicrogridConfig:
    """Configuration of 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
    """Legacy ID; if set, all gridpool relations must name it."""

    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 = field(default=None, metadata=toml_datetime_metadata())
    """Start time of the microgrid operation."""

    end_time: datetime | None = field(default=None, metadata=toml_datetime_metadata())
    """End time of the microgrid operation."""

    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[int, Self]:
        """Load microgrid configurations from table entries.

        Args:
            data: The table mapping microgrid IDs to their entries.

        Returns:
            A dict mapping microgrid IDs to MicrogridConfig instances.

        Raises:
            ValueError: If the 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 microgrid 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 int(mgrid.microgrid_id) != int(mid):
                raise ValueError(
                    f"Table reader: Microgrid ID mismatch: key {mid} != {mgrid.microgrid_id}"
                )

            mgrids[int(mid)] = mgrid

        return mgrids

Attributes

altitude class-attribute instance-attribute

altitude: float | None = None

Geographic altitude of the microgrid.

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.

end_time class-attribute instance-attribute

end_time: datetime | None = field(
    default=None, metadata=toml_datetime_metadata()
)

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

Legacy ID; if set, all gridpool relations must name it.

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.

pv class-attribute instance-attribute

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

Configuration of the PV system.

start_time class-attribute instance-attribute

start_time: datetime | None = field(
    default=None, metadata=toml_datetime_metadata()
)

Start time of the microgrid operation.

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

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 = field(default=None, metadata=toml_datetime_metadata())
    """Start time of the PV system installation."""

    end_time: datetime | None = field(default=None, metadata=toml_datetime_metadata())
    """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 = field(
    default=None, metadata=toml_datetime_metadata()
)

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 = field(
    default=None, metadata=toml_datetime_metadata()
)

Start time of the PV system installation.

frequenz.gridpool.config.RelationConfig

A relation between a gridpool, a microgrid and a market location.

It names at least two of the three. The key it is filed under is G<gridpool>M<microgrid>L<market_location> with the segments of the unset sides dropped. The key clusters the lines of one record and is never read: every value comes from the fields, and AssetsConfig.check verifies that the two agree.

A relation naming a gridpool sits in a delivery_area; the grid zone rides on the relation, not on the market location, so a gridpool-to-microgrid relation with no market location still carries one. A gridpool-free microgrid-to-market-location relation may also carry one to map that market location to its delivery area. A delivery area names its code type, which defaults to EIC.

Its validity lives in validity, each entry a period the relation applies over. A period naming a market use case tells the relation's gridpool participations apart; one without is a plain window, for a relation that serves a single use or none, such as a bare microgrid-to-market-location metering relation. A relation with no periods applies at all times.

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

@dataclass(frozen=True)
class RelationConfig:
    """A relation between a gridpool, a microgrid and a market location.

    It names at least two of the three. The key it is filed under is
    `G<gridpool>M<microgrid>L<market_location>` with the segments of the unset
    sides dropped. The key clusters the lines of one record and is never read:
    every value comes from the fields, and `AssetsConfig.check` verifies that the
    two agree.

    A relation naming a gridpool sits in a `delivery_area`; the grid zone rides
    on the relation, not on the market location, so a gridpool-to-microgrid
    relation with no market location still carries one. A gridpool-free
    microgrid-to-market-location relation may also carry one to map that market
    location to its delivery area. A delivery area names its code type, which
    defaults to EIC.

    Its validity lives in `validity`, each entry a period the relation applies
    over. A period naming a market use case tells the relation's gridpool
    participations apart; one without is a plain window, for a relation that
    serves a single use or none, such as a bare microgrid-to-market-location
    metering relation. A relation with no periods applies at all times.
    """

    gridpool_id: int | None = None
    """Gridpool participating in this relation."""

    microgrid_id: int | None = None
    """Microgrid participating in this relation."""

    market_location_id: str | None = None
    """Market location participating in this relation."""

    delivery_area: DeliveryAreaConfig | None = None
    """Delivery area this relation sits in; required once a gridpool is named."""

    validity: dict[str, ValidityConfig] = field(default_factory=dict)
    """Periods this relation applies over, each optionally a market use case."""

    Schema: ClassVar[Type[Schema]] = Schema

    def __post_init__(self) -> None:
        """Check the periods against the relation.

        A record may be incomplete here, since an override names only the fields
        it changes; `AssetsConfig.check` looks at the merged result.

        Raises:
            ValueError: If a period names a market use case without a gridpool,
                or periods of one use case overlap.
        """
        by_use: dict[MarketParticipationType | None, list[ValidityConfig]] = {}
        for period in self.validity.values():
            if period.participation is not None and self.gridpool_id is None:
                raise ValueError(
                    f"Relation {self.key}: a {period.participation.name} "
                    "participation applies only to a gridpool relation"
                )
            by_use.setdefault(period.participation, []).append(period)
        for use, group in by_use.items():
            label = use.name if use is not None else "untyped"
            for i, period in enumerate(group):
                for other in group[i + 1 :]:
                    if period.overlaps(other):
                        raise ValueError(
                            f"Relation {self.key}: {label} periods "
                            f"{period.start}..{period.end} and "
                            f"{other.start}..{other.end} overlap"
                        )

    @property
    def key(self) -> str:
        """The key this relation belongs under, derived from its own fields."""
        return _relation_key(
            self.gridpool_id, self.microgrid_id, self.market_location_id
        )

    @property
    def is_complete(self) -> bool:
        """Whether this relation names at least two of the three sides."""
        sides = (self.gridpool_id, self.microgrid_id, self.market_location_id)
        return sum(side is not None for side in sides) >= 2

    def covers(self, at: datetime) -> bool:
        """Check whether this relation applies at an instant.

        Args:
            at: The instant to check.

        Returns:
            Whether any of its periods covers the instant, or `True` when it
            lists none and so applies always.
        """
        _require_offset_aware(at, "Instant")
        if not self.validity:
            return True
        return any(period.covers(at) for period in self.validity.values())

    def matches(
        self,
        *,
        gridpool_id: int | None = None,
        microgrid_id: int | None = None,
        market_location_id: str | None = None,
        delivery_area: str | DeliveryAreaConfig | None = None,
        participation: MarketParticipationType | None = None,
        at: datetime | None = None,
    ) -> bool:
        """Check whether this relation has all the given sides.

        Args:
            gridpool_id: Gridpool to match, or `None` to ignore.
            microgrid_id: Microgrid to match, or `None` to ignore.
            market_location_id: Market location to match, or `None` to ignore.
            delivery_area: Delivery area to match, either a `DeliveryAreaConfig`
                matched on code and code type, or a bare code string matched on
                code alone; `None` to ignore.
            participation: Use case the relation must serve, or `None` to ignore.
            at: Instant the relation, or the given participation, must apply at,
                or `None` to ignore.

        Returns:
            Whether every side given matches this relation.
        """
        if at is not None:
            _require_offset_aware(at, "Instant")
        if gridpool_id is not None and gridpool_id != self.gridpool_id:
            return False
        if microgrid_id is not None and microgrid_id != self.microgrid_id:
            return False
        if market_location_id is not None and (
            market_location_id != self.market_location_id
        ):
            return False
        if delivery_area is not None:
            if self.delivery_area is None:
                return False
            if isinstance(delivery_area, DeliveryAreaConfig):
                if delivery_area != self.delivery_area:
                    return False
            elif delivery_area != self.delivery_area.code:
                return False
        if participation is not None:
            served = [
                p for p in self.validity.values() if p.participation == participation
            ]
            if not served or (at is not None and not any(p.covers(at) for p in served)):
                return False
        elif at is not None and not self.covers(at):
            return False
        return True

Attributes

delivery_area class-attribute instance-attribute

delivery_area: DeliveryAreaConfig | None = None

Delivery area this relation sits in; required once a gridpool is named.

gridpool_id class-attribute instance-attribute

gridpool_id: int | None = None

Gridpool participating in this relation.

is_complete property

is_complete: bool

Whether this relation names at least two of the three sides.

key property

key: str

The key this relation belongs under, derived from its own fields.

market_location_id class-attribute instance-attribute

market_location_id: str | None = None

Market location participating in this relation.

microgrid_id class-attribute instance-attribute

microgrid_id: int | None = None

Microgrid participating in this relation.

validity class-attribute instance-attribute

validity: dict[str, ValidityConfig] = field(
    default_factory=dict
)

Periods this relation applies over, each optionally a market use case.

Methods:

__post_init__

__post_init__() -> None

Check the periods against the relation.

A record may be incomplete here, since an override names only the fields it changes; AssetsConfig.check looks at the merged result.

RAISES	DESCRIPTION
ValueError	If a period names a market use case without a gridpool, or periods of one use case overlap.

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

def __post_init__(self) -> None:
    """Check the periods against the relation.

    A record may be incomplete here, since an override names only the fields
    it changes; `AssetsConfig.check` looks at the merged result.

    Raises:
        ValueError: If a period names a market use case without a gridpool,
            or periods of one use case overlap.
    """
    by_use: dict[MarketParticipationType | None, list[ValidityConfig]] = {}
    for period in self.validity.values():
        if period.participation is not None and self.gridpool_id is None:
            raise ValueError(
                f"Relation {self.key}: a {period.participation.name} "
                "participation applies only to a gridpool relation"
            )
        by_use.setdefault(period.participation, []).append(period)
    for use, group in by_use.items():
        label = use.name if use is not None else "untyped"
        for i, period in enumerate(group):
            for other in group[i + 1 :]:
                if period.overlaps(other):
                    raise ValueError(
                        f"Relation {self.key}: {label} periods "
                        f"{period.start}..{period.end} and "
                        f"{other.start}..{other.end} overlap"
                    )

covers

covers(at: datetime) -> bool

Check whether this relation applies at an instant.

PARAMETER	DESCRIPTION
at	The instant to check. TYPE: datetime

RETURNS	DESCRIPTION
bool	Whether any of its periods covers the instant, or True when it
bool	lists none and so applies always.

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

def covers(self, at: datetime) -> bool:
    """Check whether this relation applies at an instant.

    Args:
        at: The instant to check.

    Returns:
        Whether any of its periods covers the instant, or `True` when it
        lists none and so applies always.
    """
    _require_offset_aware(at, "Instant")
    if not self.validity:
        return True
    return any(period.covers(at) for period in self.validity.values())

matches

matches(
    *,
    gridpool_id: int | None = None,
    microgrid_id: int | None = None,
    market_location_id: str | None = None,
    delivery_area: str | DeliveryAreaConfig | None = None,
    participation: MarketParticipationType | None = None,
    at: datetime | None = None
) -> bool

Check whether this relation has all the given sides.

PARAMETER	DESCRIPTION
gridpool_id	Gridpool to match, or None to ignore. TYPE: int | None DEFAULT: None
microgrid_id	Microgrid to match, or None to ignore. TYPE: int | None DEFAULT: None
market_location_id	Market location to match, or None to ignore. TYPE: str | None DEFAULT: None
delivery_area	Delivery area to match, either a DeliveryAreaConfig matched on code and code type, or a bare code string matched on code alone; None to ignore. TYPE: str | DeliveryAreaConfig | None DEFAULT: None
participation	Use case the relation must serve, or None to ignore. TYPE: MarketParticipationType | None DEFAULT: None
at	Instant the relation, or the given participation, must apply at, or None to ignore. TYPE: datetime | None DEFAULT: None

RETURNS	DESCRIPTION
bool	Whether every side given matches this relation.

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

def matches(
    self,
    *,
    gridpool_id: int | None = None,
    microgrid_id: int | None = None,
    market_location_id: str | None = None,
    delivery_area: str | DeliveryAreaConfig | None = None,
    participation: MarketParticipationType | None = None,
    at: datetime | None = None,
) -> bool:
    """Check whether this relation has all the given sides.

    Args:
        gridpool_id: Gridpool to match, or `None` to ignore.
        microgrid_id: Microgrid to match, or `None` to ignore.
        market_location_id: Market location to match, or `None` to ignore.
        delivery_area: Delivery area to match, either a `DeliveryAreaConfig`
            matched on code and code type, or a bare code string matched on
            code alone; `None` to ignore.
        participation: Use case the relation must serve, or `None` to ignore.
        at: Instant the relation, or the given participation, must apply at,
            or `None` to ignore.

    Returns:
        Whether every side given matches this relation.
    """
    if at is not None:
        _require_offset_aware(at, "Instant")
    if gridpool_id is not None and gridpool_id != self.gridpool_id:
        return False
    if microgrid_id is not None and microgrid_id != self.microgrid_id:
        return False
    if market_location_id is not None and (
        market_location_id != self.market_location_id
    ):
        return False
    if delivery_area is not None:
        if self.delivery_area is None:
            return False
        if isinstance(delivery_area, DeliveryAreaConfig):
            if delivery_area != self.delivery_area:
                return False
        elif delivery_area != self.delivery_area.code:
            return False
    if participation is not None:
        served = [
            p for p in self.validity.values() if p.participation == participation
        ]
        if not served or (at is not None and not any(p.covers(at) for p in served)):
            return False
    elif at is not None and not self.covers(at):
        return False
    return True

frequenz.gridpool.config.ValidityConfig

A period a relation applies over, optionally a market use case.

A period naming a participation is a market participation, which applies only to a gridpool relation; one without is a plain validity window. Filed under a free label, which is never read or parsed. The half-open interval [start, end) mirrors the Assets API: the start is inclusive, the end exclusive, and an unset bound is open.

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

@dataclass(frozen=True)
class ValidityConfig:
    """A period a relation applies over, optionally a market use case.

    A period naming a `participation` is a market participation, which applies
    only to a gridpool relation; one without is a plain validity window. Filed
    under a free label, which is never read or parsed. The half-open interval
    `[start, end)` mirrors the Assets API: the start is inclusive, the end
    exclusive, and an unset bound is open.
    """

    participation: MarketParticipationType | None = None
    """The market use case, unset for a plain validity window."""

    start: datetime | None = field(default=None, metadata=toml_datetime_metadata())
    """Inclusive start of the period."""

    end: datetime | None = field(default=None, metadata=toml_datetime_metadata())
    """Exclusive end of the period."""

    Schema: ClassVar[Type[Schema]] = Schema

    def __post_init__(self) -> None:
        """Check the use case, bounds and order of the period.

        Raises:
            ValueError: If the use case is unspecified, a bound has no UTC
                offset, or the period ends at or before it starts.
        """
        if self.participation is MarketParticipationType.UNSPECIFIED:
            raise ValueError(
                "Participation is unspecified; name a use case or leave it unset "
                "for a plain period"
            )
        if self.start is not None:
            _require_offset_aware(self.start, "Period start")
        if self.end is not None:
            _require_offset_aware(self.end, "Period end")
        # Reject an empty `[t, t)` window: it covers no instant, yet `overlaps`
        # would still report it as overlapping a period around `t`.
        if self.start is not None and self.end is not None and self.end <= self.start:
            raise ValueError(
                f"Period ends {self.end} at or before it starts {self.start}"
            )

    def covers(self, at: datetime) -> bool:
        """Check whether this period applies at an instant.

        Args:
            at: The instant to check.

        Returns:
            Whether the instant falls within the half-open period.
        """
        _require_offset_aware(at, "Instant")
        return (self.start is None or self.start <= at) and (
            self.end is None or at < self.end
        )

    def overlaps(self, other: "ValidityConfig") -> bool:
        """Check whether two periods share an instant.

        Args:
            other: The period to compare with.

        Returns:
            Whether the two half-open periods overlap.
        """
        return (self.start is None or other.end is None or self.start < other.end) and (
            other.start is None or self.end is None or other.start < self.end
        )

Attributes

end class-attribute instance-attribute

end: datetime | None = field(
    default=None, metadata=toml_datetime_metadata()
)

Exclusive end of the period.

participation class-attribute instance-attribute

participation: MarketParticipationType | None = None

The market use case, unset for a plain validity window.

start class-attribute instance-attribute

start: datetime | None = field(
    default=None, metadata=toml_datetime_metadata()
)

Inclusive start of the period.

Methods:

__post_init__

__post_init__() -> None

Check the use case, bounds and order of the period.

RAISES	DESCRIPTION
ValueError	If the use case is unspecified, a bound has no UTC offset, or the period ends at or before it starts.

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

def __post_init__(self) -> None:
    """Check the use case, bounds and order of the period.

    Raises:
        ValueError: If the use case is unspecified, a bound has no UTC
            offset, or the period ends at or before it starts.
    """
    if self.participation is MarketParticipationType.UNSPECIFIED:
        raise ValueError(
            "Participation is unspecified; name a use case or leave it unset "
            "for a plain period"
        )
    if self.start is not None:
        _require_offset_aware(self.start, "Period start")
    if self.end is not None:
        _require_offset_aware(self.end, "Period end")
    # Reject an empty `[t, t)` window: it covers no instant, yet `overlaps`
    # would still report it as overlapping a period around `t`.
    if self.start is not None and self.end is not None and self.end <= self.start:
        raise ValueError(
            f"Period ends {self.end} at or before it starts {self.start}"
        )

covers

covers(at: datetime) -> bool

Check whether this period applies at an instant.

PARAMETER	DESCRIPTION
at	The instant to check. TYPE: datetime

RETURNS	DESCRIPTION
bool	Whether the instant falls within the half-open period.

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

def covers(self, at: datetime) -> bool:
    """Check whether this period applies at an instant.

    Args:
        at: The instant to check.

    Returns:
        Whether the instant falls within the half-open period.
    """
    _require_offset_aware(at, "Instant")
    return (self.start is None or self.start <= at) and (
        self.end is None or at < self.end
    )

overlaps

overlaps(other: ValidityConfig) -> bool

Check whether two periods share an instant.

PARAMETER	DESCRIPTION
other	The period to compare with. TYPE: ValidityConfig

RETURNS	DESCRIPTION
bool	Whether the two half-open periods overlap.

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

def overlaps(self, other: "ValidityConfig") -> bool:
    """Check whether two periods share an instant.

    Args:
        other: The period to compare with.

    Returns:
        Whether the two half-open periods overlap.
    """
    return (self.start is None or other.end is None or self.start < other.end) and (
        other.start is None or self.end is None or other.start < self.end
    )

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 = field(default=None, metadata=toml_datetime_metadata())
    """Start time of the wind turbine installation."""

    end_time: datetime | None = field(default=None, metadata=toml_datetime_metadata())
    """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 = field(
    default=None, metadata=toml_datetime_metadata()
)

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 = field(
    default=None, metadata=toml_datetime_metadata()
)

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,
    component_graph_config: (
        ComponentGraphConfig | None
    ) = None,
) -> AssetsConfig

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
component_graph_config	How to build the component graph and generate its formulas. See ComponentGraphConfig. Defaults to that class's own defaults. Requires an assets_client. TYPE: ComponentGraphConfig | None DEFAULT: None

RETURNS	DESCRIPTION
AssetsConfig	The merged document. Use .microgrids for just the microgrid map; the
AssetsConfig	file layers may also contribute relations and market_locations,
AssetsConfig	which the Assets API layer does not provide.

RAISES	DESCRIPTION
ValueError	If none of the three sources is provided, if microgrid_ids or component_graph_config is given without an assets_client, or if a file's assets.microgrids is not a table.

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,
    component_graph_config: ComponentGraphConfig | None = None,
) -> AssetsConfig:
    """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.
        component_graph_config:
            How to build the component graph and generate its formulas.  See
            `ComponentGraphConfig`.  Defaults to that class's own defaults.
            Requires an `assets_client`.

    Returns:
        The merged document. Use `.microgrids` for just the microgrid map; the
        file layers may also contribute `relations` and `market_locations`,
        which the Assets API layer does not provide.

    Raises:
        ValueError: If none of the three sources is provided, if `microgrid_ids`
            or `component_graph_config` is given without an `assets_client`, or
            if a file's `assets.microgrids` is not a table.
    """
    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.")

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

    default_table: dict[str, Any] = {}
    if default_files is not None:
        default_table = _merge_file_tables(default_files)

    override_table: dict[str, Any] = {}
    if override_files is not None:
        override_table = _merge_file_tables(override_files)

    merged = default_table
    if assets_client is not None:
        if microgrid_ids is None:
            file_ids: set[str] = set()
            for table in (default_table, override_table):
                microgrids = table.get("microgrids", {})
                if not isinstance(microgrids, dict):
                    raise ValueError(
                        f"`assets.microgrids` must be a table, got {type(microgrids)}"
                    )
                file_ids |= set(microgrids)
            microgrid_ids = sorted(int(mid) for mid in file_ids)
        assets_configs = await _load_microgrids_from_api(
            assets_client=assets_client,
            microgrid_ids=microgrid_ids,
            component_graph_config=component_graph_config,
        )
        schema = MicrogridConfig.Schema()
        api_table: dict[str, Any] = {
            "microgrids": {
                str(mid): schema.dump(cfg) for mid, cfg in assets_configs.items()
            }
        }
        merged = _deep_merge(merged, api_table)

    merged = _deep_merge(merged, override_table)

    loaded = AssetsConfig.Schema().load(merged)
    assert isinstance(loaded, AssetsConfig)
    loaded.check()
    return loaded