helpers

frequenz.lib.notebooks.reporting.utils.helpers

Energy flow and configuration utilities for microgrid analysis and reporting.

This module provides helper functions for

• Safe numeric extraction and aggregation from pandas DataFrames.
• Loading and validating YAML-based configuration files.
• Labeling microgrid component columns using configuration metadata.
• Computing derived energy flow metrics such as:
  • Production excess
  • Battery charge utilization
  • Grid feed-in
  • Self-consumption and self-share
• Formatting and timezone conversion utilities for reporting.

These utilities are primarily used in energy analytics pipelines and microgrid reporting notebooks to ensure consistent data preprocessing, metric calculation, and standardized output structures.

FUNCTION	DESCRIPTION
_get_numeric_series	Safely extract a numeric Series or return zeros if missing.
_sum_cols	Safely sum multiple numeric columns.
load_config	Load and validate a YAML configuration file.
_fmt_to_de_system	Format numbers using German-style decimal conventions.
_convert_timezone	Convert a DataFrame timestamp column to a target timezone.
label_component_columns	Rename numeric component columns using MicrogridConfig.
get_energy_report_columns	Determine relevant columns for energy reporting.
add_energy_flows	Compute derived production, battery, and grid metrics.

Notes

• These helpers are designed for internal use and assume well-structured DataFrames with datetime indices or timestamp columns.
• All numeric outputs are returned as float64 Series to ensure consistency.

Classes

Functions:

frequenz.lib.notebooks.reporting.utils.helpers.add_energy_flows

add_energy_flows(
    df: DataFrame,
    production_cols: list[str] | None = None,
    consumption_cols: list[str] | None = None,
    grid_cols: list[str] | None = None,
    battery_cols: list[str] | None = None,
) -> DataFrame

Compute and add derived energy flow metrics to the DataFrame.

This function aggregates production and consumption data, derives energy flow relationships such as grid feed-in, battery charging, and self-consumption, and appends these computed columns to the given DataFrame. Columns that are specified but missing or contain only null/zero values are ignored.

PARAMETER	DESCRIPTION
df	Input DataFrame containing production, consumption, and optionally battery power data. TYPE: DataFrame
production_cols	list of column names representing production sources. TYPE: list[str] | None DEFAULT: None
consumption_cols	list of column names representing consumption sources. TYPE: list[str] | None DEFAULT: None
grid_cols	list of column names representing grid import/export. TYPE: list[str] | None DEFAULT: None
battery_cols	optional list of column names for battery charging power. If None, battery-related flows are set to zero. TYPE: list[str] | None DEFAULT: None

RETURNS

DESCRIPTION

DataFrame

A DataFrame including additional columns: - "production_excess": Production exceeding consumption. - "grid_feed_in": Portion of excess fed into the grid. - "production_self_use": Self-consumed portion of production. - "production_to_battery": Production energy sent to the battery. - "grid_to_battery": Grid energy sent to the battery. - "battery_to_grid": Battery discharge exported to the grid. - "battery_to_consumption": Battery discharge used by consumption. - "production_self_share": Share of production that is self-consumed (self-consumed / total production). - "production_self_usage": Share of consumption covered by self-production (self-consumed / total consumption).

Source code in src/frequenz/lib/notebooks/reporting/utils/helpers.py

def add_energy_flows(
    df: pd.DataFrame,
    production_cols: list[str] | None = None,
    consumption_cols: list[str] | None = None,
    grid_cols: list[str] | None = None,
    battery_cols: list[str] | None = None,
) -> pd.DataFrame:
    """Compute and add derived energy flow metrics to the DataFrame.

    This function aggregates production and consumption data, derives energy flow
    relationships such as grid feed-in, battery charging, and self-consumption,
    and appends these computed columns to the given DataFrame. Columns that are
    specified but missing or contain only null/zero values are ignored.

    Args:
        df: Input DataFrame containing production, consumption, and optionally
            battery power data.
        production_cols: list of column names representing production sources.
        consumption_cols: list of column names representing consumption sources.
        grid_cols: list of column names representing grid import/export.
        battery_cols: optional list of column names for battery charging power. If None,
            battery-related flows are set to zero.

    Returns:
        A DataFrame including additional columns:
            - "production_excess": Production exceeding consumption.
            - "grid_feed_in": Portion of excess fed into the grid.
            - "production_self_use": Self-consumed portion of production.
            - "production_to_battery": Production energy sent to the battery.
            - "grid_to_battery": Grid energy sent to the battery.
            - "battery_to_grid": Battery discharge exported to the grid.
            - "battery_to_consumption": Battery discharge used by consumption.
            - "production_self_share": Share of production that is self-consumed
              (self-consumed / total production).
            - "production_self_usage": Share of consumption covered by
              self-production (self-consumed / total consumption).
    """
    df_flows = df.copy()

    # Normalize production, consumption and battery columns by removing None entries
    resolved_production_cols = [
        col for col in (production_cols or []) if _column_has_data(df_flows, col)
    ]
    resolved_consumption_cols = [
        col for col in (consumption_cols or []) if _column_has_data(df_flows, col)
    ]
    resolved_grid_cols = [
        col for col in (grid_cols or []) if _column_has_data(df_flows, col)
    ]
    resolved_battery_cols = [
        col for col in (battery_cols or []) if _column_has_data(df_flows, col)
    ]

    battery_power_series = _sum_cols(df_flows, resolved_battery_cols)
    grid_power_series = _sum_cols(df_flows, resolved_grid_cols)

    # Compute total asset production
    asset_production_cols: list[str] = []
    for col in resolved_production_cols:
        series = _get_numeric_series(
            df_flows,
            col,
        )
        asset_series = asset_production(
            series,
        )
        asset_col_name = f"{col}_asset_production"
        df_flows[asset_col_name] = asset_series
        asset_production_cols.append(asset_col_name)

    df_flows["production_total"] = _sum_cols(df_flows, asset_production_cols)

    # Compute total consumption
    consumption_series_cols: list[str] = []
    for col in resolved_consumption_cols:
        df_flows[col] = _get_numeric_series(df_flows, col)
        consumption_series_cols.append(col)

    if resolved_consumption_cols:
        df_flows["consumption_total"] = _sum_cols(df_flows, consumption_series_cols)
    else:
        # When no explicit consumption columns exist, infer it from grid info.
        df_flows["consumption_total"] = consumption(
            grid_power_series,
            production=_sum_cols(
                df_flows, resolved_production_cols
            ),  # raw production to have the negative production values
            battery=battery_power_series,
        )

        # Set consumption to total consumption if no explicit consumption columns exist
        df_flows["consumption"] = df_flows["consumption_total"].copy()

    # Surplus vs. consumption (production is already positive because of the above cleaning)
    df_flows["production_excess"] = production_excess(
        df_flows["production_total"] * -1,
        df_flows["consumption_total"],
    )

    # Split excess into battery vs. grid
    df_flows["grid_feed_in"] = grid_feed_in(
        # To convert positive production back to PSC format (where production is negative)
        df_flows["production_total"] * -1,
        df_flows["consumption_total"],
        battery=battery_power_series,
        grid=grid_power_series if resolved_grid_cols else None,
    )

    # If no production columns exist, set self-consumption metrics to zero
    if asset_production_cols:
        # Use total production for self-consumption instead of asset_production
        # (which may not exist)
        df_flows["production_self_use"] = production_self_consumption(
            df_flows["production_total"] * -1,
            df_flows["consumption_total"],
        )
        df_flows["production_self_share"] = production_self_share(
            df_flows["production_total"] * -1,
            df_flows["consumption_total"],
        )
        df_flows["production_self_usage"] = production_self_usage(
            df_flows["production_total"] * -1,
            df_flows["consumption_total"],
        )
    else:
        df_flows["production_self_use"] = 0.0
        df_flows["production_self_share"] = 0.0
        df_flows["production_self_usage"] = 0.0

    # Add grid consumption column - grid is later renamed as grid_consumption
    if "grid" not in df_flows.columns:
        df_flows["grid"] = grid_consumption(
            grid_power_series,
            # To convert positive production back to PSC format (where production is negative)
            df_flows["production_total"] * -1,
            df_flows["consumption_total"],
            battery_power_series,
        )

    (
        df_flows["production_to_battery"],
        df_flows["grid_to_battery"],
        df_flows["battery_to_grid"],
        df_flows["battery_to_consumption"],
    ) = battery_power_flows(
        grid=grid_power_series,
        production=df_flows["production_total"],
        consumption=df_flows["consumption_total"],
        battery=battery_power_series,
    )

    df_flows = df_flows.drop(
        columns=["production_total", "consumption_total"], errors="ignore"
    )
    return df_flows

frequenz.lib.notebooks.reporting.utils.helpers.build_color_map

build_color_map(
    cols: list[str],
    color_dict: dict[str, str] | None = None,
    palette: list[str] | None = None,
) -> dict[str, str]

Generate a color mapping for columns or categories.

Creates a mapping from column names (or categorical labels) to color values. Default colors are sourced from the in-code color dictionary and can be overridden via color_dict. Remaining columns are assigned distinct colors from a chosen palette, ensuring no duplicates.

PARAMETER	DESCRIPTION
cols	List of column names or category labels to assign colors to. TYPE: list[str]
color_dict	Optional dictionary of color mappings to override defaults. Columns found here are assigned these colors directly. TYPE: dict[str, str] | None DEFAULT: None
palette	Optional list of color codes to use as defaults. If None, a combined Plotly qualitative palette is used. TYPE: list[str] | None DEFAULT: None

RETURNS	DESCRIPTION
dict[str, str]	A dictionary mapping each column or category name to a unique color.

Source code in src/frequenz/lib/notebooks/reporting/utils/helpers.py

def build_color_map(
    cols: list[str],
    color_dict: dict[str, str] | None = None,
    palette: list[str] | None = None,
) -> dict[str, str]:
    """Generate a color mapping for columns or categories.

    Creates a mapping from column names (or categorical labels) to color
    values. Default colors are sourced from the in-code color dictionary
    and can be overridden via `color_dict`. Remaining columns are assigned
    distinct colors from a chosen palette, ensuring no duplicates.

    Args:
        cols: List of column names or category labels to assign colors to.
        color_dict: Optional dictionary of color mappings to override defaults.
            Columns found here are assigned these colors directly.
        palette: Optional list of color codes to use as defaults.
            If None, a combined Plotly qualitative palette is used.

    Returns:
        A dictionary mapping each column or category name to a unique color.
    """
    # Default palette
    if palette is None:
        palette = px.colors.qualitative.Plotly + px.colors.qualitative.Dark2

    def to_rgba_str(color: str) -> str:
        """Convert any color format (hex, rgb, named) to normalized rgba(R,G,B,1) string."""
        try:
            rgba = mcolors.to_rgba(color)  # returns (r,g,b,a) in 0–1 range
            rgba_255 = tuple(int(round(x * 255)) for x in rgba[:3])
            return f"rgba({rgba_255[0]},{rgba_255[1]},{rgba_255[2]},{rgba[3]:.3f})"
        except ValueError:
            # fallback if string isn't recognized (e.g. malformed rgba)
            return color.lower().strip()

    final = {}
    used = set()

    # First assign default colors, then override with user-provided
    merged_color_dict = COLOR_DICT
    if color_dict:
        merged_color_dict = {**merged_color_dict, **color_dict}

    for c, v in merged_color_dict.items():
        if c in cols:
            rgba = to_rgba_str(v)
            final[c] = rgba
            used.add(rgba)

    # Then assign defaults, skipping already-used colors
    palette_iter = iter(palette * (len(cols) // len(palette) + 1))
    for c in cols:
        if c in final:
            continue
        for p in palette_iter:
            rgba = to_rgba_str(p)
            if rgba not in used:
                final[c] = rgba
                used.add(rgba)
                break

    return final

frequenz.lib.notebooks.reporting.utils.helpers.convert_timezone

convert_timezone(
    ts: Series,
    target_tz: str = "Europe/Berlin",
    assume_tz: str = "UTC",
) -> Series

Convert a datetime Series to a target timezone.

If the Series contains timezone-naive datetimes, they are first localized to assume_tz before converting to target_tz.

PARAMETER	DESCRIPTION
ts	Input Series containing the datetime values. TYPE: Series
target_tz	Timezone name to convert the Series to. Defaults to "Europe/Berlin". TYPE: str DEFAULT: 'Europe/Berlin'
assume_tz	Timezone to assume for naive datetimes. Defaults to "UTC". TYPE: str DEFAULT: 'UTC'

RETURNS	DESCRIPTION
Series	pd.Series: The timestamp Series converted to the requested timezone.

RAISES	DESCRIPTION
ValueError	If ts is not a pandas Series.

Source code in src/frequenz/lib/notebooks/reporting/utils/helpers.py

def convert_timezone(
    ts: pd.Series,
    target_tz: str = "Europe/Berlin",
    assume_tz: str = "UTC",
) -> pd.Series:
    """Convert a datetime Series to a target timezone.

    If the Series contains timezone-naive datetimes, they are first localized to
    ``assume_tz`` before converting to ``target_tz``.

    Args:
        ts: Input Series containing the datetime values.
        target_tz: Timezone name to convert the Series to.
            Defaults to ``"Europe/Berlin"``.
        assume_tz: Timezone to assume for naive datetimes.
            Defaults to ``"UTC"``.

    Returns:
        pd.Series: The timestamp Series converted to the requested timezone.

    Raises:
        ValueError: If ``ts`` is not a pandas Series.
    """
    if not isinstance(ts, pd.Series):
        raise ValueError("Input must be a pandas Series")

    # Localize naive timestamps
    if ts.dt.tz is None:
        ts = ts.dt.tz_localize(assume_tz)

    result = ts.dt.tz_convert(target_tz)
    return cast(pd.Series, result)

frequenz.lib.notebooks.reporting.utils.helpers.fill_aggregated_component_columns

fill_aggregated_component_columns(
    df: DataFrame,
    component_types: list[str],
    config: AggregatedComponentConfig | None = None,
) -> DataFrame

Populate missing aggregate columns by summing labeled component columns.

PARAMETER	DESCRIPTION
df	Input DataFrame potentially containing aggregated component columns (e.g., "battery_power_flow") and labeled individual component columns (e.g., "Battery #1", "Battery #2"). TYPE: DataFrame
component_types	List of component types to consider for aggregation (e.g., ["battery", "pv", "chp", "wind"]). TYPE: list[str]
config	Mapping of component types to tuples containing the aggregated column name and the prefix used to identify individual component columns. TYPE: AggregatedComponentConfig | None DEFAULT: None

RETURNS	DESCRIPTION
DataFrame	DataFrame with missing aggregated component columns filled in by summing
DataFrame	the corresponding individual component columns.

Source code in src/frequenz/lib/notebooks/reporting/utils/helpers.py

def fill_aggregated_component_columns(
    df: pd.DataFrame,
    component_types: list[str],
    config: AggregatedComponentConfig | None = None,
) -> pd.DataFrame:
    """Populate missing aggregate columns by summing labeled component columns.

    Args:
        df: Input DataFrame potentially containing aggregated component columns
            (e.g., "battery_power_flow") and labeled individual component columns
            (e.g., "Battery #1", "Battery #2").
        component_types: List of component types to consider for aggregation
            (e.g., ["battery", "pv", "chp", "wind"]).

        config: Mapping of component types to tuples containing the aggregated
            column name and the prefix used to identify individual component
            columns.

    Returns:
        DataFrame with missing aggregated component columns filled in by summing
        the corresponding individual component columns.
    """
    config = config or DEFAULT_AGGREGATED_COMPONENT_CONFIG
    normalized_types = {comp.lower() for comp in component_types}

    for comp_type, (agg_col, prefix) in config.items():
        if comp_type not in normalized_types or agg_col not in df.columns:
            continue

        component_cols = [col for col in df.columns if col.startswith(prefix)]

        # Only proceed if there are components to sum and missing values to fill
        if component_cols and df[agg_col].isna().any():
            # min_count=1 ensures all-NaN rows stay NaN instead of becoming 0
            summed = df[component_cols].sum(axis=1, skipna=True, min_count=1)

            # Fill only the holes
            df[agg_col] = df[agg_col].fillna(summed)

    return df

frequenz.lib.notebooks.reporting.utils.helpers.fmt_to_de_system

fmt_to_de_system(x: float) -> str

Format a number using German-style decimal and thousands separators.

The function formats the number with two decimal places, using a comma as the decimal separator and a dot as the thousands separator.

PARAMETER	DESCRIPTION
x	The number to format. TYPE: float

RETURNS	DESCRIPTION
str	The formatted string with German number formatting applied.

Example

_fmt_to_de_system(12345.6789) '12.345,68'

Source code in src/frequenz/lib/notebooks/reporting/utils/helpers.py

def fmt_to_de_system(x: float) -> str:
    """Format a number using German-style decimal and thousands separators.

    The function formats the number with two decimal places, using a comma
    as the decimal separator and a dot as the thousands separator.

    Args:
        x: The number to format.

    Returns:
        The formatted string with German number formatting applied.

    Example:
        >>> _fmt_to_de_system(12345.6789)
        '12.345,68'
    """
    return f"{x:,.2f}".replace(",", "X").replace(".", ",").replace("X", ".")

frequenz.lib.notebooks.reporting.utils.helpers.get_energy_report_columns

get_energy_report_columns(
    component_types: list[str], single_components: list[str]
) -> list[str]

Build the list of dataframe columns for the energy report.

The selected columns depend on the available component types.

PARAMETER	DESCRIPTION
component_types	List of component types (e.g. ["pv", "wind", "chp", "battery"]) TYPE: list[str]
single_components	Extra component columns to always include. TYPE: list[str]

RETURNS	DESCRIPTION
list[str]	The full list of dataframe columns.

Source code in src/frequenz/lib/notebooks/reporting/utils/helpers.py

def get_energy_report_columns(
    component_types: list[str], single_components: list[str]
) -> list[str]:
    """Build the list of dataframe columns for the energy report.

    The selected columns depend on the available component types.

    Args:
        component_types: List of component types (e.g. ["pv", "wind", "chp", "battery"])
        single_components: Extra component columns to always include.

    Returns:
        The full list of dataframe columns.
    """
    # Base columns
    energy_report_df_cols = [
        "timestamp",
        "grid_consumption",
        "mid_consumption",
    ] + single_components

    # Map component types to the columns they enable
    component_column_map = {
        "battery": ["battery_power_flow"],
        "pv": ["pv_asset_production"],
        "chp": ["chp_asset_production"],
        "wind": ["wind_asset_production"],
    }

    production_components = set(component_column_map.keys()) - {"battery"}

    # Columns that should be included if ANY production component exists
    universal_production_cols = [
        "production_self_use",
        "grid_feed_in",
        "production_self_share",
        "production_self_usage",
    ]

    # Columns available ONLY when battery exists
    battery_dependent_cols = [
        "production_to_battery",
        "grid_to_battery",
        "battery_to_grid",
        "battery_to_consumption",
    ]

    # Check if any production component is present
    has_production = any(comp in production_components for comp in component_types)
    has_battery = "battery" in component_types

    # Add the universal production columns ONLY if production exists
    if has_production:
        energy_report_df_cols.extend(universal_production_cols)

    # Add battery-dependent production columns
    if has_battery:
        energy_report_df_cols.extend(battery_dependent_cols)

    # Add component-specific columns
    for comp in component_types:
        if comp in component_column_map:
            energy_report_df_cols.extend(component_column_map[comp])

    return energy_report_df_cols

frequenz.lib.notebooks.reporting.utils.helpers.get_meter_display_names async

get_meter_display_names(
    microgrid_id: int,
    *,
    server_url: str | None = None,
    auth_key: str | None = None,
    sign_secret: str | None = None
) -> dict[str, str]

Fetch component display names asynchronously from the Assets API.

Source code in src/frequenz/lib/notebooks/reporting/utils/helpers.py

async def get_meter_display_names(
    microgrid_id: int,
    *,
    server_url: str | None = None,
    auth_key: str | None = None,
    sign_secret: str | None = None,
) -> dict[str, str]:
    """Fetch component display names asynchronously from the Assets API."""
    (
        resolved_server_url,
        resolved_auth_key,
        resolved_sign_secret,
    ) = _resolved_assets_api_config(server_url, auth_key, sign_secret)

    client = AssetsApiClient(
        server_url=resolved_server_url,
        auth_key=resolved_auth_key,
        sign_secret=resolved_sign_secret,
    )
    components = await client.list_microgrid_electrical_components(
        MicrogridId(microgrid_id)
    )

    meter_names: dict[str, str] = {}
    for component in components:
        component_pair = _component_display_name(component)
        if component_pair is not None:
            component_id, component_name = component_pair
            meter_names[component_id] = component_name

    return meter_names

frequenz.lib.notebooks.reporting.utils.helpers.label_component_columns

label_component_columns(
    df: DataFrame,
    mcfg: MicrogridConfig,
    column_battery: str = "battery",
    column_pv: str = "pv",
    column_chp: str = "chp",
    column_ev: str = "ev",
    column_wind: str = "wind",
    component_display_names: (
        Mapping[str, str] | None
    ) = None,
) -> tuple[DataFrame, list[str]]

Rename numeric single-component columns to labeled names.

Numeric string column names like "14" are converted to "Battery #14", "PV #14", "CHP #14" or "EV #14" based on the component IDs provided by mcfg.component_type_ids(...)

PARAMETER	DESCRIPTION
df	Input DataFrame with numeric string column names. TYPE: DataFrame
mcfg	Configuration with _component_types_cfg mapping component types to a meter iterable of numeric IDs. TYPE: MicrogridConfig
column_battery	Key name for battery component type. TYPE: str DEFAULT: 'battery'
column_pv	Key name for PV component type. TYPE: str DEFAULT: 'pv'
column_chp	Key name for CHP component type. TYPE: str DEFAULT: 'chp'
column_ev	Key name for EV component type. TYPE: str DEFAULT: 'ev'
column_wind	Key name for wind component type. TYPE: str DEFAULT: 'wind'
component_display_names	Optional mapping from numeric component IDs to human-readable display names fetched from the Assets API. TYPE: Mapping[str, str] | None DEFAULT: None

RETURNS	DESCRIPTION
tuple[DataFrame, list[str]]	Tuple containing the renamed DataFrame and the list of applied labels

Source code in src/frequenz/lib/notebooks/reporting/utils/helpers.py

def label_component_columns(
    df: pd.DataFrame,
    mcfg: MicrogridConfig,
    column_battery: str = "battery",
    column_pv: str = "pv",
    column_chp: str = "chp",
    column_ev: str = "ev",
    column_wind: str = "wind",
    component_display_names: Mapping[str, str] | None = None,
) -> tuple[pd.DataFrame, list[str]]:
    """Rename numeric single-component columns to labeled names.

    Numeric string column names like ``"14"`` are converted to
    ``"Battery #14"``, ``"PV #14"``, ``"CHP #14"`` or ``"EV #14"`` based on
    the component IDs provided by ``mcfg.component_type_ids(...)``

    Args:
        df: Input DataFrame with numeric string column names.
        mcfg: Configuration with ``_component_types_cfg`` mapping component types to a
            ``meter`` iterable of numeric IDs.
        column_battery: Key name for battery component type.
        column_pv: Key name for PV component type.
        column_chp: Key name for CHP component type.
        column_ev: Key name for EV component type.
        column_wind: Key name for wind component type.
        component_display_names: Optional mapping from numeric component IDs to
            human-readable display names fetched from the Assets API.

    Returns:
        Tuple containing the renamed DataFrame and the list of applied labels
    """
    # Numeric component columns present in df
    single_components = [str(c) for c in df.columns if str(c).isdigit()]
    available_types = set(mcfg.component_types())

    # From config (empty set if missing). Union of all ID categories, since a
    # component type can have meter, inverter, and component IDs configured
    # at once, and each should be labeled if present in the DataFrame.
    def ids_if_available(t: str) -> set[str]:
        if t not in available_types:
            return set()
        ids: set[str] = set()
        for category in ("meter", "inverter", "component"):
            category_ids = mcfg.component_type_ids(t, component_category=category)
            ids.update(str(x) for x in category_ids or [])
        return ids

    battery_ids = ids_if_available(column_battery)
    pv_ids = ids_if_available(column_pv)
    chp_ids = ids_if_available(column_chp)
    ev_ids = ids_if_available(column_ev)
    wind_ids = ids_if_available(column_wind)

    display_names = component_display_names or {}

    def display_label(prefix: str, component_id: str) -> str:
        component_name = display_names.get(component_id)
        suffix = f" - {component_name}" if component_name else ""
        return f"{prefix} #{component_id}{suffix}"

    rename: dict[str, str] = {}
    rename.update(
        {
            c: display_label(column_battery.capitalize(), c)
            for c in single_components
            if c in battery_ids
        }
    )
    rename.update(
        {
            c: display_label(column_pv.upper(), c)
            for c in single_components
            if c in pv_ids
        }
    )
    rename.update(
        {
            c: display_label(column_ev.upper(), c)
            for c in single_components
            if c in ev_ids
        }
    )
    rename.update(
        {
            c: display_label(column_chp.upper(), c)
            for c in single_components
            if c in chp_ids
        }
    )
    rename.update(
        {
            c: display_label(column_wind.capitalize(), c)
            for c in single_components
            if c in wind_ids
        }
    )

    return df.rename(columns=rename), list(rename.values())

frequenz.lib.notebooks.reporting.utils.helpers.load_config

load_config(path: str) -> dict[str, Any]

Load a YAML config file and return it as a dictionary.

PARAMETER	DESCRIPTION
path	Path to the YAML file. TYPE: str

RETURNS	DESCRIPTION
dict[str, Any]	Configuration values as a dictionary.

RAISES	DESCRIPTION
TypeError	If the YAML root element is not a mapping (dict).

Source code in src/frequenz/lib/notebooks/reporting/utils/helpers.py

def load_config(path: str) -> dict[str, Any]:
    """
    Load a YAML config file and return it as a dictionary.

    Args:
        path: Path to the YAML file.

    Returns:
        Configuration values as a dictionary.

    Raises:
        TypeError: If the YAML root element is not a mapping (dict).
    """
    with open(path, "r", encoding="utf-8") as f:
        data = yaml.safe_load(f)

    if not isinstance(data, dict):
        raise TypeError(
            f"YAML root must be a mapping (dict), got {type(data).__name__}"
        )

    return data

frequenz.lib.notebooks.reporting.utils.helpers.long_to_wide

long_to_wide(
    df: DataFrame,
    *,
    time_col: str | Index = "Timestamp",
    category_col: str | None = "Battery",
    value_col: str | None = "Battery Throughput",
    sum_col_name: str | None = None,
    aggfunc: AggFuncLiteral = "sum"
) -> DataFrame

Convert a long-format DataFrame into wide format with optional aggregation.

Transforms a long-format dataset (one row per timestamp-category pair) into a wide-format table, where each category becomes a separate column. Optionally adds a total (sum) column across all categories.

PARAMETER	DESCRIPTION
df	Input DataFrame in long format. TYPE: DataFrame
time_col	Column name representing timestamps used as the index in the resulting wide table. Defaults to "Timestamp". TYPE: str | Index DEFAULT: 'Timestamp'
category_col	Column name representing category labels that become column headers in the wide table. Defaults to "Battery". TYPE: str | None DEFAULT: 'Battery'
value_col	Column name representing numeric values to aggregate and pivot into columns. Defaults to "Battery Throughput". TYPE: str | None DEFAULT: 'Battery Throughput'
sum_col_name	Optional name for a new column containing the row-wise sum of all category columns. If None, defaults to "<value_col> Sum". TYPE: str | None DEFAULT: None
aggfunc	Aggregation function applied when multiple entries exist per timestamp-category pair (e.g., "sum", "mean"). Defaults to "sum". TYPE: AggFuncLiteral DEFAULT: 'sum'

RETURNS	DESCRIPTION
DataFrame	A wide-format DataFrame with one row per timestamp, one column per category,
DataFrame	and an optional total column representing the aggregated sum across all categories.

Source code in src/frequenz/lib/notebooks/reporting/utils/helpers.py

def long_to_wide(
    df: pd.DataFrame,
    *,
    time_col: str | pd.Index = "Timestamp",
    category_col: str | None = "Battery",
    value_col: str | None = "Battery Throughput",
    sum_col_name: str | None = None,
    aggfunc: AggFuncLiteral = "sum",
) -> pd.DataFrame:
    """Convert a long-format DataFrame into wide format with optional aggregation.

    Transforms a long-format dataset (one row per timestamp-category pair)
    into a wide-format table, where each category becomes a separate column.
    Optionally adds a total (sum) column across all categories.

    Args:
        df: Input DataFrame in long format.
        time_col: Column name representing timestamps used as the index in
            the resulting wide table. Defaults to `"Timestamp"`.
        category_col: Column name representing category labels that become
            column headers in the wide table. Defaults to `"Battery"`.
        value_col: Column name representing numeric values to aggregate and
            pivot into columns. Defaults to `"Battery Throughput"`.
        sum_col_name: Optional name for a new column containing the row-wise sum
            of all category columns. If None, defaults to `"<value_col> Sum"`.
        aggfunc: Aggregation function applied when multiple entries exist per
            timestamp-category pair (e.g., `"sum"`, `"mean"`). Defaults to `"sum"`.

    Returns:
        A wide-format DataFrame with one row per timestamp, one column per category,
        and an optional total column representing the aggregated sum across all categories.
    """
    tmp = df.copy()

    wide = tmp.pivot_table(
        index=time_col,  # type: ignore [arg-type]
        columns=category_col,
        values=value_col,
        aggfunc=aggfunc,
    ).sort_index()

    wide.columns.name = None

    if sum_col_name is None:
        sum_col_name = f"{value_col} Sum"
    wide[sum_col_name] = wide.sum(axis=1, numeric_only=True)
    return wide

frequenz.lib.notebooks.reporting.utils.helpers.normalize_date_for_reporting

normalize_date_for_reporting(
    input_date: date | datetime, timezone_name: str = "UTC"
) -> datetime

Return midnight for past dates or current time for today.

If the input date is today (in the target timezone), this returns the current time in that timezone. If the date is in the future, a ValueError is raised to prevent selecting future dates. For past dates, midnight is returned via set_date_to_midnight.

PARAMETER	DESCRIPTION
input_date	Date or datetime object to evaluate. TYPE: date | datetime
timezone_name	Name of the target timezone (e.g., "Europe/Berlin"). Defaults to "UTC". Falls back to UTC if the timezone name is invalid. TYPE: str DEFAULT: 'UTC'

RETURNS	DESCRIPTION
datetime	A timezone-aware datetime object representing the current time
datetime	when the input date is today, or midnight for past dates.

RAISES	DESCRIPTION
ValueError	If the input date is in the future.

Source code in src/frequenz/lib/notebooks/reporting/utils/helpers.py

def normalize_date_for_reporting(
    input_date: date | datetime, timezone_name: str = "UTC"
) -> datetime:
    """Return midnight for past dates or current time for today.

    If the input date is today (in the target timezone), this returns the
    current time in that timezone. If the date is in the future, a ValueError
    is raised to prevent selecting future dates. For past dates, midnight is
    returned via ``set_date_to_midnight``.

    Args:
        input_date: Date or datetime object to evaluate.
        timezone_name: Name of the target timezone (e.g., "Europe/Berlin").
            Defaults to "UTC". Falls back to UTC if the timezone name
            is invalid.

    Returns:
        A timezone-aware datetime object representing the current time
        when the input date is today, or midnight for past dates.

    Raises:
        ValueError: If the input date is in the future.
    """
    input_day = input_date.date() if isinstance(input_date, datetime) else input_date

    try:
        tz = ZoneInfo(timezone_name)
    except (ZoneInfoNotFoundError, KeyError):
        warnings.warn(
            f"Unknown timezone '{timezone_name}', falling back to UTC.",
            RuntimeWarning,
        )
        tz = ZoneInfo("UTC")

    now = datetime.now(tz)
    today = now.date()

    if input_day > today:
        raise ValueError("Future dates can't be selected.")
    if input_day == today:
        return now

    return set_date_to_midnight(input_date, timezone_name)

frequenz.lib.notebooks.reporting.utils.helpers.set_date_to_midnight

set_date_to_midnight(
    input_date: date | datetime, timezone_name: str = "UTC"
) -> datetime

Return a timezone-aware datetime set to midnight of the given date.

Converts a date or datetime into a midnight timestamp localized to the specified timezone. If the input is already a datetime, only the date portion is used.

PARAMETER	DESCRIPTION
input_date	Date or datetime object to normalize to midnight. TYPE: date | datetime
timezone_name	Name of the target timezone (e.g., "Europe/Berlin"). Defaults to "UTC". Falls back to UTC if the timezone name is invalid. TYPE: str DEFAULT: 'UTC'

RETURNS	DESCRIPTION
datetime	A timezone-aware datetime object representing midnight of the
datetime	given date in the specified timezone.

Source code in src/frequenz/lib/notebooks/reporting/utils/helpers.py

def set_date_to_midnight(
    input_date: date | datetime, timezone_name: str = "UTC"
) -> datetime:
    """Return a timezone-aware datetime set to midnight of the given date.

    Converts a date or datetime into a midnight timestamp localized to
    the specified timezone. If the input is already a datetime, only the
    date portion is used.

    Args:
        input_date: Date or datetime object to normalize to midnight.
        timezone_name: Name of the target timezone (e.g., "Europe/Berlin").
            Defaults to "UTC". Falls back to UTC if the timezone name
            is invalid.

    Returns:
        A timezone-aware datetime object representing midnight of the
        given date in the specified timezone.
    """
    if isinstance(input_date, datetime):
        input_date = input_date.date()

    try:
        tz = ZoneInfo(timezone_name)
    except (ZoneInfoNotFoundError, KeyError):
        warnings.warn(
            f"Unknown timezone '{timezone_name}', falling back to UTC.",
            RuntimeWarning,
        )
        tz = ZoneInfo("UTC")

    return datetime.combine(input_date, time.min).replace(tzinfo=tz)