diff --git a/.gitignore b/.gitignore index 8801fa2d2..bbb832fcb 100644 --- a/.gitignore +++ b/.gitignore @@ -46,6 +46,7 @@ coverage/predbat_standalone/ coverage/predbat_standalone_ha/ coverage/supabase/ coverage/octopus/ +coverage/venv/ .coverage .pytest_cache/ .tox/ diff --git a/apps/predbat/config.py b/apps/predbat/config.py index 77a85cdbc..de8bdf5fc 100644 --- a/apps/predbat/config.py +++ b/apps/predbat/config.py @@ -2006,9 +2006,9 @@ "car_charging_limit": {"type": "sensor", "sensor_type": "float", "entries": "num_cars"}, "car_charging_exclusive": {"type": "boolean_list", "entries": "num_cars"}, "carbon_intensity": {"type": "sensor", "sensor_type": "string"}, - "octopus_intelligent_slot": {"type": "sensor", "sensor_type": "boolean|action"}, - "octopus_ready_time": {"type": "sensor", "sensor_type": "string"}, - "octopus_charge_limit": {"type": "sensor", "sensor_type": "float"}, + "octopus_intelligent_slot": {"type": "sensor|sensor_list", "sensor_type": "boolean|action", "entries": "num_cars"}, + "octopus_ready_time": {"type": "sensor|sensor_list", "sensor_type": "string", "entries": "num_cars"}, + "octopus_charge_limit": {"type": "sensor|sensor_list", "sensor_type": "float", "entries": "num_cars"}, "octopus_slot_low_rate": {"type": "boolean"}, "octopus_slot_max": {"type": "integer"}, "octopus_saving_session_octopoints_per_penny": {"type": "integer"}, diff --git a/apps/predbat/fetch.py b/apps/predbat/fetch.py index 98470eaec..248baf066 100644 --- a/apps/predbat/fetch.py +++ b/apps/predbat/fetch.py @@ -614,41 +614,50 @@ def fetch_sensor_data(self, save=True): # Work out current car SoC and limit self.car_charging_loss = 1 - float(self.get_arg("car_charging_loss")) - entity_id = self.get_arg("octopus_intelligent_slot", indirect=False) - ohme_automatic = self.get_arg("ohme_automatic", False) + # Get octopus intelligent slot configuration - could be single value or list for multiple cars + entity_id_config = self.get_arg("octopus_intelligent_slot", indirect=False) + + # Normalize to list for multi-car support + if entity_id_config and not isinstance(entity_id_config, list): + entity_id_list = [entity_id_config] + elif entity_id_config: + entity_id_list = entity_id_config + else: + entity_id_list = [] - if entity_id: - completed = [] - planned = [] - - if entity_id and "octopus_intelligent_slot_action_config" in self.args: - config_entry = self.get_arg("octopus_intelligent_slot_action_config", None, indirect=False) - service_name = entity_id.replace(".", "/") - result = self.call_service_wrapper(service_name, config_entry=config_entry, return_response=True) - if result and ("slots" in result): - planned = result["slots"] - else: - self.log("Warn: Unable to get data from {} - octopus_intelligent_slot using action config {}, result was {}".format(entity_id, config_entry, result)) - else: - try: - completed = self.get_state_wrapper(entity_id=entity_id, attribute="completed_dispatches") or self.get_state_wrapper(entity_id=entity_id, attribute="completedDispatches") - planned = self.get_state_wrapper(entity_id=entity_id, attribute="planned_dispatches") or self.get_state_wrapper(entity_id=entity_id, attribute="plannedDispatches") - except (ValueError, TypeError): - self.log("Warn: Unable to get data from {} - octopus_intelligent_slot may not be set correctly in apps.yaml".format(entity_id)) - self.record_status(message="Error: octopus_intelligent_slot not set correctly in apps.yaml", had_errors=True) + if entity_id_list: + # Process each car's intelligent slot configuration + for car_n in range(min(len(entity_id_list), self.num_cars)): + entity_id = entity_id_list[car_n] + if not entity_id: + continue - # Completed and planned slots - if completed: - self.octopus_slots += completed - if planned and (not self.octopus_intelligent_ignore_unplugged or self.car_charging_planned[0]): - # We only count planned slots if the car is plugged in or we are ignoring unplugged cars - self.octopus_slots += planned + completed = [] + planned = [] - # Get rate for import to compute charging costs - if self.rate_import: - self.rate_scan(self.rate_import, print=False) + if entity_id and "octopus_intelligent_slot_action_config" in self.args: + config_entry = self.get_arg("octopus_intelligent_slot_action_config", None, indirect=False) + service_name = entity_id.replace(".", "/") + result = self.call_service_wrapper(service_name, config_entry=config_entry, return_response=True) + if result and ("slots" in result): + planned = result["slots"] + else: + self.log("Warn: Unable to get data from {} for car {} - octopus_intelligent_slot using action config {}, result was {}".format(entity_id, car_n, config_entry, result)) + else: + try: + completed = self.get_state_wrapper(entity_id=entity_id, attribute="completed_dispatches") or self.get_state_wrapper(entity_id=entity_id, attribute="completedDispatches") + planned = self.get_state_wrapper(entity_id=entity_id, attribute="planned_dispatches") or self.get_state_wrapper(entity_id=entity_id, attribute="plannedDispatches") + except (ValueError, TypeError): + self.log("Warn: Unable to get data from {} for car {} - octopus_intelligent_slot may not be set correctly in apps.yaml".format(entity_id, car_n)) + self.record_status(message="Error: octopus_intelligent_slot not set correctly in apps.yaml for car {}".format(car_n), had_errors=True) + + # Completed and planned slots - merge from all cars + if completed: + self.octopus_slots += completed + if planned and (not self.octopus_intelligent_ignore_unplugged or self.car_charging_planned[car_n]): + # We only count planned slots if the car is plugged in or we are ignoring unplugged cars + self.octopus_slots += planned - if self.num_cars >= 1: # Extract vehicle data if we can get it size = self.get_state_wrapper(entity_id=entity_id, attribute="vehicle_battery_size_in_kwh") rate = self.get_state_wrapper(entity_id=entity_id, attribute="charge_point_power_in_kw") @@ -661,45 +670,53 @@ def fetch_sensor_data(self, save=True): except (ValueError, TypeError): rate = None if size: - self.car_charging_battery_size[0] = size + self.car_charging_battery_size[car_n] = size if rate: # Take the max as Octopus over reports - self.car_charging_rate[0] = max(rate, self.car_charging_rate[0]) + self.car_charging_rate[car_n] = max(rate, self.car_charging_rate[car_n]) # Get car charging limit again from car based on new battery size - self.car_charging_limit[0] = dp3((float(self.get_arg("car_charging_limit", 100.0, index=0)) * self.car_charging_battery_size[0]) / 100.0) + self.car_charging_limit[car_n] = dp3((float(self.get_arg("car_charging_limit", 100.0, index=car_n)) * self.car_charging_battery_size[car_n]) / 100.0) # Extract vehicle preference if we can get it if self.octopus_intelligent_charging: - octopus_ready_time = self.get_arg("octopus_ready_time", None) + octopus_ready_time = self.get_arg("octopus_ready_time", None, index=car_n) if isinstance(octopus_ready_time, str) and len(octopus_ready_time) == 5: octopus_ready_time += ":00" - octopus_limit = self.get_arg("octopus_charge_limit", None) + octopus_limit = self.get_arg("octopus_charge_limit", None, index=car_n) if octopus_limit: try: octopus_limit = float(octopus_limit) except (ValueError, TypeError): - self.log("Warn: octopus_charge_limit is set to a bad value {} in apps.yaml, must be a number".format(octopus_limit)) + self.log("Warn: octopus_charge_limit is set to a bad value {} for car {} in apps.yaml, must be a number".format(octopus_limit, car_n)) octopus_limit = None if octopus_limit: - octopus_limit = dp3(float(octopus_limit) * self.car_charging_battery_size[0] / 100.0) - self.car_charging_limit[0] = min(self.car_charging_limit[0], octopus_limit) + octopus_limit = dp3(float(octopus_limit) * self.car_charging_battery_size[car_n] / 100.0) + self.car_charging_limit[car_n] = min(self.car_charging_limit[car_n], octopus_limit) if octopus_ready_time: - self.car_charging_plan_time[0] = octopus_ready_time - - # Use octopus slots for charging? - self.octopus_slots = self.add_now_to_octopus_slot(self.octopus_slots, self.now_utc) - if not self.octopus_intelligent_ignore_unplugged or self.car_charging_planned[0]: - self.car_charging_slots[0] = self.load_octopus_slots(self.octopus_slots, self.octopus_intelligent_consider_full) - if self.car_charging_slots[0]: - self.log("Car 0 using Octopus Intelligent, charging planned - charging limit {}, ready time {} - battery size {}".format(self.car_charging_limit[0], self.car_charging_plan_time[0], self.car_charging_battery_size[0])) - self.car_charging_planned[0] = True + self.car_charging_plan_time[car_n] = octopus_ready_time + + # Get rate for import to compute charging costs + if self.rate_import: + self.rate_scan(self.rate_import, print=False) + + # Use octopus slots for charging - process for each car + if self.octopus_intelligent_charging: + self.octopus_slots = self.add_now_to_octopus_slot(self.octopus_slots, self.now_utc) + for car_n in range(min(len(entity_id_list), self.num_cars)): + if not entity_id_list[car_n]: + continue + if not self.octopus_intelligent_ignore_unplugged or self.car_charging_planned[car_n]: + self.car_charging_slots[car_n] = self.load_octopus_slots(self.octopus_slots, self.octopus_intelligent_consider_full) + if self.car_charging_slots[car_n]: + self.log("Car {} using Octopus Intelligent, charging planned - charging limit {}, ready time {} - battery size {}".format(car_n, self.car_charging_limit[car_n], self.car_charging_plan_time[car_n], self.car_charging_battery_size[car_n])) + self.car_charging_planned[car_n] = True else: - self.log("Car 0 using Octopus Intelligent, no charging is planned") - self.car_charging_planned[0] = False + self.log("Car {} using Octopus Intelligent, no charging is planned".format(car_n)) + self.car_charging_planned[car_n] = False else: - self.log("Car 0 using Octopus Intelligent is unplugged") - self.car_charging_planned[0] = False + self.log("Car {} using Octopus Intelligent is unplugged".format(car_n)) + self.car_charging_planned[car_n] = False else: # Disable octopus charging if we don't have the slot sensor self.octopus_intelligent_charging = False diff --git a/apps/predbat/ha.py b/apps/predbat/ha.py index 7ef312a58..d6203568b 100644 --- a/apps/predbat/ha.py +++ b/apps/predbat/ha.py @@ -480,6 +480,9 @@ def get_state(self, entity_id=None, default=None, attribute=None, refresh=False, Get state from cached HA data """ if entity_id: + if isinstance(entity_id, list): + self.log("Error: get_state called with list entity_id: {}, this should be a single entity string".format(entity_id)) + return default self.db_mirror_list[entity_id.lower()] = True if not entity_id: diff --git a/apps/predbat/tests/test_multi_car_iog.py b/apps/predbat/tests/test_multi_car_iog.py new file mode 100644 index 000000000..85b4e0014 --- /dev/null +++ b/apps/predbat/tests/test_multi_car_iog.py @@ -0,0 +1,194 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System +# Copyright Trefor Southwell 2024 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +# fmt off +# pylint: disable=consider-using-f-string +# pylint: disable=line-too-long +# pylint: disable=attribute-defined-outside-init + +from datetime import timedelta + + +def process_octopus_intelligent_slots(my_predbat): + """ + Helper function to simulate the octopus intelligent slot processing from fetch_sensor_data + """ + entity_id_config = my_predbat.get_arg("octopus_intelligent_slot", indirect=False) + + # Normalize to list + if entity_id_config and not isinstance(entity_id_config, list): + entity_id_list = [entity_id_config] + elif entity_id_config: + entity_id_list = entity_id_config + else: + entity_id_list = [] + + # Process each car + for car_n in range(min(len(entity_id_list), my_predbat.num_cars)): + entity_id = entity_id_list[car_n] + if not entity_id: + continue + + completed = my_predbat.get_state_wrapper(entity_id=entity_id, attribute="completed_dispatches") or [] + planned = my_predbat.get_state_wrapper(entity_id=entity_id, attribute="planned_dispatches") or [] + + if completed: + my_predbat.octopus_slots += completed + if planned: + my_predbat.octopus_slots += planned + + +def run_multi_car_iog_test(testname, my_predbat): + """ + Test multi-car Intelligent Octopus Go (IOG) support + """ + failed = False + print("**** Running Test: multi_car_iog {} ****".format(testname)) + + # Setup test data - similar to what fetch_sensor_data does + my_predbat.num_cars = 2 + my_predbat.car_charging_planned = [True, True] # Both cars plugged in + my_predbat.car_charging_now = [False, False] + my_predbat.car_charging_plan_smart = [False, False] + my_predbat.car_charging_plan_max_price = [0, 0] + my_predbat.car_charging_plan_time = ["07:00:00", "07:00:00"] + my_predbat.car_charging_battery_size = [100.0, 80.0] + my_predbat.car_charging_limit = [100.0, 80.0] + my_predbat.car_charging_rate = [7.4, 7.4] + my_predbat.car_charging_slots = [[], []] + my_predbat.car_charging_exclusive = [False, False] + my_predbat.car_charging_loss = 1.0 + my_predbat.octopus_intelligent_charging = True + my_predbat.octopus_intelligent_ignore_unplugged = False + my_predbat.octopus_intelligent_consider_full = False + my_predbat.octopus_slots = [] + + # Test 1: Single car config (backward compatibility) + print("Test 1: Single car config (backward compatibility)") + my_predbat.args["octopus_intelligent_slot"] = "binary_sensor.octopus_energy_intelligent_dispatching" + + # Mock entity state + slot1_start = (my_predbat.now_utc + timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M:%S%z") + slot1_end = (my_predbat.now_utc + timedelta(hours=2)).strftime("%Y-%m-%dT%H:%M:%S%z") + + my_predbat.ha_interface.set_state( + "binary_sensor.octopus_energy_intelligent_dispatching", + "on", + attributes={ + "completed_dispatches": [], + "planned_dispatches": [ + { + "start": slot1_start, + "end": slot1_end, + "charge_in_kwh": 10.0, + "source": "smart-charge", + "location": "AT_HOME" + } + ], + "vehicle_battery_size_in_kwh": 100.0, + "charge_point_power_in_kw": 7.4 + } + ) + + # Simulate the octopus intelligent slot processing from fetch_sensor_data + my_predbat.octopus_slots = [] + process_octopus_intelligent_slots(my_predbat) + + if len(my_predbat.octopus_slots) != 1: + print("ERROR: Expected 1 slot for single car, got {}".format(len(my_predbat.octopus_slots))) + print("Slots: {}".format(my_predbat.octopus_slots)) + failed = True + + # Test 2: Multi-car config + print("Test 2: Multi-car config with two cars") + my_predbat.octopus_slots = [] + my_predbat.args["octopus_intelligent_slot"] = [ + "binary_sensor.octopus_energy_intelligent_dispatching_car1", + "binary_sensor.octopus_energy_intelligent_dispatching_car2" + ] + + # Mock entity states for both cars + slot2_start = (my_predbat.now_utc + timedelta(hours=2)).strftime("%Y-%m-%dT%H:%M:%S%z") + slot2_end = (my_predbat.now_utc + timedelta(hours=3)).strftime("%Y-%m-%dT%H:%M:%S%z") + + my_predbat.ha_interface.set_state( + "binary_sensor.octopus_energy_intelligent_dispatching_car1", + "on", + attributes={ + "completed_dispatches": [], + "planned_dispatches": [ + { + "start": slot1_start, + "end": slot1_end, + "charge_in_kwh": 10.0, + "source": "smart-charge", + "location": "AT_HOME" + } + ], + "vehicle_battery_size_in_kwh": 100.0, + "charge_point_power_in_kw": 7.4 + } + ) + + my_predbat.ha_interface.set_state( + "binary_sensor.octopus_energy_intelligent_dispatching_car2", + "on", + attributes={ + "completed_dispatches": [], + "planned_dispatches": [ + { + "start": slot2_start, + "end": slot2_end, + "charge_in_kwh": 8.0, + "source": "smart-charge", + "location": "AT_HOME" + } + ], + "vehicle_battery_size_in_kwh": 80.0, + "charge_point_power_in_kw": 7.4 + } + ) + + # Simulate the octopus intelligent slot processing + process_octopus_intelligent_slots(my_predbat) + + # Should have slots from both cars merged + if len(my_predbat.octopus_slots) != 2: + print("ERROR: Expected 2 slots (one from each car), got {}".format(len(my_predbat.octopus_slots))) + print("Slots: {}".format(my_predbat.octopus_slots)) + failed = True + + # Test 3: Multi-car config with empty slot + print("Test 3: Multi-car config with one empty/None slot") + my_predbat.octopus_slots = [] + my_predbat.args["octopus_intelligent_slot"] = [ + "binary_sensor.octopus_energy_intelligent_dispatching_car1", + None + ] + + # Simulate the octopus intelligent slot processing + process_octopus_intelligent_slots(my_predbat) + + # Should have slots from first car only + if len(my_predbat.octopus_slots) != 1: + print("ERROR: Expected 1 slot (from first car only), got {}".format(len(my_predbat.octopus_slots))) + print("Slots: {}".format(my_predbat.octopus_slots)) + failed = True + + if failed: + print("Test: {} FAILED".format(testname)) + else: + print("Test: {} PASSED".format(testname)) + + return failed + + +def run_multi_car_iog_tests(my_predbat): + """ + Run all multi-car IOG tests + """ + failed = False + failed |= run_multi_car_iog_test("multi_car_iog_basic", my_predbat) + return failed diff --git a/apps/predbat/unit_test.py b/apps/predbat/unit_test.py index 94f46f258..c17266808 100644 --- a/apps/predbat/unit_test.py +++ b/apps/predbat/unit_test.py @@ -22,6 +22,7 @@ from tests.test_model import run_model_tests from tests.test_execute import run_execute_tests from tests.test_octopus_slots import run_load_octopus_slots_tests +from tests.test_multi_car_iog import run_multi_car_iog_tests from tests.test_multi_inverter import run_inverter_multi_tests from tests.test_window2minutes import test_window2minutes from tests.test_history_attribute import test_history_attribute @@ -179,6 +180,7 @@ def main(): ("web_if", run_test_web_if, "Web interface tests", False), ("nordpool", run_nordpool_test, "Nordpool tests", False), ("octopus_slots", run_load_octopus_slots_tests, "Load Octopus slots tests", False), + ("multi_car_iog", run_multi_car_iog_tests, "Multi-car IOG tests", False), ("rate_add_io_slots", run_rate_add_io_slots_tests, "Rate add IO slots tests", False), ("find_charge_rate", test_find_charge_rate, "Find charge rate tests", False), ("energydataservice", test_energydataservice, "Energy data service tests", False),