From 9407d4dedf95ad407ec03de4efaf08d5c7d60a53 Mon Sep 17 00:00:00 2001 From: CI Date: Sat, 26 Sep 2026 15:02:00 +0100 Subject: [PATCH 1/2] fix(solis): apply slot events immediately so Predbat's read-back sees them Predbat writes a slot time to the Solis Cloud select and polls it for about 20s. The select callback only queued the event (#4875), and the queue is drained at the start of run(), which runs once a minute. So the select kept its old value, Predbat logged "Trying to write ... didn't complete", and the value then went through on the next cycle. - Slot times, SoCs, powers and enables of a discovered inverter now run their handler on the calling loop. Those handlers only change the in-memory schedule and make no API calls, so #4875's loop-affinity concern does not apply. Everything that calls the cloud is still queued, as are slot events that arrive before the inverter is discovered. - The slot handlers publish only the slot controls, through a new publish_slot_entities() moved unchanged out of publish_entities(), instead of republishing every entity from the calling thread. - read_and_write_cid() checks the read-back before reporting a failure: a write whose reply errored but whose value landed is a success. - In V2 pass 2, a slot enable or time the inverter refuses is restored to the value the inverter holds, so the select shows it and Predbat writes again on its next execute. It is only restored while the schedule still holds the attempted value, so a newer request that arrives during the write is not overwritten. Pass 1, V1, and slot SoC/current keep their existing retry-every-minute behaviour. Co-Authored-By: Claude Opus 5.5 (1M context) --- apps/predbat/solis.py | 599 ++++++++++++++++++------------- apps/predbat/tests/test_solis.py | 208 ++++++++++- 2 files changed, 547 insertions(+), 260 deletions(-) diff --git a/apps/predbat/solis.py b/apps/predbat/solis.py index b9c2c34d9..48670f26f 100644 --- a/apps/predbat/solis.py +++ b/apps/predbat/solis.py @@ -1031,6 +1031,33 @@ def get_rated_current(self, inverter_sn): return None return float(int(round(power / self.get_nominal_voltage(inverter_sn), 6))) + def restore_refused_slot_write(self, inverter_sn, slot, direction, kind, cid, attempted): + """Put a slot enable or time the inverter refused back to the value it still holds. + + The slot controls then show the inverter's value, so Predbat sees its write did not take and + writes it again on its next execute - rather than this component re-sending, every minute, a + value the plan may since have moved on from. cached_values holds what the inverter reports, + refreshed by read_and_write_cid()'s read-back. + + Only restored while the schedule still holds the value that was attempted: slot events are + applied on the caller's loop, so a newer request can arrive during the write and must win. + """ + inverter_value = str(self.cached_values.get(inverter_sn, {}).get(cid, "")) + slot_data = self.charge_discharge_time_windows.get(inverter_sn, {}).get(slot) + if slot_data is None: + return + if kind == "enable": + if inverter_value not in ("0", "1") or not slot_data.get(f"{direction}_enable"): + return + restored = {f"{direction}_enable": int(inverter_value)} + else: + inverter_times = inverter_value.split("-") + if len(inverter_times) != 2 or f"{slot_data.get(f'{direction}_start_time')}-{slot_data.get(f'{direction}_end_time')}" != attempted: + return + restored = {f"{direction}_start_time": inverter_times[0], f"{direction}_end_time": inverter_times[1]} + slot_data.update(restored) + self.log(f"Warn: Solis API: {direction} slot {slot} {kind} was not accepted by {inverter_sn}, showing the inverter's {inverter_value} so Predbat writes it again") + async def write_time_windows_if_changed(self, inverter_sn): """Write charge/discharge time windows, SOC, and current to inverter, only if values changed from cache. Automatically handles V1 vs V2 modes and only writes registers that have changed. @@ -1142,6 +1169,8 @@ async def write_time_windows_if_changed(self, inverter_sn): if cached_enable != "1": result = await self.read_and_write_cid(inverter_sn, enable_cid, "1", field_description=f"charge slot {slot} enable") success &= result + if not result: + self.restore_refused_slot_write(inverter_sn, slot, "charge", "enable", enable_cid, "1") # Check and write charge time if changed if "charge_start_time" in slot_data and "charge_end_time" in slot_data: @@ -1151,6 +1180,8 @@ async def write_time_windows_if_changed(self, inverter_sn): if cached_time != new_time_str: result = await self.read_and_write_cid(inverter_sn, time_cid, new_time_str, field_description=f"charge slot {slot} time") success &= result + if not result: + self.restore_refused_slot_write(inverter_sn, slot, "charge", "time", time_cid, new_time_str) # Check and write charge SOC if changed if "charge_soc" in slot_data: @@ -1177,6 +1208,8 @@ async def write_time_windows_if_changed(self, inverter_sn): if cached_enable != "1": result = await self.read_and_write_cid(inverter_sn, enable_cid, "1", field_description=f"discharge slot {slot} enable") success &= result + if not result: + self.restore_refused_slot_write(inverter_sn, slot, "discharge", "enable", enable_cid, "1") # Check and write discharge time if changed if "discharge_start_time" in slot_data and "discharge_end_time" in slot_data: @@ -1186,6 +1219,8 @@ async def write_time_windows_if_changed(self, inverter_sn): if cached_time != new_time_str: result = await self.read_and_write_cid(inverter_sn, time_cid, new_time_str, field_description=f"discharge slot {slot} time") success &= result + if not result: + self.restore_refused_slot_write(inverter_sn, slot, "discharge", "time", time_cid, new_time_str) # Check and write discharge SOC if changed if "discharge_soc" in slot_data: @@ -1485,14 +1520,19 @@ async def read_and_write_cid(self, inverter_sn, cid, value, field_description=No # Validate what we wrote so the cache is correct. old_value, info = await self.read_cid(inverter_sn, cid) + if cid_value_matches(old_value, value): + # The write reply can time out or error after the datalogger has already applied the value, + # so what the inverter reports decides it + if not result: + self.log(f"Solis API: CID {cid} {field_description} on {inverter_sn} is set to {value}, although the write reported an error") + else: + self.log(f"Solis API: CID {cid} {field_description} on {inverter_sn} is set to {value}") + return True + if not result: self.log(f"Warn: Solis API: Failed to write CID {cid} {field_description} on {inverter_sn}") return False - if cid_value_matches(old_value, value): - self.log(f"Solis API: CID {cid} {field_description} on {inverter_sn} is set to {value}") - return True - # A read taken this soon after the write can still be reporting the value from before # it, so read once more after a pause rather than concluding the write was refused # (issue #4774). The settled read also leaves the cache holding the better answer. @@ -2233,6 +2273,255 @@ async def fetch_entity_data(self, sn): except (ValueError, TypeError): pass + def publish_slot_entities(self, inverter_sn): + """Publish one inverter's charge/discharge slot controls from the in-memory schedule. + + Shared by publish_entities() and the slot event handlers. Those handlers run on the caller's + loop, so they publish just the slots they changed rather than every entity. + """ + prefix = self.prefix + detail = self.inverter_details.get(inverter_sn, {}) + inverter_name = detail.get("inverterName", inverter_sn) + inverter_sn_lower = inverter_sn.lower() + + # Get decoded time windows if available (works with both old and new methods) + time_windows = self.charge_discharge_time_windows.get(inverter_sn, {}) + + # Publish charge slot controls (only for slots present in decoded data) + for slot_num in range(1, 7): + # Get slot data from decoded windows - skip if not available + slot_data = time_windows.get(slot_num, None) + if slot_data is None: + continue + + # Enable switch + if "charge_enable" in slot_data: + try: + charge_enable = int(slot_data["charge_enable"]) + except (ValueError, TypeError): + charge_enable = 0 + entity_id = f"switch.{prefix}_solis_{inverter_sn_lower}_charge_slot{slot_num}_enable" + self.dashboard_item( + entity_id, + state="on" if charge_enable else "off", + attributes={ + "friendly_name": f"Solis {inverter_name} Charge Slot {slot_num} Enable", + "icon": "mdi:battery-charging", + }, + app="solis" + ) + + # Start time selector + if "charge_start_time" in slot_data: + entity_id = f"select.{prefix}_solis_{inverter_sn_lower}_charge_slot{slot_num}_start_time" + start_time = slot_data["charge_start_time"] + # Convert HH:MM to HH:MM:00 format + if start_time and ":" in start_time and len(start_time.split(":")) == 2: + time_value = start_time + ":00" + else: + time_value = start_time or "00:00:00" + self.dashboard_item( + entity_id, + state=time_value, + attributes={ + "friendly_name": f"Solis {inverter_name} Charge Slot {slot_num} Start Time", + "options": OPTIONS_TIME, + "icon": "mdi:clock-start", + }, + app="solis" + ) + + # End time selector + if "charge_end_time" in slot_data: + entity_id = f"select.{prefix}_solis_{inverter_sn_lower}_charge_slot{slot_num}_end_time" + end_time = slot_data["charge_end_time"] + # Convert HH:MM to HH:MM:00 format + if end_time and ":" in end_time and len(end_time.split(":")) == 2: + end_time_value = end_time + ":00" + else: + end_time_value = end_time or "00:00:00" + self.dashboard_item( + entity_id, + state=end_time_value, + attributes={ + "friendly_name": f"Solis {inverter_name} Charge Slot {slot_num} End Time", + "options": OPTIONS_TIME, + "icon": "mdi:clock-end", + }, + app="solis" + ) + + # SOC target number + if "charge_soc" in slot_data: + entity_id = f"number.{prefix}_solis_{inverter_sn_lower}_charge_slot{slot_num}_soc" + try: + soc_value = int(slot_data["charge_soc"]) + except (ValueError, TypeError): + soc_value = 0 + self.dashboard_item( + entity_id, + state=soc_value, + attributes={ + "friendly_name": f"Solis {inverter_name} Charge Slot {slot_num} SOC", + "unit_of_measurement": "%", + "min": 0, + "max": 100, + "step": 1, + "icon": "mdi:battery", + }, + app="solis" + ) + + # Current limit number (displayed as power in watts) + if "charge_current" in slot_data: + entity_id = f"number.{prefix}_solis_{inverter_sn_lower}_charge_slot{slot_num}_power" + current_value_amps = slot_data["charge_current"] + + # Convert amps to watts for display + current_value_watts = None + if current_value_amps is not None: + try: + current_value_watts = int(float(current_value_amps) * self.get_nominal_voltage(inverter_sn)) + except (ValueError, TypeError): + self.log("Warn: Failed to convert charge current to watts for {} slot {}: {}".format(inverter_sn, slot_num, current_value_amps)) # Debug log + + # Use pre-calculated max current (convert to watts) + max_current_amps = self.max_charge_current.get(inverter_sn, 100) + max_power_watts = int(max_current_amps * self.get_nominal_voltage(inverter_sn)) + + self.dashboard_item( + entity_id, + state=current_value_watts, + attributes={ + "friendly_name": f"Solis {inverter_name} Charge Slot {slot_num} Power", + "unit_of_measurement": "W", + "min": 0, + "max": max_power_watts, + "step": self.get_nominal_voltage(inverter_sn), + "device_class": "power", + "icon": "mdi:flash", + }, + app="solis" + ) + + # Publish discharge slot controls (only for slots present in decoded data) + for slot_num in range(1, 7): + # Get slot data from decoded windows - skip if not available + slot_data = time_windows.get(slot_num, None) + if slot_data is None: + continue + + # Enable switch + if "discharge_enable" in slot_data: + entity_id = f"switch.{prefix}_solis_{inverter_sn_lower}_discharge_slot{slot_num}_enable" + try: + discharge_enable = int(slot_data["discharge_enable"]) + except (ValueError, TypeError): + discharge_enable = 0 + self.dashboard_item( + entity_id, + state="on" if discharge_enable else "off", + attributes={ + "friendly_name": f"Solis {inverter_name} Discharge Slot {slot_num} Enable", + "icon": "mdi:battery-minus", + }, + app="solis" + ) + + # Start time selector + if "discharge_start_time" in slot_data: + entity_id = f"select.{prefix}_solis_{inverter_sn_lower}_discharge_slot{slot_num}_start_time" + start_time = slot_data["discharge_start_time"] + # Convert HH:MM to HH:MM:00 format + if start_time and ":" in start_time and len(start_time.split(":")) == 2: + time_value = start_time + ":00" + else: + time_value = start_time or "00:00:00" + self.dashboard_item( + entity_id, + state=time_value, + attributes={ + "friendly_name": f"Solis {inverter_name} Discharge Slot {slot_num} Start Time", + "options": OPTIONS_TIME, + "icon": "mdi:clock-start", + }, + app="solis" + ) + + # End time selector + if "discharge_end_time" in slot_data: + entity_id = f"select.{prefix}_solis_{inverter_sn_lower}_discharge_slot{slot_num}_end_time" + end_time = slot_data["discharge_end_time"] + # Convert HH:MM to HH:MM:00 format + if end_time and ":" in end_time and len(end_time.split(":")) == 2: + end_time_value = end_time + ":00" + else: + end_time_value = end_time or "00:00:00" + self.dashboard_item( + entity_id, + state=end_time_value, + attributes={ + "friendly_name": f"Solis {inverter_name} Discharge Slot {slot_num} End Time", + "options": OPTIONS_TIME, + "icon": "mdi:clock-end", + }, + app="solis" + ) + + # SOC target number + if "discharge_soc" in slot_data: + entity_id = f"number.{prefix}_solis_{inverter_sn_lower}_discharge_slot{slot_num}_soc" + try: + soc_value = int(slot_data["discharge_soc"]) + except (ValueError, TypeError): + soc_value = 0 + self.dashboard_item( + entity_id, + state=soc_value, + attributes={ + "friendly_name": f"Solis {inverter_name} Discharge Slot {slot_num} SOC", + "unit_of_measurement": "%", + "min": 0, + "max": 100, + "step": 1, + "icon": "mdi:battery", + }, + app="solis" + ) + + # Current limit number (displayed as power in watts) + if "discharge_current" in slot_data: + entity_id = f"number.{prefix}_solis_{inverter_sn_lower}_discharge_slot{slot_num}_power" + current_value_amps = slot_data["discharge_current"] + + # Convert amps to watts for display + current_value_watts = None + if current_value_amps is not None: + try: + current_value_watts = int(float(current_value_amps) * self.get_nominal_voltage(inverter_sn)) + except (ValueError, TypeError): + self.log("Warn: Failed to convert discharge current to watts for {} slot {}: {}".format(inverter_sn, slot_num, current_value_amps)) # Debug log + + # Use pre-calculated max current (convert to watts) + max_current_amps = self.max_discharge_current.get(inverter_sn, 100) + max_power_watts = int(max_current_amps * self.get_nominal_voltage(inverter_sn)) + + self.dashboard_item( + entity_id, + state=current_value_watts, + attributes={ + "friendly_name": f"Solis {inverter_name} Discharge Slot {slot_num} Power", + "unit_of_measurement": "W", + "min": 0, + "max": max_power_watts, + "step": self.get_nominal_voltage(inverter_sn), + "device_class": "power", + "icon": "mdi:flash", + }, + app="solis" + ) + + async def publish_entities(self): """Publish all entities to Home Assistant""" prefix = self.prefix @@ -2603,242 +2892,7 @@ async def publish_entities(self): app="solis" ) - # Get decoded time windows if available (works with both old and new methods) - time_windows = self.charge_discharge_time_windows.get(inverter_sn, {}) - - # Publish charge slot controls (only for slots present in decoded data) - for slot_num in range(1, 7): - # Get slot data from decoded windows - skip if not available - slot_data = time_windows.get(slot_num, None) - if slot_data is None: - continue - - # Enable switch - if "charge_enable" in slot_data: - try: - charge_enable = int(slot_data["charge_enable"]) - except (ValueError, TypeError): - charge_enable = 0 - entity_id = f"switch.{prefix}_solis_{inverter_sn_lower}_charge_slot{slot_num}_enable" - self.dashboard_item( - entity_id, - state="on" if charge_enable else "off", - attributes={ - "friendly_name": f"Solis {inverter_name} Charge Slot {slot_num} Enable", - "icon": "mdi:battery-charging", - }, - app="solis" - ) - - # Start time selector - if "charge_start_time" in slot_data: - entity_id = f"select.{prefix}_solis_{inverter_sn_lower}_charge_slot{slot_num}_start_time" - start_time = slot_data["charge_start_time"] - # Convert HH:MM to HH:MM:00 format - if start_time and ":" in start_time and len(start_time.split(":")) == 2: - time_value = start_time + ":00" - else: - time_value = start_time or "00:00:00" - self.dashboard_item( - entity_id, - state=time_value, - attributes={ - "friendly_name": f"Solis {inverter_name} Charge Slot {slot_num} Start Time", - "options": OPTIONS_TIME, - "icon": "mdi:clock-start", - }, - app="solis" - ) - - # End time selector - if "charge_end_time" in slot_data: - entity_id = f"select.{prefix}_solis_{inverter_sn_lower}_charge_slot{slot_num}_end_time" - end_time = slot_data["charge_end_time"] - # Convert HH:MM to HH:MM:00 format - if end_time and ":" in end_time and len(end_time.split(":")) == 2: - end_time_value = end_time + ":00" - else: - end_time_value = end_time or "00:00:00" - self.dashboard_item( - entity_id, - state=end_time_value, - attributes={ - "friendly_name": f"Solis {inverter_name} Charge Slot {slot_num} End Time", - "options": OPTIONS_TIME, - "icon": "mdi:clock-end", - }, - app="solis" - ) - - # SOC target number - if "charge_soc" in slot_data: - entity_id = f"number.{prefix}_solis_{inverter_sn_lower}_charge_slot{slot_num}_soc" - try: - soc_value = int(slot_data["charge_soc"]) - except (ValueError, TypeError): - soc_value = 0 - self.dashboard_item( - entity_id, - state=soc_value, - attributes={ - "friendly_name": f"Solis {inverter_name} Charge Slot {slot_num} SOC", - "unit_of_measurement": "%", - "min": 0, - "max": 100, - "step": 1, - "icon": "mdi:battery", - }, - app="solis" - ) - - # Current limit number (displayed as power in watts) - if "charge_current" in slot_data: - entity_id = f"number.{prefix}_solis_{inverter_sn_lower}_charge_slot{slot_num}_power" - current_value_amps = slot_data["charge_current"] - - # Convert amps to watts for display - current_value_watts = None - if current_value_amps is not None: - try: - current_value_watts = int(float(current_value_amps) * self.get_nominal_voltage(inverter_sn)) - except (ValueError, TypeError): - self.log("Warn: Failed to convert charge current to watts for {} slot {}: {}".format(inverter_sn, slot_num, current_value_amps)) # Debug log - - # Use pre-calculated max current (convert to watts) - max_current_amps = self.max_charge_current.get(inverter_sn, 100) - max_power_watts = int(max_current_amps * self.get_nominal_voltage(inverter_sn)) - - self.dashboard_item( - entity_id, - state=current_value_watts, - attributes={ - "friendly_name": f"Solis {inverter_name} Charge Slot {slot_num} Power", - "unit_of_measurement": "W", - "min": 0, - "max": max_power_watts, - "step": self.get_nominal_voltage(inverter_sn), - "device_class": "power", - "icon": "mdi:flash", - }, - app="solis" - ) - - # Publish discharge slot controls (only for slots present in decoded data) - for slot_num in range(1, 7): - # Get slot data from decoded windows - skip if not available - slot_data = time_windows.get(slot_num, None) - if slot_data is None: - continue - - # Enable switch - if "discharge_enable" in slot_data: - entity_id = f"switch.{prefix}_solis_{inverter_sn_lower}_discharge_slot{slot_num}_enable" - try: - discharge_enable = int(slot_data["discharge_enable"]) - except (ValueError, TypeError): - discharge_enable = 0 - self.dashboard_item( - entity_id, - state="on" if discharge_enable else "off", - attributes={ - "friendly_name": f"Solis {inverter_name} Discharge Slot {slot_num} Enable", - "icon": "mdi:battery-minus", - }, - app="solis" - ) - - # Start time selector - if "discharge_start_time" in slot_data: - entity_id = f"select.{prefix}_solis_{inverter_sn_lower}_discharge_slot{slot_num}_start_time" - start_time = slot_data["discharge_start_time"] - # Convert HH:MM to HH:MM:00 format - if start_time and ":" in start_time and len(start_time.split(":")) == 2: - time_value = start_time + ":00" - else: - time_value = start_time or "00:00:00" - self.dashboard_item( - entity_id, - state=time_value, - attributes={ - "friendly_name": f"Solis {inverter_name} Discharge Slot {slot_num} Start Time", - "options": OPTIONS_TIME, - "icon": "mdi:clock-start", - }, - app="solis" - ) - - # End time selector - if "discharge_end_time" in slot_data: - entity_id = f"select.{prefix}_solis_{inverter_sn_lower}_discharge_slot{slot_num}_end_time" - end_time = slot_data["discharge_end_time"] - # Convert HH:MM to HH:MM:00 format - if end_time and ":" in end_time and len(end_time.split(":")) == 2: - end_time_value = end_time + ":00" - else: - end_time_value = end_time or "00:00:00" - self.dashboard_item( - entity_id, - state=end_time_value, - attributes={ - "friendly_name": f"Solis {inverter_name} Discharge Slot {slot_num} End Time", - "options": OPTIONS_TIME, - "icon": "mdi:clock-end", - }, - app="solis" - ) - - # SOC target number - if "discharge_soc" in slot_data: - entity_id = f"number.{prefix}_solis_{inverter_sn_lower}_discharge_slot{slot_num}_soc" - try: - soc_value = int(slot_data["discharge_soc"]) - except (ValueError, TypeError): - soc_value = 0 - self.dashboard_item( - entity_id, - state=soc_value, - attributes={ - "friendly_name": f"Solis {inverter_name} Discharge Slot {slot_num} SOC", - "unit_of_measurement": "%", - "min": 0, - "max": 100, - "step": 1, - "icon": "mdi:battery", - }, - app="solis" - ) - - # Current limit number (displayed as power in watts) - if "discharge_current" in slot_data: - entity_id = f"number.{prefix}_solis_{inverter_sn_lower}_discharge_slot{slot_num}_power" - current_value_amps = slot_data["discharge_current"] - - # Convert amps to watts for display - current_value_watts = None - if current_value_amps is not None: - try: - current_value_watts = int(float(current_value_amps) * self.get_nominal_voltage(inverter_sn)) - except (ValueError, TypeError): - self.log("Warn: Failed to convert discharge current to watts for {} slot {}: {}".format(inverter_sn, slot_num, current_value_amps)) # Debug log - - # Use pre-calculated max current (convert to watts) - max_current_amps = self.max_discharge_current.get(inverter_sn, 100) - max_power_watts = int(max_current_amps * self.get_nominal_voltage(inverter_sn)) - - self.dashboard_item( - entity_id, - state=current_value_watts, - attributes={ - "friendly_name": f"Solis {inverter_name} Discharge Slot {slot_num} Power", - "unit_of_measurement": "W", - "min": 0, - "max": max_power_watts, - "step": self.get_nominal_voltage(inverter_sn), - "device_class": "power", - "icon": "mdi:flash", - }, - app="solis" - ) + self.publish_slot_entities(inverter_sn) # Storage mode selector entity_id = f"select.{prefix}_solis_{inverter_sn_lower}_storage_mode" @@ -3184,14 +3238,49 @@ def _calculate_toggle_value(self, service, current_value): # handler swallows it, and the user is told a write succeeded that never # reached the inverter. Queue instead, and let run() do the work on the loop # that owns the session. Same approach as Ohme and Octopus. + # + # Slot times, SoCs, powers and enables are the exception: their handlers only + # change the in-memory schedule, which run() writes to the inverter later, so + # there is no API work to protect. They are applied here, because Predbat reads + # the entity back for about 20s after writing it and the queue is only drained + # once a minute - a queued slot time was reported as a failed write that then + # went through. + SLOT_SCHEDULE_FIELDS = {"select": ("start_time", "end_time"), "number": ("soc", "power"), "switch": ("enable",)} + + def is_slot_schedule_event(self, entity_id): + """True if entity_id is a slot schedule control of an inverter that has already been discovered.""" + domain, _, name = entity_id.partition(".") + entity_prefix = f"{self.base.get_arg('prefix', 'predbat')}_solis_" + if domain not in self.SLOT_SCHEDULE_FIELDS or not name.startswith(entity_prefix): + return False + inverter_sn, _, field = name[len(entity_prefix):].partition("_") + # Until the first run() has discovered the inverter the handler would drop the event, so it waits in the queue + if self.find_inverter_by_sn(inverter_sn) is None: + return False + direction, _, rest = field.partition("_slot") + slot, _, suffix = rest.partition("_") + return direction in ("charge", "discharge") and slot.isdigit() and suffix in self.SLOT_SCHEDULE_FIELDS[domain] + async def select_event(self, entity_id, value): - self.queued_events.append((self.select_event_handler, entity_id, value)) + """Apply a slot time now, queue anything else for the Solis loop.""" + if self.is_slot_schedule_event(entity_id): + await self.select_event_handler(entity_id, value) + else: + self.queued_events.append((self.select_event_handler, entity_id, value)) async def number_event(self, entity_id, value): - self.queued_events.append((self.number_event_handler, entity_id, value)) + """Apply a slot SoC or power now, queue anything else for the Solis loop.""" + if self.is_slot_schedule_event(entity_id): + await self.number_event_handler(entity_id, value) + else: + self.queued_events.append((self.number_event_handler, entity_id, value)) async def switch_event(self, entity_id, service): - self.queued_events.append((self.switch_event_handler, entity_id, service)) + """Apply a slot enable now, queue anything else for the Solis loop.""" + if self.is_slot_schedule_event(entity_id): + await self.switch_event_handler(entity_id, service) + else: + self.queued_events.append((self.switch_event_handler, entity_id, service)) async def select_event_handler(self, entity_id, value): """Handle select entity changes""" @@ -3257,8 +3346,7 @@ async def select_event_handler(self, entity_id, value): # Write will happen in the main loop - # Re-publish entities - await self.publish_entities() + self.publish_slot_entities(inverter_sn) return # Handle discharge slot times @@ -3290,8 +3378,7 @@ async def select_event_handler(self, entity_id, value): # Write will happen in the main loop - # Re-publish entities - await self.publish_entities() + self.publish_slot_entities(inverter_sn) return except Exception as e: @@ -3350,8 +3437,7 @@ async def number_event_handler(self, entity_id, value): # Write will happen in the main loop - # Re-publish entities - await self.publish_entities() + self.publish_slot_entities(inverter_sn) return # Handle charge slot power (user provides watts, convert to amps) @@ -3380,8 +3466,7 @@ async def number_event_handler(self, entity_id, value): # Write will happen in the main loop - # Re-publish entities - await self.publish_entities() + self.publish_slot_entities(inverter_sn) return # Handle discharge slot SOC @@ -3407,8 +3492,7 @@ async def number_event_handler(self, entity_id, value): # Write will happen in the main loop - # Re-publish entities - await self.publish_entities() + self.publish_slot_entities(inverter_sn) return # Handle discharge slot power (user provides watts, convert to amps) @@ -3437,8 +3521,7 @@ async def number_event_handler(self, entity_id, value): # Write will happen in the main loop - # Re-publish entities - await self.publish_entities() + self.publish_slot_entities(inverter_sn) return # Handle battery SOC limits @@ -3567,8 +3650,7 @@ async def switch_event_handler(self, entity_id, service): # Write will happen in the main loop - # Re-publish entities - await self.publish_entities() + self.publish_slot_entities(inverter_sn) return # Handle discharge slot enables @@ -3605,8 +3687,7 @@ async def switch_event_handler(self, entity_id, service): # Write will happen in the main loop not here - # Re-publish entities - await self.publish_entities() + self.publish_slot_entities(inverter_sn) return # Handle battery reserve switch diff --git a/apps/predbat/tests/test_solis.py b/apps/predbat/tests/test_solis.py index 906183c69..29357b37e 100644 --- a/apps/predbat/tests/test_solis.py +++ b/apps/predbat/tests/test_solis.py @@ -1941,6 +1941,13 @@ def run_solis_tests(my_predbat): failed |= asyncio.run(test_event_queued_not_executed_on_calling_loop()) failed |= asyncio.run(test_queued_event_failure_does_not_stop_the_queue()) failed |= asyncio.run(test_queued_event_drained_after_startup()) + failed |= asyncio.run(test_slot_select_applied_before_queue_drains()) + failed |= asyncio.run(test_slot_number_and_switch_applied_before_queue_drains()) + failed |= asyncio.run(test_slot_event_before_discovery_still_queued()) + failed |= asyncio.run(test_read_and_write_cid_trusts_read_back_after_write_error()) + failed |= asyncio.run(test_failed_slot_time_write_restores_inverter_value()) + failed |= asyncio.run(test_failed_slot_enable_write_restores_inverter_value()) + failed |= asyncio.run(test_failed_slot_write_keeps_newer_request()) failed |= asyncio.run(test_select_event_storage_mode()) failed |= asyncio.run(test_select_event_charge_time()) failed |= asyncio.run(test_select_event_discharge_time()) @@ -6049,8 +6056,10 @@ async def test_rate_setpoints_do_not_drift_with_battery_voltage(): max_discharge_entity = f"number.{prefix}_solis_{sn.lower()}_max_discharge_power" slot_entity = f"number.{prefix}_solis_{sn.lower()}_discharge_slot1_power" - # Predbat writes a discharge rate in watts; the handler converts it to amps and republishes + # Predbat writes a discharge rate in watts; the handler converts it to amps and republishes the + # slots, and run() republishes everything else await api.number_event_handler(slot_entity, 3437) + await api.publish_entities() written_amps = api.charge_discharge_time_windows[sn][1]["discharge_current"] before = {entity: api.dashboard_items[entity]["state"] for entity in (max_charge_entity, max_discharge_entity, slot_entity)} @@ -6428,6 +6437,203 @@ async def spy(entity_id, value): return 0 +def _slot_event_api(): + """A discovered V2 inverter whose slot 1 is idle, as Predbat finds it before programming an export.""" + api = MockSolisAPI() + api._test_v2_mode = True + api._mock_storage_mode = True + sn = "1031730261290196" + api.inverter_sn = [sn] + api.charge_discharge_time_windows[sn] = { + 1: { + "charge_enable": 0, + "charge_start_time": "00:00", + "charge_end_time": "00:00", + "charge_soc": 100.0, + "charge_current": 75.0, + "discharge_enable": 0, + "discharge_start_time": "00:00", + "discharge_end_time": "00:00", + "discharge_soc": 10.0, + "discharge_current": 75.0, + } + } + return api, sn + + +async def test_slot_select_applied_before_queue_drains(): + """A slot time Predbat writes shows on its select straight away, not at the next run(). + + Predbat writes the select and polls it for about 20s; the queue is only drained once a + minute, so a queued slot time left the select on the old value and Predbat logged a write + that then went through as failed. The slot handlers only touch the in-memory schedule - + there is no API work to keep off the calling loop (issue #4875). + """ + api, sn = _slot_event_api() + entity_id = f"select.predbat_solis_{sn}_discharge_slot1_start_time" + + await api.select_event(entity_id, "11:30:00") + + if api.queued_events: + print("ERROR: slot time should be applied, not queued: {}".format(api.queued_events)) + return 1 + if api.charge_discharge_time_windows[sn][1]["discharge_start_time"] != "11:30": + print("ERROR: schedule not updated, got {}".format(api.charge_discharge_time_windows[sn][1]["discharge_start_time"])) + return 1 + state = api.dashboard_items.get(entity_id, {}).get("state") + if state != "11:30:00": + print("ERROR: select should read back 11:30:00 for Predbat's poll, got {}".format(state)) + return 1 + return 0 + + +async def test_slot_number_and_switch_applied_before_queue_drains(): + """Slot SoC numbers and slot enable switches are schedule-only too, so they apply straight away.""" + api, sn = _slot_event_api() + soc_entity = f"number.predbat_solis_{sn}_discharge_slot1_soc" + enable_entity = f"switch.predbat_solis_{sn}_discharge_slot1_enable" + + await api.number_event(soc_entity, 15) + await api.switch_event(enable_entity, "turn_on") + + if api.queued_events: + print("ERROR: slot number/switch should be applied, not queued: {}".format(api.queued_events)) + return 1 + slot = api.charge_discharge_time_windows[sn][1] + if slot["discharge_soc"] != 15.0 or slot["discharge_enable"] != 1: + print("ERROR: schedule not updated, got soc {} enable {}".format(slot["discharge_soc"], slot["discharge_enable"])) + return 1 + if api.dashboard_items.get(soc_entity, {}).get("state") != 15.0: + print("ERROR: SoC number should read back 15, got {}".format(api.dashboard_items.get(soc_entity))) + return 1 + if api.dashboard_items.get(enable_entity, {}).get("state") != "on": + print("ERROR: enable switch should read back on, got {}".format(api.dashboard_items.get(enable_entity))) + return 1 + return 0 + + +async def test_slot_event_before_discovery_still_queued(): + """Before the first run() has discovered the inverter a slot event cannot be applied, so it waits in the queue.""" + api, sn = _slot_event_api() + api.inverter_sn = [] + + await api.select_event(f"select.predbat_solis_{sn}_discharge_slot1_start_time", "11:30:00") + + if len(api.queued_events) != 1: + print("ERROR: expected the event to be queued until discovery, got {}".format(api.queued_events)) + return 1 + return 0 + + +async def test_read_and_write_cid_trusts_read_back_after_write_error(): + """A write whose reply errored but whose value is on the inverter is a success, not a failure. + + SolisCloud relays a control write to the datalogger, and the reply can time out or error + after the value has landed. read_and_write_cid already reads the value back; it must use + that read before reporting failure. + """ + api = MockSolisAPI() + api.read_and_write_cid = SolisAPI.read_and_write_cid.__get__(api, MockSolisAPI) + cid_state = {5964: "00:00-00:00"} + + async def mock_read_cid(inv_sn, cid): + return cid_state[cid], {"msg": cid_state[cid]} + + async def mock_write_cid_landed_but_errored(inv_sn, cid, value, old_value=None, field_description=None): + cid_state[cid] = value + return False + + api.read_cid = mock_read_cid + api.write_cid = mock_write_cid_landed_but_errored + + result = await api.read_and_write_cid("SN1", 5964, "11:30-12:00", field_description="discharge slot 1 time") + + if result is not True: + print("ERROR: a value confirmed by the read-back should report success, got {}".format(result)) + return 1 + if any("Failed to write" in msg for msg in api.log_messages): + print("ERROR: should not log a write failure for a value that landed: {}".format(api.log_messages)) + return 1 + return 0 + + +def _fail_slot_cids(api, failing_cids, on_attempt=None): + """Make read_and_write_cid refuse the given CIDs, leaving the inverter (cached_values) unchanged.""" + original = api.read_and_write_cid + + async def read_and_write_cid(inverter_sn, cid, value, field_description=None): + if cid in failing_cids: + api.read_and_write_cid_calls.append({"inverter_sn": inverter_sn, "cid": cid, "value": str(value), "field_description": field_description}) + if on_attempt: + on_attempt(cid) + return False + return await original(inverter_sn, cid, value, field_description=field_description) + + api.read_and_write_cid = read_and_write_cid + + +async def test_failed_slot_time_write_restores_inverter_value(): + """A slot time the inverter refused goes back to what the inverter holds, so Predbat sees it and writes again.""" + from solis import SOLIS_CID_DISCHARGE_TIME + + api, sn = _slot_event_api() + api.cached_values[sn] = {SOLIS_CID_DISCHARGE_TIME[0]: "00:00-00:00", SOLIS_CID_DISCHARGE_ENABLE_BASE: "0"} + slot = api.charge_discharge_time_windows[sn][1] + slot.update({"discharge_enable": 1, "discharge_start_time": "11:30", "discharge_end_time": "12:00"}) + _fail_slot_cids(api, {SOLIS_CID_DISCHARGE_TIME[0]}) + + result = await api.write_time_windows_if_changed(sn) + + if result is not False: + print("ERROR: a refused write should still report failure, got {}".format(result)) + return 1 + if (slot["discharge_start_time"], slot["discharge_end_time"]) != ("00:00", "00:00"): + print("ERROR: refused time should be restored to the inverter's 00:00-00:00, got {}-{}".format(slot["discharge_start_time"], slot["discharge_end_time"])) + return 1 + if slot["discharge_enable"] != 1: + print("ERROR: the enable write succeeded and should be left alone, got {}".format(slot["discharge_enable"])) + return 1 + return 0 + + +async def test_failed_slot_enable_write_restores_inverter_value(): + """A slot enable the inverter refused goes back to what the inverter holds.""" + api, sn = _slot_event_api() + api.cached_values[sn] = {SOLIS_CID_DISCHARGE_ENABLE_BASE: "0"} + slot = api.charge_discharge_time_windows[sn][1] + slot.update({"discharge_enable": 1, "discharge_start_time": "11:30", "discharge_end_time": "12:00"}) + _fail_slot_cids(api, {SOLIS_CID_DISCHARGE_ENABLE_BASE}) + + await api.write_time_windows_if_changed(sn) + + if slot["discharge_enable"] != 0: + print("ERROR: refused enable should be restored to the inverter's 0, got {}".format(slot["discharge_enable"])) + return 1 + return 0 + + +async def test_failed_slot_write_keeps_newer_request(): + """A request that arrives while the old value is being written is not overwritten when that write fails.""" + from solis import SOLIS_CID_DISCHARGE_TIME + + api, sn = _slot_event_api() + api.cached_values[sn] = {SOLIS_CID_DISCHARGE_TIME[0]: "00:00-00:00", SOLIS_CID_DISCHARGE_ENABLE_BASE: "1"} + slot = api.charge_discharge_time_windows[sn][1] + slot.update({"discharge_enable": 1, "discharge_start_time": "11:30", "discharge_end_time": "12:00"}) + + def newer_request_arrives(cid): + slot["discharge_start_time"] = "13:00" + + _fail_slot_cids(api, {SOLIS_CID_DISCHARGE_TIME[0]}, on_attempt=newer_request_arrives) + + await api.write_time_windows_if_changed(sn) + + if (slot["discharge_start_time"], slot["discharge_end_time"]) != ("13:00", "12:00"): + print("ERROR: the newer 13:00 request should survive the failed write, got {}-{}".format(slot["discharge_start_time"], slot["discharge_end_time"])) + return 1 + return 0 + + # ==================== Daily API allowance (issue #5087) ==================== From c84f71199a5e4ab96a25ab735df163690d6125ee Mon Sep 17 00:00:00 2001 From: CI Date: Sat, 26 Sep 2026 16:28:11 +0100 Subject: [PATCH 2/2] fix(solis): address #5247 review - lock the slot schedule, settle errored writes - Slot events now change the schedule through set_slot_schedule_field(), under a threading.Lock that write_time_windows_if_changed() also holds while it deep-copies the schedule and that restore_refused_slot_write() holds for its check-and-update. Slot events arrive on another loop, so the Solis loop is no longer the only writer. None of the locked sections await. - Slot events are applied immediately only once the inverter's schedule has been decoded (schedules_loaded). The first run() discovers the inverter before it decodes, and the decode replaces the schedule wholesale, so an event applied in between was lost; until then it is queued. - read_and_write_cid() takes the settled re-read on the write-error path too. A value that lands a moment after a timed-out reply is a success, instead of being rolled back and then overwritten with the old value. - One route_event() helper replaces the three copies of the routing branch, and routing uses self.prefix instead of get_arg('prefix') on every event. - restore_refused_slot_write() documents why the whole start-end pair is restored, and the remaining window: a start event queued before the first decode. - The rollback tests now run the real read_and_write_cid() and read_cid() against a fake inverter with a stale cache. Removing read_cid()'s cache refresh makes them fail. Co-Authored-By: Claude Opus 5.5 (1M context) --- apps/predbat/solis.py | 160 +++++++++++++--------------- apps/predbat/tests/test_solis.py | 173 +++++++++++++++++++++++++++---- 2 files changed, 225 insertions(+), 108 deletions(-) diff --git a/apps/predbat/solis.py b/apps/predbat/solis.py index 48670f26f..e9c266ddb 100644 --- a/apps/predbat/solis.py +++ b/apps/predbat/solis.py @@ -14,6 +14,7 @@ import json import time import copy +import threading from datetime import datetime, timedelta, UTC from predbat_metrics import record_api_call from component_base import ComponentBase @@ -440,6 +441,9 @@ def initialize(self, api_key=None, api_secret=None, inverter_sn=None, automatic= self.automatic = automatic self.session = None self.queued_events = [] + # Slot events change the schedule from the caller's loop while run() copies and restores it on this one + self.schedule_lock = threading.Lock() + self.schedules_loaded = set() # Inverters whose schedule has been decoded, so slot events can apply to it # Last-resort fallback, used only for an LV pack that reports no BMS charge voltage to be # classified by - matches the previous hard-coded assumption (issue #4493). # get_nominal_voltage() below is the real source of truth for the full priority order. @@ -932,6 +936,7 @@ async def decode_time_windows_v2(self, inverter_sn): "field_length": 0, # Indicate v2 format } self.charge_discharge_time_windows[inverter_sn] = result + self.schedules_loaded.add(inverter_sn) self.log("Solis API: Decoded time windows v2 for {}: {}".format(inverter_sn, result)) # Debug log return result @@ -1041,21 +1046,32 @@ def restore_refused_slot_write(self, inverter_sn, slot, direction, kind, cid, at Only restored while the schedule still holds the value that was attempted: slot events are applied on the caller's loop, so a newer request can arrive during the write and must win. + The check and the update are made under schedule_lock so no event lands between them. + + A time is restored as a whole start-end pair, because that is how the inverter holds it. + Predbat programs a slot as two select writes, start then end, back to back in + adjust_force_export(); with slot events applied immediately the end lands well under a + second after the start. The restore only happens once a refused cloud write returns, some + 15-20s after run() copied the schedule, so by then the schedule holds the whole new window + and the check above leaves it alone. The one way the end can arrive after a restore is for + the start event to have been queued - only before the first decode - in which case the + slot shows the inverter's start with Predbat's end until Predbat's next execute rewrites both. """ inverter_value = str(self.cached_values.get(inverter_sn, {}).get(cid, "")) - slot_data = self.charge_discharge_time_windows.get(inverter_sn, {}).get(slot) - if slot_data is None: - return - if kind == "enable": - if inverter_value not in ("0", "1") or not slot_data.get(f"{direction}_enable"): - return - restored = {f"{direction}_enable": int(inverter_value)} - else: - inverter_times = inverter_value.split("-") - if len(inverter_times) != 2 or f"{slot_data.get(f'{direction}_start_time')}-{slot_data.get(f'{direction}_end_time')}" != attempted: + with self.schedule_lock: + slot_data = self.charge_discharge_time_windows.get(inverter_sn, {}).get(slot) + if slot_data is None: return - restored = {f"{direction}_start_time": inverter_times[0], f"{direction}_end_time": inverter_times[1]} - slot_data.update(restored) + if kind == "enable": + if inverter_value not in ("0", "1") or not slot_data.get(f"{direction}_enable"): + return + restored = {f"{direction}_enable": int(inverter_value)} + else: + inverter_times = inverter_value.split("-") + if len(inverter_times) != 2 or f"{slot_data.get(f'{direction}_start_time')}-{slot_data.get(f'{direction}_end_time')}" != attempted: + return + restored = {f"{direction}_start_time": inverter_times[0], f"{direction}_end_time": inverter_times[1]} + slot_data.update(restored) self.log(f"Warn: Solis API: {direction} slot {slot} {kind} was not accepted by {inverter_sn}, showing the inverter's {inverter_value} so Predbat writes it again") async def write_time_windows_if_changed(self, inverter_sn): @@ -1074,7 +1090,8 @@ async def write_time_windows_if_changed(self, inverter_sn): return True #  Make a copy as we will locally modify the data before writing it - time_windows = copy.deepcopy(self.charge_discharge_time_windows[inverter_sn]) + with self.schedule_lock: + time_windows = copy.deepcopy(self.charge_discharge_time_windows[inverter_sn]) if self.is_tou_v2_mode(inverter_sn): # V2 mode: check and write individual registers for changed values only @@ -1497,6 +1514,7 @@ async def decode_time_windows(self, inverter_sn): self.log(f"Solis API: Decoded time windows for {inverter_sn}: {result}") self.charge_discharge_time_windows[inverter_sn] = result + self.schedules_loaded.add(inverter_sn) return result async def read_and_write_cid(self, inverter_sn, cid, value, field_description=None): @@ -1529,19 +1547,21 @@ async def read_and_write_cid(self, inverter_sn, cid, value, field_description=No self.log(f"Solis API: CID {cid} {field_description} on {inverter_sn} is set to {value}") return True - if not result: - self.log(f"Warn: Solis API: Failed to write CID {cid} {field_description} on {inverter_sn}") - return False - # A read taken this soon after the write can still be reporting the value from before # it, so read once more after a pause rather than concluding the write was refused - # (issue #4774). The settled read also leaves the cache holding the better answer. + # (issue #4774). The settled read also leaves the cache holding the better answer. It + # applies when the reply errored too: the value can land a moment after a reply that + # timed out, and a failure here rolls the slot schedule back to the old value. await asyncio.sleep(self.verify_settle_seconds) settled_value, info = await self.read_cid(inverter_sn, cid) if cid_value_matches(settled_value, value): self.log(f"Solis API: CID {cid} {field_description} on {inverter_sn} is set to {value}, read back {old_value} until it settled {self.verify_settle_seconds}s later") return True + if not result: + self.log(f"Warn: Solis API: Failed to write CID {cid} {field_description} on {inverter_sn}") + return False + self.log(f"Warn: Solis API: Failed to verify CID {cid} {field_description} on {inverter_sn}, wrote {value} but read back {old_value} and still {settled_value} after {self.verify_settle_seconds}s") return False @@ -3244,43 +3264,49 @@ def _calculate_toggle_value(self, service, current_value): # there is no API work to protect. They are applied here, because Predbat reads # the entity back for about 20s after writing it and the queue is only drained # once a minute - a queued slot time was reported as a failed write that then - # went through. + # went through. The schedule is shared with run() on the Solis loop, so every + # change to it goes through set_slot_schedule_field() under schedule_lock. SLOT_SCHEDULE_FIELDS = {"select": ("start_time", "end_time"), "number": ("soc", "power"), "switch": ("enable",)} def is_slot_schedule_event(self, entity_id): - """True if entity_id is a slot schedule control of an inverter that has already been discovered.""" + """True if entity_id is a slot schedule control of an inverter whose schedule has been decoded.""" domain, _, name = entity_id.partition(".") - entity_prefix = f"{self.base.get_arg('prefix', 'predbat')}_solis_" + entity_prefix = f"{self.prefix}_solis_" if domain not in self.SLOT_SCHEDULE_FIELDS or not name.startswith(entity_prefix): return False inverter_sn, _, field = name[len(entity_prefix):].partition("_") - # Until the first run() has discovered the inverter the handler would drop the event, so it waits in the queue - if self.find_inverter_by_sn(inverter_sn) is None: + # The first run() discovers the inverter before it decodes the schedule, and the decode replaces the schedule + # wholesale - an event applied in between would be lost, so until then it waits in the queue + inverter_sn = self.find_inverter_by_sn(inverter_sn) + if inverter_sn is None or inverter_sn not in self.schedules_loaded: return False direction, _, rest = field.partition("_slot") slot, _, suffix = rest.partition("_") return direction in ("charge", "discharge") and slot.isdigit() and suffix in self.SLOT_SCHEDULE_FIELDS[domain] - async def select_event(self, entity_id, value): - """Apply a slot time now, queue anything else for the Solis loop.""" + async def route_event(self, handler, entity_id, value): + """Apply a slot schedule event now, queue anything else for the Solis loop.""" if self.is_slot_schedule_event(entity_id): - await self.select_event_handler(entity_id, value) + await handler(entity_id, value) else: - self.queued_events.append((self.select_event_handler, entity_id, value)) + self.queued_events.append((handler, entity_id, value)) + + async def select_event(self, entity_id, value): + """Handle a select change from Home Assistant.""" + await self.route_event(self.select_event_handler, entity_id, value) async def number_event(self, entity_id, value): - """Apply a slot SoC or power now, queue anything else for the Solis loop.""" - if self.is_slot_schedule_event(entity_id): - await self.number_event_handler(entity_id, value) - else: - self.queued_events.append((self.number_event_handler, entity_id, value)) + """Handle a number change from Home Assistant.""" + await self.route_event(self.number_event_handler, entity_id, value) async def switch_event(self, entity_id, service): - """Apply a slot enable now, queue anything else for the Solis loop.""" - if self.is_slot_schedule_event(entity_id): - await self.switch_event_handler(entity_id, service) - else: - self.queued_events.append((self.switch_event_handler, entity_id, service)) + """Handle a switch service call from Home Assistant.""" + await self.route_event(self.switch_event_handler, entity_id, service) + + def set_slot_schedule_field(self, inverter_sn, slot_num, field, value): + """Set one field of a slot in the in-memory schedule, under the lock run() copies and restores it under.""" + with self.schedule_lock: + self.charge_discharge_time_windows.setdefault(inverter_sn, {}).setdefault(slot_num, {})[field] = value async def select_event_handler(self, entity_id, value): """Handle select entity changes""" @@ -3334,15 +3360,10 @@ async def select_event_handler(self, entity_id, value): new_hhmm = value[:5] if len(value) >= 5 else value # Update charge_discharge_time_windows cache - if inverter_sn not in self.charge_discharge_time_windows: - self.charge_discharge_time_windows[inverter_sn] = {} - if slot_num not in self.charge_discharge_time_windows[inverter_sn]: - self.charge_discharge_time_windows[inverter_sn][slot_num] = {} - if "start_time" in field: - self.charge_discharge_time_windows[inverter_sn][slot_num]["charge_start_time"] = new_hhmm + self.set_slot_schedule_field(inverter_sn, slot_num, "charge_start_time", new_hhmm) elif "end_time" in field: - self.charge_discharge_time_windows[inverter_sn][slot_num]["charge_end_time"] = new_hhmm + self.set_slot_schedule_field(inverter_sn, slot_num, "charge_end_time", new_hhmm) # Write will happen in the main loop @@ -3366,15 +3387,10 @@ async def select_event_handler(self, entity_id, value): new_hhmm = value[:5] if len(value) >= 5 else value # Update charge_discharge_time_windows cache - if inverter_sn not in self.charge_discharge_time_windows: - self.charge_discharge_time_windows[inverter_sn] = {} - if slot_num not in self.charge_discharge_time_windows[inverter_sn]: - self.charge_discharge_time_windows[inverter_sn][slot_num] = {} - if "start_time" in field: - self.charge_discharge_time_windows[inverter_sn][slot_num]["discharge_start_time"] = new_hhmm + self.set_slot_schedule_field(inverter_sn, slot_num, "discharge_start_time", new_hhmm) elif "end_time" in field: - self.charge_discharge_time_windows[inverter_sn][slot_num]["discharge_end_time"] = new_hhmm + self.set_slot_schedule_field(inverter_sn, slot_num, "discharge_end_time", new_hhmm) # Write will happen in the main loop @@ -3428,12 +3444,7 @@ async def number_event_handler(self, entity_id, value): return # Update charge_discharge_time_windows cache - if inverter_sn not in self.charge_discharge_time_windows: - self.charge_discharge_time_windows[inverter_sn] = {} - if slot_num not in self.charge_discharge_time_windows[inverter_sn]: - self.charge_discharge_time_windows[inverter_sn][slot_num] = {} - - self.charge_discharge_time_windows[inverter_sn][slot_num]["charge_soc"] = float(value_str) + self.set_slot_schedule_field(inverter_sn, slot_num, "charge_soc", float(value_str)) # Write will happen in the main loop @@ -3457,12 +3468,7 @@ async def number_event_handler(self, entity_id, value): amps = int(value / self.get_nominal_voltage(inverter_sn)) # Update charge_discharge_time_windows cache - if inverter_sn not in self.charge_discharge_time_windows: - self.charge_discharge_time_windows[inverter_sn] = {} - if slot_num not in self.charge_discharge_time_windows[inverter_sn]: - self.charge_discharge_time_windows[inverter_sn][slot_num] = {} - - self.charge_discharge_time_windows[inverter_sn][slot_num]["charge_current"] = float(amps) + self.set_slot_schedule_field(inverter_sn, slot_num, "charge_current", float(amps)) # Write will happen in the main loop @@ -3483,12 +3489,7 @@ async def number_event_handler(self, entity_id, value): return # Update charge_discharge_time_windows cache - if inverter_sn not in self.charge_discharge_time_windows: - self.charge_discharge_time_windows[inverter_sn] = {} - if slot_num not in self.charge_discharge_time_windows[inverter_sn]: - self.charge_discharge_time_windows[inverter_sn][slot_num] = {} - - self.charge_discharge_time_windows[inverter_sn][slot_num]["discharge_soc"] = float(value_str) + self.set_slot_schedule_field(inverter_sn, slot_num, "discharge_soc", float(value_str)) # Write will happen in the main loop @@ -3512,12 +3513,7 @@ async def number_event_handler(self, entity_id, value): amps = round(float(value) / self.get_nominal_voltage(inverter_sn), 1) # Update charge_discharge_time_windows cache - if inverter_sn not in self.charge_discharge_time_windows: - self.charge_discharge_time_windows[inverter_sn] = {} - if slot_num not in self.charge_discharge_time_windows[inverter_sn]: - self.charge_discharge_time_windows[inverter_sn][slot_num] = {} - - self.charge_discharge_time_windows[inverter_sn][slot_num]["discharge_current"] = amps + self.set_slot_schedule_field(inverter_sn, slot_num, "discharge_current", amps) # Write will happen in the main loop @@ -3641,12 +3637,7 @@ async def switch_event_handler(self, entity_id, service): return # Update charge_discharge_time_windows cache - if inverter_sn not in self.charge_discharge_time_windows: - self.charge_discharge_time_windows[inverter_sn] = {} - if slot_num not in self.charge_discharge_time_windows[inverter_sn]: - self.charge_discharge_time_windows[inverter_sn][slot_num] = {} - - self.charge_discharge_time_windows[inverter_sn][slot_num]["charge_enable"] = int(value) + self.set_slot_schedule_field(inverter_sn, slot_num, "charge_enable", int(value)) # Write will happen in the main loop @@ -3678,12 +3669,7 @@ async def switch_event_handler(self, entity_id, service): return # Update charge_discharge_time_windows cache - if inverter_sn not in self.charge_discharge_time_windows: - self.charge_discharge_time_windows[inverter_sn] = {} - if slot_num not in self.charge_discharge_time_windows[inverter_sn]: - self.charge_discharge_time_windows[inverter_sn][slot_num] = {} - - self.charge_discharge_time_windows[inverter_sn][slot_num]["discharge_enable"] = int(value) + self.set_slot_schedule_field(inverter_sn, slot_num, "discharge_enable", int(value)) # Write will happen in the main loop not here diff --git a/apps/predbat/tests/test_solis.py b/apps/predbat/tests/test_solis.py index 29357b37e..43dd96f7b 100644 --- a/apps/predbat/tests/test_solis.py +++ b/apps/predbat/tests/test_solis.py @@ -9,6 +9,8 @@ # pylint: disable=attribute-defined-outside-init import asyncio +import threading +import time import solis as solis_module from datetime import datetime, timedelta, UTC from unittest.mock import MagicMock, patch @@ -61,6 +63,9 @@ def __init__(self, prefix="predbat"): # Entity callbacks queue here for the component loop to drain; the real # __init__ (skipped above) sets this. self.queued_events = [] + # Guards the slot schedule, which slot events change from the caller's loop; the real __init__ sets both + self.schedule_lock = threading.Lock() + self.schedules_loaded = set() self.nominal_voltage = 48.0 self.live_voltage_last_known = {} self.nominal_voltage_reported = {} @@ -1948,6 +1953,10 @@ def run_solis_tests(my_predbat): failed |= asyncio.run(test_failed_slot_time_write_restores_inverter_value()) failed |= asyncio.run(test_failed_slot_enable_write_restores_inverter_value()) failed |= asyncio.run(test_failed_slot_write_keeps_newer_request()) + failed |= asyncio.run(test_read_and_write_cid_settles_after_write_error()) + failed |= asyncio.run(test_slot_event_before_schedule_decoded_is_queued()) + failed |= asyncio.run(test_slot_event_waits_for_schedule_lock()) + failed |= asyncio.run(test_schedule_copied_under_lock()) failed |= asyncio.run(test_select_event_storage_mode()) failed |= asyncio.run(test_select_event_charge_time()) failed |= asyncio.run(test_select_event_discharge_time()) @@ -6458,6 +6467,7 @@ def _slot_event_api(): "discharge_current": 75.0, } } + api.schedules_loaded.add(sn) return api, sn @@ -6557,30 +6567,49 @@ async def mock_write_cid_landed_but_errored(inv_sn, cid, value, old_value=None, return 0 -def _fail_slot_cids(api, failing_cids, on_attempt=None): - """Make read_and_write_cid refuse the given CIDs, leaving the inverter (cached_values) unchanged.""" - original = api.read_and_write_cid +def _fake_inverter(api, inverter_state, refused_cids=(), on_attempt=None): + """Drive the real read_and_write_cid() and read_cid() against a fake inverter. - async def read_and_write_cid(inverter_sn, cid, value, field_description=None): - if cid in failing_cids: - api.read_and_write_cid_calls.append({"inverter_sn": inverter_sn, "cid": cid, "value": str(value), "field_description": field_description}) - if on_attempt: - on_attempt(cid) + Only the network request and the control write are replaced, so cached_values is refreshed by + read_cid()'s own read-back - the value restore_refused_slot_write() relies on. Writes to + refused_cids are rejected and leave the inverter as it was. + """ + api.read_and_write_cid = SolisAPI.read_and_write_cid.__get__(api, MockSolisAPI) + # read_cid() creates cached_values[sn] and cached_infos[sn] together; tests seed cached_values alone + for sn in api.cached_values: + api.cached_infos.setdefault(sn, {}) + + async def execute_request(endpoint, payload): + assert endpoint == SOLIS_READ_ENDPOINT, endpoint + return {"msg": inverter_state.get(payload["cid"], "")} + + async def write_cid(inverter_sn, cid, value, old_value=None, field_description=None): + api.read_and_write_cid_calls.append({"inverter_sn": inverter_sn, "cid": cid, "value": str(value), "field_description": field_description}) + if on_attempt: + on_attempt(cid) + if cid in refused_cids: return False - return await original(inverter_sn, cid, value, field_description=field_description) + inverter_state[cid] = str(value) + return True - api.read_and_write_cid = read_and_write_cid + api._execute_request = execute_request + api.write_cid = write_cid async def test_failed_slot_time_write_restores_inverter_value(): - """A slot time the inverter refused goes back to what the inverter holds, so Predbat sees it and writes again.""" + """A slot time the inverter refused goes back to what the inverter holds, so Predbat sees it and writes again. + + cached_values starts stale (10:00-10:30): the restore must use the read-back taken after the + refused write (00:00-00:00), not whatever the cache held before it. + """ from solis import SOLIS_CID_DISCHARGE_TIME api, sn = _slot_event_api() - api.cached_values[sn] = {SOLIS_CID_DISCHARGE_TIME[0]: "00:00-00:00", SOLIS_CID_DISCHARGE_ENABLE_BASE: "0"} + time_cid = SOLIS_CID_DISCHARGE_TIME[0] + api.cached_values[sn] = {time_cid: "10:00-10:30", SOLIS_CID_DISCHARGE_ENABLE_BASE: "1"} slot = api.charge_discharge_time_windows[sn][1] slot.update({"discharge_enable": 1, "discharge_start_time": "11:30", "discharge_end_time": "12:00"}) - _fail_slot_cids(api, {SOLIS_CID_DISCHARGE_TIME[0]}) + _fake_inverter(api, {time_cid: "00:00-00:00", SOLIS_CID_DISCHARGE_ENABLE_BASE: "1"}, refused_cids={time_cid}) result = await api.write_time_windows_if_changed(sn) @@ -6588,21 +6617,21 @@ async def test_failed_slot_time_write_restores_inverter_value(): print("ERROR: a refused write should still report failure, got {}".format(result)) return 1 if (slot["discharge_start_time"], slot["discharge_end_time"]) != ("00:00", "00:00"): - print("ERROR: refused time should be restored to the inverter's 00:00-00:00, got {}-{}".format(slot["discharge_start_time"], slot["discharge_end_time"])) + print("ERROR: refused time should be restored to the inverter's 00:00-00:00 from the read-back, got {}-{}".format(slot["discharge_start_time"], slot["discharge_end_time"])) return 1 if slot["discharge_enable"] != 1: - print("ERROR: the enable write succeeded and should be left alone, got {}".format(slot["discharge_enable"])) + print("ERROR: the enable was already on and should be left alone, got {}".format(slot["discharge_enable"])) return 1 return 0 async def test_failed_slot_enable_write_restores_inverter_value(): - """A slot enable the inverter refused goes back to what the inverter holds.""" + """A slot enable the inverter refused goes back to what the inverter holds, taken from the read-back.""" api, sn = _slot_event_api() - api.cached_values[sn] = {SOLIS_CID_DISCHARGE_ENABLE_BASE: "0"} + api.cached_values[sn] = {SOLIS_CID_DISCHARGE_ENABLE_BASE: "stale"} slot = api.charge_discharge_time_windows[sn][1] slot.update({"discharge_enable": 1, "discharge_start_time": "11:30", "discharge_end_time": "12:00"}) - _fail_slot_cids(api, {SOLIS_CID_DISCHARGE_ENABLE_BASE}) + _fake_inverter(api, {SOLIS_CID_DISCHARGE_ENABLE_BASE: "0"}, refused_cids={SOLIS_CID_DISCHARGE_ENABLE_BASE}) await api.write_time_windows_if_changed(sn) @@ -6617,14 +6646,16 @@ async def test_failed_slot_write_keeps_newer_request(): from solis import SOLIS_CID_DISCHARGE_TIME api, sn = _slot_event_api() - api.cached_values[sn] = {SOLIS_CID_DISCHARGE_TIME[0]: "00:00-00:00", SOLIS_CID_DISCHARGE_ENABLE_BASE: "1"} + time_cid = SOLIS_CID_DISCHARGE_TIME[0] + api.cached_values[sn] = {time_cid: "00:00-00:00", SOLIS_CID_DISCHARGE_ENABLE_BASE: "1"} slot = api.charge_discharge_time_windows[sn][1] slot.update({"discharge_enable": 1, "discharge_start_time": "11:30", "discharge_end_time": "12:00"}) def newer_request_arrives(cid): - slot["discharge_start_time"] = "13:00" + if cid == time_cid: + slot["discharge_start_time"] = "13:00" - _fail_slot_cids(api, {SOLIS_CID_DISCHARGE_TIME[0]}, on_attempt=newer_request_arrives) + _fake_inverter(api, {time_cid: "00:00-00:00", SOLIS_CID_DISCHARGE_ENABLE_BASE: "1"}, refused_cids={time_cid}, on_attempt=newer_request_arrives) await api.write_time_windows_if_changed(sn) @@ -6634,6 +6665,106 @@ def newer_request_arrives(cid): return 0 +async def test_read_and_write_cid_settles_after_write_error(): + """A write whose reply errored and whose value lands a moment later is a success, not a failure to roll back. + + The immediate read-back can still show the old value; the same settled re-read the successful + reply path uses decides it. Otherwise the rollback would revert a value that had landed. + """ + api = MockSolisAPI() + api.read_and_write_cid = SolisAPI.read_and_write_cid.__get__(api, MockSolisAPI) + reads = [] + + async def mock_read_cid(inv_sn, cid): + reads.append(cid) + # pre-read and immediate read-back show the old value; the settled read shows the new one + value = "11:30-12:00" if len(reads) >= 3 else "00:00-00:00" + return value, {"msg": value} + + async def mock_write_cid_errored(inv_sn, cid, value, old_value=None, field_description=None): + return False + + api.read_cid = mock_read_cid + api.write_cid = mock_write_cid_errored + + result = await api.read_and_write_cid("SN1", 5964, "11:30-12:00", field_description="discharge slot 1 time") + + if result is not True: + print("ERROR: a value that landed by the settled read should report success, got {}".format(result)) + return 1 + if len(reads) != 3: + print("ERROR: expected pre-read, read-back and settled read, got {} reads".format(len(reads))) + return 1 + return 0 + + +async def test_slot_event_before_schedule_decoded_is_queued(): + """A slot event for a discovered inverter whose schedule has not been decoded yet waits in the queue. + + The first run() discovers the inverter before it decodes the schedule, and the decode replaces + the schedule wholesale - a slot event applied in between would be lost. + """ + api, sn = _slot_event_api() + api.schedules_loaded.clear() + + await api.select_event(f"select.predbat_solis_{sn}_discharge_slot1_start_time", "11:30:00") + + if len(api.queued_events) != 1: + print("ERROR: expected the event to be queued until the schedule is decoded, got {}".format(api.queued_events)) + return 1 + if api.charge_discharge_time_windows[sn][1]["discharge_start_time"] != "00:00": + print("ERROR: schedule should not change before decode, got {}".format(api.charge_discharge_time_windows[sn][1]["discharge_start_time"])) + return 1 + return 0 + + +def _run_in_thread(coroutine_factory): + """Run a coroutine on its own loop in another thread, as the HA event loop calls the component.""" + thread = threading.Thread(target=lambda: asyncio.run(coroutine_factory())) + thread.start() + return thread + + +async def test_slot_event_waits_for_schedule_lock(): + """A slot event changes the schedule only while holding the schedule lock that run() copies it under.""" + api, sn = _slot_event_api() + api.schedule_lock.acquire() + thread = _run_in_thread(lambda: api.select_event(f"select.predbat_solis_{sn}_discharge_slot1_start_time", "11:30:00")) + time.sleep(0.2) + blocked = api.charge_discharge_time_windows[sn][1]["discharge_start_time"] + api.schedule_lock.release() + thread.join(2) + + if blocked != "00:00": + print("ERROR: the slot event changed the schedule without the lock, got {}".format(blocked)) + return 1 + if api.charge_discharge_time_windows[sn][1]["discharge_start_time"] != "11:30": + print("ERROR: the slot event should apply once the lock is free, got {}".format(api.charge_discharge_time_windows[sn][1]["discharge_start_time"])) + return 1 + return 0 + + +async def test_schedule_copied_under_lock(): + """write_time_windows_if_changed() takes its copy of the schedule while holding the schedule lock.""" + api, sn = _slot_event_api() + api.charge_discharge_time_windows[sn][1].update({"discharge_enable": 1, "discharge_start_time": "11:30", "discharge_end_time": "12:00"}) + api.cached_values[sn] = {} + api.schedule_lock.acquire() + thread = _run_in_thread(lambda: api.write_time_windows_if_changed(sn)) + time.sleep(0.2) + calls_while_locked = len(api.read_and_write_cid_calls) + api.schedule_lock.release() + thread.join(2) + + if calls_while_locked: + print("ERROR: the schedule was copied and written without the lock ({} writes)".format(calls_while_locked)) + return 1 + if not api.read_and_write_cid_calls: + print("ERROR: the write should go ahead once the lock is free") + return 1 + return 0 + + # ==================== Daily API allowance (issue #5087) ====================