refactor: split py

v2/architecture.py

@@ -2,121 +2,17 @@ import numpy as np
 import pandas as pd
 import copy
 import time
-from enum import IntEnum
 import matplotlib.pyplot as plt
 
 # it's really just a network pricing model
+from decider import Decider, Decision
 from supplier import Supplier, precalculate_supplier
 from data_processing import read_datasets, add_production_field, interpolate_and_join, SolarParameters
+from bess import BatteryModel
 from evolution_strategies import evolution_strategies_optimizer
 
-DO_VIS = False
-
-STEPS_PER_HOUR = 12
-
-
-# SOC is normalized so that minimal_depth_of_discharge = 0 and maximal_depth_of_discharge = 1.
-# please set capacity_Ah = nominal_capacity_Ah * (max_dod - min_dod)
-#
-# TODO efficiency multiplier is not currently used, where best to put it?
-class BatteryModel:
-    def __init__(self, capacity_Ah, time_interval_h):
-        self.capacity_Ah = capacity_Ah
-        self.efficiency = 0.9 # [dimensionless]
-        self.voltage_V = 600
-        self.charge_kW = 50
-        self.discharge_kW = 60
-        self.time_interval_h = time_interval_h
-
-        # the only non-constant member variable!
-        # ratio of self.current_capacity_kWh and self.maximal_capacity_kWh
-        self.soc = 0.0 
-
-    @property
-    def maximal_capacity_kWh(self):
-        return self.capacity_Ah * self.voltage_V / 1000
-
-    @property
-    def current_capacity_kWh(self):
-        return self.soc * self.maximal_capacity_kWh
-
-    def satisfy_demand(self, demand_kW):
-        assert 0 <= self.soc <= 1
-        assert demand_kW >= 0
-        # rate limited:
-        possible_discharge_in_timestep_kWh = self.discharge_kW * self.time_interval_h
-        # limited by current capacity:
-        possible_discharge_in_timestep_kWh = min((possible_discharge_in_timestep_kWh, self.current_capacity_kWh))
-        # limited by need:
-        discharge_in_timestep_kWh = min((possible_discharge_in_timestep_kWh, demand_kW * self.time_interval_h))
-        consumption_from_bess_kW = discharge_in_timestep_kWh / self.time_interval_h
-        unsatisfied_demand_kW = demand_kW - consumption_from_bess_kW
-        cap_of_battery_kWh = self.current_capacity_kWh - discharge_in_timestep_kWh
-        soc = cap_of_battery_kWh / self.maximal_capacity_kWh
-        assert 0 <= soc <= self.soc <= 1
-        self.soc = soc
-        return unsatisfied_demand_kW
-
-    def charge(self, charge_kW):
-        assert 0 <= self.soc <= 1
-        assert charge_kW >= 0
-        # rate limited:
-        possible_charge_in_timestep_kWh = self.charge_kW * self.time_interval_h
-        # limited by current capacity:
-        possible_charge_in_timestep_kWh = min((possible_charge_in_timestep_kWh, self.maximal_capacity_kWh - self.current_capacity_kWh))
-        # limited by supply:
-        charge_in_timestep_kWh = min((possible_charge_in_timestep_kWh, charge_kW * self.time_interval_h))
-        actual_charge_kW = charge_in_timestep_kWh / self.time_interval_h
-        unused_charge_kW = charge_kW - actual_charge_kW
-        cap_of_battery_kWh = self.current_capacity_kWh + charge_in_timestep_kWh
-        soc = cap_of_battery_kWh / self.maximal_capacity_kWh
-        assert 0 <= self.soc <= soc <= 1
-        self.soc = soc
-        return unused_charge_kW
-
-
-class Decision(IntEnum):
-    # use solar to satisfy consumption,
-    # and if it is not enough, use network.
-    # BESS is not discharged in this mode,
-    # but might be charged if solar has surplus.
-    PASSIVE = 0
-
-    # use the battery if possible and necessary.
-    # the possible part means that there's charge in it,
-    # and the necessary part means that the consumption
-    # is not already covered by solar.
-    DISCHARGE = 1
-
-    # use the network to charge the battery
-    # this is similar to PASSIVE, but forces the
-    # BESS to be charged, even if solar does not cover
-    # the whole of consumption plus BESS.
-    NETWORK_CHARGE = 2
-
-
-# mock class as usual
-# output_window_size is not yet used, always decides one timestep.
-class Decider:
-    def __init__(self, precalculated_supplier):
-        self.input_window_size = STEPS_PER_HOUR * 24 # day long window.
-        self.random_seed = 0
-        self.precalculated_supplier = precalculated_supplier
-
-    def decide(self, prod_pred, cons_pred, fees, battery_model):
-        # TODO 15 minutes demand charge window hardwired at this weird place
-        DEMAND_CHARGE_WINDOW_MIN = 15
-        time_interval_min = self.precalculated_supplier.time_index.freq.n
-        assert DEMAND_CHARGE_WINDOW_MIN % time_interval_min == 0
-        peak_shaving_window = DEMAND_CHARGE_WINDOW_MIN // time_interval_min
-        step_in_hour = time_interval_min / 60 # [hour], the length of a time step.
-        deficit_kw = (cons_pred[:peak_shaving_window] - prod_pred[:peak_shaving_window]).clip(min=0)
-        deficit_kwh = (step_in_hour * deficit_kw).sum()
-        if deficit_kwh > self.precalculated_supplier.peak_demand:
-            return Decision.DISCHARGE
-        else:
-            return Decision.PASSIVE
 
+DO_VIS = False
 
 # we predict last week same time for consumption, and yesterday same time for production.
 # adds fields in-place
@@ -310,7 +206,6 @@ def main():
     all_data = interpolate_and_join(met_2021_data, cons_2021_data)
 
     time_interval_min = all_data.index.freq.n
-    print("time_interval_min", time_interval_min)
     time_interval_h = time_interval_min / 60
 
     all_data_with_predictions = all_data.copy()

v2/bess.py

@@ -0,0 +1,58 @@
+# SOC is normalized so that minimal_depth_of_discharge = 0 and maximal_depth_of_discharge = 1.
+# please set capacity_Ah = nominal_capacity_Ah * (max_dod - min_dod)
+#
+# TODO efficiency multiplier is not currently used, where best to put it?
+class BatteryModel:
+    def __init__(self, capacity_Ah, time_interval_h):
+        self.capacity_Ah = capacity_Ah
+        self.efficiency = 0.9 # [dimensionless]
+        self.voltage_V = 600
+        self.charge_kW = 50
+        self.discharge_kW = 60
+        self.time_interval_h = time_interval_h
+
+        # the only non-constant member variable!
+        # ratio of self.current_capacity_kWh and self.maximal_capacity_kWh
+        self.soc = 0.0 
+
+    @property
+    def maximal_capacity_kWh(self):
+        return self.capacity_Ah * self.voltage_V / 1000
+
+    @property
+    def current_capacity_kWh(self):
+        return self.soc * self.maximal_capacity_kWh
+
+    def satisfy_demand(self, demand_kW):
+        assert 0 <= self.soc <= 1
+        assert demand_kW >= 0
+        # rate limited:
+        possible_discharge_in_timestep_kWh = self.discharge_kW * self.time_interval_h
+        # limited by current capacity:
+        possible_discharge_in_timestep_kWh = min((possible_discharge_in_timestep_kWh, self.current_capacity_kWh))
+        # limited by need:
+        discharge_in_timestep_kWh = min((possible_discharge_in_timestep_kWh, demand_kW * self.time_interval_h))
+        consumption_from_bess_kW = discharge_in_timestep_kWh / self.time_interval_h
+        unsatisfied_demand_kW = demand_kW - consumption_from_bess_kW
+        cap_of_battery_kWh = self.current_capacity_kWh - discharge_in_timestep_kWh
+        soc = cap_of_battery_kWh / self.maximal_capacity_kWh
+        assert 0 <= soc <= self.soc <= 1
+        self.soc = soc
+        return unsatisfied_demand_kW
+
+    def charge(self, charge_kW):
+        assert 0 <= self.soc <= 1
+        assert charge_kW >= 0
+        # rate limited:
+        possible_charge_in_timestep_kWh = self.charge_kW * self.time_interval_h
+        # limited by current capacity:
+        possible_charge_in_timestep_kWh = min((possible_charge_in_timestep_kWh, self.maximal_capacity_kWh - self.current_capacity_kWh))
+        # limited by supply:
+        charge_in_timestep_kWh = min((possible_charge_in_timestep_kWh, charge_kW * self.time_interval_h))
+        actual_charge_kW = charge_in_timestep_kWh / self.time_interval_h
+        unused_charge_kW = charge_kW - actual_charge_kW
+        cap_of_battery_kWh = self.current_capacity_kWh + charge_in_timestep_kWh
+        soc = cap_of_battery_kWh / self.maximal_capacity_kWh
+        assert 0 <= self.soc <= soc <= 1
+        self.soc = soc
+        return unused_charge_kW

v2/decider.py

@@ -0,0 +1,50 @@
+from enum import IntEnum
+
+
+
+STEPS_PER_HOUR = 12
+
+
+class Decision(IntEnum):
+    # use solar to satisfy consumption,
+    # and if it is not enough, use network.
+    # BESS is not discharged in this mode,
+    # but might be charged if solar has surplus.
+    PASSIVE = 0
+
+    # use the battery if possible and necessary.
+    # the possible part means that there's charge in it,
+    # and the necessary part means that the consumption
+    # is not already covered by solar.
+    DISCHARGE = 1
+
+    # use the network to charge the battery
+    # this is similar to PASSIVE, but forces the
+    # BESS to be charged, even if solar does not cover
+    # the whole of consumption plus BESS.
+    NETWORK_CHARGE = 2
+
+
+# mock class as usual
+# output_window_size is not yet used, always decides one timestep.
+class Decider:
+    def __init__(self, precalculated_supplier):
+        self.input_window_size = STEPS_PER_HOUR * 24 # day long window.
+        self.random_seed = 0
+        self.precalculated_supplier = precalculated_supplier
+
+    def decide(self, prod_pred, cons_pred, fees, battery_model):
+        # TODO 15 minutes demand charge window hardwired at this weird place
+        DEMAND_CHARGE_WINDOW_MIN = 15
+        time_interval_min = self.precalculated_supplier.time_index.freq.n
+        assert DEMAND_CHARGE_WINDOW_MIN % time_interval_min == 0
+        peak_shaving_window = DEMAND_CHARGE_WINDOW_MIN // time_interval_min
+        step_in_hour = time_interval_min / 60 # [hour], the length of a time step.
+        deficit_kw = (cons_pred[:peak_shaving_window] - prod_pred[:peak_shaving_window]).clip(min=0)
+        deficit_kwh = (step_in_hour * deficit_kw).sum()
+        if deficit_kwh > self.precalculated_supplier.peak_demand:
+            return Decision.DISCHARGE
+        else:
+            return Decision.PASSIVE
+
+