Merge pull request #221 from rl-institut/feature/battery_strategy

battery strategies

Author: stefansc1

.gitignore

@@ -2,6 +2,7 @@
 /__pycache__
 *.pyc
 .DS_Store
+spice_ev.egg-info/
 
 .coverage
 htmlcov/

doc/source/charging_strategies_incentives.rst

@@ -112,6 +112,23 @@ again, the process is repeated.
 .. image:: _files/example_strategies.png
    :width: 80 %
 
+Battery Strategies
+==================
+
+Stationary batteries may be managed through a special battery strategy, instead of relying on a charging strategy's behavior. The following battery strategies are supported:
+
+surplus
+-------
+Charge stationary batteries if there is a surplus of energy, for example through feed-in from photovoltaics. Discharge to support a GC in times of high load (specifically, if the maximum power at a GC would be exceeded).
+
+peak_shaving
+------------
+Charge stationary batteries in times of low load (maximum power of a GC not yet reached, includes surplus) and discharge in times of high load (maximum power of a GC exceeded).
+
+peak_load_window
+----------------
+Similar to peak_shaving, but only charge from GC outside of peak load windows. May use surplus inside of peak load windows. Discharge to support GC as needed.
+
 Cost calculation
 ================
 

doc/source/simulating_with_spiceev.rst

@@ -209,7 +209,9 @@ stationary batteries or V2G, you need to set the target SOC parameter of the bat
     +---------------------+----------------------------+---------------------------------------------------------+-------------+--------------+---------------------+--------------+------------------+---------------------+------------------+-----------------+
     | **Strategy option** | **Default**                |              **Explanation**                            | **Greedy**  | **Balanced** | **Balanced Market** | **Schedule** | **Peak Shaving** | **Peak Load Window**| **Flex Window**  | **Distributed** |
     +---------------------+----------------------------+---------------------------------------------------------+-------------+--------------+---------------------+--------------+------------------+---------------------+------------------+-----------------+
-    | Perfect_Foresight   | True                       | All events and loads are known at start of simulation.  |             |              |                     |              | x                |                     |                  |                 |
+    | battery_strategy    | None                       | Use special strategy for stationary batteries.          | x           | x            | x                   |              | x                | x                   | x                | x               |
+    +---------------------+----------------------------+---------------------------------------------------------+-------------+--------------+---------------------+--------------+------------------+---------------------+------------------+-----------------+
+    | perfect_foresight   | True                       | All events and loads are known at start of simulation.  |             |              |                     |              | x                |                     |                  |                 |
     +---------------------+----------------------------+---------------------------------------------------------+-------------+--------------+---------------------+--------------+------------------+---------------------+------------------+-----------------+
     | CONCURRENCY         | 1.0                        | Reduce maximum available power at each charging station.| x           |              | x                   |              |                  |                     | x                |                 |
     |                     |                            |                                                         |             |              |                     |              |                  |                     |                  |                 |

spice_ev/generate/generate_schedule.py

@@ -99,7 +99,7 @@ def clamp_to_gc(power):
     for step_i in range(scenario.n_intervals):
         with warnings.catch_warnings():
             warnings.simplefilter('ignore', UserWarning)
-            s.step(event_steps[step_i])
+            s.pre_step(event_steps[step_i])
 
         current_datetime = scenario.start_time + scenario.interval * step_i
         currently_in_core_standing_time = \

spice_ev/scenario.py

@@ -125,7 +125,7 @@ def run(self, strategy_name, options):
 
             # process events
             try:
-                super(type(strat), strat).step(event_steps[step_i])
+                strat.pre_step(event_steps[step_i])
             except Exception:
                 error = traceback.format_exc()
 
@@ -183,6 +183,7 @@ def run(self, strategy_name, options):
             try:
                 if error is None:
                     res = strat.step()
+                    strat.post_step()
             except Exception:
                 # error during strategy: add dummy result and abort
                 error = traceback.format_exc() if error is None else error

spice_ev/strategies/distributed.py

@@ -55,32 +55,12 @@ def __init__(self, comps, start_time, **kwargs):
 
         # prepare batteries
         for b_id, bat in self.world_state.batteries.items():
-            # make note to run GC even if no vehicles are connected
+            # create look-up-table for GC ID -> battery dict
             if self.gc_battery.get(bat.parent):
                 self.gc_battery[bat.parent][b_id] = bat
             else:
                 self.gc_battery[bat.parent] = {b_id: bat}
 
-            station_type = self.strategies.get(bat.parent)
-            if station_type is None or station_type[0] == "deps":
-                # only batteries at opportunity stations need preparation
-                continue
-            name = f"stationary_{b_id}"
-            # create new vehicle type equivalent to battery (only charging, no discharging)
-            self.virtual_vt[name] = components.VehicleType({
-                "name": name,
-                "capacity": bat.capacity,
-                "charging_curve": bat.charging_curve.points,
-                "min_charging_power": bat.min_charging_power,
-                "battery_efficiency": bat.efficiency,
-            })
-            # set up virtual charging station
-            self.virtual_cs[name] = components.ChargingStation({
-                "parent": bat.parent,
-                "max_power": bat.charging_curve.max_power,
-                "min_power": bat.min_charging_power,
-            })
-
     def step(self):
         """ Calculates charging power in each timestep.
 
@@ -189,6 +169,7 @@ def step(self):
                 # copy reference of current GC and relevant vehicles
                 # changes during simulation reflect back to original!
                 new_world_state.grid_connectors = {gc_id: gc}
+                strat.gc_power[gc_id] = gc.cur_max_power
 
                 # filter future events for this GC (within event horizon)
                 new_world_state.future_events = []
@@ -221,71 +202,25 @@ def step(self):
                             new_world_state.future_events.append(event)
 
                 # stationary batteries
-                avail_bat_power = dict()
-                if station_type == "deps":
-                    # depot: use stationary batteries according to selected strategy
-                    new_world_state.batteries = self.gc_battery.get(gc_id, {})
-                else:
-                    # opportunity station:
-                    # - charge according to strategy (simulate by creating equivalent vehicle)
-                    # - discharge when needed power is above GC max power
-                    for b_id, battery in self.gc_battery.get(gc_id, {}).items():
-                        if connected_vehicles:
-                            # vehicle present: support GC (increase GC max power)
-                            power = battery.get_available_power(self.interval)
-                            if power < battery.min_charging_power:
-                                # below minimum (dis)charging power
-                                continue
-                            total_time = self.interval.total_seconds() / 3600
-                            energy_delta = power / battery.efficiency * total_time
-                            soc_delta = energy_delta / battery.capacity
-                            if soc_delta < self.EPS:
-                                # remaining power too small
-                                continue
-                            avail_bat_power[b_id] = (power, gc.cur_max_power)
-                            gc.cur_max_power += power
-                        else:
-                            # vacant station: charge with strategy until vehicle arrives
-                            name = f"stationary_{b_id}"
-                            arrive = next_arrival.get(gc_id, self.current_time+self.ARRIVAL_HORIZON)
-                            bat_vehicle = components.Vehicle({
-                                "vehicle_type": name,
-                                "connected_charging_station": name,
-                                "soc": battery.soc,
-                                "desired_soc": 1,
-                                "estimated_time_of_departure": str(arrive),
-                            }, self.virtual_vt)
-                            new_world_state.vehicle_types[name] = self.virtual_vt[name]
-                            new_world_state.charging_stations[name] = self.virtual_cs[name]
-                            new_world_state.vehicles[b_id] = bat_vehicle
+                gc_batteries = self.gc_battery.get(gc_id, dict())
+                new_world_state.batteries = gc_batteries
+                if strat.battery_strategy is not None:
+                    # special stationary battery strategy: ignore charging strategy
+                    # remove batteries from GC, so they can't charge
+                    # add available battery power to GC power
+                    for bat in gc_batteries.values():
+                        available_power = bat.get_available_power(self.interval)
+                        gc.cur_max_power += available_power
+                        bat.parent = None
 
                 # update world state of strategy
                 strat.current_time = self.current_time
                 strat.world_state = new_world_state
                 # run sub-strategy
                 commands = strat.step()["commands"]
-                # update stationary batteries
-                if station_type == "opps":
-                    for b_id, battery in self.gc_battery.get(gc_id, {}).items():
-                        power = avail_bat_power.get(b_id)
-                        if power is not None:
-                            # battery used to support GC -> revert max_power, discharge
-                            gc.cur_max_power = power[1]
-                            power_needed = gc.get_current_load() - gc.cur_max_power
-                            power = battery.unload(self.interval, target_power=max(power_needed, 0))
-                            gc.add_load(b_id, -power['avg_power'])
-                            continue
-                        name = f"stationary_{b_id}"
-                        if name in commands:
-                            # battery is simulated as vehicle -> apply changes
-                            # remove from commands
-                            del commands[name]
-                            # and add as battery
-                            # this will crash if virtual CS power has not been added correctly to GC
-                            gc.add_load(b_id, gc.current_loads.pop(name))
-                            # update battery SoC
-                            battery.soc = strat.world_state.vehicles[b_id].battery.soc
                 charging_stations.update(commands)
+                # update batteries according to battery strategy
+                strat.post_step()
 
         # all vehicles charged
         charging_stations.update(self.distribute_surplus_power())

spice_ev/strategies/peak_load_window.py

@@ -1,6 +1,5 @@
 from copy import deepcopy
 import datetime
-import json
 import warnings
 
 from spice_ev import events, util
@@ -23,54 +22,9 @@ def __init__(self, components, start_time, **kwargs):
 
         if self.time_windows is None:
             raise Exception("Need time windows for Peak Load Window strategy")
-        with open(self.time_windows, 'r') as f:
-            self.time_windows = json.load(f)
-
-        # check time windows
-        # start year in time windows?
-        years = set()
-        for grid_operator in self.time_windows.values():
-            for window in grid_operator.values():
-                years.add(int(window["start"][:4]))
-        assert len(years) > 0, "No time windows given"
-        # has the scenario year to be replaced because it is not in time windows?
-        replace_year = start_time.year not in years
-        if replace_year:
-            replace_year = start_time.year
-            old_year = sorted(years)[0]
-            warnings.warn("Time windows do not include scenario year,"
-                          f"replacing {old_year} with {replace_year}")
-        # cast strings to dates/times, maybe replacing year
-        grid_operator = None
-        for grid_operator, grid_operator_seasons in self.time_windows.items():
-            for season, info in grid_operator_seasons.items():
-                start_date = datetime.date.fromisoformat(info["start"])
-                if replace_year and start_date.year == old_year:
-                    start_date = start_date.replace(year=replace_year)
-                info["start"] = start_date
-                end_date = datetime.date.fromisoformat(info["end"])
-                if replace_year and end_date.year == old_year:
-                    end_date = end_date.replace(year=replace_year)
-                info["end"] = end_date
-                for level, windows in info.get("windows", {}).items():
-                    # cast times to datetime.time, store as tuples
-                    info["windows"][level] = [
-                        (datetime.time.fromisoformat(t[0]), datetime.time.fromisoformat(t[1]))
-                        for t in windows]
-                self.time_windows[grid_operator][season] = info
-
-        gcs = self.world_state.grid_connectors
-
-        for gc_id, gc in gcs.items():
-            if gc.voltage_level is None:
-                warnings.warn(f"GC {gc_id} has no voltage level, might not find time window")
-                warnings.warn("SETTING VOLTAGE LEVEL TO MV")
-                gc.voltage_level = "MV"  # TODO remove
-            if gc.grid_operator is None:
-                warnings.warn(f"GC {gc_id} has no grid operator, might not find time window")
-                # take the first grid operator from time windows
-                warnings.warn(f"SETTING GRID OPERATOR TO {grid_operator}")
-                gc.grid_operator = grid_operator  # TODO remove
+        if type(self.time_windows) is not dict:
+            # parse time windows file at path into time windows dict
+            util.parse_time_windows(self, self.time_windows, start_time=start_time, check_gc=True)
 
         # perfect foresight for grid and local load events
         local_events = [e for e in self.events.grid_operator_signals
@@ -97,6 +51,7 @@ def __init__(self, components, start_time, **kwargs):
             elif event.event_type == "departure":
                 stop_time = max(stop_time, event.start_time)
 
+        gcs = self.world_state.grid_connectors
         # restructure events (like event_steps): list with events for each timestep
         # also, find highest peak of GC power within time windows
         self.events = []