battery_usecase_plot

frequenz.lib.notebooks.reporting.utils.battery_usecase_plot

Helpers for battery-usecase plotting.

Functions:

frequenz.lib.notebooks.reporting.utils.battery_usecase_plot.add_battery_usecase_overlay_traces

add_battery_usecase_overlay_traces(
    fig: Figure,
    source_df: DataFrame,
    *,
    color_dict: dict[str, str],
    yaxis_title: str,
    stack_mode: BatteryUsecaseStackMode
) -> None

Hide base traces and re-add them using viz_plotly-style stackgroups.

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

def add_battery_usecase_overlay_traces(
    fig: go.Figure,
    source_df: pd.DataFrame,
    *,
    color_dict: dict[str, str],
    yaxis_title: str,
    stack_mode: BatteryUsecaseStackMode,
) -> None:
    """Hide base traces and re-add them using viz_plotly-style stackgroups."""
    production_columns = _get_production_columns(source_df)

    hidden_names = {
        _DISPLAY_LABELS["battery_power_flow"],
        _DISPLAY_LABELS["battery_discharge"],
        _DISPLAY_LABELS["battery_charge"],
        "Battery Charge",
        "Battery Discharge",
        *production_columns,
    }
    fig.data = tuple(
        trace
        for trace in fig.data
        if not (isinstance(trace, go.Scatter) and trace.name in hidden_names)
    )

    display_grid = _DISPLAY_LABELS["grid_consumption"]
    display_consumption = _DISPLAY_LABELS["consumption"]
    entladung_name = _DISPLAY_LABELS["battery_discharge"]  # "Batterie Entladung"
    beladung_name = _DISPLAY_LABELS["battery_charge"]  # "Batterie Beladung"

    required_production = [
        _DISPLAY_LABELS[c] for c in _REQUIRED_PRODUCTION_OVERLAY_COLUMNS
    ]
    grid_with_battery: pd.Series | None = None
    consumption: pd.Series | None = None
    if all(c in source_df.columns for c in required_production):
        grid_with_battery = pd.to_numeric(source_df[display_grid], errors="coerce")
    if display_consumption in source_df.columns:
        consumption = pd.to_numeric(source_df[display_consumption], errors="coerce")

    required_battery = [_DISPLAY_LABELS[c] for c in _REQUIRED_BATTERY_OVERLAY_COLUMNS]
    if grid_with_battery is not None and consumption is not None:
        if stack_mode == "energy_balance":
            supply_columns = [entladung_name, *production_columns]
            supply_baseline = grid_with_battery
            finite_mask = grid_with_battery.notna()
            supply_direction: Literal["up", "down"] = "up"
        else:
            preferred_supply_order = [
                _DISPLAY_LABELS["wind"],
                _DISPLAY_LABELS["chp"],
                _DISPLAY_LABELS["pv"],
                entladung_name,
            ]
            supply_columns = [
                column
                for column in preferred_supply_order
                if column == entladung_name or column in production_columns
            ]
            supply_baseline = consumption
            finite_mask = consumption.notna()
            supply_direction = "down"
        _add_cumulative_supply_stack_traces(
            fig,
            source_df,
            baseline=supply_baseline,
            finite_mask=finite_mask,
            columns=supply_columns,
            color_dict=color_dict,
            yaxis_title=yaxis_title,
            direction=supply_direction,
            opacity_overrides={entladung_name: _BATTERY_OVERLAY_ALPHA},
            min_power_overrides={entladung_name: _MIN_BATTERY_OVERLAY_POWER},
        )

    if (
        all(c in source_df.columns for c in required_battery)
        and consumption is not None
    ):
        _add_battery_charge_overlay(
            fig,
            source_df,
            column=beladung_name,
            grid=grid_with_battery,
            consumption=consumption,
            color_dict=color_dict,
            yaxis_title=yaxis_title,
            stack_mode=stack_mode,
        )

    fig.update_layout(legend={"groupclick": "togglegroup"})
    _move_grid_traces_to_top(fig)

frequenz.lib.notebooks.reporting.utils.battery_usecase_plot.prepare_battery_usecase_plot

prepare_battery_usecase_plot(
    df: DataFrame,
    *,
    cols: list[str] | None,
    fill_cols: list[str] | None,
    dotted_cols: list[str] | None,
    plot_order: list[str] | None,
    secondary_y_cols: Sequence[str] | None,
    color_dict: dict[str, str] | None,
    time_col: str | None,
    battery_power_flow: str,
    battery_charging: str,
    battery_discharging: str,
    pv_col: str,
    consumption_col: str,
    grid_consumption: str
) -> tuple[
    DataFrame,
    list[str] | None,
    list[str] | None,
    list[str] | None,
    list[str] | None,
    list[str] | None,
    dict[str, str],
]

Normalize battery-usecase inputs, apply plot defaults, and build color map.

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

def prepare_battery_usecase_plot(
    df: pd.DataFrame,
    *,
    cols: list[str] | None,
    fill_cols: list[str] | None,
    dotted_cols: list[str] | None,
    plot_order: list[str] | None,
    secondary_y_cols: Sequence[str] | None,
    color_dict: dict[str, str] | None,
    time_col: str | None,
    battery_power_flow: str,
    battery_charging: str,
    battery_discharging: str,
    pv_col: str,
    consumption_col: str,
    grid_consumption: str,
) -> tuple[
    pd.DataFrame,
    list[str] | None,
    list[str] | None,
    list[str] | None,
    list[str] | None,
    list[str] | None,
    dict[str, str],
]:
    """Normalize battery-usecase inputs, apply plot defaults, and build color map."""
    normalize_map: dict[str, str] = {v: k for k, v in _DISPLAY_LABELS.items()}
    normalize_map.update(
        {
            battery_power_flow: "battery_power_flow",
            battery_charging: "battery_discharge",
            battery_discharging: "battery_charge",
            pv_col: "pv",
            consumption_col: "consumption",
            grid_consumption: "grid_consumption",
        }
    )

    def _rename(seq: list[str] | None, mapping: dict[str, str]) -> list[str] | None:
        return None if seq is None else [mapping.get(item, item) for item in seq]

    def _ensure(seq: list[str] | None, items: list[str]) -> list[str] | None:
        if seq is None:
            return None
        result = list(seq)
        for item in items:
            if item not in result:
                result.append(item)
        return result

    df = df.rename(columns=normalize_map)
    cols = _rename(cols, normalize_map)
    fill_cols = _rename(fill_cols, normalize_map)
    dotted_cols = _rename(dotted_cols, normalize_map)
    plot_order = _rename(plot_order, normalize_map)
    secondary_y_cols = _rename(
        list(secondary_y_cols) if secondary_y_cols is not None else None, normalize_map
    )

    for canonical_name in ("pv", "chp", "wind"):
        if canonical_name in df.columns:
            fill_cols = (
                [canonical_name]
                if fill_cols is None and canonical_name == "pv"
                else _ensure(fill_cols, [canonical_name])
            )
            plot_order = _ensure(plot_order, [canonical_name])

    if "consumption" in df.columns:
        # Plain reference line, not a fill — just make sure it's plotted.
        plot_order = _ensure(plot_order, ["consumption"])

    peak_columns = [c for c in _PEAK_COLUMNS if c in df.columns]
    cols = _ensure(cols, peak_columns)
    plot_order = _ensure(plot_order, peak_columns)
    dotted_cols = _ensure(dotted_cols, peak_columns) or peak_columns

    # Localize canonical names → display labels
    df = df.rename(columns=_DISPLAY_LABELS)
    cols = _rename(cols, _DISPLAY_LABELS)
    fill_cols = _rename(fill_cols, _DISPLAY_LABELS)
    dotted_cols = _rename(dotted_cols, _DISPLAY_LABELS)
    plot_order = _rename(plot_order, _DISPLAY_LABELS)
    secondary_y_cols = _rename(secondary_y_cols, _DISPLAY_LABELS)

    # Build color map with defaults
    colors = dict(color_dict or {})
    colors.setdefault(_DISPLAY_LABELS["grid_consumption"], COLOR_DICT["Netzbezug"])
    colors.setdefault(
        _DISPLAY_LABELS["battery_discharge"], COLOR_DICT["Battery Charge"]
    )
    colors.setdefault(
        _DISPLAY_LABELS["battery_charge"], COLOR_DICT["Battery Discharge"]
    )
    for peak_col in peak_columns:
        colors.setdefault(_DISPLAY_LABELS[peak_col], COLOR_DICT["peak"])

    for display_name, default_color in _PRODUCTION_DEFAULT_COLORS.items():
        if display_name in df.columns:
            if cols is None:
                cols = [
                    c
                    for c in df.select_dtypes(include="number").columns
                    if c != time_col
                ]
            elif display_name not in cols:
                cols = [*cols, display_name]
            colors.setdefault(display_name, default_color)

    display_consumption = _DISPLAY_LABELS["consumption"]
    if display_consumption in df.columns:
        if cols is None:
            cols = [
                c for c in df.select_dtypes(include="number").columns if c != time_col
            ]
        elif display_consumption not in cols:
            cols = [*cols, display_consumption]
        colors.setdefault(
            display_consumption,
            COLOR_DICT.get("Verbrauch") or COLOR_DICT.get("Consumption") or "#6c757d",
        )

    return df, cols, fill_cols, dotted_cols, plot_order, secondary_y_cols, colors