holidays/weekends for peak load window

examples/data/time_windows.json

@@ -37,5 +37,6 @@
                 "LV": [["10:30", "12:00"],  ["17:45", "19:00"]]
             }
         }
-    }
+    },
+    "holidays": ["2020-01-01", "2020-04-10", "2020-04-13", "2020-05-01", "2020-05-21", "2020-10-03", "2020-12-25", "2020-12-26"]
 }

examples/output/scenario_generate.json

@@ -2,7 +2,8 @@
   "scenario": {
     "start_time": "2023-01-01T01:00:00+02:00",
     "interval": 15.0,
-    "stop_time": "2023-01-08T01:00:00+02:00"
+    "stop_time": "2023-01-08T01:00:00+02:00",
+    "holidays": []
   },
   "components": {
     "vehicle_types": {

examples/output/scenario_generate_from_simbev.json

@@ -2,7 +2,8 @@
   "scenario": {
     "start_time": "2021-09-17T00:00:00",
     "interval": 15,
-    "n_intervals": 1345
+    "n_intervals": 1345,
+    "holidays": []
   },
   "components": {
     "vehicle_types": {

examples/output/scenario_generate_from_statistics.json

@@ -2,7 +2,10 @@
   "scenario": {
     "start_time": "2023-01-01T00:00:00+02:00",
     "interval": 15.0,
-    "stop_time": "2023-01-08T00:00:00+02:00"
+    "stop_time": "2023-01-08T00:00:00+02:00",
+    "holidays": [
+      "2023-01-01"
+    ]
   },
   "components": {
     "vehicle_types": {

spice_ev/generate/generate_from_csv.py

@@ -300,6 +300,7 @@ def generate_from_csv(args):
             "start_time": start.isoformat(),
             "interval": interval.total_seconds() // 60,
             "stop_time": stop.isoformat(),
+            "holidays": vars(args).get("holidays", []),
         },
         "components": {
             "vehicle_types": vehicle_types,

spice_ev/generate/generate_from_simbev.py

@@ -417,6 +417,7 @@ def datetime_from_timestep(timestep):
             "start_time": start.isoformat(),
             "interval": args.interval,
             "n_intervals": n_intervals,
+            "holidays": vars(args).get("holidays", []),
         },
         "components": {
             "vehicle_types": vehicle_types,

spice_ev/generate/generate_from_statistics.py

@@ -298,6 +298,7 @@ def generate_from_statistics(args):
             "start_time": start.isoformat(),
             "interval": interval.total_seconds() // 60,
             "stop_time": stop.isoformat(),
+            "holidays": vars(args).get("holidays", []),
         },
         "components": {
             "vehicle_types": vehicle_types,

spice_ev/scenario.py

@@ -47,6 +47,9 @@ def __init__(self, json_dict, dir_path=''):
 
         # only relevant for schedule strategy
         self.core_standing_time = scenario.get('core_standing_time', None)
+        # only relevant for peak load window strategy
+        # holidays might also be defined in time_windows file
+        self.holiday = scenario.get('holidays', [])
 
         # compute average load for each timeslot
         for fixed_load_list in self.events.fixed_load_lists.values():
@@ -67,6 +70,7 @@ def run(self, strategy_name, options):
         """
 
         options['events'] = self.events
+        options['holiday'] = self.holiday
         options['interval'] = self.interval
         options['stop_time'] = self.stop_time
         options['n_intervals'] = self.n_intervals

spice_ev/strategies/peak_load_window.py

@@ -25,6 +25,11 @@ def __init__(self, components, start_time, **kwargs):
         with open(self.time_windows, 'r') as f:
             self.time_windows = json.load(f)
 
+        # get holidays from time windows and scenario
+        holidays = self.time_windows.pop("holidays", [])
+        holidays += kwargs.get('holiday', [])
+        self.holidays = [datetime.date.fromisoformat(date) for date in holidays]
+
         # check time windows
         # start year in time windows?
         years = set()
@@ -96,6 +101,13 @@ def __init__(self, components, start_time, **kwargs):
             elif event.event_type == "departure":
                 stop_time = max(stop_time, event.start_time)
 
+        # check if holidays are within scenario range
+        filtered_holidays = [
+            d for d in self.holidays if start_time.date() <= d <= stop_time.date()]
+        if filtered_holidays != self.holidays:
+            warnings.warn(f"{len(self.holidays) - len(filtered_holidays)} holidays ignored")
+            self.holidays = filtered_holidays
+
         # restructure events (like event_steps): list with events for each timestep
         # also, find highest peak of GC power within time windows
         self.events = []
@@ -130,10 +142,8 @@ def __init__(self, components, start_time, **kwargs):
             # end of events for this timestep
             # update peak power
             for gc_id, gc in gcs.items():
-                is_window = util.datetime_within_time_window(
-                    cur_time, self.time_windows[gc.grid_operator], gc.voltage_level)
                 gc_sum_loads = sum(current_loads[gc_id].values())
-                if is_window and gc_sum_loads > peak_power[gc_id]:
+                if self.within_window(gc, cur_time) and gc_sum_loads > peak_power[gc_id]:
                     # new peak power
                     peak_power[gc_id] = gc_sum_loads
                     peak_time[gc_id] = cur_time
@@ -144,6 +154,11 @@ def __init__(self, components, start_time, **kwargs):
                 warnings.warn(f"Peak power of {peak_power[gc_id]} kW at {gc_id} "
                               f"is not within simulation time, but at {t}")
 
+    def within_window(self, gc, dt):
+        return util.datetime_within_time_window(
+            dt, self.time_windows[gc.grid_operator], gc.voltage_level
+        ) and util.is_workday(dt, self.holidays)
+
     def step(self):
         """ Calculate charging power in each timestep.
 
@@ -192,16 +207,12 @@ def step_gc(self, gc_id, gc):
         stationary_batteries = {
             bid: b for bid, b in self.world_state.batteries.items() if b.parent == gc_id}
 
-        def within_window(dt):
-            return util.datetime_within_time_window(
-                dt, self.time_windows[gc.grid_operator], gc.voltage_level)
-
-        gc.window = within_window(self.current_time)
+        gc.window = self.within_window(gc, self.current_time)
         if stationary_batteries:
             # stat. batteries present: find next change of time window (or end of scenario)
             cur_time = self.current_time + self.interval
             ts_until_window_change = 1
-            while within_window(cur_time) == gc.window and cur_time <= self.stop_time:
+            while self.within_window(gc, cur_time) == gc.window and cur_time <= self.stop_time:
                 cur_time += self.interval
                 ts_until_window_change += 1
             if gc.window:
@@ -231,13 +242,10 @@ def within_window(dt):
                 {
                     "power": sum(cur_loads.values()),
                     "max_power": cur_max_power,
-                    "window": util.datetime_within_time_window(
-                        cur_time, self.time_windows[gc.grid_operator], gc.voltage_level)
+                    "window": self.within_window(gc, cur_time),
                 }
             )
 
-        gc.window = util.datetime_within_time_window(
-            self.current_time, self.time_windows[gc.grid_operator], gc.voltage_level)
         peak_power = self.peak_power[gc_id]
 
         # sort vehicles by length of standing time

spice_ev/util.py

@@ -50,6 +50,44 @@ def datetime_within_time_window(dt, time_windows, voltage_level):
     return False
 
 
+def is_workday(dt, holidays):
+    """
+    Check if a given datetime is during a workday.
+
+    - must not be weekend
+    - must not be in holidays list
+    - must not be single day between holiday and weekend
+    - must not be between Christmas and New Year
+
+    :param dt: time
+    :type dt: datetime
+    :param holidays: holiday dates
+    :return: is datetime a workday?
+    :rtype: bool
+    """
+    if dt.weekday() > 4:
+        # weekend
+        return False
+    for d in holidays:
+        if dt.date() == d:
+            # exact holiday
+            return False
+        if dt.weekday() == 0 and (dt + datetime.timedelta(days=1)).date() == d:
+            # Monday before holiday
+            return False
+        if dt.weekday() == 4 and (dt - datetime.timedelta(days=1)).date() == d:
+            # Friday after holiday
+            return False
+    if dt.month == 12:
+        if dt.day >= 25:
+            # between Christmas and New Year
+            return False
+        if dt.day == 24 and dt.weekday() == 0:
+            # Monday before Christmas
+            return False
+    return True
+
+
 def dt_within_core_standing_time(dt, core_standing_time):
     """ Check if datetime dt is in inside core standing time.
 

tests/test_calculate_costs.py

@@ -90,8 +90,9 @@ def test_calculate_costs_basic(self):
                                price_sheet_path=str(price_sheet_path))
 
         # check returned values
-        result = cc.calculate_costs(supported_strategies[0], "MV", s.interval,
-                                    *timeseries_lists, price_sheet_path=str(price_sheet_path))
+        result = cc.calculate_costs(
+            supported_strategies[0], "MV", s.interval,
+            *timeseries_lists, price_sheet_path=str(price_sheet_path))
         assert result["total_costs_per_year"] == 78.18
         assert result["commodity_costs_eur_per_year"] == 0
         assert result["capacity_costs_eur"] == 65.7
@@ -175,8 +176,9 @@ def test_calculate_costs_balanced_market_A(self):
         pv = sum([pv.nominal_power for pv in s.components.photovoltaics.values()])
 
         # check returned values
-        result = cc.calculate_costs("balanced_market", "MV", s.interval, *timeseries_lists,
-                                    str(price_sheet_path), grid_operator, None, pv)
+        result = cc.calculate_costs(
+            "balanced_market", "MV", s.interval, *timeseries_lists,
+            str(price_sheet_path), grid_operator, None, pv)
         assert result["total_costs_per_year"] == 323.14
         assert result["commodity_costs_eur_per_year"] == 14.41
         assert result["capacity_costs_eur"] == 0
@@ -229,14 +231,15 @@ def test_peak_load_window_C1(self):
         pv_power = j["components"]["photovoltaics"]["PV1"]["nominal_power"]
 
         # check returned values
-        result = cc.calculate_costs("peak_load_window", "MV", s.interval, *timeseries_lists,
-                                    str(price_sheet_path), power_pv_nominal=pv_power)
-        assert result["total_costs_per_year"] == 32206.19
-        assert result["commodity_costs_eur_per_year"] == 5699.49
+        result = cc.calculate_costs(
+            "peak_load_window", "MV", s.interval, *timeseries_lists,
+            str(price_sheet_path), power_pv_nominal=pv_power)
+        assert result["total_costs_per_year"] == 31627.88
+        assert result["commodity_costs_eur_per_year"] == 5541.06
         assert result["capacity_costs_eur"] == 1497.21
-        assert result["power_procurement_costs_per_year"] == 12574.80
-        assert result["levies_fees_and_taxes_per_year"] == 12709.74
-        assert result["feed_in_remuneration_per_year"] == 275.03
+        assert result["power_procurement_costs_per_year"] == 12225.26
+        assert result["levies_fees_and_taxes_per_year"] == 12364.34
+        assert result["feed_in_remuneration_per_year"] == 0.0
 
     def test_calculate_costs_balanced_market_C(self):
         scen_path = TEST_REPO_PATH / 'test_data/input_test_strategies/scenario_C1.json'
@@ -256,8 +259,9 @@ def test_calculate_costs_balanced_market_C(self):
         pv = sum([pv.nominal_power for pv in s.components.photovoltaics.values()])
 
         # check returned values
-        result = cc.calculate_costs("balanced_market", "MV", s.interval, *timeseries_lists,
-                                    str(price_sheet_path), grid_operator, None, pv)
+        result = cc.calculate_costs(
+            "balanced_market", "MV", s.interval, *timeseries_lists,
+            str(price_sheet_path), grid_operator, None, pv)
         assert result["total_costs_per_year"] == 24.02
         assert result["commodity_costs_eur_per_year"] == 4.56
         assert result["capacity_costs_eur"] == 4.42

tests/test_examples.py

@@ -1,5 +1,7 @@
+import difflib
 from pathlib import Path
 import subprocess
+import sys
 
 TEST_REPO_PATH = Path(__file__).parent
 EXAMPLE_PATH = TEST_REPO_PATH.parent / "examples"
@@ -11,11 +13,16 @@ def compare_files(p1, p2):
     # => replace different line break with universal linebreak (by reading with newline=None)
     try:
         with p1.open('r', newline=None) as f1:
-            content1 = f1.read()
+            content1 = f1.readlines()
         with p2.open('r', newline=None) as f2:
-            content2 = f2.read()
+            content2 = f2.readlines()
+        if content1 != content2:
+            # text content differs: write diff to stderr
+            diff = difflib.unified_diff(content1, content2, fromfile=str(p1), tofile=str(p2), n=0)
+            sys.stderr.writelines(diff)
         return content1 == content2
-    except Exception:
+    except Exception as e:
+        sys.stderr.write(f"Error while comparing: {str(e)}")
         return False
 
 

tests/test_util.py

@@ -67,6 +67,24 @@ def test_time_window(self):
             # wrong voltage level
             assert not util.datetime_within_time_window(dt, time_windows, "not lvl")
 
+    def test_is_workday(self):
+        holidays = [datetime.date.fromisoformat(d) for d in ["2020-01-01", "2020-05-21"]]
+        dates = [
+            ("2020-01-01", False),  # in holidays list
+            ("2020-01-03", True),   # is workday (Friday)
+            ("2020-01-04", False),  # Saturday
+            ("2020-01-05", False),  # Sunday
+            ("2020-05-22", False),  # Friday after holiday
+            ("2020-12-24", True),   # Christmas Eve is working day
+            ("2018-12-24", False),  # except when it's Monday
+            ("2020-12-25", False),  # Christmas day is not a working day
+            ("2020-12-29", False),  # between Christmas and New Year
+            ("2020-12-31", False),  # New Year
+        ]
+        for d, expect in dates:
+            assert util.is_workday(datetime.datetime.fromisoformat(d), holidays) == expect, (
+                f"Workday mismatch: {d}")
+
     def test_core_window(self):
         dt = datetime.datetime(day=1, month=1, year=2020)
         assert util.dt_within_core_standing_time(dt, None)