From f9a1bf81d15f108821e4292e68b6a889b48d4c01 Mon Sep 17 00:00:00 2001 From: Moritz Schloesser Date: Wed, 8 Jul 2026 11:59:22 +0200 Subject: [PATCH] feat: configurable timestep selection for the run pipeline (#663) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a `select_timesteps` pipeline task that reduces the analysed time index to a selected subset, in two modes: - manual: an explicit set/range of timesteps (positioned before the imports so egon_data downloads are restricted to the selected steps); - auto: the most critical time intervals, via either a power flow (`get_most_critical_time_intervals(by="power_flow")`) or the residual load (`by="residual_load"`, no power flow — intervals centered on the highest/lowest residual-load steps, snapped to `time_step_day_start`). Auto selection may yield two disconnected intervals. `pm_optimize` now detects a non-contiguous time index and runs a separate, independent OPF per contiguous interval (storage/heat state does not carry across the gap), merging the per-interval results and reporting per-interval solve status. `reinforce` handles the reduced index unchanged. Architecture: pipeline tasks stay thin (mode selection + data transfer); the computation lives in edisgo core: - selection logic in tools.temporal_complexity_reduction (`get_most_critical_time_steps`/`_intervals` gain a `by` mode; public `select_two_intervals`/`intervals_overlap`); - the multi-interval OPF split/merge in opf.powermodels_opf.pm_optimize. Also: - EV flexibility bands built in a dedicated `build_flexibility_bands` task and aligned (year + frequency) to the analysis index in `reduce_timeseries_data_to_given_timeindex`; - `OPFResults.interval_results` for the per-interval solve report; - config surface: a top-level `timeseries_selection` block + `uc5_select_timesteps` preset + `run_example_05.py`; - tests in tests/run, tests/tools, tests/opf. --- edisgo/io/powermodels_io.py | 33 +- edisgo/opf/powermodels_opf.py | 274 ++++++++++++++- edisgo/opf/results/opf_result_class.py | 8 + edisgo/run/presets/uc4_example.yaml | 6 +- edisgo/run/presets/uc5_select_timesteps.yaml | 117 +++++++ edisgo/run/tasks/__init__.py | 17 +- edisgo/run/tasks/analysis.py | 20 +- edisgo/run/tasks/flex.py | 123 +++++-- edisgo/run/tasks/timeseries.py | 306 +++++++++++++++- edisgo/tools/temporal_complexity_reduction.py | 331 +++++++++++++++++- edisgo/tools/tools.py | 21 +- run_example_05.py | 39 +++ tests/opf/test_powermodels_opf.py | 191 ++++++++++ tests/run/test_tasks.py | 247 ++++++++++++- tests/run/test_validator.py | 96 +++-- .../test_temporal_complexity_reduction.py | 89 +++++ 16 files changed, 1832 insertions(+), 86 deletions(-) create mode 100644 edisgo/run/presets/uc5_select_timesteps.yaml create mode 100644 run_example_05.py diff --git a/edisgo/io/powermodels_io.py b/edisgo/io/powermodels_io.py index a978a0805..223d96cf3 100644 --- a/edisgo/io/powermodels_io.py +++ b/edisgo/io/powermodels_io.py @@ -370,7 +370,7 @@ def from_powermodels( # calculate relative error df2 = deepcopy(df) for flex in df2.columns: - if type(hv_flex_dict[flex]) == pd.Series: + if isinstance(hv_flex_dict[flex], pd.Series): abs_error = abs(df2[flex].values - hv_flex_dict[flex].values) rel_error = [ abs_error[i] / hv_flex_dict[flex].iloc[i] @@ -379,10 +379,15 @@ def from_powermodels( for i in range(len(abs_error)) ] else: - abs_error = abs(df2[flex].values - hv_flex_dict[flex].sum(axis=1).values) + abs_error = abs( + df2[flex].values - hv_flex_dict[flex].sum(axis=1).values + ) rel_error = [ abs_error[i] / hv_flex_dict[flex].sum(axis=1).iloc[i] - if ((abs_error > 0.01)[i] & (hv_flex_dict[flex].sum(axis=1).iloc[i] != 0)) + if ( + (abs_error > 0.01)[i] + & (hv_flex_dict[flex].sum(axis=1).iloc[i] != 0) + ) else 0 for i in range(len(abs_error)) ] @@ -1061,6 +1066,18 @@ def _build_battery_storage( * edisgo_obj.topology.storage_units_df.max_hours ) + # The end-of-period SoC step (timeindex[-1] + freq) is only used as the OPF + # boundary (soc_end) and is not an optimized time step. When the time index + # is a reduced, non-contiguous selection, that step can fall in a gap and be + # missing from the source SoC series (which only carried a trailing step for + # the very last interval), leaving it NaN. A NaN boundary makes the Julia OPF + # fail with "Inf - Inf". Forward-fill (then back-fill) so the boundary takes + # the interval's last valid SoC — a harmless approximation for a throwaway + # scaffolding step. + edisgo_obj.overlying_grid.storage_units_soc = ( + edisgo_obj.overlying_grid.storage_units_soc.ffill().bfill() + ) + for stor_i in np.arange(len(flexible_storage_units)): idx_bus = _mapping( psa_net, @@ -1357,6 +1374,12 @@ def _build_heat_storage(psa_net, pm, edisgo_obj, s_base, flexible_hps, opf_versi edisgo_obj.overlying_grid.heat_storage_units_soc = pd.concat( [df_decentral, df_central], axis=1 ) + # Fill the end-of-period boundary SoC step (see storage note above) so a + # reduced, non-contiguous time index does not leave a NaN boundary that + # breaks the Julia OPF. + edisgo_obj.overlying_grid.heat_storage_units_soc = ( + edisgo_obj.overlying_grid.heat_storage_units_soc.ffill().bfill() + ) heat_storage_df = heat_storage_df.loc[flexible_hps] for stor_i in np.arange(len(flexible_hps)): @@ -1616,7 +1639,7 @@ def _build_hv_requirements( ) for i in np.arange(len(opf_flex)): - if type(hv_flex_dict[opf_flex[i]]) == pd.DataFrame: + if isinstance(hv_flex_dict[opf_flex[i]], pd.DataFrame): pm["HV_requirements"][str(i + 1)] = { "P": hv_flex_dict[opf_flex[i]].sum(axis=1).iloc[0], "name": opf_flex[i], @@ -1957,7 +1980,7 @@ def _build_component_timeseries( if (kind == "HV_requirements") & (pm["opf_version"] in [3, 4]): for i in np.arange(len(opf_flex)): - if type(hv_flex_dict[opf_flex[i]])==pd.DataFrame: + if isinstance(hv_flex_dict[opf_flex[i]], pd.DataFrame): pm_comp[(str(i + 1))] = { "P": hv_flex_dict[opf_flex[i]].sum(axis=1).round(20).tolist(), } diff --git a/edisgo/opf/powermodels_opf.py b/edisgo/opf/powermodels_opf.py index 22841f78b..a22b9c3e6 100644 --- a/edisgo/opf/powermodels_opf.py +++ b/edisgo/opf/powermodels_opf.py @@ -9,6 +9,7 @@ # # SPDX-License-Identifier: AGPL-3.0-or-later +import copy import json import logging import os @@ -16,6 +17,7 @@ import sys import numpy as np +import pandas as pd from edisgo.flex_opt import exceptions from edisgo.io.powermodels_io import from_powermodels @@ -23,6 +25,120 @@ logger = logging.getLogger(__name__) +# Time-indexed opf_results attributes that from_powermodels overwrites on each +# call. When the OPF is run separately per interval, these must be concatenated +# across intervals so opf_results covers the full (reduced) time index. Nested +# containers (LineVariables etc.) are listed via their sub-frame attribute names. +_OPF_FLAT_TIME_FRAMES = ( + "slack_generator_t", + "hv_requirement_slacks_t", +) +_OPF_NESTED_TIME_FRAMES = ( + "lines_t", + "heat_storage_t", + "grid_slacks_t", + "battery_storage_t", +) + + +def _with_freq(index): + """Return the DatetimeIndex with its frequency inferred/attached if regular. + + Reducing/uniting time indices drops the ``freq`` attribute; several + downstream consumers (notably the powermodels OPF) do + ``timeindex[-1] + timeindex.freq`` and break on ``freq is None``. This + re-attaches the freq when the index is regularly spaced (a no-op otherwise). + """ + if index.freq is not None or len(index) < 2: + return index + inferred = pd.infer_freq(index) + if inferred is not None: + try: + return pd.DatetimeIndex(index, freq=inferred) + except (ValueError, TypeError): + return index + return index + + +def _contiguous_intervals(timeindex): + """ + Split a time index into contiguous intervals. + + Automatic timestep selection can reduce the time index to disconnected + intervals (e.g. one load-case and one feed-in-case week). This helper detects + the gap(s) so the OPF can be run separately per interval — storage/heat state + does not carry across a gap, so a single OPF over the concatenated steps would + be wrong. + + A boundary is placed wherever the spacing between two consecutive time steps + exceeds the regular step (the smallest spacing in the index). A contiguous + index therefore yields a single interval. Each returned interval has its + ``freq`` restored (set operations that produced the reduced index drop it). + + Parameters + ---------- + timeindex : pandas.DatetimeIndex + + Returns + ------- + list of pandas.DatetimeIndex + One entry per contiguous interval, in chronological order. Empty index + in -> empty list out; a single time step -> one interval. + """ + timeindex = timeindex.sort_values() + if len(timeindex) <= 1: + return [timeindex] if len(timeindex) else [] + diffs = timeindex[1:] - timeindex[:-1] + step = diffs.min() + breaks = [i + 1 for i, d in enumerate(diffs) if d > step] + starts = [0] + breaks + ends = breaks + [len(timeindex)] + return [_with_freq(timeindex[s:e]) for s, e in zip(starts, ends)] + + +def _snapshot_opf_time_frames(opf_results): + """Copy the time-indexed opf_results frames produced by one interval's OPF.""" + snap = {} + for attr in _OPF_FLAT_TIME_FRAMES: + snap[attr] = getattr(opf_results, attr).copy() + for attr in _OPF_NESTED_TIME_FRAMES: + container = getattr(opf_results, attr) + snap[attr] = { + sub: getattr(container, sub).copy() for sub in container._attributes() + } + return snap + + +def _merge_opf_time_frames(opf_results, snapshots): + """ + Concatenate per-interval opf_results snapshots by time index and write them + back onto ``opf_results``, so its detailed frames cover the full reduced + index rather than only the last interval's. + """ + + def _concat(frames): + frames = [f for f in frames if f is not None and not f.empty] + if not frames: + return pd.DataFrame() + return pd.concat(frames).sort_index() + + for attr in _OPF_FLAT_TIME_FRAMES: + setattr(opf_results, attr, _concat([s[attr] for s in snapshots])) + for attr in _OPF_NESTED_TIME_FRAMES: + container = getattr(opf_results, attr) + for sub in container._attributes(): + setattr(container, sub, _concat([s[attr][sub] for s in snapshots])) + + # Recompute the overlying_grid summary (opf_version 3/4) from the merged HV + # requirement slacks, since it is a reduction over the whole time index. + hv = opf_results.hv_requirement_slacks_t + if not hv.empty: + opf_results.overlying_grid = pd.DataFrame( + columns=["Highest error", "Mean error", "Sum error"], + index=hv.columns, + data=pd.concat([hv.max(), hv.mean(), hv.sum()], axis=1).values, + ) + def pm_optimize( edisgo_obj, @@ -37,8 +153,162 @@ def pm_optimize( silence_moi: bool = False, ) -> None: """ - Run OPF for edisgo object in julia subprocess and write results of OPF to edisgo - object. Results of OPF are time series of operation schedules of flexibilities. + Run OPF for the edisgo object and write results back to it. + + If the time index is a single contiguous interval, this runs one OPF + (:func:`_pm_optimize_single`). If the time index is NON-contiguous + (disconnected intervals, e.g. from automatic timestep selection), each + contiguous interval is optimized separately and independently — storage/heat + state does not carry across the gap — and the results are combined: + + * per-interval operation schedules accumulate in ``edisgo.timeseries``; + * the detailed ``edisgo.opf_results`` frames are merged by time index; + * a per-interval solve report is stored in + ``edisgo.opf_results.interval_results``; + * if any interval was infeasible, an + :class:`~.flex_opt.exceptions.InfeasibleModelError` is raised after the + feasible intervals' results have been stored. + + The overlying-grid SOC attributes and reactive-power time series (which + ``to_powermodels`` / ``from_powermodels`` mutate or replace on the current + interval) are snapshotted and restored pristine before each interval so a + later interval sees intact input. Parameters are as for + :func:`_pm_optimize_single`. + """ + opf_kwargs = dict( + s_base=s_base, + flexible_cps=flexible_cps, + flexible_hps=flexible_hps, + flexible_loads=flexible_loads, + flexible_storage_units=flexible_storage_units, + opf_version=opf_version, + method=method, + warm_start=warm_start, + silence_moi=silence_moi, + ) + + intervals = _contiguous_intervals(edisgo_obj.timeseries.timeindex) + if len(intervals) <= 1: + # single contiguous optimization. Re-set the (freq-restored) interval so + # the OPF sees a time index with a frequency — set operations upstream + # (e.g. timestep selection) drop it, and the OPF needs timeindex.freq. + if intervals: + edisgo_obj.set_timeindex(intervals[0]) + _pm_optimize_single(edisgo_obj, **opf_kwargs) + return + + logger.info( + f"pm_optimize: time index has {len(intervals)} disconnected intervals; " + f"running a separate OPF per interval." + ) + full_timeindex = edisgo_obj.timeseries.timeindex + + # Snapshot the shared input state that per-interval OPF runs mutate: + # * overlying-grid SOC attributes are rewritten in place by to_powermodels; + # * the reactive-power time series are fully REPLACED (not .loc-updated) by + # the set_time_series_reactive_power_control() call inside from_powermodels. + # Reactive power was set on the full reduced index before this call; restore + # this input pristine before each interval. Active-power frames are NOT + # restored — they accumulate each interval's OPF results via .loc. + og = edisgo_obj.overlying_grid + og_snapshot = {attr: copy.deepcopy(getattr(og, attr)) for attr in og._attributes} + reactive_attrs = [ + "_generators_reactive_power", + "_loads_reactive_power", + "_storage_units_reactive_power", + ] + reactive_snapshot = { + attr: copy.deepcopy(getattr(edisgo_obj.timeseries, attr, None)) + for attr in reactive_attrs + } + + def _restore_pristine_inputs(): + for attr, value in og_snapshot.items(): + setattr(og, attr, copy.deepcopy(value)) + for attr, value in reactive_snapshot.items(): + if value is not None: + setattr(edisgo_obj.timeseries, attr, copy.deepcopy(value)) + + # Pre-allocate the storage active-power schedule over the FULL reduced index + # so from_powermodels .loc-accumulates each interval's storage result instead + # of replacing the frame with an interval-only one (which would drop earlier + # intervals' storage schedules). + su_names = edisgo_obj.topology.storage_units_df.index + if len(su_names) > 0 and edisgo_obj.timeseries.storage_units_active_power.empty: + edisgo_obj.timeseries.storage_units_active_power = pd.DataFrame( + 0.0, index=full_timeindex, columns=su_names + ) + + snapshots = [] + report = [] + try: + for interval in intervals: + _restore_pristine_inputs() + edisgo_obj.set_timeindex(interval) + entry = { + "start": interval[0], + "end": interval[-1], + "status": None, + "solver": None, + "solution_time": None, + } + try: + _pm_optimize_single(edisgo_obj, **opf_kwargs) + entry["status"] = edisgo_obj.opf_results.status + entry["solver"] = edisgo_obj.opf_results.solver + entry["solution_time"] = edisgo_obj.opf_results.solution_time + snapshots.append(_snapshot_opf_time_frames(edisgo_obj.opf_results)) + except exceptions.InfeasibleModelError: + entry["status"] = "infeasible" + logger.warning( + f"pm_optimize: OPF infeasible for interval " + f"{interval[0]}..{interval[-1]}." + ) + report.append(entry) + finally: + # restore the full (reduced) index so all intervals' schedules are exposed + # and undo the per-interval mutations of the overlying-grid/reactive input. + _restore_pristine_inputs() + edisgo_obj.set_timeindex(full_timeindex) + + _merge_opf_time_frames(edisgo_obj.opf_results, snapshots) + edisgo_obj.opf_results.interval_results = report + solution_times = [ + e["solution_time"] for e in report if e["solution_time"] is not None + ] + edisgo_obj.opf_results.solution_time = ( + sum(solution_times) if solution_times else None + ) + statuses = [e["status"] for e in report] + infeasible = [e for e in report if e["status"] == "infeasible"] + edisgo_obj.opf_results.status = ( + "infeasible" if infeasible else (statuses[0] if statuses else None) + ) + if infeasible: + raise exceptions.InfeasibleModelError( + f"OPF infeasible for {len(infeasible)} of {len(intervals)} time " + f"intervals; see edisgo.opf_results.interval_results. Results for " + f"feasible intervals have been stored." + ) + + +def _pm_optimize_single( + edisgo_obj, + s_base: int = 1, + flexible_cps: np.ndarray | None = None, + flexible_hps: np.ndarray | None = None, + flexible_loads: np.ndarray | None = None, + flexible_storage_units: np.ndarray | None = None, + opf_version: int = 1, + method: str = "soc", + warm_start: bool = False, + silence_moi: bool = False, +) -> None: + """ + Run a single-interval OPF for the edisgo object in a julia subprocess and + write results back to the edisgo object. Assumes the time index is a single + contiguous interval; :func:`pm_optimize` is the public entry point that + handles non-contiguous indices by calling this per interval. Parameters ---------- diff --git a/edisgo/opf/results/opf_result_class.py b/edisgo/opf/results/opf_result_class.py index f109c681e..83ac1e1a7 100644 --- a/edisgo/opf/results/opf_result_class.py +++ b/edisgo/opf/results/opf_result_class.py @@ -176,6 +176,13 @@ class OPFResults: Aggregated exchange with the overlying grid. battery_storage_t : :class:`~.opf.results.opf_result_class.BatteryStorage` Battery-storage results. + interval_results : list of dict + Per-interval solve report, populated when the OPF is run separately over + several disconnected time intervals (see the ``optimize`` pipeline task, + which splits a non-contiguous time index — e.g. from automatic timestep + selection — into independent optimizations). Each entry has keys + ``start``, ``end``, ``status``, ``solver`` and ``solution_time``. Empty + for a single contiguous optimization. """ @@ -190,6 +197,7 @@ def __init__(self): self.grid_slacks_t = GridSlacks() self.overlying_grid = pd.DataFrame() self.battery_storage_t = BatteryStorage() + self.interval_results = [] def to_csv(self, directory, attributes=None): """ diff --git a/edisgo/run/presets/uc4_example.yaml b/edisgo/run/presets/uc4_example.yaml index 5ec024f6c..02b26ee53 100644 --- a/edisgo/run/presets/uc4_example.yaml +++ b/edisgo/run/presets/uc4_example.yaml @@ -31,9 +31,9 @@ database: timeindex: {start: "2035-01-01", periods: 24, freq: h} overlying_grid: - enabled: false # master switch — set true to activate import_overlying_grid_data + enabled: true # master switch — set true to activate import_overlying_grid_data source: csv # "csv" (load from path) or "etrago" (consume overlying_grid_data kwarg) - path: "/path/to/overlying_grid_csv_dir" # required when source == csv; full leaf dir for ONE grid (like ding0_path) + path: "/storage/JoDa/edisgo_playground/overlying_grid_data" # required when source == csv; full leaf dir for ONE grid (like ding0_path) results: directory: results/uc4_example @@ -58,7 +58,7 @@ pipeline: - optimize: flexible: [heat_pumps, storage, charging_points, dsm] method: soc - opf_version: 3 + opf_version: 2 - reinforce - save: archive: true diff --git a/edisgo/run/presets/uc5_select_timesteps.yaml b/edisgo/run/presets/uc5_select_timesteps.yaml new file mode 100644 index 000000000..426ce1231 --- /dev/null +++ b/edisgo/run/presets/uc5_select_timesteps.yaml @@ -0,0 +1,117 @@ +_comment: | + UC5 — OPF with configurable timestep selection (manual OR auto): + Like UC4 (full flexibility OPF with HV requirements), but the time index + is reduced to a selected subset instead of a fixed window. The mode is + chosen in the timeseries_selection block below (typically overridden in + the run script). + + The pipeline carries TWO select_timesteps steps, each with a `position`: + - position: pre_import (before import_heat_pumps) — acts only in MANUAL + mode. It sets the explicit time index, which the heat-pump/DSM imports + and oedb_ts then use to fetch only the selected steps (cheap). + - position: post_grid (after import_overlying_grid_data, before + reactive_power) — acts only in AUTO mode. It needs all active-power + time series (incl. overlying-grid generation) set to run the scoring + power flow via get_most_critical_time_intervals. + Whichever mode is configured, the other positioned step is a no-op. + + Auto mode normally yields two disconnected intervals (one overloading, + one voltage). They are kept separate (a gap in the time index); if they + overlap, a non-overlapping pair is chosen if possible, otherwise they are + concatenated into one interval. A later optimize step can detect the gap + and run separate optimizations per interval. + +_workflow: + - setup_grid: load ding0 topology + - import_generators / import_home_batteries + - select_timesteps (pre_import): manual only — set explicit time index + - import_heat_pumps / import_dsm: fetch only selected steps (manual) + - import_electromobility: dumb charging, flex bands + - oedb_ts: real wind/solar + load time series + - apply_charging_strategy / apply_heat_pump_strategy + - build_flexibility_bands: EV bands on the fixed hourly index + - import_overlying_grid_data: HV constraints from CSV dir + - select_timesteps (post_grid): auto only — reduce to critical intervals + - reactive_power: fixed cosphi on the reduced index + - optimize: pm_optimize with flex assets + - reinforce / save + +# Self-contained (no `extends`): everything uc4_example provided is inlined +# below, so this preset can be run on its own via +# run_edisgo({"extends": "uc5_select_timesteps", "grid": {"ding0_path": ...}}) +scenario: eGon2035 + +grid: + ding0_path: "/path/to/ding0_grid" + legacy_ding0_grids: false + +database: + ssh: + enabled: false + +# No explicit base time index is set. oedb_ts falls back to a full year derived +# from the scenario when none is given, which is what auto interval selection +# needs (week-long critical intervals to pick from). For manual selection the +# pre-import select_timesteps step sets the index instead. + +overlying_grid: + enabled: true # set true to activate import_overlying_grid_data + source: csv # "csv" (load from path) or "etrago" (kwarg) + path: "/storage/JoDa/edisgo_playground/overlying_grid_data" + +results: + directory: results/uc5_select_timesteps + +# Top-level block read by the select_timesteps task via ctx.raw_config. +# eGo can inject this block the same way it injects overlying_grid. +# Set `mode` (and its parameters) here or override it in the run script. +timeseries_selection: + mode: auto + # auto method: "power_flow" (default, scores intervals via a power flow) or + # "residual_load" (no power flow — the weeks ending at the max/min residual-load + # time steps; requires overlying-grid data). + method: residual_load + # --- shared auto parameters (both methods) --- + time_steps_per_time_interval: 168 # one week (must be a multiple of 24) + time_step_day_start: 4 # hour of day the intervals start/end on + # --- power_flow method parameters --- + percentage: 1.0 + save_steps: true # write selected intervals CSV to results_dir + use_troubleshooting_mode: true # handle power-flow non-convergence + overloading_factor: 0.95 + voltage_deviation_factor: 0.95 + # --- manual parameters (used when mode: manual) --- + # timestamps: ["2035-01-15 08:00", "2035-01-15 9:00", "2035-01-15 10:00"] + # or a range instead of `timestamps`: + start: "2035-01-15 00:00" + periods: 24 + freq: h + +pipeline: + - setup_grid + - import_generators + - import_home_batteries + - select_timesteps: {position: pre_import} # acts in manual mode only + - import_heat_pumps + - import_dsm + - import_electromobility: + charging_strategy: null + # flexibility bands are built later (build_flexibility_bands), once the + # analysis time index is fixed, so they are resampled to it + - oedb_ts: + dispatchable: {other: 0.7} + - apply_charging_strategy: {strategy: dumb} + - apply_heat_pump_strategy: {strategy: uncontrolled} + - build_flexibility_bands # hourly bands on the 2035 index + - import_overlying_grid_data + - select_timesteps: {position: post_grid} # acts in auto mode only + - reactive_power + - optimize: + flexible: [heat_pumps, storage, charging_points, dsm] + method: soc + opf_version: 2 + - reinforce: + catch_convergence_problems: true + - save: + archive: true + save_opf_results: true diff --git a/edisgo/run/tasks/__init__.py b/edisgo/run/tasks/__init__.py index 0d59ea02a..9a1be82a8 100644 --- a/edisgo/run/tasks/__init__.py +++ b/edisgo/run/tasks/__init__.py @@ -1,3 +1,13 @@ +# This file is part of eDisGo (Electrical Distribution Grid Optimization), +# a Python package for analyzing flexibility options in distribution grids. +# +# Copyright (c) Reiner Lemoine Institut gGmbH +# Contributors are listed in the version control history: +# https://github.com/openego/eDisGo/ +# +# Documentation: https://edisgo.readthedocs.io/ +# +# SPDX-License-Identifier: AGPL-3.0-or-later """ Task implementations for the eDisGo pipeline runner. @@ -10,9 +20,9 @@ ``set_timeindex``, ``reactive_power`` * :mod:`.flex` — flex imports (``import_heat_pumps``, ``import_home_batteries``, ``import_dsm``, - ``import_electromobility``, ``import_generators``) and operating - strategies (``apply_charging_strategy``, - ``apply_heat_pump_strategy``) + ``import_electromobility``, ``import_generators``), + ``build_flexibility_bands``, and operating strategies + (``apply_charging_strategy``, ``apply_heat_pump_strategy``) * :mod:`.analysis` — ``check_integrity``, ``analyze``, ``reinforce``, ``base_reinforce``, ``optimize`` * :mod:`.io` — ``save``, ``load_charging_from_files`` @@ -22,4 +32,5 @@ returned value, if non-None, replaces the current one in the runner's loop). """ + from edisgo.run.tasks import analysis, flex, grid, io, timeseries # noqa: F401 diff --git a/edisgo/run/tasks/analysis.py b/edisgo/run/tasks/analysis.py index a0d0f4ebd..7bbb4ba3f 100644 --- a/edisgo/run/tasks/analysis.py +++ b/edisgo/run/tasks/analysis.py @@ -1,3 +1,13 @@ +# This file is part of eDisGo (Electrical Distribution Grid Optimization), +# a Python package for analyzing flexibility options in distribution grids. +# +# Copyright (c) Reiner Lemoine Institut gGmbH +# Contributors are listed in the version control history: +# https://github.com/openego/eDisGo/ +# +# Documentation: https://edisgo.readthedocs.io/ +# +# SPDX-License-Identifier: AGPL-3.0-or-later """ Power-flow, reinforcement, and optimization tasks. @@ -286,12 +296,18 @@ def task_optimize( Explicit ``flexible_*`` kwargs override the shortcut. + This task only performs the ``flexible`` shortcut expansion (mode selection) + and calls :meth:`EDisGo.pm_optimize`. Handling of a non-contiguous (reduced) + time index — running a separate OPF per contiguous interval and merging the + results — lives in :func:`~.opf.powermodels_opf.pm_optimize`. + Parameters ---------- edisgo : edisgo.EDisGo EDisGo instance to optimize. ctx : RunContext - Run context (unused). + Run context. Used for logging and, for multi-interval runs, nothing + else is required from it. flexible : list of str, optional High-level selector, subset of ``{"heat_pumps", "charging_points", "storage"}``. If ``None``, nothing is @@ -343,6 +359,8 @@ def task_optimize( if flexible_storage_units is None: flexible_storage_units = [] + # pm_optimize handles a non-contiguous (reduced) time index internally: + # it runs one OPF per contiguous interval and merges the results. edisgo.pm_optimize( flexible_cps=flexible_cps, flexible_hps=flexible_hps, diff --git a/edisgo/run/tasks/flex.py b/edisgo/run/tasks/flex.py index 87ba4a35e..3e33ffa3f 100644 --- a/edisgo/run/tasks/flex.py +++ b/edisgo/run/tasks/flex.py @@ -1,3 +1,13 @@ +# This file is part of eDisGo (Electrical Distribution Grid Optimization), +# a Python package for analyzing flexibility options in distribution grids. +# +# Copyright (c) Reiner Lemoine Institut gGmbH +# Contributors are listed in the version control history: +# https://github.com/openego/eDisGo/ +# +# Documentation: https://edisgo.readthedocs.io/ +# +# SPDX-License-Identifier: AGPL-3.0-or-later """ Flex-asset import and operation-strategy tasks. @@ -8,6 +18,7 @@ and typically BEFORE the time-series step, so the time series can cover the new assets. """ + from __future__ import annotations from edisgo.run.registry import register_task @@ -30,7 +41,10 @@ def task_import_heat_pumps(edisgo, ctx, *, import_types=None, timeindex=None): Subset of ``["individual_heat_pumps", "central_heat_pumps"]``; default imports both. timeindex : pandas.DatetimeIndex, optional - Restrict COP / heat-demand time series to this index. + Restrict COP / heat-demand time series to this index. If None, + falls back to ``ctx.flags['selected_timeindex']`` (set by a + preceding ``select_timesteps`` manual step) so the download is + restricted to the selected steps. Returns ------- @@ -38,17 +52,18 @@ def task_import_heat_pumps(edisgo, ctx, *, import_types=None, timeindex=None): The modified EDisGo instance. """ + if timeindex is None: + timeindex = ctx.flags.get("selected_timeindex") edisgo.import_heat_pumps( scenario=ctx.scenario, engine=ctx.ensure_engine(), timeindex=timeindex, import_types=import_types, ) - ctx.flags["has_heat_pumps"] = len( - edisgo.topology.loads_df.loc[ - edisgo.topology.loads_df.type == "heat_pump" - ] - ) > 0 + ctx.flags["has_heat_pumps"] = ( + len(edisgo.topology.loads_df.loc[edisgo.topology.loads_df.type == "heat_pump"]) + > 0 + ) return edisgo @@ -72,12 +87,8 @@ def task_import_home_batteries(edisgo, ctx): The modified EDisGo instance. """ - edisgo.import_home_batteries( - scenario=ctx.scenario, engine=ctx.ensure_engine() - ) - ctx.flags["has_home_batteries"] = ( - not edisgo.topology.storage_units_df.empty - ) + edisgo.import_home_batteries(scenario=ctx.scenario, engine=ctx.ensure_engine()) + ctx.flags["has_home_batteries"] = not edisgo.topology.storage_units_df.empty return edisgo @@ -94,7 +105,10 @@ def task_import_dsm(edisgo, ctx, *, timeindex=None): Run context. Uses ``ctx.scenario`` and ``ctx.ensure_engine()``. Sets ``ctx.flags['has_dsm']``. timeindex : pandas.DatetimeIndex, optional - Restrict DSM availability time series to this index. + Restrict DSM availability time series to this index. If None, + falls back to ``ctx.flags['selected_timeindex']`` (set by a + preceding ``select_timesteps`` manual step) so the download is + restricted to the selected steps. Returns ------- @@ -102,23 +116,28 @@ def task_import_dsm(edisgo, ctx, *, timeindex=None): The modified EDisGo instance. """ + if timeindex is None: + timeindex = ctx.flags.get("selected_timeindex") edisgo.import_dsm( scenario=ctx.scenario, engine=ctx.ensure_engine(), timeindex=timeindex, ) - ctx.flags["has_dsm"] = ( - edisgo.dsm.p_max is not None and not edisgo.dsm.p_max.empty - ) + ctx.flags["has_dsm"] = edisgo.dsm.p_max is not None and not edisgo.dsm.p_max.empty return edisgo @register_task("import_electromobility", requires={"grid"}, provides={"flex"}) -def task_import_electromobility(edisgo, ctx, *, data_source="oedb", - charging_strategy="dumb", - flexibility_bands_ucs = None, - import_electromobility_data_kwds=None, - allocate_charging_demand_kwds=None): +def task_import_electromobility( + edisgo, + ctx, + *, + data_source="oedb", + charging_strategy="dumb", + flexibility_bands_ucs=None, + import_electromobility_data_kwds=None, + allocate_charging_demand_kwds=None, +): """ Import electromobility data (charging processes + parks). @@ -148,7 +167,11 @@ def task_import_electromobility(edisgo, ctx, *, data_source="oedb", and charging-strategy application. Valid entries: ``"home"``, ``"work"``, ``"public"``, ``"hpc"``. Pass a single string for one use case or a list for multiple. ``None`` - (default) skips flexibility-band computation. + (default) skips flexibility-band computation — build them later + with the standalone :func:`task_build_flexibility_bands` once the + analysis time index is fixed, so the bands are resampled to it + (mirrors heat-pump handling, where the HP time series are not set + inside ``import_heat_pumps``). import_electromobility_data_kwds : dict, optional Extra kwargs passed through to the underlying importer. allocate_charging_demand_kwds : dict, optional @@ -178,9 +201,47 @@ def task_import_electromobility(edisgo, ctx, *, data_source="oedb", return edisgo +@register_task("build_flexibility_bands", requires={"flex"}) +def task_build_flexibility_bands(edisgo, ctx, *, use_case=None): + """ + Build EV charging flexibility bands from imported electromobility data. + + Standalone variant of the band computation that + :func:`task_import_electromobility` can do inline. Running it as a + separate step lets it execute *after* the analysis time index is fixed + (e.g. after ``oedb_ts`` / timestep selection), so + :meth:`Electromobility.get_flexibility_bands` resamples the bands to the + edisgo time-series frequency instead of leaving them at the raw SimBEV + resolution. This mirrors how the heat-pump time series are set outside + ``import_heat_pumps``, and is more efficient than building bands over a + non-final index. + + Parameters + ---------- + edisgo : edisgo.EDisGo + EDisGo instance to modify in place. + ctx : RunContext + Run context. + use_case : str or list of str, optional + Charging-point use case(s) to compute bands for. Valid entries: + ``"home"``, ``"work"``, ``"public"``, ``"hpc"``. Defaults to all + four. + + Returns + ------- + edisgo.EDisGo + The modified EDisGo instance. + """ + if use_case is None: + use_case = ["home", "work", "public", "hpc"] + edisgo.electromobility.get_flexibility_bands(edisgo, use_case=use_case) + return edisgo + + @register_task("apply_charging_strategy") -def task_apply_charging_strategy(edisgo, ctx, *, strategy="dumb", - charging_park_ids=None): +def task_apply_charging_strategy( + edisgo, ctx, *, strategy="dumb", charging_park_ids=None +): """ Apply a charging strategy to the already-imported EV fleet. @@ -212,8 +273,9 @@ def task_apply_charging_strategy(edisgo, ctx, *, strategy="dumb", @register_task("apply_heat_pump_strategy") -def task_apply_heat_pump_strategy(edisgo, ctx, *, strategy="uncontrolled", - heat_pump_names=None): +def task_apply_heat_pump_strategy( + edisgo, ctx, *, strategy="uncontrolled", heat_pump_names=None +): """ Apply a heat-pump operating strategy. @@ -239,10 +301,7 @@ def task_apply_heat_pump_strategy(edisgo, ctx, *, strategy="uncontrolled", """ if not ctx.flags.get("has_heat_pumps"): - ctx.logger.info( - "Skipping 'apply_heat_pump_strategy': no heat pumps " - "present." - ) + ctx.logger.info("Skipping 'apply_heat_pump_strategy': no heat pumps present.") return edisgo edisgo.apply_heat_pump_operating_strategy( strategy=strategy, heat_pump_names=heat_pump_names @@ -276,7 +335,5 @@ def task_import_generators(edisgo, ctx, *, generator_scenario=None): The modified EDisGo instance. """ - edisgo.import_generators( - generator_scenario=generator_scenario or ctx.scenario - ) + edisgo.import_generators(generator_scenario=generator_scenario or ctx.scenario) return edisgo diff --git a/edisgo/run/tasks/timeseries.py b/edisgo/run/tasks/timeseries.py index d4a6c8a30..bbe776398 100644 --- a/edisgo/run/tasks/timeseries.py +++ b/edisgo/run/tasks/timeseries.py @@ -1,3 +1,13 @@ +# This file is part of eDisGo (Electrical Distribution Grid Optimization), +# a Python package for analyzing flexibility options in distribution grids. +# +# Copyright (c) Reiner Lemoine Institut gGmbH +# Contributors are listed in the version control history: +# https://github.com/openego/eDisGo/ +# +# Documentation: https://edisgo.readthedocs.io/ +# +# SPDX-License-Identifier: AGPL-3.0-or-later """ Time-series tasks — set active/reactive power profiles on EDisGo. @@ -8,7 +18,10 @@ 1. Set the time index and active-power profiles with one of :func:`task_worst_case_ts`, :func:`task_oedb_ts`, :func:`task_manual_ts`, possibly :func:`task_set_timeindex`. -2. Finally call :func:`task_reactive_power` to fix reactive power +2. Optionally reduce the time index to a selected subset with + :func:`task_select_timesteps` (manual before the imports, auto after + ``import_overlying_grid_data``). +3. Finally call :func:`task_reactive_power` to fix reactive power control — this MUST come last because it overwrites whatever reactive power was set by the earlier steps. """ @@ -181,6 +194,20 @@ def task_oedb_ts( freq=timeindex.get("freq", "h"), ) edisgo.set_timeindex(ti_df) + elif edisgo.timeseries.timeindex.empty: + # No explicit timeindex and none set yet (e.g. no manual time series + # earlier): fall back to a full year derived from the scenario, the + # same default the flex imports use. + from edisgo.tools.tools import get_year_based_on_scenario + + year = get_year_based_on_scenario(ctx.scenario) + if year is None: + raise ValueError( + f"Cannot derive a default time index: invalid scenario " + f"{ctx.scenario!r}. Provide a 'timeindex' or a valid scenario " + f"('eGon2035', 'eGon100RE')." + ) + edisgo.set_timeindex(pd.date_range(f"1/1/{year}", periods=8760, freq="h")) dispatchable_df = None if dispatchable is not None: @@ -277,6 +304,283 @@ def _as_df(obj): return edisgo +def _set_default_full_year_timeindex(edisgo, ctx): + """ + Set a full-year hourly time index derived from the scenario. + + Used as a fallback so time-index-dependent imports (notably the EV + flexibility bands built in ``import_electromobility``) run on an hourly, + full-year index rather than their raw source resolution. The year is only a + label — the DB imports fetch scenario-correct data regardless — and the index + can be overridden later (e.g. by ``oedb_ts`` or the auto ``select_timesteps`` + step). + """ + from edisgo.tools.tools import get_year_based_on_scenario + + year = get_year_based_on_scenario(ctx.scenario) or 2011 + edisgo.set_timeindex(pd.date_range(f"1/1/{year}", periods=8760, freq="h")) + ctx.logger.info( + f"select_timesteps: no time index set; using default full year " + f"{year} (8760 h) so imports build hourly full-year data." + ) + + +@register_task("select_timesteps", provides={"timeseries"}, ts_altering=True) +def task_select_timesteps(edisgo, ctx, **overrides): + """ + Select the time steps the grid is analyzed/optimized for. + + Reduces the time index to a configurable subset. Configuration is + read from the top-level ``timeseries_selection:`` config block (so + eGo can inject it the same way it injects ``overlying_grid``); + inline step params override individual keys of that block. Two + modes: + + ``manual`` + Reduce to an explicit set of time steps. Positioned *before* + the data-import tasks so ``import_heat_pumps`` / ``import_dsm`` + download only the requested steps. The selected index is + stashed in ``ctx.flags['selected_timeindex']`` for those + imports to pick up. + + ``auto`` + Determine the two most critical time intervals and reduce to + them. Must be positioned *after* ``import_overlying_grid_data`` + (needs all active-power time series) and *before* + ``reactive_power``. Two ``method`` options: + + * ``power_flow`` (default) — score intervals via a power flow + (:func:`~.tools.temporal_complexity_reduction.get_most_critical_time_intervals`). + A reactive-power series is set internally to run the scoring + power flow, but ``ctx.flags['reactive_power_set']`` is left + unset so the pipeline's own ``reactive_power`` step still runs + on the reduced index. + * ``residual_load`` — no power flow. The overlying-grid dispatch + is distributed onto the components and the residual load is + ranked over the whole year; intervals are centered on the + highest (load case) and lowest (feed-in case) residual-load + steps. Requires overlying-grid data to be present. + + Both methods delegate to + :func:`~.tools.temporal_complexity_reduction.get_most_critical_time_intervals` + (via its ``by`` parameter) and reduce to a non-overlapping pair + chosen by + :func:`~.tools.temporal_complexity_reduction.select_two_intervals`. + + The auto mode normally yields two disconnected intervals (one for + overloading, one for voltage issues). These are kept separate in the + resulting time index (there is a gap between them). If they overlap, + a non-overlapping pair is chosen if possible, otherwise they are + concatenated into one interval. The intervals themselves are not + stored — a later ``optimize`` step can detect the gap in the time + index and run separate optimizations per interval. + + Parameters + ---------- + edisgo : edisgo.EDisGo + EDisGo instance to modify in place. + ctx : RunContext + Run context. Reads ``ctx.raw_config['timeseries_selection']``. + Sets ``ctx.flags['selected_timeindex']`` (manual) and + ``ctx.flags['timesteps_selected'] = True``. + **overrides + Inline step params overriding keys of the + ``timeseries_selection`` block. Recognized keys: + ``position`` (``"pre_import"`` | ``"post_grid"``, optional) — the + step only acts when the configured ``mode`` matches this + position (``pre_import`` ↔ ``manual``, ``post_grid`` ↔ ``auto``), + otherwise it is a no-op; this lets one pipeline carry both a + pre-import and a post-grid ``select_timesteps`` step and support + either mode via config. When omitted, the step always acts. + ``mode`` (``"manual"`` | ``"auto"``); + for manual: ``timestamps`` (list) or ``start`` / + ``periods`` / ``end`` / ``freq``; + for auto: ``method`` (``"power_flow"`` (default) | + ``"residual_load"``), ``time_steps_per_time_interval``; + for ``method="power_flow"`` additionally ``percentage``, + ``time_step_day_start`` (default 4), ``save_steps`` (default + True; CSV written to ``ctx.results_dir``), + ``use_troubleshooting_mode``, ``overloading_factor``, + ``voltage_deviation_factor``. + + Returns + ------- + edisgo.EDisGo + The modified EDisGo instance. + + Raises + ------ + ValueError + If ``mode`` is missing/unknown, if manual mode has neither + ``timestamps`` nor a range, or if auto mode runs before active + power time series are set. + """ + from edisgo.tools.temporal_complexity_reduction import ( + get_most_critical_time_intervals, + select_two_intervals, + ) + from edisgo.tools.tools import reduce_timeseries_data_to_given_timeindex + + cfg = {**ctx.raw_config.get("timeseries_selection", {}), **overrides} + mode = cfg.get("mode") + if mode not in ("manual", "auto"): + raise ValueError( + f"select_timesteps needs mode 'manual' or 'auto', got {mode!r}." + ) + + # A pipeline may include two select_timesteps steps — one before the + # imports (``position: pre_import``, where manual selection belongs) and + # one after import_overlying_grid_data (``position: post_grid``, where auto + # selection belongs) — so the same preset supports both modes. Each step + # only acts when the configured mode matches its position; otherwise it is + # a no-op. When ``position`` is omitted (single-step usage) the step always + # acts. + position = overrides.get("position") + expected_mode = {"pre_import": "manual", "post_grid": "auto"} + if position is not None: + if position not in expected_mode: + raise ValueError( + f"select_timesteps 'position' must be 'pre_import' or " + f"'post_grid', got {position!r}." + ) + if mode != expected_mode[position]: + # This positioned step is not the active selector. If it is the + # pre-import step and no time index has been set yet (i.e. manual + # selection is not driving the index), establish a full-year default + # so the following imports build their time-index-dependent data + # (e.g. EV flexibility bands) on an hourly full-year index. Only a + # label — DB imports fetch scenario-correct data regardless — and it + # is overridden later by oedb_ts / the auto select_timesteps step. + if position == "pre_import" and edisgo.timeseries.timeindex.empty: + _set_default_full_year_timeindex(edisgo, ctx) + ctx.logger.debug( + f"select_timesteps at position {position!r} is a no-op for " + f"mode {mode!r}." + ) + return edisgo + + if mode == "manual": + timestamps = cfg.get("timestamps") + if timestamps is not None: + timeindex = pd.DatetimeIndex(pd.to_datetime(list(timestamps))) + elif cfg.get("end") is not None: + timeindex = pd.date_range( + start=cfg["start"], end=cfg["end"], freq=cfg.get("freq", "h") + ) + elif cfg.get("periods") is not None: + timeindex = pd.date_range( + start=cfg["start"], + periods=cfg["periods"], + freq=cfg.get("freq", "h"), + ) + else: + raise ValueError( + "select_timesteps manual mode needs 'timestamps' or a " + "'start' plus 'periods'/'end' range." + ) + timeindex = timeindex.sort_values().unique() + if not edisgo.timeseries.timeindex.empty: + # A time index is already set (manual selection reducing an existing + # full time series): align the user-supplied timestamps to that + # index's year so date-based slicing matches even if the user wrote + # them in a different (e.g. scenario) year than the internally used + # reference year. + year_diff = edisgo.timeseries.timeindex[0].year - timeindex[0].year + if year_diff != 0: + timeindex = timeindex + pd.DateOffset(years=year_diff) + ctx.flags["selected_timeindex"] = timeindex + if edisgo.timeseries.timeindex.empty: + # positioned before imports: just set the index so HP/DSM + # imports restrict their downloads to it + edisgo.set_timeindex(timeindex) + else: + reduce_timeseries_data_to_given_timeindex(edisgo, timeindex) + ctx.logger.info( + f"select_timesteps (manual): selected {len(timeindex)} time steps." + ) + ctx.flags["timesteps_selected"] = True + return edisgo + + # auto mode + if not ctx.flags.get("timeseries_set"): + raise ValueError( + "select_timesteps mode 'auto' needs active-power time series to " + "be set first (e.g. run oedb_ts before it)." + ) + + method = cfg.get("method", "power_flow") + if method not in ("power_flow", "residual_load"): + raise ValueError( + f"select_timesteps auto 'method' must be 'power_flow' or " + f"'residual_load', got {method!r}." + ) + tsp = cfg.get("time_steps_per_time_interval", 168) + + if method == "residual_load": + # residual-load selection requires overlying-grid data (the dispatch + # distributed onto the components); guard here (mode selection) before + # delegating the computation to the tools function. + og = edisgo.overlying_grid + if all( + s.empty + for s in ( + og.electromobility_active_power, + og.storage_units_active_power, + og.heat_pump_central_active_power, + og.heat_pump_decentral_active_power, + og.dsm_active_power, + og.renewables_curtailment, + ) + ): + raise ValueError( + "select_timesteps method 'residual_load' needs overlying-grid " + "data to be present (run import_overlying_grid_data before it)." + ) + col_a, col_b = "time_steps_load_case", "time_steps_feedin_case" + else: # power_flow + # throwaway reactive power so the scoring power flow yields meaningful + # voltages; do NOT mark reactive_power_set — the pipeline's own + # reactive_power step runs afterwards on the reduced index. + edisgo.set_time_series_reactive_power_control(control="fixed_cosphi") + col_a, col_b = "time_steps_overloading", "time_steps_voltage_issues" + + intervals_df = get_most_critical_time_intervals( + edisgo, + by=method, + percentage=cfg.get("percentage", 1.0), + time_steps_per_time_interval=tsp, + time_step_day_start=cfg.get("time_step_day_start", 4), + save_steps=cfg.get("save_steps", True), + path=str(ctx.results_dir) if ctx.results_dir is not None else "", + use_troubleshooting_mode=cfg.get("use_troubleshooting_mode", True), + overloading_factor=cfg.get("overloading_factor", 0.95), + voltage_deviation_factor=cfg.get("voltage_deviation_factor", 0.95), + ) + intervals = select_two_intervals( + list(intervals_df.get(col_a, [])), + list(intervals_df.get(col_b, [])), + ) + + if not intervals: + raise ValueError( + "select_timesteps mode 'auto' found no critical time intervals; " + "cannot reduce the time index." + ) + + timeindex = intervals[0] + for interval in intervals[1:]: + timeindex = timeindex.union(interval) + timeindex = timeindex.sort_values() + + reduce_timeseries_data_to_given_timeindex(edisgo, timeindex) + ctx.logger.info( + f"select_timesteps (auto): selected {len(intervals)} interval(s), " + f"{len(timeindex)} time steps total." + ) + ctx.flags["timesteps_selected"] = True + return edisgo + + @register_task("reactive_power") def task_reactive_power( edisgo, diff --git a/edisgo/tools/temporal_complexity_reduction.py b/edisgo/tools/temporal_complexity_reduction.py index 2adc269f9..df4bb0730 100644 --- a/edisgo/tools/temporal_complexity_reduction.py +++ b/edisgo/tools/temporal_complexity_reduction.py @@ -647,6 +647,189 @@ def _troubleshooting_mode( return edisgo_obj +def intervals_overlap(a, b): + """ + Return True if two contiguous time-step intervals overlap. + + Each interval is a :pandas:`pandas.DatetimeIndex` of + contiguous, sorted time steps. Overlap is checked on the closed + ``[min, max]`` ranges, so intervals that merely touch (share an end point) + count as overlapping. + """ + return (a.min() <= b.max()) and (b.min() <= a.max()) + + +def select_two_intervals(load_case, feedin_case): + """ + Pick the time intervals to analyze from two ranked candidate lists. + + Used to reduce the ranked most-critical intervals (e.g. from + :func:`get_most_critical_time_intervals`) to the intervals actually analyzed: + + * Start from the most critical interval of each list (index 0). + * If they do not overlap, both are kept — two disconnected intervals are + returned (a later optimization can detect the gap and optimize each + interval separately). + * If they overlap, walk down the second list to the highest ranked interval + that does not overlap the top interval of the first list, and keep that + pair instead. + * If no non-overlapping pair exists, the two most critical intervals are + concatenated into a single contiguous interval and only that one is + returned. + + Parameters + ---------- + load_case : list of pandas.DatetimeIndex + First ranked list of intervals (most critical first). May be empty. + feedin_case : list of pandas.DatetimeIndex + Second ranked list of intervals (most critical first). May be empty. + + Returns + ------- + list of pandas.DatetimeIndex + One or two contiguous, non-overlapping intervals. Empty if both input + lists are empty. + """ + if not load_case and not feedin_case: + return [] + if not load_case: + return [feedin_case[0]] + if not feedin_case: + return [load_case[0]] + + top = load_case[0] + for cand in feedin_case: + if not intervals_overlap(top, cand): + return [top, cand] + + # no non-overlapping pair -> concatenate the two most critical intervals + merged = top.union(feedin_case[0]).sort_values() + start, end = merged.min(), merged.max() + freq = pd.infer_freq(top) or "H" + return [pd.date_range(start=start, end=end, freq=freq)] + + +def _build_centered_interval( + timestep, timeindex, time_steps_per_time_interval, time_step_day_start +): + """ + Build a contiguous interval centered on a critical time step. + + The interval has ``time_steps_per_time_interval`` steps, is centered on + ``timestep``, and its start is snapped to the ``time_step_day_start`` hour of + day (so intervals begin on that hour). The interval is clipped to lie within + ``timeindex``; if centering would run past either end, it is shifted inward. + Centering guarantees the critical step is not the last step of the interval + (where a storage state-of-charge-end constraint would force zero power). + + Parameters + ---------- + timestep : pandas.Timestamp + Critical time step to center on. + timeindex : pandas.DatetimeIndex + The full (sorted) time index the interval must lie within. + time_steps_per_time_interval : int + Interval length in steps. + time_step_day_start : int + Hour of day the interval should start on. + + Returns + ------- + pandas.DatetimeIndex + The contiguous interval. + """ + n = int(time_steps_per_time_interval) + step = timeindex[1] - timeindex[0] + half = (n // 2) * step + # center on the critical step, then snap the start back to the day-start hour + start = timestep - half + while start.hour != int(time_step_day_start): + start = start - step + end = start + (n - 1) * step + # keep the interval within the available time index; shift inward if needed + if start < timeindex[0]: + start = timeindex[0] + end = start + (n - 1) * step + if end > timeindex[-1]: + end = timeindex[-1] + start = end - (n - 1) * step + if start < timeindex[0]: + start = timeindex[0] + return pd.date_range(start=start, end=end, freq=step) + + +def _most_critical_time_intervals_residual_load( + edisgo_obj, + num_time_intervals=None, + percentage=1.0, + time_steps_per_time_interval=168, + time_step_day_start=0, + save_steps=False, + path="", +): + """ + Determine the most critical time intervals from the residual load. + + Ranks the critical single time steps by residual load (via + :func:`get_most_critical_time_steps` with ``by="residual_load"``) and wraps + each into an interval centered on the step and snapped to the + ``time_step_day_start`` hour (see :func:`_build_centered_interval`). Returns a + DataFrame ranked by residual magnitude with per-case columns + ``time_steps_load_case`` (highest residual) and ``time_steps_feedin_case`` + (lowest residual). Overlaps between intervals are allowed (mirroring the + power-flow interval selection); a downstream :func:`select_two_intervals` + picks a non-overlapping pair. + """ + timeindex = edisgo_obj.timeseries.timeindex + + # number of ranked intervals per case + if num_time_intervals is None: + num_time_intervals = int(np.ceil(len(timeindex) * percentage)) + + from edisgo.network.overlying_grid import ( + distribute_overlying_grid_requirements, + ) + + distributed = distribute_overlying_grid_requirements(edisgo_obj) + residual = distributed.timeseries.residual_load + + load_steps = residual.sort_values(ascending=False).index[:num_time_intervals] + feedin_steps = residual.sort_values(ascending=True).index[:num_time_intervals] + + load_intervals = [ + _build_centered_interval( + t, timeindex, time_steps_per_time_interval, time_step_day_start + ) + for t in load_steps + ] + feedin_intervals = [ + _build_centered_interval( + t, timeindex, time_steps_per_time_interval, time_step_day_start + ) + for t in feedin_steps + ] + + steps = pd.DataFrame( + { + "time_steps_load_case": load_intervals, + "time_steps_feedin_case": feedin_intervals, + } + ) + if len(steps) == 0: + logger.info("No critical steps detected. No network expansion required.") + + if save_steps: + abs_path = os.path.abspath(path) + steps.to_csv( + os.path.join( + abs_path, + f"{edisgo_obj.topology.id}_t_{time_steps_per_time_interval}" + f"_residual_load.csv", + ) + ) + return steps + + def get_most_critical_time_intervals( edisgo_obj, num_time_intervals=None, @@ -659,6 +842,7 @@ def get_most_critical_time_intervals( overloading_factor=0.95, voltage_deviation_factor=0.95, weight_by_costs=True, + by="power_flow", ): """ Get time intervals sorted by severity of overloadings as well as voltage issues. @@ -747,15 +931,32 @@ def get_most_critical_time_intervals( time intervals. Default: True. + by : str + Criticality measure used to determine the intervals. Options: + + * "power_flow" (default): run a power flow and score rolling windows by + overloading and voltage violations. Returns columns + ``time_steps_overloading`` / ``time_steps_voltage_issues``. + * "residual_load": no power flow — rank the critical single steps by + residual load (see :func:`get_most_critical_time_steps` with + ``by="residual_load"``) and center an interval on each, snapped to the + ``time_step_day_start`` hour. Returns columns ``time_steps_load_case`` + (highest residual) / ``time_steps_feedin_case`` (lowest residual). + Overlaps between intervals are allowed. + + Default: "power_flow". Returns -------- :pandas:`pandas.DataFrame` - Contains time intervals in which grid expansion needs due to overloading and - voltage issues are detected. The time intervals are determined independently - for overloading and voltage issues and sorted descending by the expected - cumulated grid expansion costs, so that the time intervals with the highest - expected costs correspond to index 0. + Contains time intervals in which grid expansion needs are detected, + ranked most-critical first. Column names depend on ``by`` (see above): + ``time_steps_overloading``/``time_steps_voltage_issues`` for + ``power_flow``, ``time_steps_load_case``/``time_steps_feedin_case`` for + ``residual_load``. For ``power_flow`` the intervals are determined + independently for overloading and voltage issues and sorted descending by + the expected cumulated grid expansion costs, so that the time intervals + with the highest expected costs correspond to index 0. In case of overloading, the time steps in the respective time interval are given in column "time_steps_overloading" and the share of components for which the maximum overloading is reached during the time interval is given in column @@ -766,6 +967,28 @@ def get_most_critical_time_intervals( "percentage_buses_max_voltage_deviation". """ + if by not in ("power_flow", "residual_load"): + raise ValueError( + f"get_most_critical_time_intervals: 'by' must be 'power_flow' or " + f"'residual_load', got {by!r}." + ) + + if by == "residual_load": + # No power flow: rank the critical single steps by residual load (via + # get_most_critical_time_steps(by="residual_load")) and wrap each into an + # interval centered on the step and snapped to the time_step_day_start + # block. Returns per-case columns time_steps_load_case / + # time_steps_feedin_case (ranked, overlaps allowed). + return _most_critical_time_intervals_residual_load( + edisgo_obj, + num_time_intervals=num_time_intervals, + percentage=percentage, + time_steps_per_time_interval=time_steps_per_time_interval, + time_step_day_start=time_step_day_start, + save_steps=save_steps, + path=path, + ) + # check frequency of time series data timeindex = edisgo_obj.timeseries.timeindex timedelta = timeindex[1] - timeindex[0] @@ -845,6 +1068,64 @@ def get_most_critical_time_intervals( return steps +def _most_critical_time_steps_residual_load( + edisgo_obj, + num_steps_loading=None, + num_steps_voltage=None, + percentage=1.0, +): + """ + Rank time steps by residual load, without running a power flow. + + Distributes the overlying-grid dispatch onto the grid components (via + :func:`~.network.overlying_grid.distribute_overlying_grid_requirements`) and + evaluates the residual load (load minus generation minus storage) over the + whole time index. The load-case steps are those with the highest residual + load, the feed-in-case steps those with the lowest (most negative). + + Parameters + ---------- + edisgo_obj : :class:`~.EDisGo` + num_steps_loading : int or None + Number of highest-residual (load-case) steps to select. If None, + ``percentage`` of all steps is used. + num_steps_voltage : int or None + Number of lowest-residual (feed-in-case) steps to select. If None, + ``percentage`` of all steps is used. + percentage : float + Fraction of all time steps to select per case when the corresponding + ``num_steps_*`` is None. Default: 1.0. + + Returns + ------- + :pandas:`pandas.DatetimeIndex` + Unique union of the selected load-case and feed-in-case time steps. + """ + from edisgo.network.overlying_grid import ( + distribute_overlying_grid_requirements, + ) + + distributed = distribute_overlying_grid_requirements(edisgo_obj) + residual = distributed.timeseries.residual_load + + n = len(residual) + if num_steps_loading is None: + num_steps_loading = int(n * percentage) + if num_steps_voltage is None: + num_steps_voltage = int(n * percentage) + num_steps_loading = min(num_steps_loading, n) + num_steps_voltage = min(num_steps_voltage, n) + + # highest residual = worst load case; lowest residual = worst feed-in case + load_case = residual.sort_values(ascending=False).index[:num_steps_loading] + feedin_case = residual.sort_values(ascending=True).index[:num_steps_voltage] + + steps = load_case.append(feedin_case) + if len(steps) == 0: + logger.warning("No critical steps detected. No network expansion required.") + return pd.DatetimeIndex(steps.unique()) + + def get_most_critical_time_steps( edisgo_obj: EDisGo, mode=None, @@ -857,6 +1138,7 @@ def get_most_critical_time_steps( use_troubleshooting_mode=True, run_initial_analyze=True, weight_by_costs=True, + by="power_flow", ) -> pd.DatetimeIndex: """ Get the time steps with the most critical overloading and voltage issues. @@ -926,14 +1208,51 @@ def get_most_critical_time_steps( If False, only the relative overloading is used. Default: True. + by : str + Criticality measure used to rank time steps. Options: + + * "power_flow" (default): run a power flow and score steps by overloading + and voltage violations (the parameters `mode`, `timesteps`, + `lv_grid_id`, `scale_timeseries`, `use_troubleshooting_mode`, + `run_initial_analyze`, `weight_by_costs` apply to this measure). + * "residual_load": no power flow — rank steps by the residual load + (load minus generation minus storage) after distributing the + overlying-grid dispatch onto the components. The highest residual + steps are the critical load cases, the lowest (most negative) the + critical feed-in cases. `num_steps_loading` / `num_steps_voltage` / + `percentage` control how many of each are selected; the power-flow + parameters are ignored. + + Default: "power_flow". Returns -------- :pandas:`pandas.DatetimeIndex` Time index with unique time steps where maximum overloading or maximum - voltage deviation is reached for at least one component respectively bus. + voltage deviation is reached for at least one component respectively bus + (``by="power_flow"``), or with the highest/lowest residual load + (``by="residual_load"``). """ + if by not in ("power_flow", "residual_load"): + raise ValueError( + f"get_most_critical_time_steps: 'by' must be 'power_flow' or " + f"'residual_load', got {by!r}." + ) + + if by == "residual_load": + # No power flow needed: rank time steps by the residual load (load minus + # generation minus storage) after distributing the overlying-grid + # dispatch onto the components. The most critical load-case steps have + # the highest residual load, the most critical feed-in-case steps the + # lowest (most negative). Returns the union of both, deduplicated. + return _most_critical_time_steps_residual_load( + edisgo_obj, + num_steps_loading=num_steps_loading, + num_steps_voltage=num_steps_voltage, + percentage=percentage, + ) + # Run power flow if run_initial_analyze: if use_troubleshooting_mode: diff --git a/edisgo/tools/tools.py b/edisgo/tools/tools.py index 8bbce99aa..ead0219e5 100644 --- a/edisgo/tools/tools.py +++ b/edisgo/tools/tools.py @@ -1228,6 +1228,25 @@ def reduce_timeseries_data_to_given_timeindex( ) # Battery electric vehicle timeseries if electromobility: + # The EV flexibility bands are built in import_electromobility from the + # raw SimBEV grid (typically 15-min and in the reference year 2011), + # independently of the analysis time index. Before slicing by datetime, + # align them to the target index: first resample to its frequency + # (Electromobility.resample uses the correct per-band aggregation — + # mean for power, max for energy), then shift the year and reindex via + # align_series_to_timeindex so datetime .loc lookups below succeed. + _bands = edisgo_obj.electromobility.flexibility_bands + _band0 = next((b for b in _bands.values() if not b.empty), None) + if _band0 is not None and len(_band0.index) > 1: + band_freq = _band0.index[1] - _band0.index[0] + if band_freq != frequency: + edisgo_obj.electromobility.resample(freq=frequency) + # year-align every (now correctly-sampled) band onto the timeindex + for key, df in edisgo_obj.electromobility.flexibility_bands.items(): + if not df.empty: + edisgo_obj.electromobility.flexibility_bands[key] = ( + align_series_to_timeindex(df, timeindex) + ) if save_ev_soc_initial: # timestep EV SOC from timestep before if possible ts_before = timeindex[0] - frequency @@ -1423,7 +1442,7 @@ def reduce_memory_usage(df: pd.DataFrame, show_reduction: bool = False) -> pd.Da for col in df.columns: col_type = df[col].dtype - if col_type != object and str(col_type) != "category": + if not pd.api.types.is_object_dtype(col_type) and str(col_type) != "category": c_min = df[col].min() c_max = df[col].max() diff --git a/run_example_05.py b/run_example_05.py new file mode 100644 index 000000000..552082d3d --- /dev/null +++ b/run_example_05.py @@ -0,0 +1,39 @@ +# This file is part of eDisGo (Electrical Distribution Grid Optimization), +# a Python package for analyzing flexibility options in distribution grids. +# +# Copyright (c) Reiner Lemoine Institut gGmbH +# Contributors are listed in the version control history: +# https://github.com/openego/eDisGo/ +# +# Documentation: https://edisgo.readthedocs.io/ +# +# SPDX-License-Identifier: AGPL-3.0-or-later +"""Runner für uc5_select_timesteps.yaml — einfach ``python run_example_05.py``.""" + +import logging + +from edisgo.run.runner import run_edisgo + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(name)s %(levelname)s: %(message)s", +) + +# edisgo = run_edisgo("/storage/JoDa/ego/edisgo_run_edisgo/eDisGo/edisgo/run/presets/uc4_example_MS.yaml") # noqa: E501 +edisgo = run_edisgo( + { + "extends": "uc5_select_timesteps.yaml", + # "grid": {"ding0_path": "/home/gurobi/.ding0/run_hetzner_59763_2023_04_06/ding0_grids/32355"} # noqa: E501 + "grid": { + "ding0_path": "/home/gurobi/.ding0/2024-07-25T17:38:34_new_planning_new_edisgo/ding0_grids/32377" # noqa: E501 + }, + # OG path must be the leaf dir for THIS grid (like ding0_path), not the parent. + "overlying_grid": { + "path": "/storage/JoDa/edisgo_playground/overlying_grid_data/32377" + }, + } +) + +print("\n=== Fertig ===") +print("Ausbaukosten:\n", edisgo.results.grid_expansion_costs) +print("\nUngelöste Probleme:\n", edisgo.results.unresolved_issues) diff --git a/tests/opf/test_powermodels_opf.py b/tests/opf/test_powermodels_opf.py index c79cacb06..9b4300e6d 100644 --- a/tests/opf/test_powermodels_opf.py +++ b/tests/opf/test_powermodels_opf.py @@ -339,3 +339,194 @@ def test_pm_optimize(self): ) ) ) + + +class TestContiguousIntervals: + def test_contiguous_index_is_one_interval(self): + from edisgo.opf.powermodels_opf import _contiguous_intervals + + ti = pd.date_range("2035-01-01", periods=48, freq="h") + result = _contiguous_intervals(ti) + assert len(result) == 1 and result[0].equals(ti) + + def test_two_disconnected_intervals(self): + from edisgo.opf.powermodels_opf import _contiguous_intervals + + a = pd.date_range("2035-01-01", periods=24, freq="h") + b = pd.date_range("2035-07-01", periods=24, freq="h") + result = _contiguous_intervals(a.union(b)) + assert len(result) == 2 + assert result[0].equals(a) and result[1].equals(b) + + def test_freq_restored_on_freqless_index(self): + from edisgo.opf.powermodels_opf import _contiguous_intervals + + idx = pd.DatetimeIndex(pd.date_range("2035-01-01", periods=48, freq="h").values) + assert idx.freq is None + result = _contiguous_intervals(idx) + assert len(result) == 1 and result[0].freq is not None + + def test_single_and_empty(self): + from edisgo.opf.powermodels_opf import _contiguous_intervals + + assert len(_contiguous_intervals(pd.date_range("2035-01-01", periods=1))) == 1 + assert _contiguous_intervals(pd.DatetimeIndex([])) == [] + + +class TestMergeOpfTimeFrames: + @staticmethod + def _empty_snapshot(opf, slack_generator_t): + return { + "slack_generator_t": slack_generator_t, + "hv_requirement_slacks_t": pd.DataFrame(), + "lines_t": {k: pd.DataFrame() for k in opf.lines_t._attributes()}, + "heat_storage_t": { + k: pd.DataFrame() for k in opf.heat_storage_t._attributes() + }, + "grid_slacks_t": { + k: pd.DataFrame() for k in opf.grid_slacks_t._attributes() + }, + "battery_storage_t": { + k: pd.DataFrame() for k in opf.battery_storage_t._attributes() + }, + } + + def test_flat_frame_concatenated_and_sorted(self): + from edisgo.opf.powermodels_opf import _merge_opf_time_frames + from edisgo.opf.results.opf_result_class import OPFResults + + opf = OPFResults() + a = pd.DataFrame({"x": [1.0]}, index=pd.date_range("2035-01-01", periods=1)) + b = pd.DataFrame({"x": [2.0]}, index=pd.date_range("2035-07-01", periods=1)) + _merge_opf_time_frames( + opf, [self._empty_snapshot(opf, a), self._empty_snapshot(opf, b)] + ) + assert len(opf.slack_generator_t) == 2 + assert list(opf.slack_generator_t["x"]) == [1.0, 2.0] + + +class TestPmOptimizeIntervalSplit: + """pm_optimize's multi-interval split, with the single-interval OPF stubbed + (no Julia/DB). Patches powermodels_opf._pm_optimize_single.""" + + @pytest.fixture + def edisgo_obj(self): + e = EDisGo(ding0_grid=pytest.ding0_test_network_path) + e.set_timeindex(pd.date_range("2035-01-01", periods=24, freq="h")) + return e + + def test_single_interval_calls_once_with_freq(self, edisgo_obj, monkeypatch): + import edisgo.opf.powermodels_opf as pmo + + ti = pd.DatetimeIndex(pd.date_range("2035-01-01", periods=24, freq="h").values) + assert ti.freq is None + edisgo_obj.set_timeindex(ti) + seen = [] + monkeypatch.setattr( + pmo, + "_pm_optimize_single", + lambda e, **kw: seen.append(e.timeseries.timeindex.freq), + ) + pmo.pm_optimize(edisgo_obj) + assert seen == [pd.tseries.frequencies.to_offset("h")] + + def test_two_intervals_run_separately_and_restore(self, edisgo_obj, monkeypatch): + import edisgo.opf.powermodels_opf as pmo + + a = pd.date_range("2035-01-01", periods=24, freq="h") + b = pd.date_range("2035-07-01", periods=24, freq="h") + full = a.union(b) + edisgo_obj.set_timeindex(full) + seen = [] + + def fake_single(e, **kw): + seen.append(e.timeseries.timeindex) + e.opf_results.status = "OPTIMAL" + e.opf_results.solver = "Gurobi" + e.opf_results.solution_time = 1.0 + + monkeypatch.setattr(pmo, "_pm_optimize_single", fake_single) + pmo.pm_optimize(edisgo_obj) + assert len(seen) == 2 and seen[0].equals(a) and seen[1].equals(b) + assert edisgo_obj.timeseries.timeindex.equals(full) + assert len(edisgo_obj.opf_results.interval_results) == 2 + assert edisgo_obj.opf_results.solution_time == 2.0 + assert edisgo_obj.opf_results.status == "OPTIMAL" + + def test_overlying_grid_state_restored(self, edisgo_obj, monkeypatch): + import edisgo.opf.powermodels_opf as pmo + + a = pd.date_range("2035-01-01", periods=24, freq="h") + b = pd.date_range("2035-07-01", periods=24, freq="h") + full = a.union(b) + edisgo_obj.set_timeindex(full) + edisgo_obj.overlying_grid.storage_units_soc = pd.Series(1.0, index=full) + seen_types = [] + + def fake_single(e, **kw): + og = e.overlying_grid + seen_types.append(type(og.storage_units_soc).__name__) + og.storage_units_soc = pd.DataFrame( + 0.0, index=e.timeseries.timeindex, columns=["s1"] + ) + e.opf_results.status = "OPTIMAL" + + monkeypatch.setattr(pmo, "_pm_optimize_single", fake_single) + pmo.pm_optimize(edisgo_obj) + assert seen_types == ["Series", "Series"] + assert isinstance(edisgo_obj.overlying_grid.storage_units_soc, pd.Series) + + def test_reactive_power_restored(self, edisgo_obj, monkeypatch): + import edisgo.opf.powermodels_opf as pmo + + a = pd.date_range("2035-01-01", periods=24, freq="h") + b = pd.date_range("2035-07-01", periods=24, freq="h") + full = a.union(b) + edisgo_obj.set_timeindex(full) + gen = edisgo_obj.topology.generators_df.index[0] + edisgo_obj.timeseries._generators_reactive_power = pd.DataFrame( + 0.0, index=full, columns=[gen] + ) + seen_ok = [] + + def fake_single(e, **kw): + ti = e.timeseries.timeindex + q = e.timeseries.generators_reactive_power + seen_ok.append(not q.empty and q.index.equals(ti)) + e.timeseries._generators_reactive_power = pd.DataFrame( + 0.0, index=ti, columns=[gen] + ) + e.opf_results.status = "OPTIMAL" + + monkeypatch.setattr(pmo, "_pm_optimize_single", fake_single) + pmo.pm_optimize(edisgo_obj) + assert seen_ok == [True, True] + assert edisgo_obj.timeseries._generators_reactive_power.index.equals(full) + + def test_infeasible_interval_stores_report_and_raises( + self, edisgo_obj, monkeypatch + ): + import edisgo.opf.powermodels_opf as pmo + + from edisgo.flex_opt.exceptions import InfeasibleModelError + + a = pd.date_range("2035-01-01", periods=24, freq="h") + b = pd.date_range("2035-07-01", periods=24, freq="h") + full = a.union(b) + edisgo_obj.set_timeindex(full) + + def fake_single(e, **kw): + if e.timeseries.timeindex[0] == a[0]: + e.opf_results.status = "OPTIMAL" + e.opf_results.solution_time = 1.0 + else: + raise InfeasibleModelError("stub") + + monkeypatch.setattr(pmo, "_pm_optimize_single", fake_single) + with pytest.raises(InfeasibleModelError): + pmo.pm_optimize(edisgo_obj) + report = edisgo_obj.opf_results.interval_results + assert len(report) == 2 + assert report[0]["status"] == "OPTIMAL" + assert report[1]["status"] == "infeasible" + assert edisgo_obj.timeseries.timeindex.equals(full) diff --git a/tests/run/test_tasks.py b/tests/run/test_tasks.py index c6c12631e..1f1db2237 100644 --- a/tests/run/test_tasks.py +++ b/tests/run/test_tasks.py @@ -7,6 +7,7 @@ enough, and the DB-free branches of ``import_overlying_grid_data`` are exercised directly. """ + import glob import os @@ -18,9 +19,17 @@ from edisgo.edisgo import EDisGo from edisgo.run.config import load_config from edisgo.run.context import RunContext +from edisgo.run.tasks.analysis import task_optimize from edisgo.run.tasks.io import task_import_overlying_grid_data -from edisgo.run.tasks.timeseries import task_manual_ts +from edisgo.run.tasks.timeseries import ( + task_manual_ts, + task_select_timesteps, +) from edisgo.run.validator import validate +from edisgo.tools.temporal_complexity_reduction import ( + intervals_overlap, + select_two_intervals, +) @pytest.fixture @@ -69,8 +78,7 @@ def test_unknown_source_warns(self, edisgo_obj, caplog): assert "unknown source" in caplog.text def test_etrago_without_data_warns(self, edisgo_obj, caplog): - ctx = self._ctx({"enabled": True, "source": "etrago"}, - overlying_grid_data=None) + ctx = self._ctx({"enabled": True, "source": "etrago"}, overlying_grid_data=None) result = task_import_overlying_grid_data(edisgo_obj, ctx) assert result is edisgo_obj assert "no" in caplog.text.lower() @@ -80,8 +88,7 @@ def test_etrago_empty_data_does_not_crash(self, edisgo_obj): A partial/empty etrago dict must not raise (regression: the task used to call .empty on dict.get() results that were None). """ - ctx = self._ctx({"enabled": True, "source": "etrago"}, - overlying_grid_data={}) + ctx = self._ctx({"enabled": True, "source": "etrago"}, overlying_grid_data={}) # must simply return without AttributeError assert task_import_overlying_grid_data(edisgo_obj, ctx) is edisgo_obj @@ -92,6 +99,236 @@ def test_csv_without_path_warns(self, edisgo_obj, caplog): assert "path" in caplog.text.lower() +class TestSelectTimestepsHelpers: + """Pure helpers for auto interval selection — no DB, no power flow.""" + + @staticmethod + def _week(start): + return pd.date_range(start=start, periods=168, freq="h") + + def test_overlap_detection(self): + a = self._week("2035-01-01") + assert intervals_overlap(a, self._week("2035-01-04")) # overlaps + assert not intervals_overlap(a, self._week("2035-06-01")) # disjoint + + def test_disjoint_top_intervals_kept_as_two(self): + load = [self._week("2035-01-01")] + volt = [self._week("2035-06-01")] + result = select_two_intervals(load, volt) + assert len(result) == 2 + assert not intervals_overlap(result[0], result[1]) + + def test_overlap_falls_to_next_ranked_voltage(self): + load = [self._week("2035-01-01")] + # first voltage candidate overlaps the top loading interval, second does not + volt = [self._week("2035-01-03"), self._week("2035-09-01")] + result = select_two_intervals(load, volt) + assert len(result) == 2 + assert result[1].equals(volt[1]) + + def test_all_overlap_concatenates_to_one(self): + load = [self._week("2035-01-01")] + volt = [self._week("2035-01-03")] # only candidate, overlaps + result = select_two_intervals(load, volt) + assert len(result) == 1 + merged = result[0] + # merged interval is contiguous and spans both inputs + assert merged.min() == load[0].min() + assert merged.max() == volt[0].max() + assert (merged[1:] - merged[:-1]).nunique() == 1 # regular spacing + + def test_single_side_only(self): + week = self._week("2035-01-01") + for result in ( + select_two_intervals([week], []), + select_two_intervals([], [week]), + ): + assert len(result) == 1 + assert result[0].equals(week) + assert select_two_intervals([], []) == [] + + +class TestSelectTimestepsManual: + def test_manual_explicit_timestamps_before_imports(self, edisgo_obj): + """ + Manual mode with an empty timeseries (positioned before imports) sets + the index and stashes it for HP/DSM imports. + """ + # start from an empty time index to mimic the pre-import position + edisgo_obj.set_timeindex(pd.DatetimeIndex([])) + ts = ["2011-01-01 00:00", "2011-01-01 02:00"] + ctx = RunContext( + raw_config={"timeseries_selection": {"mode": "manual", "timestamps": ts}} + ) + result = task_select_timesteps(edisgo_obj, ctx) + assert list(result.timeseries.timeindex) == list(pd.to_datetime(ts)) + assert list(ctx.flags["selected_timeindex"]) == list(pd.to_datetime(ts)) + assert ctx.flags["timesteps_selected"] is True + + def test_manual_range_reduces_existing_timeseries(self, edisgo_obj): + """Manual range with an existing 3-step index reduces to the range.""" + ctx = RunContext( + raw_config={ + "timeseries_selection": { + "mode": "manual", + "start": "2011-01-01 00:00", + "periods": 2, + "freq": "h", + } + } + ) + result = task_select_timesteps(edisgo_obj, ctx) + assert len(result.timeseries.timeindex) == 2 + + def test_missing_mode_raises(self, edisgo_obj): + with pytest.raises(ValueError, match="mode 'manual' or 'auto'"): + task_select_timesteps(edisgo_obj, RunContext(raw_config={})) + + def test_auto_without_active_power_raises(self, edisgo_obj): + ctx = RunContext(raw_config={"timeseries_selection": {"mode": "auto"}}) + with pytest.raises(ValueError, match="active-power time series"): + task_select_timesteps(edisgo_obj, ctx) + + def test_auto_unknown_method_raises(self, edisgo_obj): + ctx = RunContext( + raw_config={ + "timeseries_selection": { + "mode": "auto", + "method": "bogus", + } + } + ) + ctx.flags["timeseries_set"] = True + with pytest.raises(ValueError, match="power_flow.*residual_load"): + task_select_timesteps(edisgo_obj, ctx) + + def test_residual_load_requires_overlying_grid(self, edisgo_obj): + """residual_load method must raise when no overlying-grid data is set.""" + ctx = RunContext( + raw_config={ + "timeseries_selection": { + "mode": "auto", + "method": "residual_load", + } + } + ) + ctx.flags["timeseries_set"] = True + with pytest.raises(ValueError, match="overlying-grid data"): + task_select_timesteps(edisgo_obj, ctx) + + +class TestSelectTimestepsPosition: + """The `position` param lets one pipeline carry both a pre-import and a + post-grid select_timesteps step; each no-ops off its mode.""" + + def test_pre_import_noops_in_auto_mode(self, edisgo_obj): + """ + A pre_import step in auto mode must be a no-op — crucially it must NOT + hit the auto guard (no active power set yet), it just returns. + """ + before = edisgo_obj.timeseries.timeindex + ctx = RunContext(raw_config={"timeseries_selection": {"mode": "auto"}}) + result = task_select_timesteps(edisgo_obj, ctx, position="pre_import") + assert result is edisgo_obj + assert result.timeseries.timeindex.equals(before) + assert "timesteps_selected" not in ctx.flags + + def test_post_grid_noops_in_manual_mode(self, edisgo_obj): + """A post_grid step in manual mode must be a no-op (manual already ran + earlier at pre_import).""" + before = edisgo_obj.timeseries.timeindex + ctx = RunContext( + raw_config={ + "timeseries_selection": { + "mode": "manual", + "timestamps": ["2011-01-01 00:00"], + } + } + ) + result = task_select_timesteps(edisgo_obj, ctx, position="post_grid") + assert result is edisgo_obj + assert result.timeseries.timeindex.equals(before) + + def test_pre_import_acts_in_manual_mode(self, edisgo_obj): + edisgo_obj.set_timeindex(pd.DatetimeIndex([])) + ctx = RunContext( + raw_config={ + "timeseries_selection": { + "mode": "manual", + "timestamps": ["2011-01-01 00:00"], + } + } + ) + task_select_timesteps(edisgo_obj, ctx, position="pre_import") + assert len(edisgo_obj.timeseries.timeindex) == 1 + assert ctx.flags["timesteps_selected"] is True + + def test_bad_position_raises(self, edisgo_obj): + ctx = RunContext(raw_config={"timeseries_selection": {"mode": "manual"}}) + with pytest.raises(ValueError, match="position"): + task_select_timesteps(edisgo_obj, ctx, position="bogus") + + def test_pre_import_sets_default_index_when_empty_in_auto(self, edisgo_obj): + """ + A pre_import step in auto mode with no time index set must establish a + full-year default (so later imports build hourly full-year data), even + though it otherwise no-ops. + """ + edisgo_obj.set_timeindex(pd.DatetimeIndex([])) + ctx = RunContext( + scenario="eGon2035", + raw_config={"timeseries_selection": {"mode": "auto"}}, + ) + task_select_timesteps(edisgo_obj, ctx, position="pre_import") + ti = edisgo_obj.timeseries.timeindex + assert len(ti) == 8760 + assert ti[0].year == 2035 + # still a no-op for actual selection + assert "timesteps_selected" not in ctx.flags + + def test_manual_shifts_user_timestamps_to_timeseries_year(self, edisgo_obj): + """ + Manual mode reducing an existing (differently-yeared) time series shifts + the user timestamps to the time-series year so slicing matches. + """ + # existing time series in 2011 + edisgo_obj.set_timeindex(pd.date_range("2011-06-01", periods=5, freq="h")) + # user selects timestamps written in the scenario year 2035 + ctx = RunContext( + raw_config={ + "timeseries_selection": { + "mode": "manual", + "timestamps": ["2035-06-01 01:00", "2035-06-01 03:00"], + } + } + ) + task_select_timesteps(edisgo_obj, ctx) + ti = edisgo_obj.timeseries.timeindex + assert list(ti) == [ + pd.Timestamp("2011-06-01 01:00"), + pd.Timestamp("2011-06-01 03:00"), + ] + + +class TestOptimizeTaskDelegation: + """task_optimize is thin: expand the `flexible` shortcut and call + edisgo.pm_optimize. The multi-interval split lives in pm_optimize and is + tested in tests/opf/test_powermodels_opf.py.""" + + def test_expands_flexible_shortcut_and_calls_pm_optimize(self, edisgo_obj): + edisgo_obj.set_timeindex(pd.date_range("2035-01-01", periods=24, freq="h")) + captured = {} + + def fake_pm_optimize(**kw): + captured.update(kw) + + edisgo_obj.pm_optimize = fake_pm_optimize + task_optimize(edisgo_obj, RunContext(), flexible=["heat_pumps", "storage"]) + # shortcut expanded to explicit name lists (empty ok if grid lacks type) + assert "flexible_hps" in captured and "flexible_storage_units" in captured + assert isinstance(captured["flexible_hps"], list) + + def test_all_bundled_presets_validate(): """ Every bundled preset must pass the (metadata-driven) validator — this diff --git a/tests/run/test_validator.py b/tests/run/test_validator.py index 1d375f10d..9e830e0bc 100644 --- a/tests/run/test_validator.py +++ b/tests/run/test_validator.py @@ -5,6 +5,7 @@ without TS, optimize without flex, flex import without grid, and the stage-level ``load_from`` constraints. """ + import pytest from edisgo.run.validator import validate @@ -30,8 +31,9 @@ def _wrap(pipeline): def test_valid_pipeline(): """A well-formed pipeline must pass validation without raising.""" - validate(_wrap(["setup_grid", "worst_case_ts", "reactive_power", - "reinforce", "save"])) + validate( + _wrap(["setup_grid", "worst_case_ts", "reactive_power", "reinforce", "save"]) + ) def test_unknown_task_rejected(): @@ -46,6 +48,33 @@ def test_reactive_before_ts_rejected(): validate(_wrap(["setup_grid", "reactive_power", "worst_case_ts"])) +def test_select_timesteps_after_reactive_rejected(): + """ + select_timesteps is ts_altering (it reduces the time index), so it must + not appear after reactive_power. + """ + with pytest.raises(ValueError, match="reactive_power"): + validate( + _wrap(["setup_grid", "worst_case_ts", "reactive_power", "select_timesteps"]) + ) + + +def test_select_timesteps_before_reactive_ok(): + """select_timesteps before reactive_power is the intended ordering.""" + validate( + _wrap( + [ + "setup_grid", + "worst_case_ts", + "select_timesteps", + "reactive_power", + "reinforce", + "save", + ] + ) + ) + + def test_reinforce_without_ts_rejected(): """reinforce without any prior time-series step must fail.""" with pytest.raises(ValueError, match="time series"): @@ -66,37 +95,48 @@ def test_flex_import_before_grid_rejected(): def test_stage_load_from_missing_rejected(): """``load_from: X`` where X has not run must fail.""" - cfg = {"stages": [ - {"name": "a", "pipeline": ["setup_grid", "worst_case_ts", - "reinforce"]}, - {"name": "b", "load_from": "nonexistent", - "pipeline": ["reinforce"]}, - ]} + cfg = { + "stages": [ + {"name": "a", "pipeline": ["setup_grid", "worst_case_ts", "reinforce"]}, + {"name": "b", "load_from": "nonexistent", "pipeline": ["reinforce"]}, + ] + } with pytest.raises(ValueError, match="load_from"): validate(cfg) def test_stage_load_from_requires_save_in_source(): """A stage consumed by ``load_from`` must itself end with ``save``.""" - cfg = {"stages": [ - {"name": "a", "pipeline": ["setup_grid", "worst_case_ts", - "reinforce"]}, # no save - {"name": "b", "load_from": "a", "pipeline": ["reinforce"]}, - ]} + cfg = { + "stages": [ + { + "name": "a", + "pipeline": ["setup_grid", "worst_case_ts", "reinforce"], + }, # no save + {"name": "b", "load_from": "a", "pipeline": ["reinforce"]}, + ] + } with pytest.raises(ValueError, match="load_from"): validate(cfg) def test_stage_load_from_with_save_ok(): """Stage chain with a save in the source must validate successfully.""" - cfg = {"stages": [ - {"name": "a", "pipeline": ["setup_grid", "worst_case_ts", - "reinforce", "save"]}, - # load_from reloads the grid with import_timeseries=False, so the - # consuming stage must set time series itself before reinforce. - {"name": "b", "load_from": "a", - "pipeline": ["worst_case_ts", "reinforce", "save"]}, - ]} + cfg = { + "stages": [ + { + "name": "a", + "pipeline": ["setup_grid", "worst_case_ts", "reinforce", "save"], + }, + # load_from reloads the grid with import_timeseries=False, so the + # consuming stage must set time series itself before reinforce. + { + "name": "b", + "load_from": "a", + "pipeline": ["worst_case_ts", "reinforce", "save"], + }, + ] + } validate(cfg) @@ -106,10 +146,14 @@ def test_stage_load_from_without_ts_rejected(): reloaded with import_timeseries=False, so reinforce after a bare load_from (no time-series task in the stage) must be rejected. """ - cfg = {"stages": [ - {"name": "a", "pipeline": ["setup_grid", "worst_case_ts", - "reinforce", "save"]}, - {"name": "b", "load_from": "a", "pipeline": ["reinforce", "save"]}, - ]} + cfg = { + "stages": [ + { + "name": "a", + "pipeline": ["setup_grid", "worst_case_ts", "reinforce", "save"], + }, + {"name": "b", "load_from": "a", "pipeline": ["reinforce", "save"]}, + ] + } with pytest.raises(ValueError, match="requires time series"): validate(cfg) diff --git a/tests/tools/test_temporal_complexity_reduction.py b/tests/tools/test_temporal_complexity_reduction.py index 1063093f4..4c32daa01 100644 --- a/tests/tools/test_temporal_complexity_reduction.py +++ b/tests/tools/test_temporal_complexity_reduction.py @@ -183,3 +183,92 @@ def test_get_most_critical_time_intervals(self): steps.loc[0, "time_steps_voltage_issues"] == pd.date_range("1/1/2018", periods=24, freq="H") ).all() + + +class TestIntervalHelpers: + """Relocated pure helpers + residual_load selection (no DB / no power flow).""" + + @staticmethod + def _week(start): + return pd.date_range(start=start, periods=168, freq="h") + + def test_intervals_overlap(self): + a = self._week("2035-01-01") + assert temp_red.intervals_overlap(a, self._week("2035-01-04")) + assert not temp_red.intervals_overlap(a, self._week("2035-06-01")) + + def test_select_two_intervals_disjoint(self): + result = temp_red.select_two_intervals( + [self._week("2035-01-01")], [self._week("2035-06-01")] + ) + assert len(result) == 2 + assert not temp_red.intervals_overlap(result[0], result[1]) + + def test_select_two_intervals_next_ranked(self): + load = [self._week("2035-01-01")] + volt = [self._week("2035-01-03"), self._week("2035-09-01")] + result = temp_red.select_two_intervals(load, volt) + assert len(result) == 2 and result[1].equals(volt[1]) + + def test_select_two_intervals_concatenate(self): + result = temp_red.select_two_intervals( + [self._week("2035-01-01")], [self._week("2035-01-03")] + ) + assert len(result) == 1 + assert (result[0][1:] - result[0][:-1]).nunique() == 1 # contiguous + + def test_select_two_intervals_single_and_empty(self): + week = self._week("2035-01-01") + assert temp_red.select_two_intervals([week], [])[0].equals(week) + assert temp_red.select_two_intervals([], [week])[0].equals(week) + assert temp_red.select_two_intervals([], []) == [] + + def test_build_centered_interval(self): + idx = pd.date_range("2035-01-01 00:00", periods=8760, freq="h") + t = pd.Timestamp("2035-02-10 12:00") + iv = temp_red._build_centered_interval(t, idx, 168, 4) + assert len(iv) == 168 + assert iv[0].hour == 4 # starts on the day-start hour + assert t in iv # critical step contained + assert iv[-1] != t # centered -> not the last step + + def test_residual_load_steps_and_intervals(self, monkeypatch): + import types + + idx = pd.date_range("2035-01-01 00:00", periods=8760, freq="h") + residual = pd.Series(range(8760), index=idx, dtype=float) + fake = types.SimpleNamespace( + timeseries=types.SimpleNamespace(residual_load=residual) + ) + monkeypatch.setattr( + "edisgo.network.overlying_grid.distribute_overlying_grid_requirements", + lambda e: fake, + ) + # steps: top-3 highest + bottom-2 lowest residual + steps = temp_red.get_most_critical_time_steps( + object(), by="residual_load", num_steps_loading=3, num_steps_voltage=2 + ) + assert idx[-1] in steps and idx[0] in steps and len(steps) == 5 + + # intervals: per-case columns, centered on the residual max/min steps + e = types.SimpleNamespace( + timeseries=types.SimpleNamespace(timeindex=idx), + topology=types.SimpleNamespace(id="g"), + ) + df = temp_red.get_most_critical_time_intervals( + e, + by="residual_load", + num_time_intervals=2, + time_steps_per_time_interval=168, + time_step_day_start=4, + ) + assert list(df.columns) == ["time_steps_load_case", "time_steps_feedin_case"] + assert len(df) == 2 + # top load-case interval is centered on the global max residual step + assert idx[-1] in df.loc[0, "time_steps_load_case"] + + def test_bad_by_raises(self): + with pytest.raises(ValueError, match="power_flow.*residual_load"): + temp_red.get_most_critical_time_steps(object(), by="bogus") + with pytest.raises(ValueError, match="power_flow.*residual_load"): + temp_red.get_most_critical_time_intervals(object(), by="bogus")