Merge pull request #30 from Open-ISP/representative-weeks

Representative weeks

Author: nick-gorman

dodo.py

@@ -42,6 +42,7 @@
 )
 from ispypsa.translator.lines import translate_flow_paths_to_lines
 from ispypsa.translator.snapshot import create_complete_snapshot_index
+from ispypsa.translator.temporal_filters import filter_snapshot
 
 root_folder = Path("ispypsa_runs")
 
@@ -143,12 +144,15 @@ def create_pypsa_inputs_from_config_and_ispypsa_inputs(
 ) -> None:
     create_or_clean_task_output_folder(pypsa_inputs_location)
     pypsa_inputs = {}
-    pypsa_inputs["snapshot"] = create_complete_snapshot_index(
+    snapshot = create_complete_snapshot_index(
         start_year=config.temporal.start_year,
         end_year=config.temporal.end_year,
         operational_temporal_resolution_min=config.temporal.operational_temporal_resolution_min,
         year_type=config.temporal.year_type,
     )
+    pypsa_inputs["snapshot"] = filter_snapshot(
+        config=config.temporal, snapshot=snapshot
+    )
     pypsa_inputs["generators"] = _translate_ecaa_generators(
         ispypsa_inputs_location, config.network.nodes.regional_granularity
     )

ispypsa_runs/development/ispypsa_inputs/ispypsa_config.yaml

@@ -32,6 +32,14 @@ temporal:
   start_year: 2025
   end_year: 2025
   reference_year_cycle: [2018]
+  aggregation:
+    # Representative weeks to use instead of full yearly temporal representation.
+    # Options:
+    #   "None": Full yearly temporal representation is used.
+    #   list[int]: a list of integers specifying weeks of year to use as representative. Weeks of year are defined as
+    #   full weeks (Monday-Sunday) falling within the year. For example, if the list is "[1]" the model will only use the
+    #   first full week of each modelled year.
+    representative_weeks: [1, 12, 25, 38]
 # External solver to use
 # Options (refer to https://pypsa.readthedocs.io/en/latest/getting-started/installation.html):
 #   Free, and by default, installed with ISPyPSA:

src/ispypsa/config/validators.py

@@ -16,13 +16,18 @@ class NetworkConfig(BaseModel):
     nodes: NodesConfig
 
 
+class TemporalAggregationConfig(BaseModel):
+    representative_weeks: list[int] | None
+
+
 class TemporalConfig(BaseModel):
     operational_temporal_resolution_min: int
     path_to_parsed_traces: str
     year_type: Literal["fy", "calendar"]
     start_year: int
     end_year: int
     reference_year_cycle: list[int]
+    aggregation: TemporalAggregationConfig
 
     @field_validator("operational_temporal_resolution_min")
     @classmethod

src/ispypsa/translator/buses.py

@@ -5,6 +5,7 @@
 from isp_trace_parser import get_data
 
 from ispypsa.translator.mappings import _BUS_ATTRIBUTES
+from ispypsa.translator.temporal_filters import time_series_filter
 from ispypsa.translator.time_series_checker import check_time_series
 
 
@@ -111,6 +112,7 @@ def _translate_buses_demand_timeseries(
         node_trace = node_traces.groupby("Datetime", as_index=False)["Value"].sum()
         # datetime in nanoseconds required by PyPSA
         node_trace["Datetime"] = node_trace["Datetime"].astype("datetime64[ns]")
+        node_trace = time_series_filter(node_trace, snapshot)
         check_time_series(
             node_trace["Datetime"],
             pd.Series(snapshot.index),

src/ispypsa/translator/generators.py

@@ -5,6 +5,7 @@
 from isp_trace_parser import get_data
 
 from ispypsa.translator.mappings import _GENERATOR_ATTRIBUTES
+from ispypsa.translator.temporal_filters import time_series_filter
 from ispypsa.translator.time_series_checker import check_time_series
 
 
@@ -117,7 +118,8 @@ def _translate_generator_timeseries(
         )
         # datetime in nanoseconds required by PyPSA
         trace["Datetime"] = trace["Datetime"].astype("datetime64[ns]")
+        trace = time_series_filter(trace, snapshot)
         check_time_series(
-            trace["Datetime"], pd.Series(snapshot.index), "generator trace data", gen
+            trace["Datetime"], snapshot.index.to_series(), "generator trace data", gen
         )
         trace.to_parquet(Path(output_trace_path, f"{gen}.parquet"), index=False)

src/ispypsa/translator/helpers.py

@@ -0,0 +1,13 @@
+def get_iteration_start_and_end_time(year_type: str, start_year: int, end_year: int):
+    """Get the model start year, end year, and start/end month for iteration, which depend on
+    financial vs calendar year.
+    """
+    if year_type == "fy":
+        start_year = start_year - 1
+        end_year = end_year
+        month = 7
+    else:
+        start_year = start_year
+        end_year = end_year + 1
+        month = 1
+    return start_year, end_year, month

src/ispypsa/translator/snapshot.py

@@ -2,6 +2,8 @@
 
 import pandas as pd
 
+from ispypsa.translator.helpers import get_iteration_start_and_end_time
+
 
 def create_complete_snapshot_index(
     start_year: int,
@@ -25,14 +27,9 @@ def create_complete_snapshot_index(
     Returns:
         pd.DataFrame
     """
-    if year_type == "fy":
-        start_year = start_year - 1
-        end_year = end_year
-        month = 7
-    else:
-        start_year = start_year
-        end_year = end_year + 1
-        month = 1
+    start_year, end_year, month = get_iteration_start_and_end_time(
+        year_type, start_year, end_year
+    )
 
     if operational_temporal_resolution_min < 60:
         hour = 0

src/ispypsa/translator/temporal_filters.py

@@ -0,0 +1,182 @@
+from datetime import datetime, timedelta
+
+import pandas as pd
+
+from ispypsa.config.validators import TemporalConfig
+from ispypsa.translator.helpers import get_iteration_start_and_end_time
+
+
+def time_series_filter(time_series_data: pd.DataFrame, snapshot: pd.DataFrame):
+    """Filters a timeseries pandas DataFrame based using the datetime values in
+     the snapshot index.
+
+    Examples:
+
+    >>> datetime_index = pd.date_range('2020-01-01', '2020-01-03', freq='h')
+    >>> time_series_data = pd.DataFrame({'Datetime': datetime_index, 'Value': range(len(datetime_index))})
+    >>> snapshot = pd.DataFrame(index=datetime_index[::12])  # Every 12 hours
+    >>> time_series_filter(time_series_data, snapshot)
+                  Datetime  Value
+    0  2020-01-01 00:00:00      0
+    12 2020-01-01 12:00:00     12
+    24 2020-01-02 00:00:00     24
+    36 2020-01-02 12:00:00     36
+    48 2020-01-03 00:00:00     48
+
+    Args:
+        time_series_data: pd.DataFrame with time series column called 'Datetime'
+        snapshot: pd.DataFrame with datetime index
+
+    """
+    return time_series_data[time_series_data["Datetime"].isin(snapshot.index)]
+
+
+def filter_snapshot(config: TemporalConfig, snapshot: pd.DataFrame):
+    """Appy filter to the snapshot based on the model config.
+
+    - If config.representative_weeks is not None then filter the
+      snapshot based on the supplied list of representative weeks.
+
+    Examples:
+
+    # Create dummy config class with just data need for example.
+
+    >>> from dataclasses import dataclass
+
+    >>> @dataclass
+    ... class TemporalAggregationConfig:
+    ...     representative_weeks: list[int]
+
+    >>> @dataclass
+    ... class TemporalConfig:
+    ...     start_year: int
+    ...     end_year: int
+    ...     year_type: str
+    ...     aggregation: TemporalAggregationConfig
+
+    >>> config = TemporalConfig(
+    ...     start_year=2024,
+    ...     end_year=2024,
+    ...     year_type='calendar',
+    ...     aggregation=TemporalAggregationConfig(
+    ...        representative_weeks=[1],
+    ...     )
+    ... )
+
+    >>> snapshot = pd.DataFrame(index=pd.date_range('2024-01-01', '2024-12-31', freq='h'))
+
+    >>> snapshot = filter_snapshot(config, snapshot)
+
+    >>> snapshot.index[0]
+    Timestamp('2024-01-01 01:00:00')
+
+    >>> snapshot.index[-1]
+    Timestamp('2024-01-08 00:00:00')
+
+    Args:
+         config: TemporalConfig defining snapshot filtering.
+         snapshot: pd.DataFrame with datetime index containing the snapshot
+    """
+    if config.aggregation.representative_weeks is not None:
+        snapshot = filter_snapshot_for_representative_weeks(
+            representative_weeks=config.aggregation.representative_weeks,
+            snapshot=snapshot,
+            start_year=config.start_year,
+            end_year=config.end_year,
+            year_type=config.year_type,
+        )
+    return snapshot
+
+
+def filter_snapshot_for_representative_weeks(
+    representative_weeks: list[int],
+    snapshot: pd.DataFrame,
+    start_year: int,
+    end_year: int,
+    year_type: str,
+):
+    """Filters a snapshot by a list of weeks.
+
+    A snapshot is provided as a pandas DatFrame with a datetime index. The
+    snapshot may be multiple years in length. The snapshot is filtered for
+    date times that fall within the weeks defined in representative_weeks.
+    The weeks are defined as full weeks within a financial or calendar year,
+    depending on the year_type provided.
+
+    Examples:
+    >>> # Filter for first and last full weeks of each calendar year from 2020-2022
+    >>> df = pd.DataFrame(index=pd.date_range('2020-01-01', '2022-12-31', freq='h'))
+    >>> filter_snapshot_for_representative_weeks(
+    ...     representative_weeks=[1],
+    ...     snapshot=df,
+    ...     start_year=2020,
+    ...     end_year=2022,
+    ...     year_type='calendar'
+    ... ).head(3)
+    Empty DataFrame
+    Columns: []
+    Index: [2020-01-06 01:00:00, 2020-01-06 02:00:00, 2020-01-06 03:00:00]
+
+    >>> # Filter for weeks 1, 26 of financial years 2021-2022 (July 2020 - June 2022)
+    >>> df = pd.DataFrame(index=pd.date_range('2020-07-01', '2022-06-30', freq='h'))
+    >>> filter_snapshot_for_representative_weeks(
+    ...     representative_weeks=[2],
+    ...     snapshot=df,
+    ...     start_year=2021,
+    ...     end_year=2022,
+    ...     year_type='fy'
+    ... ).head(3)
+    Empty DataFrame
+    Columns: []
+    Index: [2020-07-13 01:00:00, 2020-07-13 02:00:00, 2020-07-13 03:00:00]
+
+    Args:
+        representative_weeks: list[int] of full weeks to filter for. The
+            week 1 refers to the first full week (Monday-Sunday) falling
+            with in the year.
+        snapshot: pd.DataFrame with datetime index containing the snapshot
+        start_year: int defining the start year of the snapshot (inclusive)
+        end_year: int defining the end year of the snapshot (inclusive)
+        year_type: str defining year the 'fy' for financial year or 'calendar'
+
+    Raises: ValueError if the end of week falls outside after the year end i.e.
+        for all weeks 53 or greater and for some years the week 52.
+    """
+    start_year, end_year, month = get_iteration_start_and_end_time(
+        year_type, start_year, end_year
+    )
+
+    snapshot = snapshot.index.to_series()
+
+    filtered_snapshot = []
+
+    for year in range(start_year, end_year):
+        start_of_year_date_time = datetime(
+            year=year, month=month, day=1, hour=0, minute=0
+        )
+        end_of_year_date_time = datetime(
+            year=year + 1, month=month, day=1, hour=0, minute=0
+        )
+        days_until_monday = (7 - start_of_year_date_time.weekday()) % 7
+        first_monday = start_of_year_date_time + timedelta(days=days_until_monday)
+        for week_number in representative_weeks:
+            nth_week_start = first_monday + timedelta(weeks=week_number - 1)
+            nth_week_end = nth_week_start + timedelta(days=7)
+
+            if nth_week_end - timedelta(seconds=1) > end_of_year_date_time:
+                raise ValueError(
+                    f"Representative week {week_number} ends after end of model year {year}."
+                    " Adjust config to use a smaller week_number for representative_weeks."
+                )
+
+            filtered_snapshot.append(
+                snapshot[
+                    (snapshot > nth_week_start) & (snapshot <= nth_week_end)
+                ].copy()
+            )
+
+    filtered_snapshot = pd.concat(filtered_snapshot)
+
+    filtered_snapshot = pd.DataFrame(index=filtered_snapshot)
+
+    return filtered_snapshot