Support multiple Octopus Intelligent Go vehicles

apps/predbat/config.py

@@ -2029,6 +2029,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_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"},
     "carbon_postcode": {"type": "string", "empty": False},
     "carbon_automatic": {"type": "boolean"},
     "axle_api_key": {"type": "string", "empty": False},

apps/predbat/fetch.py

@@ -616,41 +616,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")
@@ -663,45 +672,57 @@ 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

apps/predbat/ha.py

@@ -580,6 +580,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:

apps/predbat/tests/test_multi_car_iog.py

@@ -0,0 +1,159 @@
+# -----------------------------------------------------------------------------
+# 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

apps/predbat/unit_test.py

@@ -21,6 +21,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_fetch_config_options import test_fetch_config_options
 from tests.test_multi_inverter import run_inverter_multi_tests
 from tests.test_window2minutes import test_window2minutes
@@ -185,6 +186,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),
         ("rate_replicate", test_rate_replicate, "Rate replicate comprehensive tests (missing slots, IO, offsets, gas)", False),
         ("find_charge_rate", test_find_charge_rate, "Find charge rate tests", False),