config

frequenz.lib.notebooks.config

Migration module for MicrogridConfig.

Classes

frequenz.lib.notebooks.config.MicrogridConfig

Configuration of a microgrid.

Source code in frequenz/data/microgrid/config.py

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

    meta: Metadata
    """Metadata of the microgrid."""

    pv: dict[str, PVConfig] | None = None
    """Configuration of the PV system."""

    wind: dict[str, WindConfig] | None = None
    """Configuration of the wind turbines."""

    battery: dict[str, BatteryConfig] | None = None
    """Configuration of the batteries."""

    ctype: dict[str, ComponentTypeConfig] = field(default_factory=dict)
    """Mapping of component category types to ac power component config."""

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

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

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

        Returns:
            List of component IDs for this component type.

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

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

        return cfg.cids(metric)

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

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

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

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

        return formula

    Schema: ClassVar[Type[Schema]] = Schema

    @classmethod
    def _load_table_entries(cls, data: dict[str, Any]) -> dict[str, Self]:
        """Load microgrid configurations from table entries.

        Args:
            data: The loaded TOML data.

        Returns:
            A dict mapping microgrid IDs to MicrogridConfig instances.

        Raises:
            ValueError: If top-level keys are not numeric microgrid IDs
                or if there is a microgrid ID mismatch.
            TypeError: If microgrid data is not a dict.
        """
        if not all(str(k).isdigit() for k in data.keys()):
            raise ValueError("All top-level keys must be numeric microgrid IDs.")

        mgrids = {}
        for mid, entry in data.items():
            if not mid.isdigit():
                raise ValueError(
                    f"Table reader: Microgrid ID key must be numeric, got {mid}"
                )
            if not isinstance(entry, dict):
                raise TypeError("Table reader: Each microgrid entry must be a dict")

            mgrid = cls.Schema().load(entry)
            if mgrid.meta is None or mgrid.meta.microgrid_id is None:
                raise ValueError(
                    "Table reader: Each microgrid entry must have a meta.microgrid_id"
                )
            if int(mgrid.meta.microgrid_id) != int(mid):
                raise ValueError(
                    f"Table reader: Microgrid ID mismatch: key {mid} != {mgrid.meta.microgrid_id}"
                )

            mgrids[mid] = mgrid

        return mgrids

    @classmethod
    def load_from_file(cls, config_path: Path) -> dict[int, Self]:
        """
        Load and validate configuration settings from a TOML file.

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

        Returns:
            A dict mapping microgrid IDs to MicrogridConfig instances.
        """
        with config_path.open("rb") as f:
            data = tomllib.load(f)

        assert isinstance(data, dict)

        return cls._load_table_entries(data)

    @staticmethod
    def load_configs(
        microgrid_config_files: str | Path | list[str | Path] | None = None,
        microgrid_config_dir: str | Path | None = None,
    ) -> dict[str, "MicrogridConfig"]:
        """Load multiple microgrid configurations from a file.

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

        Args:
            microgrid_config_files: Path to a single microgrid config file or list of paths.
            microgrid_config_dir: Directory containing multiple microgrid config files.

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

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

        config_files: list[Path] = []

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

        if microgrid_config_dir:
            if Path(microgrid_config_dir).is_dir():
                config_files += list(Path(microgrid_config_dir).glob("*.toml"))
            else:
                raise ValueError(
                    f"Microgrid config directory {microgrid_config_dir} "
                    "is not a directory"
                )

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

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

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

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

        return microgrid_configs

Attributes

battery class-attribute instance-attribute

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

Configuration of the batteries.

ctype class-attribute instance-attribute

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

Mapping of component category types to ac power component config.

meta instance-attribute

meta: Metadata

Metadata of the microgrid.

pv class-attribute instance-attribute

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

Configuration of the PV system.

wind class-attribute instance-attribute

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

Configuration of the wind turbines.

Functions

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 frequenz/data/microgrid/config.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 frequenz/data/microgrid/config.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 frequenz/data/microgrid/config.py

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

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

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

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

    return formula

load_configs staticmethod

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

Load multiple microgrid configurations from a file.

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

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

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

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

Source code in frequenz/data/microgrid/config.py

@staticmethod
def load_configs(
    microgrid_config_files: str | Path | list[str | Path] | None = None,
    microgrid_config_dir: str | Path | None = None,
) -> dict[str, "MicrogridConfig"]:
    """Load multiple microgrid configurations from a file.

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

    Args:
        microgrid_config_files: Path to a single microgrid config file or list of paths.
        microgrid_config_dir: Directory containing multiple microgrid config files.

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

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

    config_files: list[Path] = []

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

    if microgrid_config_dir:
        if Path(microgrid_config_dir).is_dir():
            config_files += list(Path(microgrid_config_dir).glob("*.toml"))
        else:
            raise ValueError(
                f"Microgrid config directory {microgrid_config_dir} "
                "is not a directory"
            )

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

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

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

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

    return microgrid_configs

load_from_file classmethod

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

Load and validate configuration settings from a TOML file.

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

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

Source code in frequenz/data/microgrid/config.py

@classmethod
def load_from_file(cls, config_path: Path) -> dict[int, Self]:
    """
    Load and validate configuration settings from a TOML file.

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

    Returns:
        A dict mapping microgrid IDs to MicrogridConfig instances.
    """
    with config_path.open("rb") as f:
        data = tomllib.load(f)

    assert isinstance(data, dict)

    return cls._load_table_entries(data)