Add market data handling, battery state updates, and unit tests

src/batterymodel/models/__main__.py

@@ -0,0 +1,26 @@
+import argparse
+import datetime
+from pathlib import Path
+
+from .market import MarketState
+from .battery import BatteryState
+
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--market1", type=Path, help="Market state file")
+    parser.add_argument("--market2", type=Path, help="Market state file")
+
+    args = parser.parse_args()
+    market1 = MarketState().load_market_data_from_csv(args.market1)
+    market2 = MarketState().load_market_data_from_csv(args.market2)
+
+    battery = BatteryState()
+
+    increment = datetime.timedelta(minutes=10)
+    market_1_start = market1.get_start_time()
+    market2_start = market2.get_start_time()
+    start_time = min(market_1_start, market2_start)
+
+

src/batterymodel/models/battery.py

@@ -1,6 +1,8 @@
 import datetime
 from enum import Enum
 
+from batterymodel.models.market import MarketDataIncrement
+
 
 class BatteryState(Enum):
     IDLE = 0
@@ -19,12 +21,59 @@ class BatteryModel:
         self._storage_volume_degradation_rate = 0.001  # %/cycle
         self._fixed_operational_costs = 50000  # £/year
         self._runtime = datetime.timedelta(0)
+        self._charge_discharge_cycles = 0
         self._decision_log_file_handle = open("battery_log.txt", "w+")
 
         self._battery_state = BatteryState.IDLE
+        self._current_charge_rate = 0.0
+        self._current_discharge_rate = 0.0
         self._current_market_rate = 0.0
         self._current_charge_value = 0.0
+        self._current_charge = 0.0
         self._total_profit = 0.0
 
-    def update(self, increment: datetime.timedelta):
-        pass
+    def update(self, increment: datetime.timedelta, previous_state: "BatteryModel", market_data_increment: MarketDataIncrement):
+        pass
+
+    def handle_charge_or_discharge(self, increment: datetime.timedelta, previous_state: "BatteryModel"):
+        if self._battery_state != BatteryState.CHARGING:
+            total_charge_this_increment = self._current_charge_rate * (increment.total_seconds()/60.0)
+            self._current_charge = self._current_charge + total_charge_this_increment
+
+            # We filled the battery this increment
+            if self._current_charge > self._battery_capacity:
+                self._current_charge = self._battery_capacity
+                total_charge_this_increment = self._battery_capacity - previous_state._current_charge
+                self._battery_state = BatteryState.IDLE
+                time_of_charge = datetime.timedelta(seconds=total_charge_this_increment / self._current_charge_rate)
+
+        elif self._battery_state == BatteryState.DISCHARGING:
+
+            total_discharge_this_increment = self._current_discharge_rate * (increment.total_seconds() / 60.0)
+            self._current_charge = self._current_charge - total_discharge_this_increment
+
+            # emptied the battery this increment
+            if self._current_charge <= 0:
+                self._battery_state = BatteryState.IDLE
+                # might have to deal with penalties here for not providing enough?
+                time_of_charge = datetime.timedelta(seconds=previous_state._current_charge / self._current_discharge_rate)
+
+        if self._battery_state != BatteryState.IDLE:
+            self._current_charge_value = self._current_charge_value + (self._current_market_rate * time_of_charge.total_seconds())
+            self._total_profit += self._current_charge_value
+            self.log_state_change("handle_charge_or_discharge_charge", previous_state._current_charge, self._current_charge)
+            self.log_state_change("handle_charge_or_discharge_value", previous_state._current_charge_value,
+                                  self._current_charge_value)
+
+    def log_state_change(self, fn, original_state, new_state):
+        self._decision_log_file_handle.write(f"{self._runtime},{fn},{original_state},{new_state}\n")
+
+    def handle_storage_capacity_check(self, previous_state: "BatteryModel"):
+        if self._charge_discharge_cycles > previous_state._charge_discharge_cycles and self._charge_discharge_cycles / int(self._charge_discharge_cycles) == 1:
+            storage_volume_lost = (self._max_storage_volume/100) * self._storage_volume_degradation_rate
+            self._max_storage_volume -= storage_volume_lost
+            self.log_state_change("handle_storage_capacity_check_cycles_check", previous_state._max_storage_volume, self._max_storage_volume)
+
+
+
+

src/batterymodel/models/market.py

@@ -0,0 +1,40 @@
+import csv
+import datetime
+from collections import OrderedDict
+from typing import List, Tuple
+from xmlrpc.client import DateTime
+
+
+class MarketState:
+    def __init__(self):
+        self._market_data = OrderedDict()
+
+    def load_market_data_from_csv(self, csv_file):
+        with open(csv_file, 'r') as file:
+            reader = csv.reader(file)
+            for row in reader:
+                self._market_data[datetime.datetime.strptime(row[0], '%d-%m-%Y %H:%M')] = float(row[1])
+
+    def get_start_time(self):
+        return next(iter(self._market_data))
+
+    def get_market_rate_for_timestamp(self, timestamp: datetime.datetime):
+        last_timestamp = None
+        for key in self._market_data:
+            if last_timestamp is not None:
+                delta = timestamp - last_timestamp
+                key_delta = key - last_timestamp
+            if key == timestamp:
+                return self._market_data[key]
+            elif last_timestamp is not None and last_timestamp < key and key_delta > delta:
+
+                return self._market_data[last_timestamp]
+            else:
+                last_timestamp = key
+        return None
+
+
+class MarketDataIncrement:
+    def __init__(self, timestamp: datetime.datetime, markets: List[Tuple[datetime.datetime, float]]):
+        self.timestamp = timestamp
+        self.markets = markets

tests/test_battery.py

@@ -1,4 +1,6 @@
 from batterymodel import BatteryModel
+from batterymodel.models.battery import BatteryState
+
 
 def test_battery_stats():
     assert BatteryModel()._max_charge_rate == 2000000 # Watts
@@ -10,3 +12,7 @@ def test_battery_stats():
     assert BatteryModel()._lifetime_cycles == 50000
     assert BatteryModel()._storage_volume_degradation_rate == 0.001 # %/cycle
     assert BatteryModel()._fixed_operational_costs == 50000 # £/year
+
+def test_battery_update():
+    battery = BatteryModel()
+    assert battery._battery_state == BatteryState.IDLE

tests/test_market.py

@@ -0,0 +1,16 @@
+import datetime
+
+from batterymodel.models.market import MarketState
+
+
+def test_market():
+    this_market = MarketState()
+    this_market._market_data[datetime.datetime.strptime('01-01-2022 00:00', '%d-%m-%Y %H:%M')] = 11
+    this_market._market_data[datetime.datetime.strptime('01-01-2022 00:01', '%d-%m-%Y %H:%M')] = 12
+    this_market._market_data[datetime.datetime.strptime('01-01-2022 00:02', '%d-%m-%Y %H:%M')] = 13
+    this_market._market_data[datetime.datetime.strptime('01-01-2022 00:05', '%d-%m-%Y %H:%M')] = 14
+
+    assert this_market.get_market_rate_for_timestamp(datetime.datetime.strptime('01-01-2022 00:02', '%d-%m-%Y %H:%M')) == 13
+    assert this_market.get_market_rate_for_timestamp(datetime.datetime.strptime('01-01-2022 00:03', '%d-%m-%Y %H:%M')) == 13
+
+