matt/batterymodel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor BatteryModel and market handling for improved time tracking, logging, and market rate calculations

Browse Source 
Something is slow, I suspect its the logging, but time is tight, so I'm going to wrap up to make it at least useable
This commit is contained in:
matt
2026-09-22 17:55:58 +01:00
parent b2eaa6394a
commit 3000b461ec
6 changed files with 150 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/batterymodel/__main__.py
+47
View File
@@ -0,0 +1,47 @@
import argparse
import datetime
import sys
from copy import copy
from pathlib import Path
from batterymodel import BatteryModel
from batterymodel.models.market import MarketState, MarketDataIncrement
from batterymodel.models.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)
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)
print(f"Start Time: {start_time}")
end_time = max(market1.get_end_time(), market2.get_end_time())
print(f"End Time: {end_time}")
battery = BatteryModel(start_time)
current_time = copy(start_time)
while True:
this_market_info_1 = market1.get_market_rate_for_timestamp(current_time)
this_market_info_2 = market2.get_market_rate_for_timestamp(current_time)
if current_time > end_time:
sys.exit()
market_increment = MarketDataIncrement(current_time, [this_market_info_1, this_market_info_2])
battery.update(increment, market_increment)
current_time = current_time + increment
if __name__ == "__main__":
main()
src/batterymodel/models/__main__.py
-32
View File
@@ -1,32 +0,0 @@
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)
for item in market_1_start:
print(item)
battery.update(increment, None, market1)
if __name__ == "__main__":
main()
src/batterymodel/models/battery.py
+49 -17
View File
@@ -1,7 +1,7 @@
import datetime
from enum import Enum
from batterymodel.models.market import MarketDataIncrement
from ..models.market import MarketDataIncrement
class BatteryState(Enum):
@@ -10,7 +10,7 @@ class BatteryState(Enum):
DISCHARGING = 2
class BatteryModel:
def __init__(self):
def __init__(self, start_time: datetime.datetime):
self._max_charge_rate = 2000000 # Watts
self._max_discharge_rate = 2000000 # Watts
self._max_storage_volume = 4000000 # Wh
@@ -22,29 +22,61 @@ class BatteryModel:
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._decision_log_file = "battery.log"
self._decision_log_handle = open(self._decision_log_file, "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_market_rate = ()
self._current_charge_value = 0.0
self._current_charge = 0.0
self._total_profit = 0.0
def update(self, increment: datetime.timedelta, previous_state: "BatteryModel", market_data_increment: MarketDataIncrement):
self.market_decision(increment, previous_state, market_data_increment)
self.start_time = start_time
self.current_time = start_time
def update(self, increment: datetime.timedelta, market_data_increment: MarketDataIncrement):
self.current_time += increment
self.market_decision(increment, market_data_increment)
if not self._battery_state == BatteryState.IDLE:
self._runtime += increment
pass
def market_decision(self, increment: datetime.timedelta, previous_state: "BatteryModel", market_data_increment: MarketDataIncrement):
def market_decision(self, increment: datetime.timedelta, market_data_increment: MarketDataIncrement):
self.log_state_change("market_decision", None, None)
if self._battery_state == BatteryState.IDLE and self._current_charge > 0:
# Sell, I guess, to whoever offers more
for market_data_increment in market_data_increment.markets:
self._battery_state = BatteryState.DISCHARGING
self._current_charge_rate = self._max_discharge_rate
self._current_market_rate = market_data_increment.get_max_market_rate()
self._current_charge_value = self._current_charge_rate * self._current_charge
return
self._battery_state = BatteryState.DISCHARGING
self._current_charge_rate = self._max_discharge_rate
self._current_market_rate = market_data_increment.get_max_market_rate()
self._current_charge_value = self._current_charge_rate * self._current_charge
self.log_state_change("market_decision_battery", BatteryState.IDLE, self._battery_state)
elif self._battery_state == BatteryState.DISCHARGING and self._current_market_rate[0] < self.current_time:
# We've committed to discharging, so do figure out what we've earned this period, do the discharge calculation
# and continue
discharge_time = (increment.total_seconds() / 60.0)
self._total_profit = self._current_market_rate[1] * discharge_time
elif self._battery_state == BatteryState.DISCHARGING and self._current_market_rate[0] > self.current_time:
# We've finished discharging for this time period, calculate profit assuming we only discharged for this increment
# and remember to chop any time we wouldn't have been charging if our increments have overlapped.
discharge_time = (increment.total_seconds()/60.0) - ((self.current_time-self._current_market_rate[0])/60.0)
# Could we actually discharge for this time
remaining_capacity = self._current_charge - (self._max_discharge_rate * discharge_time)
if remaining_capacity > 0:
self._total_profit += (self._current_market_rate[1] * discharge_time)
else:
discharge_time = remaining_capacity / self._max_discharge_rate
self._total_profit += (self._current_market_rate[1] * discharge_time)
self._battery_state = BatteryState.IDLE
self._current_charge = self._max_discharge_rate * discharge_time
elif self._battery_state == BatteryState.IDLE and self._current_charge == 0.0:
# No charge at all, normally I'd assume some kind of settable boundary here so we don't buy power too high
# but I'll just assume we buy to make life easy
self._battery_state = BatteryState.CHARGING
selected_market = market_data_increment.get_min_market_rate()
self._current_market_rate = selected_market
def handle_charge_or_discharge(self, increment: datetime.timedelta, previous_state: "BatteryModel"):
@@ -53,9 +85,9 @@ class BatteryModel:
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
if self._current_charge > self._max_storage_volume:
self._current_charge = self._max_storage_volume
total_charge_this_increment = self._max_storage_volume - previous_state._current_charge
self._battery_state = BatteryState.IDLE
time_of_charge = datetime.timedelta(seconds=total_charge_this_increment / self._current_charge_rate)
@@ -78,7 +110,7 @@ class BatteryModel:
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")
self._decision_log_handle.write(f"{self.current_time},{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:
src/batterymodel/models/market.py
+28 -7
View File
@@ -8,29 +8,41 @@ from xmlrpc.client import DateTime
class MarketState:
def __init__(self):
self._market_data = OrderedDict()
self._last_key = None
def load_market_data_from_csv(self, csv_file):
with open(csv_file, 'r') as file:
reader = csv.reader(file)
next(reader, None) # skip the headers
for row in reader:
self._market_data[datetime.datetime.strptime(row[0], '%d-%m-%Y %H:%M')] = float(row[1])
self._market_data[datetime.datetime.strptime(row[0], '%d/%m/%Y %H:%M')] = float(row[1])
self._last_key = datetime.datetime.strptime(row[0], '%d/%m/%Y %H:%M')
return self
def get_start_time(self):
return next(iter(self._market_data))
return next(iter(self._market_data.items()))[0]
def get_end_time(self):
end = None
for i in iter(self._market_data.items()):
end = i[0]
return end
def get_market_rate_for_timestamp(self, timestamp: datetime.datetime):
last_timestamp = None
if timestamp >= self._last_key:
return 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]
return timestamp, self._market_data[key]
elif last_timestamp is not None and last_timestamp < key and key_delta > delta:
return self._market_data[last_timestamp]
return last_timestamp, self._market_data[last_timestamp]
else:
last_timestamp = key
last_timestamp = key
return None
@@ -46,4 +58,13 @@ class MarketDataIncrement:
if i[1] > rate:
rate = i[1]
highest = i
return highest
return highest
def get_min_market_rate(self):
rate = 99999999999999
lowest = ()
for i in self.markets:
if i[1] < rate:
rate = i[1]
lowest = i
return lowest
tests/test_battery.py
+24
View File
@@ -1,5 +1,8 @@
import datetime
import copy
from batterymodel import BatteryModel
from batterymodel.models.battery import BatteryState
from batterymodel.models.market import MarketDataIncrement
def test_battery_stats():
@@ -20,3 +23,24 @@ def test_battery_update():
def test_market_decision_idle():
battery = BatteryModel()
battery._current_charge=100000
time_increment = datetime.timedelta(minutes=30)
market_increment = MarketDataIncrement(
datetime.datetime.strptime('01-01-2022 00:00', '%d-%m-%Y %H:%M'),
[
(datetime.datetime.strptime('01-01-2022 00:00', '%d-%m-%Y %H:%M'),43.0),
(datetime.datetime.strptime('01-01-2022 00:00', '%d-%m-%Y %H:%M'),14.0)
]
)
market_increment_current = MarketDataIncrement(
datetime.datetime.strptime('01-01-2022 00:30', '%d-%m-%Y %H:%M'),
[
(datetime.datetime.strptime('01-01-2022 00:00', '%d-%m-%Y %H:%M'),43.0),
(datetime.datetime.strptime('01-01-2022 00:30', '%d-%m-%Y %H:%M'),13.0)
]
)
current_battery = copy.deepcopy(battery)
battery.market_decision(time_increment, current_battery, market_increment_current)
assert current_battery._battery_state == BatteryState.IDLE
assert battery._battery_state == BatteryState.DISCHARGING
tests/test_market.py
+2 -2
View File
@@ -10,7 +10,7 @@ def test_market():
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
assert this_market.get_market_rate_for_timestamp(datetime.datetime.strptime('01-01-2022 00:02', '%d-%m-%Y %H:%M')) == (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')) == (datetime.datetime.strptime('01-01-2022 00:02', '%d-%m-%Y %H:%M'),13)