Skip to content

microgrid

frequenz.gridpool.config.microgrid ¤

Data model for microgrid configurations.

Attributes¤

frequenz.gridpool.config.microgrid.ComponentCategory module-attribute ¤

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

Valid component categories.

frequenz.gridpool.config.microgrid.ComponentType module-attribute ¤

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

Valid component types.

Classes¤

frequenz.gridpool.config.microgrid.BatteryConfig ¤

Configuration of a battery in a microgrid.

Source code in src/frequenz/gridpool/config/microgrid.py
@dataclass(frozen=True)
class BatteryConfig:
    """Configuration of a battery in a microgrid."""

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

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

    capacity: float | None = None
    """Capacity of the battery in Wh."""
Attributes¤
capacity class-attribute instance-attribute ¤
capacity: float | None = None

Capacity of the battery in Wh.

end_time class-attribute instance-attribute ¤
end_time: datetime | None = None

End time of the battery installation.

start_time class-attribute instance-attribute ¤
start_time: datetime | None = None

Start time of the battery installation.

frequenz.gridpool.config.microgrid.ComponentTypeConfig ¤

Configuration of a microgrid component type.

Source code in src/frequenz/gridpool/config/microgrid.py
@dataclass
class ComponentTypeConfig:
    """Configuration of a microgrid component type."""

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

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

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

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

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

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

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

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

        Args:
            metric: Metric name of the formula.

        Returns:
            List of component IDs for this component.

        Raises:
            ValueError: If the metric is not supported or improperly set.
        """
        if metric:
            if not isinstance(self.formula, dict):
                raise ValueError("Formula must be a dictionary.")
            formula = self.formula.get(metric)
            if not formula:
                raise ValueError(f"{metric} does not have a formula")
            # Extract component IDs from the formula which are given as e.g. #123
            pattern = r"#(\d+)"
            return [int(e) for e in re.findall(pattern, self.formula[metric])]

        return self._default_cids()

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

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

        Returns:
            List of component IDs for this component.

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

        raise ValueError("No IDs available")

    @classmethod
    def is_valid_type(cls, ctype: str) -> bool:
        """Check if `ctype` is a valid enum value."""
        return ctype in get_args(ComponentType)
Attributes¤
component class-attribute instance-attribute ¤
component: list[int] | None = None

List of component IDs for this component.

formula class-attribute instance-attribute ¤
formula: dict[str, str] | None = None

Formula to calculate the power of this component.

inverter class-attribute instance-attribute ¤
inverter: list[int] | None = None

List of inverter IDs for this component.

meter class-attribute instance-attribute ¤
meter: list[int] | None = None

List of meter IDs for this component.

Methods:¤
__post_init__ ¤
__post_init__() -> None

Set the default formula if none is provided.

Source code in src/frequenz/gridpool/config/microgrid.py
def __post_init__(self) -> None:
    """Set the default formula if none is provided."""
    self.formula = self.formula or {}
    if "AC_ACTIVE_POWER" in self.formula:
        _logger.warning(
            "ComponentTypeConfig: 'AC_ACTIVE_POWER' formula is deprecated, "
            "please use 'AC_POWER_ACTIVE' instead."
        )
cids ¤
cids(metric: str = '') -> list[int]

Get component IDs for this component.

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

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

PARAMETER DESCRIPTION
metric

Metric name of the formula.

TYPE: str DEFAULT: ''

RETURNS DESCRIPTION
list[int]

List of component IDs for this component.

RAISES DESCRIPTION
ValueError

If the metric is not supported or improperly set.

Source code in src/frequenz/gridpool/config/microgrid.py
def cids(self, metric: str = "") -> list[int]:
    """Get component IDs for this component.

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

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

    Args:
        metric: Metric name of the formula.

    Returns:
        List of component IDs for this component.

    Raises:
        ValueError: If the metric is not supported or improperly set.
    """
    if metric:
        if not isinstance(self.formula, dict):
            raise ValueError("Formula must be a dictionary.")
        formula = self.formula.get(metric)
        if not formula:
            raise ValueError(f"{metric} does not have a formula")
        # Extract component IDs from the formula which are given as e.g. #123
        pattern = r"#(\d+)"
        return [int(e) for e in re.findall(pattern, self.formula[metric])]

    return self._default_cids()
is_valid_type classmethod ¤
is_valid_type(ctype: str) -> bool

Check if ctype is a valid enum value.

Source code in src/frequenz/gridpool/config/microgrid.py
@classmethod
def is_valid_type(cls, ctype: str) -> bool:
    """Check if `ctype` is a valid enum value."""
    return ctype in get_args(ComponentType)

frequenz.gridpool.config.microgrid.Metadata ¤

Metadata for a microgrid.

Source code in src/frequenz/gridpool/config/microgrid.py
@dataclass(frozen=True)
class Metadata:
    """Metadata for a microgrid."""

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

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

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

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

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

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

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

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

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

    end_time: datetime | None = None
    """End time of the microgrid operation."""
Attributes¤
altitude class-attribute instance-attribute ¤
altitude: float | None = None

Geographic altitude of the microgrid.

delivery_area class-attribute instance-attribute ¤
delivery_area: str | None = None

Delivery area of the microgrid.

end_time class-attribute instance-attribute ¤
end_time: datetime | None = None

End time of the microgrid operation.

enterprise_id class-attribute instance-attribute ¤
enterprise_id: int | None = None

Enterprise ID of the microgrid.

gid class-attribute instance-attribute ¤
gid: int | None = None

Gridpool ID of the microgrid.

latitude class-attribute instance-attribute ¤
latitude: float | None = None

Geographic latitude of the microgrid.

longitude class-attribute instance-attribute ¤
longitude: float | None = None

Geographic longitude of the microgrid.

microgrid_id instance-attribute ¤
microgrid_id: int

ID of the microgrid.

name class-attribute instance-attribute ¤
name: str | None = None

Name of the microgrid.

start_time class-attribute instance-attribute ¤
start_time: datetime | None = None

Start time of the microgrid operation.

frequenz.gridpool.config.microgrid.MicrogridConfig ¤

Configuration of a microgrid.

Source code in src/frequenz/gridpool/config/microgrid.py
@dataclass
class MicrogridConfig:
    """Configuration of a microgrid."""

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

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

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

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

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

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

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

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

        Returns:
            List of component IDs for this component type.

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

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

        return cfg.cids(metric)

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

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

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

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

        return formula

    Schema: ClassVar[Type[Schema]] = Schema

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

        Args:
            data: The loaded TOML data.

        Returns:
            A dict mapping microgrid IDs to MicrogridConfig instances.

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

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

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

            mgrids[mid] = mgrid

        return mgrids

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

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

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

        assert isinstance(data, dict)

        return cls._load_table_entries(data)
Attributes¤
battery class-attribute instance-attribute ¤
battery: dict[str, BatteryConfig] | None = None

Configuration of the batteries.

ctype class-attribute instance-attribute ¤
ctype: dict[str, ComponentTypeConfig] = field(
    default_factory=dict
)

Mapping of component category types to ac power component config.

meta instance-attribute ¤
meta: Metadata

Metadata of the microgrid.

pv class-attribute instance-attribute ¤
pv: dict[str, PVConfig] | None = None

Configuration of the PV system.

wind class-attribute instance-attribute ¤
wind: dict[str, WindConfig] | None = None

Configuration of the wind turbines.

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

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

PARAMETER DESCRIPTION
component_type

Component type to be aggregated.

TYPE: str

component_category

Specific category of component IDs to retrieve (e.g., "meter", "inverter", or "component"). If not provided, the default logic is used.

TYPE: str | None DEFAULT: None

metric

Metric name of the formula if CIDs should be extracted from the formula.

TYPE: str DEFAULT: ''

RETURNS DESCRIPTION
list[int]

List of component IDs for this component type.

RAISES DESCRIPTION
ValueError

If the component type is unknown.

KeyError

If component_category is invalid.

Source code in src/frequenz/gridpool/config/microgrid.py
def component_type_ids(
    self,
    component_type: str,
    component_category: str | None = None,
    metric: str = "",
) -> list[int]:
    """Get a list of all component IDs for a component type.

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

    Returns:
        List of component IDs for this component type.

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

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

    return cfg.cids(metric)
component_types ¤
component_types() -> list[str]

Get a list of all component types in the configuration.

Source code in src/frequenz/gridpool/config/microgrid.py
def component_types(self) -> list[str]:
    """Get a list of all component types in the configuration."""
    return list(self.ctype.keys())
formula ¤
formula(component_type: str, metric: str) -> str

Get the formula for a component type.

PARAMETER DESCRIPTION
component_type

Component type to be aggregated.

TYPE: str

metric

Metric to be aggregated.

TYPE: str

RETURNS DESCRIPTION
str

Formula to be used for this aggregated component as string.

RAISES DESCRIPTION
ValueError

If the component type is unknown or formula is missing.

Source code in src/frequenz/gridpool/config/microgrid.py
def formula(self, component_type: str, metric: str) -> str:
    """Get the formula for a component type.

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

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

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

    return formula
load_from_file classmethod ¤
load_from_file(config_path: Path) -> dict[str, Self]

Load and validate configuration settings from a TOML file.

PARAMETER DESCRIPTION
config_path

the path to the TOML configuration file.

TYPE: Path

RETURNS DESCRIPTION
dict[str, Self]

A dict mapping microgrid IDs to MicrogridConfig instances.

Source code in src/frequenz/gridpool/config/microgrid.py
@classmethod
def load_from_file(cls, config_path: Path) -> dict[str, Self]:
    """
    Load and validate configuration settings from a TOML file.

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

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

    assert isinstance(data, dict)

    return cls._load_table_entries(data)

frequenz.gridpool.config.microgrid.PVConfig ¤

Configuration of a PV system in a microgrid.

Source code in src/frequenz/gridpool/config/microgrid.py
@dataclass(frozen=True)
class PVConfig:
    """Configuration of a PV system in a microgrid."""

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

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

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

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

    curtailable: bool | None = None
    """Flag to indicate if PV system can be curtailed."""
Attributes¤
curtailable class-attribute instance-attribute ¤
curtailable: bool | None = None

Flag to indicate if PV system can be curtailed.

end_time class-attribute instance-attribute ¤
end_time: datetime | None = None

End time of the PV system installation.

peak_power class-attribute instance-attribute ¤
peak_power: float | None = None

Peak power of the PV system in Watt.

rated_power class-attribute instance-attribute ¤
rated_power: float | None = None

Rated power of the inverters in Watt.

start_time class-attribute instance-attribute ¤
start_time: datetime | None = None

Start time of the PV system installation.

frequenz.gridpool.config.microgrid.WindConfig ¤

Configuration of a wind turbine in a microgrid.

Source code in src/frequenz/gridpool/config/microgrid.py
@dataclass(frozen=True)
class WindConfig:
    # pylint: disable=too-many-instance-attributes
    """Configuration of a wind turbine in a microgrid."""

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

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

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

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

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

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

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

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

    latitude: float | None = None
    """Geographic latitude of the wind turbine."""
Attributes¤
end_time class-attribute instance-attribute ¤
end_time: datetime | None = None

End time of the wind turbine installation.

hellmann_exponent class-attribute instance-attribute ¤
hellmann_exponent: float | None = None

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

latitude class-attribute instance-attribute ¤
latitude: float | None = None

Geographic latitude of the wind turbine.

longitude class-attribute instance-attribute ¤
longitude: float | None = None

Geographic longitude of the wind turbine.

number_of_turbines class-attribute instance-attribute ¤
number_of_turbines: int = 1

Number of wind turbines.

rated_power class-attribute instance-attribute ¤
rated_power: float | None = None

Rated power of the wind turbine in Watt.

start_time class-attribute instance-attribute ¤
start_time: datetime | None = None

Start time of the wind turbine installation.

turbine_height class-attribute instance-attribute ¤
turbine_height: float | None = None

Height of the wind turbine in meters.

turbine_model class-attribute instance-attribute ¤
turbine_model: str | None = None

Model name of the wind turbine.

Functions:¤

frequenz.gridpool.config.microgrid.merge_config_maps ¤

merge_config_maps(
    base: dict[str, MicrogridConfig],
    override: dict[str, MicrogridConfig],
) -> dict[str, MicrogridConfig]

Merge two dictionaries of MicrogridConfig objects.

For microgrid IDs present in both maps the configs are merged via merge_microgrid_configs. IDs that exist only in one map are included unchanged.

PARAMETER DESCRIPTION
base

The base dictionary of MicrogridConfig objects.

TYPE: dict[str, MicrogridConfig]

override

The overriding dictionary of MicrogridConfig objects.

TYPE: dict[str, MicrogridConfig]

RETURNS DESCRIPTION
dict[str, MicrogridConfig]

A new dictionary representing the merged result.

Source code in src/frequenz/gridpool/config/microgrid.py
def merge_config_maps(
    base: dict[str, MicrogridConfig],
    override: dict[str, MicrogridConfig],
) -> dict[str, MicrogridConfig]:
    """Merge two dictionaries of `MicrogridConfig` objects.

    For microgrid IDs present in both maps the configs are merged via
    `merge_microgrid_configs`.  IDs that exist only in one map are
    included unchanged.

    Args:
        base: The base dictionary of MicrogridConfig objects.
        override: The overriding dictionary of MicrogridConfig objects.

    Returns:
        A new dictionary representing the merged result.
    """
    merged = dict(base)
    for mid, cfg in override.items():
        if mid in merged:
            merged[mid] = merge_microgrid_configs(merged[mid], cfg)
        else:
            merged[mid] = cfg
    return merged

frequenz.gridpool.config.microgrid.merge_microgrid_configs ¤

merge_microgrid_configs(
    base: MicrogridConfig, override: MicrogridConfig
) -> MicrogridConfig

Merge two MicrogridConfig objects.

The override config takes precedence over base. Nested dictionaries are merged recursively. If a field in override is None the value from base is retained, so partial overrides never nullify existing data.

PARAMETER DESCRIPTION
base

The base MicrogridConfig.

TYPE: MicrogridConfig

override

The overriding MicrogridConfig.

TYPE: MicrogridConfig

RETURNS DESCRIPTION
MicrogridConfig

A new MicrogridConfig representing the merged result.

Source code in src/frequenz/gridpool/config/microgrid.py
def merge_microgrid_configs(
    base: MicrogridConfig,
    override: MicrogridConfig,
) -> MicrogridConfig:
    """Merge two `MicrogridConfig` objects.

    The *override* config takes precedence over *base*.  Nested dictionaries
    are merged recursively.  If a field in *override* is `None` the value
    from *base* is retained, so partial overrides never nullify existing data.

    Args:
        base: The base MicrogridConfig.
        override: The overriding MicrogridConfig.

    Returns:
        A new MicrogridConfig representing the merged result.
    """
    schema = MicrogridConfig.Schema()
    base_dict = schema.dump(base)
    override_dict = schema.dump(override)

    def _deep_merge(a: dict[Any, Any], b: dict[Any, Any]) -> dict[Any, Any]:
        result = deepcopy(a)
        for k, v in b.items():
            if v is None:
                continue
            if isinstance(v, dict) and isinstance(result.get(k), dict):
                result[k] = _deep_merge(result[k], v)
            else:
                result[k] = v
        return result

    merged = schema.load(_deep_merge(base_dict, override_dict))
    assert isinstance(merged, MicrogridConfig)
    return merged