Skip to content

viz_data

frequenz.lib.notebooks.reporting.asset_optimization.viz_data ¤

Data preparation helpers for asset optimization visualizations.

Classes¤

frequenz.lib.notebooks.reporting.asset_optimization.viz_data.BatteryPowerData dataclass ¤

Prepared series for battery power plotting.

Source code in src/frequenz/lib/notebooks/reporting/asset_optimization/viz_data.py 
40
41
42
43
44
45
46
47
48
49
50
@dataclass(frozen=True)
class BatteryPowerData:
    """Prepared series for battery power plotting."""

    index: pd.Index
    soc: pd.Series
    available: pd.Series
    battery: pd.Series
    charge: pd.Series
    discharge: pd.Series
    max_abs_battery: float

frequenz.lib.notebooks.reporting.asset_optimization.viz_data.EnergyTradeData dataclass ¤

Prepared series for energy trade plotting.

Source code in src/frequenz/lib/notebooks/reporting/asset_optimization/viz_data.py 
31
32
33
34
35
36
37
@dataclass(frozen=True)
class EnergyTradeData:
    """Prepared series for energy trade plotting."""

    index: pd.Index
    buy: pd.Series
    sell: pd.Series

frequenz.lib.notebooks.reporting.asset_optimization.viz_data.MonthlyAggregateData dataclass ¤

Prepared data for monthly aggregate plotting.

Source code in src/frequenz/lib/notebooks/reporting/asset_optimization/viz_data.py 
53
54
55
56
57
58
59
@dataclass(frozen=True)
class MonthlyAggregateData:
    """Prepared data for monthly aggregate plotting."""

    months: pd.DataFrame
    positive: pd.DataFrame
    negative: pd.DataFrame

frequenz.lib.notebooks.reporting.asset_optimization.viz_data.PowerFlowData dataclass ¤

Prepared series for power flow plotting.

Source code in src/frequenz/lib/notebooks/reporting/asset_optimization/viz_data.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
@dataclass(frozen=True)
class PowerFlowData:
    """Prepared series for power flow plotting."""

    index: pd.Index
    consumption: pd.Series
    chp: pd.Series
    production: pd.Series
    has_chp: bool
    has_pv: bool
    battery_consumption: pd.Series | None
    charge: pd.Series | None
    discharge: pd.Series | None
    grid: pd.Series | None

Functions¤

frequenz.lib.notebooks.reporting.asset_optimization.viz_data.prepare_battery_power_data ¤

prepare_battery_power_data(
    df: DataFrame,
) -> BatteryPowerData

Prepare normalized battery power data for visualization.

Extracts battery-related signals from a power-flow DataFrame and computes derived series used for plotting, including charge, discharge, available battery power, and state-of-charge (SOC). The function also determines the maximum absolute battery value for consistent axis scaling.

PARAMETER DESCRIPTION
df

Input DataFrame containing at least the columns "battery", "grid", and "soc". Values are assumed to represent power time-series data.

TYPE: DataFrame

RETURNS DESCRIPTION
BatteryPowerData

A structured container including: - timestamp index, - state of charge (soc), - available battery power (available), - raw battery power values, - separated charge and discharge series, - maximum absolute battery power for plot scaling.
Source code in src/frequenz/lib/notebooks/reporting/asset_optimization/viz_data.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def prepare_battery_power_data(df: pd.DataFrame) -> BatteryPowerData:
    """Prepare normalized battery power data for visualization.

    Extracts battery-related signals from a power-flow DataFrame and computes
    derived series used for plotting, including charge, discharge, available
    battery power, and state-of-charge (SOC). The function also determines the
    maximum absolute battery value for consistent axis scaling.

    Args:
        df:
            Input DataFrame containing at least the columns ``"battery"``,
            ``"grid"``, and ``"soc"``. Values are assumed to represent power
            time-series data.

    Returns:
        A structured container including:
            - timestamp index,
            - state of charge (``soc``),
            - available battery power (``available``),
            - raw battery power values,
            - separated charge and discharge series,
            - maximum absolute battery power for plot scaling.
    """
    require_columns(df, "battery", "grid", "soc")

    soc = df["soc"]
    available = df["battery"] - df["grid"]

    max_abs_bat = max(
        abs(df["battery"].min()),
        abs(df["battery"].max()),
        abs(available.min()),
        abs(available.max()),
    )

    charge = _masked(df["battery"], df["battery"] > 0)
    discharge = _masked(df["battery"], df["battery"] <= 0)

    return BatteryPowerData(
        index=df.index,
        soc=soc,
        available=available,
        battery=df["battery"],
        charge=charge,
        discharge=discharge,
        max_abs_battery=max_abs_bat,
    )

frequenz.lib.notebooks.reporting.asset_optimization.viz_data.prepare_energy_trade_data ¤

prepare_energy_trade_data(df: DataFrame) -> EnergyTradeData

Prepare normalized energy-trade data for plotting.

Processes raw power time-series data to derive buy and sell energy flows relative to site consumption and local production. The function applies sign normalization, subtracts on-site generation (PV and CHP if present), and converts the resulting power values into resampled energy quantities.

PARAMETER DESCRIPTION
df

Input DataFrame containing at least the "consumption" column. Optional columns such as "chp" and "pv" are used if available to adjust traded energy values.

TYPE: DataFrame

RETURNS DESCRIPTION
EnergyTradeData

A structured container including: - resampled timestamp index (15-minute resolution), - positive energy purchases (buy), - negative energy sales (sell).
Source code in src/frequenz/lib/notebooks/reporting/asset_optimization/viz_data.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def prepare_energy_trade_data(df: pd.DataFrame) -> EnergyTradeData:
    """Prepare normalized energy-trade data for plotting.

    Processes raw power time-series data to derive buy and sell energy flows
    relative to site consumption and local production. The function applies
    sign normalization, subtracts on-site generation (PV and CHP if present),
    and converts the resulting power values into resampled energy quantities.

    Args:
        df:
            Input DataFrame containing at least the ``"consumption"`` column.
            Optional columns such as ``"chp"`` and ``"pv"`` are used if available
            to adjust traded energy values.

    Returns:
        A structured container including:
            - resampled timestamp index (15-minute resolution),
            - positive energy purchases (``buy``),
            - negative energy sales (``sell``).
    """
    require_columns(df, "consumption")
    d = -df.copy()
    cons = -d["consumption"]
    trade = cons.copy()

    has_chp = "chp" in d.columns
    has_pv = "pv" in d.columns
    chp = d["chp"] if has_chp else 0 * cons
    prod = chp + (d["pv"].clip(lower=0) if has_pv else 0)
    trade -= prod

    g = trade.resample("15min").mean() / 4

    return EnergyTradeData(
        index=g.index,
        buy=g.clip(lower=0),
        sell=g.clip(upper=0),
    )

frequenz.lib.notebooks.reporting.asset_optimization.viz_data.prepare_monthly_data ¤

prepare_monthly_data(df: DataFrame) -> MonthlyAggregateData

Prepare monthly aggregate data.

Source code in src/frequenz/lib/notebooks/reporting/asset_optimization/viz_data.py 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
def prepare_monthly_data(df: pd.DataFrame) -> MonthlyAggregateData:
    """Prepare monthly aggregate data."""
    if len(df.index) < 2:
        raise ValueError(
            "At least two data points are required for monthly aggregation."
        )

    months: pd.DataFrame = df.resample("1MS").sum()
    resolution = (df.index[1] - df.index[0]).total_seconds()
    kW2MWh = resolution / 3600 / 1000  # pylint: disable=invalid-name
    months *= kW2MWh

    if not isinstance(months.index, pd.DatetimeIndex):
        months.index = pd.to_datetime(months.index)
    months.index = pd.Index(months.index.date)

    pos = months[[c for c in months.columns if "_pos" in c]].rename(
        columns={
            "grid_pos": "Grid Consumption",
            "battery_pos": "Battery Charge",
            "consumption_pos": "Consumption",
            "pv_pos": "PV Consumption",
            "chp_pos": "CHP Consumption",
        }
    )
    neg = months[[c for c in months.columns if "_neg" in c]].rename(
        columns={
            "grid_neg": "Grid Feed-in",
            "battery_neg": "Battery Discharge",
            "consumption_neg": "Unknown Production",
            "pv_neg": "PV Production",
            "chp_neg": "CHP Production",
        }
    )

    if pos.empty and neg.empty:
        pos = months.clip(lower=0).add_suffix("_pos")
        neg = months.clip(upper=0).add_suffix("_neg")
        months = pd.concat([months, neg, pos], axis=1)
        pos = pos.rename(
            columns={
                "grid_pos": "Grid Consumption",
                "battery_pos": "Battery Charge",
                "consumption_pos": "Consumption",
                "pv_pos": "PV Consumption",
                "chp_pos": "CHP Consumption",
                "soc_pos": "SOC",
            }
        )
        neg = neg.rename(
            columns={
                "grid_neg": "Grid Feed-in",
                "battery_neg": "Battery Discharge",
                "consumption_neg": "Unknown Production",
                "pv_neg": "PV Production",
                "chp_neg": "CHP Production",
                "soc_neg": "SOC",
            }
        )

    pos = pos.loc[:, pos.abs().sum(axis=0) > 0]
    neg = neg.loc[:, neg.abs().sum(axis=0) > 0]

    return MonthlyAggregateData(months=months, positive=pos, negative=neg)

frequenz.lib.notebooks.reporting.asset_optimization.viz_data.prepare_power_flow_data ¤

prepare_power_flow_data(df: DataFrame) -> PowerFlowData

Prepare normalized power-flow data for plotting.

Transforms raw microgrid power columns into a structured PowerFlowData object suitable for visualization. The function ensures required columns exist, applies sign normalization, derives production totals, and splits battery behavior into charge and discharge flows.

PARAMETER DESCRIPTION
df

Input DataFrame containing at least the columns "consumption", "battery", and "grid". Optional columns such as "chp" and "pv" are used if present.

TYPE: DataFrame

RETURNS DESCRIPTION
PowerFlowData

A structured container including: - normalized consumption series, - CHP and PV production signals, - combined production values, - battery charge and discharge components (if available), - grid exchange values, - metadata flags indicating available production sources.
Source code in src/frequenz/lib/notebooks/reporting/asset_optimization/viz_data.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def prepare_power_flow_data(df: pd.DataFrame) -> PowerFlowData:
    """Prepare normalized power-flow data for plotting.

    Transforms raw microgrid power columns into a structured ``PowerFlowData``
    object suitable for visualization. The function ensures required columns
    exist, applies sign normalization, derives production totals, and splits
    battery behavior into charge and discharge flows.

    Args:
        df:
            Input DataFrame containing at least the columns
            ``"consumption"``, ``"battery"``, and ``"grid"``.
            Optional columns such as ``"chp"`` and ``"pv"`` are used if present.

    Returns:
        A structured container including:
            - normalized consumption series,
            - CHP and PV production signals,
            - combined production values,
            - battery charge and discharge components (if available),
            - grid exchange values,
            - metadata flags indicating available production sources.
    """
    require_columns(df, "consumption", "battery", "grid")
    d = -df.copy()
    cons = -d["consumption"]

    has_chp = "chp" in d.columns
    has_pv = "pv" in d.columns
    chp = d["chp"] if has_chp else pd.Series(0.0, index=cons.index)
    pv = d["pv"].clip(lower=0) if has_pv else pd.Series(0.0, index=cons.index)
    prod = chp + pv

    battery_consumption = None
    charge = None
    discharge = None
    if "battery" in d.columns:
        battery_consumption = -(d["consumption"] + d["battery"])
        charge = _masked(battery_consumption, battery_consumption > cons)
        discharge = _masked(battery_consumption, battery_consumption < cons)

    grid = -d["grid"] if "grid" in d.columns else None

    return PowerFlowData(
        index=d.index,
        consumption=cons,
        chp=chp,
        production=prod,
        has_chp=has_chp,
        has_pv=has_pv,
        battery_consumption=battery_consumption,
        charge=charge,
        discharge=discharge,
        grid=grid,
    )

frequenz.lib.notebooks.reporting.asset_optimization.viz_data.require_columns ¤

require_columns(df: DataFrame, *columns: str) -> None

Validate that a DataFrame contains all required columns.

Checks whether the provided column names exist in the DataFrame. If any required columns are missing, a ValueError is raised listing all missing column names.

PARAMETER DESCRIPTION
df

The DataFrame to validate.

TYPE: DataFrame

*columns

One or more column names that must be present in df.

TYPE: str DEFAULT: ()

RAISES DESCRIPTION
ValueError

If one or more required columns are not found in the DataFrame.
Source code in src/frequenz/lib/notebooks/reporting/asset_optimization/viz_data.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def require_columns(df: pd.DataFrame, *columns: str) -> None:
    """Validate that a DataFrame contains all required columns.

    Checks whether the provided column names exist in the DataFrame. If any
    required columns are missing, a ``ValueError`` is raised listing all
    missing column names.

    Args:
        df:
            The DataFrame to validate.
        *columns:
            One or more column names that must be present in ``df``.

    Raises:
        ValueError:
            If one or more required columns are not found in the DataFrame.
    """
    missing = [col for col in columns if col not in df.columns]
    if missing:
        raise ValueError(f"DataFrame is missing required columns: {', '.join(missing)}")