Index

frequenz.data.microgrid

Initialize the microgrid data module.

Classes

frequenz.data.microgrid.MicrogridConfig

Configuration of a microgrid.

Source code in 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 = None
    """Start time of the microgrid operation."""

    end_time: datetime | None = None
    """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 = 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

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 = None

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 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 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 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.data.microgrid.MicrogridData

Fetch power data for component types of a microgrid.

Source code in src/frequenz/data/microgrid/component_data.py

class MicrogridData:
    """Fetch power data for component types of a microgrid."""

    def __init__(
        self,
        server_url: str,
        auth_key: str,
        sign_secret: str,
        microgrid_configs: (
            _MicrogridConfigMapping | _MicrogridConfigsContainer | None
        ) = None,
    ) -> None:
        """Initialize microgrid data.

        Args:
            server_url: URL of the reporting service.
            auth_key: Authentication key to the service.
            sign_secret: Secret for signing requests.
            microgrid_configs: A mapping of microgrid IDs to configurations, or a
                gridpool configuration document containing that mapping.
        """
        self._microgrid_configs = (
            None
            if microgrid_configs is None
            else _normalize_microgrid_configs(microgrid_configs)
        )
        self._client = ReportingApiClient(
            server_url=server_url, auth_key=auth_key, sign_secret=sign_secret
        )

    @property
    def microgrid_ids(self) -> list[int]:
        """Get the microgrid IDs.

        Returns:
            List of microgrid IDs.
        """
        if self._microgrid_configs is None:
            return []
        return list(self._microgrid_configs.keys())

    @property
    def microgrid_configs(self) -> dict[int, MicrogridConfig] | None:
        """Return the microgrid configurations."""
        return self._microgrid_configs

    # pylint: disable=too-many-locals
    async def metric_data(  # pylint: disable=too-many-arguments
        self,
        *,
        microgrid_id: int | str,
        start: datetime,
        end: datetime,
        component_types: tuple[str, ...] = ("grid", "pv", "wind", "battery"),
        resampling_period: timedelta = timedelta(seconds=10),
        metric: str = "AC_POWER_ACTIVE",
        keep_components: bool = False,
        splits: bool = False,
    ) -> pd.DataFrame | None:
        """Power data for component types of a microgrid.

        Args:
            microgrid_id: Microgrid ID. Numeric strings are accepted for notebook
                compatibility.
            start: Start timestamp.
            end: End timestamp.
            component_types: List of component types to be aggregated.
            resampling_period: Data resampling period.
            metric: Metric to be fetched.
            keep_components: Include individual components in output.
            splits: Include columns for positive and negative power values for components.

        Returns:
            DataFrame with power data of aggregated components
            or None if no data is available

        Raises:
            ValueError: If microgrid configurations are not loaded.
        """
        if self._microgrid_configs is None:
            raise ValueError("Microgrid configurations are not loaded.")
        microgrid_id = int(microgrid_id)
        mcfg = self._microgrid_configs[microgrid_id]

        formulas = {
            ctype: mcfg.formula(ctype, metric.upper()) for ctype in component_types
        }

        logging.debug("Formulas: %s", formulas)

        metric_enum = Metric[metric.upper()]
        data = [
            sample
            for ctype, formula in formulas.items()
            async for sample in self._client.receive_aggregated_data(
                microgrid_id=microgrid_id,
                metric=metric_enum,
                aggregation_formula=formula,
                start_time=start,
                end_time=end,
                resampling_period=resampling_period,
            )
        ]

        all_cids = []
        if keep_components:
            all_cids = [
                cid
                for ctype in component_types
                for cid in mcfg.component_type_ids(ctype, metric=metric)
            ]
            _logger.debug("CIDs: %s", all_cids)
            microgrid_components = [
                (microgrid_id, all_cids),
            ]
            data_comp = [
                sample
                async for sample in self._client.receive_microgrid_components_data(
                    microgrid_components=microgrid_components,
                    metrics=metric_enum,
                    start_time=start,
                    end_time=end,
                    resampling_period=resampling_period,
                )
            ]
            data.extend(data_comp)

        if len(data) == 0:
            _logger.warning("No data found")
            return None

        df = pd.DataFrame(data)
        df["timestamp"] = pd.to_datetime(df["timestamp"], utc=True)
        assert df["timestamp"].dt.tz is not None, "Timestamps are not tz-aware"

        # Remove duplicates
        dup_mask = df.duplicated(keep="first")
        if not dup_mask.empty:
            _logger.info("Found %s rows that have duplicates", dup_mask.sum())
        df = df[~dup_mask]

        # Pivot table
        df = df.pivot_table(index="timestamp", columns="component_id", values="value")
        # Rename formula columns
        rename_cols: dict[str, str] = {}
        for ctype, formula in formulas.items():
            if formula in rename_cols:
                _logger.warning(
                    "Ignoring %s since formula %s exists already for %s",
                    ctype,
                    formula,
                    rename_cols[formula],
                )
                continue
            rename_cols[formula] = ctype

        df = df.rename(columns=rename_cols)
        if keep_components:
            # Set missing columns to NaN
            for cid in all_cids:
                if cid not in df.columns:
                    _logger.warning(
                        "Component ID %s not found in data, setting zero", cid
                    )
                    df.loc[:, cid] = np.nan

        # Make string columns
        df.columns = [str(e) for e in df.columns]  # type: ignore

        cols = df.columns
        if splits:
            pos_cols = [f"{col}_pos" for col in cols]
            neg_cols = [f"{col}_neg" for col in cols]
            df[pos_cols] = df[cols].clip(lower=0)
            df[neg_cols] = df[cols].clip(upper=0)

        # Sort columns
        ctypes = list(rename_cols.values())
        new_cols = [e for e in ctypes if e in df.columns] + sorted(
            [e for e in df.columns if e not in ctypes]
        )
        df = df[new_cols]

        return df

    async def ac_active_power(  # pylint: disable=too-many-arguments
        self,
        *,
        microgrid_id: int | str,
        start: datetime,
        end: datetime,
        component_types: tuple[str, ...] = ("grid", "pv", "wind", "battery"),
        resampling_period: timedelta = timedelta(seconds=10),
        keep_components: bool = False,
        splits: bool = False,
        unit: str = "kW",
    ) -> pd.DataFrame | None:
        """Power data for component types of a microgrid."""
        df = await self.metric_data(
            microgrid_id=microgrid_id,
            start=start,
            end=end,
            component_types=component_types,
            resampling_period=resampling_period,
            metric="AC_POWER_ACTIVE",
            keep_components=keep_components,
            splits=splits,
        )
        if df is None:
            return df

        if unit == "W":
            pass
        if unit == "kW":
            df = df / 1000
        elif unit == "MW":
            df = df / 1e6
        else:
            raise ValueError(f"Unknown unit: {unit}")
        return df

    async def soc(  # pylint: disable=too-many-arguments
        self,
        *,
        microgrid_id: int | str,
        start: datetime,
        end: datetime,
        resampling_period: timedelta = timedelta(seconds=10),
        keep_components: bool = False,
    ) -> pd.DataFrame | None:
        """Soc data for component types of a microgrid."""
        df = await self.metric_data(
            microgrid_id=microgrid_id,
            start=start,
            end=end,
            component_types=("battery",),
            resampling_period=resampling_period,
            metric="BATTERY_SOC_PCT",
            keep_components=keep_components,
        )
        return df

Attributes

microgrid_configs property

microgrid_configs: dict[int, MicrogridConfig] | None

Return the microgrid configurations.

microgrid_ids property

microgrid_ids: list[int]

Get the microgrid IDs.

RETURNS	DESCRIPTION
list[int]	List of microgrid IDs.

Methods:

__init__

__init__(
    server_url: str,
    auth_key: str,
    sign_secret: str,
    microgrid_configs: (
        _MicrogridConfigMapping
        | _MicrogridConfigsContainer
        | None
    ) = None,
) -> None

Initialize microgrid data.

PARAMETER	DESCRIPTION
server_url	URL of the reporting service. TYPE: str
auth_key	Authentication key to the service. TYPE: str
sign_secret	Secret for signing requests. TYPE: str
microgrid_configs	A mapping of microgrid IDs to configurations, or a gridpool configuration document containing that mapping. TYPE: _MicrogridConfigMapping | _MicrogridConfigsContainer | None DEFAULT: None

Source code in src/frequenz/data/microgrid/component_data.py

def __init__(
    self,
    server_url: str,
    auth_key: str,
    sign_secret: str,
    microgrid_configs: (
        _MicrogridConfigMapping | _MicrogridConfigsContainer | None
    ) = None,
) -> None:
    """Initialize microgrid data.

    Args:
        server_url: URL of the reporting service.
        auth_key: Authentication key to the service.
        sign_secret: Secret for signing requests.
        microgrid_configs: A mapping of microgrid IDs to configurations, or a
            gridpool configuration document containing that mapping.
    """
    self._microgrid_configs = (
        None
        if microgrid_configs is None
        else _normalize_microgrid_configs(microgrid_configs)
    )
    self._client = ReportingApiClient(
        server_url=server_url, auth_key=auth_key, sign_secret=sign_secret
    )

ac_active_power async

ac_active_power(
    *,
    microgrid_id: int | str,
    start: datetime,
    end: datetime,
    component_types: tuple[str, ...] = (
        "grid",
        "pv",
        "wind",
        "battery",
    ),
    resampling_period: timedelta = timedelta(seconds=10),
    keep_components: bool = False,
    splits: bool = False,
    unit: str = "kW"
) -> DataFrame | None

Power data for component types of a microgrid.

Source code in src/frequenz/data/microgrid/component_data.py

async def ac_active_power(  # pylint: disable=too-many-arguments
    self,
    *,
    microgrid_id: int | str,
    start: datetime,
    end: datetime,
    component_types: tuple[str, ...] = ("grid", "pv", "wind", "battery"),
    resampling_period: timedelta = timedelta(seconds=10),
    keep_components: bool = False,
    splits: bool = False,
    unit: str = "kW",
) -> pd.DataFrame | None:
    """Power data for component types of a microgrid."""
    df = await self.metric_data(
        microgrid_id=microgrid_id,
        start=start,
        end=end,
        component_types=component_types,
        resampling_period=resampling_period,
        metric="AC_POWER_ACTIVE",
        keep_components=keep_components,
        splits=splits,
    )
    if df is None:
        return df

    if unit == "W":
        pass
    if unit == "kW":
        df = df / 1000
    elif unit == "MW":
        df = df / 1e6
    else:
        raise ValueError(f"Unknown unit: {unit}")
    return df

metric_data async

metric_data(
    *,
    microgrid_id: int | str,
    start: datetime,
    end: datetime,
    component_types: tuple[str, ...] = (
        "grid",
        "pv",
        "wind",
        "battery",
    ),
    resampling_period: timedelta = timedelta(seconds=10),
    metric: str = "AC_POWER_ACTIVE",
    keep_components: bool = False,
    splits: bool = False
) -> DataFrame | None

Power data for component types of a microgrid.

PARAMETER	DESCRIPTION
microgrid_id	Microgrid ID. Numeric strings are accepted for notebook compatibility. TYPE: int | str
start	Start timestamp. TYPE: datetime
end	End timestamp. TYPE: datetime
component_types	List of component types to be aggregated. TYPE: tuple[str, ...] DEFAULT: ('grid', 'pv', 'wind', 'battery')
resampling_period	Data resampling period. TYPE: timedelta DEFAULT: timedelta(seconds=10)
metric	Metric to be fetched. TYPE: str DEFAULT: 'AC_POWER_ACTIVE'
keep_components	Include individual components in output. TYPE: bool DEFAULT: False
splits	Include columns for positive and negative power values for components. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
DataFrame | None	DataFrame with power data of aggregated components
DataFrame | None	or None if no data is available

RAISES	DESCRIPTION
ValueError	If microgrid configurations are not loaded.

Source code in src/frequenz/data/microgrid/component_data.py

async def metric_data(  # pylint: disable=too-many-arguments
    self,
    *,
    microgrid_id: int | str,
    start: datetime,
    end: datetime,
    component_types: tuple[str, ...] = ("grid", "pv", "wind", "battery"),
    resampling_period: timedelta = timedelta(seconds=10),
    metric: str = "AC_POWER_ACTIVE",
    keep_components: bool = False,
    splits: bool = False,
) -> pd.DataFrame | None:
    """Power data for component types of a microgrid.

    Args:
        microgrid_id: Microgrid ID. Numeric strings are accepted for notebook
            compatibility.
        start: Start timestamp.
        end: End timestamp.
        component_types: List of component types to be aggregated.
        resampling_period: Data resampling period.
        metric: Metric to be fetched.
        keep_components: Include individual components in output.
        splits: Include columns for positive and negative power values for components.

    Returns:
        DataFrame with power data of aggregated components
        or None if no data is available

    Raises:
        ValueError: If microgrid configurations are not loaded.
    """
    if self._microgrid_configs is None:
        raise ValueError("Microgrid configurations are not loaded.")
    microgrid_id = int(microgrid_id)
    mcfg = self._microgrid_configs[microgrid_id]

    formulas = {
        ctype: mcfg.formula(ctype, metric.upper()) for ctype in component_types
    }

    logging.debug("Formulas: %s", formulas)

    metric_enum = Metric[metric.upper()]
    data = [
        sample
        for ctype, formula in formulas.items()
        async for sample in self._client.receive_aggregated_data(
            microgrid_id=microgrid_id,
            metric=metric_enum,
            aggregation_formula=formula,
            start_time=start,
            end_time=end,
            resampling_period=resampling_period,
        )
    ]

    all_cids = []
    if keep_components:
        all_cids = [
            cid
            for ctype in component_types
            for cid in mcfg.component_type_ids(ctype, metric=metric)
        ]
        _logger.debug("CIDs: %s", all_cids)
        microgrid_components = [
            (microgrid_id, all_cids),
        ]
        data_comp = [
            sample
            async for sample in self._client.receive_microgrid_components_data(
                microgrid_components=microgrid_components,
                metrics=metric_enum,
                start_time=start,
                end_time=end,
                resampling_period=resampling_period,
            )
        ]
        data.extend(data_comp)

    if len(data) == 0:
        _logger.warning("No data found")
        return None

    df = pd.DataFrame(data)
    df["timestamp"] = pd.to_datetime(df["timestamp"], utc=True)
    assert df["timestamp"].dt.tz is not None, "Timestamps are not tz-aware"

    # Remove duplicates
    dup_mask = df.duplicated(keep="first")
    if not dup_mask.empty:
        _logger.info("Found %s rows that have duplicates", dup_mask.sum())
    df = df[~dup_mask]

    # Pivot table
    df = df.pivot_table(index="timestamp", columns="component_id", values="value")
    # Rename formula columns
    rename_cols: dict[str, str] = {}
    for ctype, formula in formulas.items():
        if formula in rename_cols:
            _logger.warning(
                "Ignoring %s since formula %s exists already for %s",
                ctype,
                formula,
                rename_cols[formula],
            )
            continue
        rename_cols[formula] = ctype

    df = df.rename(columns=rename_cols)
    if keep_components:
        # Set missing columns to NaN
        for cid in all_cids:
            if cid not in df.columns:
                _logger.warning(
                    "Component ID %s not found in data, setting zero", cid
                )
                df.loc[:, cid] = np.nan

    # Make string columns
    df.columns = [str(e) for e in df.columns]  # type: ignore

    cols = df.columns
    if splits:
        pos_cols = [f"{col}_pos" for col in cols]
        neg_cols = [f"{col}_neg" for col in cols]
        df[pos_cols] = df[cols].clip(lower=0)
        df[neg_cols] = df[cols].clip(upper=0)

    # Sort columns
    ctypes = list(rename_cols.values())
    new_cols = [e for e in ctypes if e in df.columns] + sorted(
        [e for e in df.columns if e not in ctypes]
    )
    df = df[new_cols]

    return df

soc async

soc(
    *,
    microgrid_id: int | str,
    start: datetime,
    end: datetime,
    resampling_period: timedelta = timedelta(seconds=10),
    keep_components: bool = False
) -> DataFrame | None

Soc data for component types of a microgrid.

Source code in src/frequenz/data/microgrid/component_data.py

async def soc(  # pylint: disable=too-many-arguments
    self,
    *,
    microgrid_id: int | str,
    start: datetime,
    end: datetime,
    resampling_period: timedelta = timedelta(seconds=10),
    keep_components: bool = False,
) -> pd.DataFrame | None:
    """Soc data for component types of a microgrid."""
    df = await self.metric_data(
        microgrid_id=microgrid_id,
        start=start,
        end=end,
        component_types=("battery",),
        resampling_period=resampling_period,
        metric="BATTERY_SOC_PCT",
        keep_components=keep_components,
    )
    return df

frequenz.data.microgrid.StatefulDataFetcher

A helper class to handle fetching of microgrid component data.

This class provides methods to query new data for a specific microgrid and its components, and to write the updated data to a temporary file. The file is written only temporarily and needs to be committed or rolled back in case of success or failure, respectively.

Source code in src/frequenz/data/microgrid/_stateful_data_fetcher.py

class StatefulDataFetcher:
    """A helper class to handle fetching of microgrid component data.

    This class provides methods to query new data for a specific microgrid and
    its components, and to write the updated data to a temporary file.
    The file is written only temporarily and needs to be committed or rolled back
    in case of success or failure, respectively.
    """

    def __init__(  # pylint: disable=too-many-arguments, too-many-positional-arguments
        self,
        microgrid_data: MicrogridData,
        data_buffer_dir: Path,
        resampling_period: timedelta,
        end_time_delta: timedelta = END_TIME_DELTA,
        initial_period: timedelta = timedelta(hours=1),
    ) -> None:
        """Initialize the TransactionalDataExport class.

        Args:
            microgrid_data: An instance of MicrogridData to query data from.
            data_buffer_dir: The path to the directory for buffering data.
            resampling_period: The period to resample the data.
            end_time_delta: The time subtracted from now to be passed as
                            end_time to the API.
            initial_period: The initial period to use for the first data fetch.
        """
        _logger.debug("Initializing StatefulDataFetcher instance.")
        self._microgrid_data = microgrid_data
        self._data_buffer = data_buffer_dir
        self._resampling_period = resampling_period
        self._end_time_delta = end_time_delta
        self._initial_period = initial_period

        self._data_buffer.mkdir(parents=True, exist_ok=True)

        # Maps temp_path -> final_path for files that need to be committed
        self._temp_files_to_commit: dict[Path, Path] = {}

    @staticmethod
    def _generate_filename(
        microgrid_id: int, components: tuple[str, ...], metric: str
    ) -> str:
        """Generate a unique, sanitized filename.

        Args:
            microgrid_id: The ID of the microgrid.
            components: The component types to include in the filename.
            metric: The name of the metric to include in the filename.

        Returns:
            A sanitized string suitable for use as a filename.
        """
        sorted_components = sorted([c.lower().replace(" ", "_") for c in components])
        return (
            f"mid{microgrid_id}_{'_'.join(sorted_components)}_{metric.lower()}.parquet"
        )

    async def receive_microgrid_data(
        self,
        microgrid_id: int,
        components: tuple[str, ...],
        metric: str,
    ) -> pd.DataFrame | None:
        """Query new microgrid data and write the updated buffer to a temp file.

        Reads the last timestamp from the main buffer file, queries for new data,
        and writes the combined data to a new temporary buffer file.

        Args:
            microgrid_id: The ID of the microgrid to query.
            components: The component types to include in the query.
            metric: The metric to query.

        Returns:
            A pandas DataFrame with only the new data points, or None.
        """
        _logger.debug(
            "Processing data for microgrid ID %s, components %s",
            microgrid_id,
            components,
        )

        final_parquet_file = self._data_buffer / self._generate_filename(
            microgrid_id, components, metric
        )
        temp_parquet_file = final_parquet_file.with_suffix(
            final_parquet_file.suffix + ".tmp"
        )

        now = datetime.now(timezone.utc)
        end_time = now - self._end_time_delta
        on_disk_buffer = pd.DataFrame()
        start_time: datetime

        try:
            on_disk_buffer = pd.read_parquet(final_parquet_file)
            if not on_disk_buffer.empty and isinstance(
                on_disk_buffer.index, pd.DatetimeIndex
            ):
                last_timestamp = on_disk_buffer.index.max().to_pydatetime()
                start_time = last_timestamp + timedelta(microseconds=1)
                _logger.info("Read buffer. Last timestamp: %s.", last_timestamp)
            else:
                _logger.info(
                    "Buffer file is empty. Defaulting to: %s.", self._initial_period
                )
                start_time = now - self._initial_period
        except FileNotFoundError:
            _logger.info(
                "No buffer file at '%s'. Defaulting to %s.",
                final_parquet_file,
                self._initial_period,
            )
            start_time = now - self._initial_period

        if start_time >= end_time:
            _logger.info("Buffer is up to date. No new data to fetch.")
            return None

        _logger.info("Fetching new data from %s to %s...", start_time, end_time)

        new_df = await self._microgrid_data.metric_data(
            microgrid_id=microgrid_id,
            start=start_time,
            end=end_time,
            component_types=components,
            resampling_period=self._resampling_period,
            metric=metric,
        )

        if new_df is not None and not new_df.empty:
            _logger.info("Fetched %d new data points.", len(new_df))
            combined_df = pd.concat([on_disk_buffer, new_df])
            combined_df = combined_df[~combined_df.index.duplicated(keep="last")]
            combined_df.sort_index(inplace=True)
            updated_buffer = combined_df.tail(BUFFER_SIZE)

            # Write to temporary file instead of final destination
            updated_buffer.to_parquet(temp_parquet_file, engine="pyarrow")

            _logger.info(
                "Wrote updated buffer with %d points to temporary file '%s'.",
                len(updated_buffer),
                temp_parquet_file,
            )

            self._temp_files_to_commit[temp_parquet_file] = final_parquet_file
        else:
            _logger.info("No new data was returned from the API.")

        return new_df

    def commit(self) -> None:
        """Commit the temporary files to their final locations."""
        _logger.info("Reports sent successfully. Committing buffer updates.")
        for temp_path, final_path in self._temp_files_to_commit.items():
            try:
                os.replace(temp_path, final_path)
                _logger.debug(
                    "Committed buffer update: %s -> %s", temp_path, final_path
                )
            except OSError as e:
                _logger.error(
                    "Failed to commit buffer update from %s to %s: %s",
                    temp_path,
                    final_path,
                    e,
                )

        self._temp_files_to_commit.clear()

    def rollback(self) -> None:
        """Rollback the temporary files if the transaction fails."""
        for temp_path in self._temp_files_to_commit:
            try:
                temp_path.unlink()
                _logger.debug("Rolled back by deleting temp file: %s", temp_path)
            except OSError as unlink_e:
                _logger.error(
                    "Failed to clean up temporary buffer file %s during rollback: %s",
                    temp_path,
                    unlink_e,
                )

        self._temp_files_to_commit.clear()

Methods:

__init__

__init__(
    microgrid_data: MicrogridData,
    data_buffer_dir: Path,
    resampling_period: timedelta,
    end_time_delta: timedelta = END_TIME_DELTA,
    initial_period: timedelta = timedelta(hours=1),
) -> None

Initialize the TransactionalDataExport class.

PARAMETER	DESCRIPTION
microgrid_data	An instance of MicrogridData to query data from. TYPE: MicrogridData
data_buffer_dir	The path to the directory for buffering data. TYPE: Path
resampling_period	The period to resample the data. TYPE: timedelta
end_time_delta	The time subtracted from now to be passed as end_time to the API. TYPE: timedelta DEFAULT: END_TIME_DELTA
initial_period	The initial period to use for the first data fetch. TYPE: timedelta DEFAULT: timedelta(hours=1)

Source code in src/frequenz/data/microgrid/_stateful_data_fetcher.py

def __init__(  # pylint: disable=too-many-arguments, too-many-positional-arguments
    self,
    microgrid_data: MicrogridData,
    data_buffer_dir: Path,
    resampling_period: timedelta,
    end_time_delta: timedelta = END_TIME_DELTA,
    initial_period: timedelta = timedelta(hours=1),
) -> None:
    """Initialize the TransactionalDataExport class.

    Args:
        microgrid_data: An instance of MicrogridData to query data from.
        data_buffer_dir: The path to the directory for buffering data.
        resampling_period: The period to resample the data.
        end_time_delta: The time subtracted from now to be passed as
                        end_time to the API.
        initial_period: The initial period to use for the first data fetch.
    """
    _logger.debug("Initializing StatefulDataFetcher instance.")
    self._microgrid_data = microgrid_data
    self._data_buffer = data_buffer_dir
    self._resampling_period = resampling_period
    self._end_time_delta = end_time_delta
    self._initial_period = initial_period

    self._data_buffer.mkdir(parents=True, exist_ok=True)

    # Maps temp_path -> final_path for files that need to be committed
    self._temp_files_to_commit: dict[Path, Path] = {}

commit

commit() -> None

Commit the temporary files to their final locations.

Source code in src/frequenz/data/microgrid/_stateful_data_fetcher.py

def commit(self) -> None:
    """Commit the temporary files to their final locations."""
    _logger.info("Reports sent successfully. Committing buffer updates.")
    for temp_path, final_path in self._temp_files_to_commit.items():
        try:
            os.replace(temp_path, final_path)
            _logger.debug(
                "Committed buffer update: %s -> %s", temp_path, final_path
            )
        except OSError as e:
            _logger.error(
                "Failed to commit buffer update from %s to %s: %s",
                temp_path,
                final_path,
                e,
            )

    self._temp_files_to_commit.clear()

receive_microgrid_data async

receive_microgrid_data(
    microgrid_id: int,
    components: tuple[str, ...],
    metric: str,
) -> DataFrame | None

Query new microgrid data and write the updated buffer to a temp file.

Reads the last timestamp from the main buffer file, queries for new data, and writes the combined data to a new temporary buffer file.

PARAMETER	DESCRIPTION
microgrid_id	The ID of the microgrid to query. TYPE: int
components	The component types to include in the query. TYPE: tuple[str, ...]
metric	The metric to query. TYPE: str

RETURNS	DESCRIPTION
DataFrame | None	A pandas DataFrame with only the new data points, or None.

Source code in src/frequenz/data/microgrid/_stateful_data_fetcher.py

async def receive_microgrid_data(
    self,
    microgrid_id: int,
    components: tuple[str, ...],
    metric: str,
) -> pd.DataFrame | None:
    """Query new microgrid data and write the updated buffer to a temp file.

    Reads the last timestamp from the main buffer file, queries for new data,
    and writes the combined data to a new temporary buffer file.

    Args:
        microgrid_id: The ID of the microgrid to query.
        components: The component types to include in the query.
        metric: The metric to query.

    Returns:
        A pandas DataFrame with only the new data points, or None.
    """
    _logger.debug(
        "Processing data for microgrid ID %s, components %s",
        microgrid_id,
        components,
    )

    final_parquet_file = self._data_buffer / self._generate_filename(
        microgrid_id, components, metric
    )
    temp_parquet_file = final_parquet_file.with_suffix(
        final_parquet_file.suffix + ".tmp"
    )

    now = datetime.now(timezone.utc)
    end_time = now - self._end_time_delta
    on_disk_buffer = pd.DataFrame()
    start_time: datetime

    try:
        on_disk_buffer = pd.read_parquet(final_parquet_file)
        if not on_disk_buffer.empty and isinstance(
            on_disk_buffer.index, pd.DatetimeIndex
        ):
            last_timestamp = on_disk_buffer.index.max().to_pydatetime()
            start_time = last_timestamp + timedelta(microseconds=1)
            _logger.info("Read buffer. Last timestamp: %s.", last_timestamp)
        else:
            _logger.info(
                "Buffer file is empty. Defaulting to: %s.", self._initial_period
            )
            start_time = now - self._initial_period
    except FileNotFoundError:
        _logger.info(
            "No buffer file at '%s'. Defaulting to %s.",
            final_parquet_file,
            self._initial_period,
        )
        start_time = now - self._initial_period

    if start_time >= end_time:
        _logger.info("Buffer is up to date. No new data to fetch.")
        return None

    _logger.info("Fetching new data from %s to %s...", start_time, end_time)

    new_df = await self._microgrid_data.metric_data(
        microgrid_id=microgrid_id,
        start=start_time,
        end=end_time,
        component_types=components,
        resampling_period=self._resampling_period,
        metric=metric,
    )

    if new_df is not None and not new_df.empty:
        _logger.info("Fetched %d new data points.", len(new_df))
        combined_df = pd.concat([on_disk_buffer, new_df])
        combined_df = combined_df[~combined_df.index.duplicated(keep="last")]
        combined_df.sort_index(inplace=True)
        updated_buffer = combined_df.tail(BUFFER_SIZE)

        # Write to temporary file instead of final destination
        updated_buffer.to_parquet(temp_parquet_file, engine="pyarrow")

        _logger.info(
            "Wrote updated buffer with %d points to temporary file '%s'.",
            len(updated_buffer),
            temp_parquet_file,
        )

        self._temp_files_to_commit[temp_parquet_file] = final_parquet_file
    else:
        _logger.info("No new data was returned from the API.")

    return new_df

rollback

rollback() -> None

Rollback the temporary files if the transaction fails.

Source code in src/frequenz/data/microgrid/_stateful_data_fetcher.py

def rollback(self) -> None:
    """Rollback the temporary files if the transaction fails."""
    for temp_path in self._temp_files_to_commit:
        try:
            temp_path.unlink()
            _logger.debug("Rolled back by deleting temp file: %s", temp_path)
        except OSError as unlink_e:
            _logger.error(
                "Failed to clean up temporary buffer file %s during rollback: %s",
                temp_path,
                unlink_e,
            )

    self._temp_files_to_commit.clear()

Functions:

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