feat: add shared Deye Solarman template and observed service sequences

Author: arturonaredo

apps/predbat/inverter.py

@@ -56,6 +56,10 @@
 _NOT_COMMITTED = object()
 
 
+class ServiceSequenceError(ValueError):
+    """A local validation/observation failure whose diagnostic is safe to log."""
+
+
 def within_fuzzy(value, target, fuzzy, fuzzy_below=None):
     """
     Whether a numeric read counts as target: within fuzzy either side, or within fuzzy_below when it is
@@ -3306,19 +3310,153 @@ def set_current_from_power(self, direction, power):
         new_current = round(power / self.battery_voltage, self.inv_current_dp)
         self.write_and_poll_value(f"timed_{direction}_current", self.base.get_arg(f"timed_{direction}_current", indirect=False, index=self.id), new_current, fuzzy=1)
 
+    def service_sequence_number(self, value):
+        """Parse a finite numeric service value without accepting booleans as zero/one."""
+        if isinstance(value, bool):
+            raise ServiceSequenceError("boolean is not a numeric control value")
+        number = float(value)
+        if not math.isfinite(number):
+            raise ServiceSequenceError("non-finite control value")
+        return number
+
+    def service_sequence_matches(self, entity_id, expected):
+        """Compare an HA observation, not a hardware acknowledgement, with a service target."""
+        state = self.base.get_state_wrapper(entity_id, refresh=True)
+        return self.service_sequence_state_matches(state, expected)
+
+    def service_sequence_state_matches(self, state, expected):
+        """Match a raw state without substituting a default for unavailable numeric data."""
+        if isinstance(expected, float):
+            try:
+                return self.service_sequence_number(state) == expected
+            except (ValueError, TypeError, OverflowError):
+                return False
+        return state == expected
+
+    def prepare_service_sequence(self, service_list, data, extra_data):
+        """Resolve and validate every observed step before allowing the first write."""
+        actions = {
+            "number/set_value": ("value", None),
+            "input_number/set_value": ("value", None),
+            "select/select_option": ("option", None),
+            "input_select/select_option": ("option", None),
+            "switch/turn_on": (None, "on"),
+            "switch/turn_off": (None, "off"),
+            "input_boolean/turn_on": (None, "on"),
+            "input_boolean/turn_off": (None, "off"),
+        }
+        full_data = {**data, **extra_data, "reserve_soc": math.ceil(self.service_sequence_number(self.reserve_percent))}
+        steps = []
+        for template in service_list:
+            if not isinstance(template, dict) or template.get("verify") is not True:
+                raise ServiceSequenceError("every step must have verify: true")
+            resolved = {}
+            for key, value in template.items():
+                if key != "verify":
+                    resolved[key] = self.base.resolve_arg(template, value, indirect=False, index=self.id if isinstance(value, list) else None, default=None, extra_args=full_data)
+            service = resolved.get("service")
+            if not isinstance(service, str) or service.replace(".", "/") not in actions:
+                raise ServiceSequenceError("unsupported observed service")
+            service = service.replace(".", "/")
+            field, expected = actions[service]
+            if set(resolved) != {"service", "entity_id"} | ({field} if field else set()):
+                raise ServiceSequenceError("observed services require only service, entity_id and their value/option")
+            entity_ids = resolved["entity_id"]
+            if not isinstance(entity_ids, str):
+                raise ServiceSequenceError("entity_id must resolve to a string (comma-separated for a group)")
+            entities = [entity.strip() for entity in entity_ids.split(",")]
+            entity_domain = service.split("/")[0]
+            if len(set(entities)) != len(entities) or any(not re.fullmatch(entity_domain + r"\.[a-z0-9_]+", entity) for entity in entities):
+                raise ServiceSequenceError("invalid or duplicate entity_id for observed service")
+            if field == "value":
+                expected = self.service_sequence_number(resolved[field])
+            elif field == "option":
+                expected = resolved[field]
+                if not isinstance(expected, str) or not expected:
+                    raise ServiceSequenceError("option must be a non-empty string")
+            for entity in entities:
+                state = self.base.get_state_wrapper(entity, refresh=True)
+                if state is None or str(state).lower() in ("unknown", "unavailable", "none", ""):
+                    raise ServiceSequenceError("{} has no available state".format(entity))
+                if field == "value":
+                    self.service_sequence_number(state)
+                    lower, upper, step = [self.service_sequence_number(self.base.get_state_wrapper(entity, attribute=attribute)) for attribute in ("min", "max", "step")]
+                    if lower > upper or step <= 0 or not lower <= expected <= upper:
+                        raise ServiceSequenceError("{} target is outside its numeric bounds".format(entity))
+                    ticks = (expected - lower) / step
+                    if not math.isfinite(ticks) or not math.isclose(ticks, round(ticks), rel_tol=0, abs_tol=1e-7):
+                        raise ServiceSequenceError("{} target does not match its numeric step".format(entity))
+                elif field == "option":
+                    options = self.base.get_state_wrapper(entity, attribute="options")
+                    if not isinstance(options, list) or expected not in options or state not in options:
+                        raise ServiceSequenceError("{} has an unsupported option".format(entity))
+                elif state not in ("on", "off"):
+                    raise ServiceSequenceError("{} has an invalid switch state".format(entity))
+            service_data = {"entity_id": ",".join(entities)}
+            if field:
+                service_data[field] = expected
+            steps.append((service, service_data, entities, expected))
+        return steps
+
+    def call_service_sequence(self, service, service_list, data, extra_data):
+        """Apply opt-in observed steps in order, stopping on failure without speculative rollback.
+
+        Repeated entities are intentional (disable, set target, enable). Compare their FINAL
+        values first so an unchanged charge does not disable/re-enable on every plan cycle.
+        HA can expose cached/optimistic state: this is not a physical confirmation or transaction.
+        """
+        try:
+            steps = self.prepare_service_sequence(service_list, data, extra_data)
+            final = {entity: expected for _, _, entities, expected in steps for entity in entities}
+            if all(self.service_sequence_matches(entity, expected) for entity, expected in final.items()):
+                return
+            observed = {}
+            for service_name, service_data, entities, expected in steps:
+                if not all(self.service_sequence_matches(entity, value) for entity, value in observed.items()):
+                    raise ServiceSequenceError("an earlier step no longer reads as observed")
+                if not all(self.service_sequence_matches(entity, expected) for entity in entities):
+                    if not self.base.call_service_wrapper(service_name, **service_data):
+                        raise ServiceSequenceError("{} was not accepted".format(service_name))
+                    # Wait on HA only; this does not request faster hardware polling.
+                    for entity in entities:
+                        state = self._poll_after_write(entity, lambda state, expected=expected: self.service_sequence_state_matches(state, expected))
+                        if not self.service_sequence_state_matches(state, expected):
+                            raise ServiceSequenceError("{} was not observed".format(entity))
+                    self.count_register_writes += len(entities)
+                observed.update({entity: expected for entity in entities})
+                if not all(self.service_sequence_matches(entity, value) for entity, value in observed.items()):
+                    raise ServiceSequenceError("{} was not observed".format(service_name))
+            self.log("Inverter {} service {} final state observed in Home Assistant".format(self.id, service))
+        except Exception as error:
+            # A transport exception must also stop the rest of the sequence. Avoid logging its
+            # potentially credential-bearing response; validation errors above are local text.
+            reason = str(error) if isinstance(error, ServiceSequenceError) else type(error).__name__
+            message = "Warn: Inverter {} service {} sequence stopped: {}".format(self.id, service, reason)
+            self.log(message)
+            self.base.record_status(message, had_errors=True)
+
     def call_service_template(self, service, data, domain="charge", extra_data=None):
         """
         Call a service template with data
         """
         if extra_data is None:
             extra_data = {}
         service_list = self.base.args.get(service, "")
+        # An explicit empty list handles a deliberate no-op, suppressing the caller's fallback.
+        if service_list == []:
+            return True
         if not service_list:
             return False
 
         if not isinstance(service_list, list):
             service_list = [service_list]
 
+        if any(isinstance(step, dict) and "verify" in step for step in service_list):
+            self.base.last_service_hash.pop(domain, None)
+            self.call_service_sequence(service, service_list, data, extra_data)
+            # Keep the configured/handled contract, even on failure (never downgrade a freeze).
+            return True
+
         hash_index = domain
         last_service_hash = self.base.last_service_hash.get(hash_index, "")
         this_service_hash = hash(str(service) + "_" + str(data) + "_" + str(extra_data))

apps/predbat/tests/test_deye_solarman_template.py

@@ -0,0 +1,178 @@
+# -----------------------------------------------------------------------------
+# Predbat Home Battery System
+# Copyright Trefor Southwell 2026 - All Rights Reserved
+# This application maybe used for personal use only and not for commercial use
+# -----------------------------------------------------------------------------
+"""Exercise the shared YAML through real native inverter and executor paths, without hardware."""
+
+import copy
+from datetime import timedelta
+from pathlib import Path
+from unittest.mock import patch
+
+import yaml
+
+from config import INVERTER_DEF
+from inverter import Inverter
+from tests.test_infra import reset_inverter
+
+
+def run_deye_solarman_template_tests(my_predbat):
+    """Validate six-row charge/demand, failure barriers and the absence of unrelated writes."""
+    template_path = Path(__file__).resolve().parents[3] / "templates" / "deye_solarman.yaml"
+    with template_path.open() as stream:
+        template = yaml.safe_load(stream)["pred_bat"]
+    saved_definition = copy.deepcopy(INVERTER_DEF.get("DeyeSolarman"))
+    saved_call = my_predbat.call_service_wrapper
+    writes = []
+    rejected = set()
+    failed = False
+    ha = my_predbat.ha_interface
+    sources = ["select.test_deye_program_{}_charging".format(row) for row in range(1, 7)]
+    targets = ["number.test_deye_program_{}_soc".format(row) for row in range(1, 7)]
+    protected = {"time.test_deye_program_{}_time".format(row): "{:02d}:00:00".format((row - 1) * 4) for row in range(1, 7)}
+    protected.update({"number.test_deye_program_{}_power".format(row): 3000 for row in range(1, 7)})
+    protected.update(
+        {
+            "select.test_deye_work_mode": "Zero Export To CT",
+            "select.test_deye_time_of_use": "Week",
+            "number.test_deye_battery_max_charging_current": 59,
+            "number.test_deye_battery_max_discharging_current": 160,
+            "number.test_deye_battery_shutdown_soc": 5,
+        }
+    )
+
+    def check(condition, message):
+        """Collect assertions so one failing case does not hide the remaining evidence."""
+        nonlocal failed
+        if not condition:
+            print("ERROR: Deye/Solarman template: " + message)
+            failed = True
+
+    def service_call(service, **data):
+        """Model HA's grouped state-setting services, retaining entity attributes."""
+        entities = data.get("entity_id", "").split(",")
+        writes.append((service, copy.deepcopy(data)))
+        if rejected.intersection(entities):
+            return False
+        for entity in entities:
+            check(entity in sources + targets, "unexpected physical write to " + entity)
+            if entity in sources + targets:
+                ha.dummy_items[entity]["state"] = data.get("option", data.get("value"))
+        return True
+
+    def rows(source, target):
+        """Check all six observations, never just the first row."""
+        return all(ha.get_state(entity) == source for entity in sources) and all(float(ha.get_state(entity)) == target for entity in targets)
+
+    try:
+        reset_inverter(my_predbat)
+        my_predbat.args.clear()
+        my_predbat.args.update(template)
+        my_predbat.args["deye_entity_prefix"] = "test_deye"
+        my_predbat.args_from_apps_yaml = copy.deepcopy(my_predbat.args)
+        my_predbat.last_service_hash = {}
+        ha.dummy_items.update(protected)
+        for entity in sources:
+            ha.dummy_items[entity] = {"state": "Grid", "options": ["Disabled", "Grid", "Generator", "Both"]}
+        for entity in targets:
+            ha.dummy_items[entity] = {"state": 40, "min": 0, "max": 100, "step": 1, "unit_of_measurement": "%"}
+        for suffix, value in (("battery", 30), ("battery_voltage", 51.2), ("battery_temperature", 20), ("battery_power", 0), ("pv_power", 0), ("load_power", 500), ("grid_power", 500)):
+            ha.dummy_items["sensor.test_deye_" + suffix] = value
+        my_predbat.call_service_wrapper = service_call
+        inv = Inverter(my_predbat, 0, quiet=True)
+        inv.sleep = lambda seconds: None
+        inv.inv_write_and_poll_sleep = 0
+        inv.update_status(my_predbat.minutes_now)
+        my_predbat.inverters = [inv]
+        check(not writes, "construction must not send hardware services")
+        check(template["template"] is True, "shipped example must stay in template mode")
+        check(not inv.inv_support_charge_freeze and not inv.inv_support_discharge_freeze and not inv.inv_has_timed_pause, "unsupported freeze/pause must remain disabled")
+        check(inv.inv_output_charge_control == "none", "global rate controls must remain unmapped")
+        check("discharge_start_service" not in template, "forced export must be absent")
+
+        # Both window bookkeeping and rate/target updates are private sensors, not ToU rows.
+        inv.adjust_charge_window(my_predbat.midnight_utc + timedelta(hours=23), my_predbat.midnight_utc + timedelta(hours=25), 23 * 60)
+        inv.adjust_charge_rate(1000)
+        inv.adjust_discharge_rate(0)
+        inv.adjust_battery_target(80, isCharging=True)
+        check(not writes, "native bookkeeping must not write physical time, current or SOC controls")
+        for key in ("charge_start_time", "charge_end_time", "discharge_start_time", "discharge_end_time"):
+            check(my_predbat.get_arg(key, indirect=False, index=0) in inv.created_attributes, "time bookkeeping must use created sensors")
+
+        inv.adjust_charge_immediate(80)
+        check(rows("Grid", 80), "charge must set all six targets and sources")
+        check([data.get("option", data.get("value")) for _, data in writes] == ["Disabled", 80.0, "Grid"], "charge must observe disable → SOC → enable")
+        writes.clear()
+        inv.adjust_charge_immediate(80)
+        inv.adjust_export_immediate(100)
+        check(not writes and rows("Grid", 80), "repeated charge and idle export hook must not clobber a charge")
+
+        inv.adjust_charge_immediate(0)
+        check(rows("Disabled", 10), "demand must disable charging and restore the domestic floor, not shutdown")
+        my_predbat.expose_config("set_reserve_min", 25)
+        inv.refresh_config()
+        inv.adjust_charge_immediate(0)
+        check(rows("Disabled", 25), "demand must follow the current manual domestic minimum")
+
+        writes.clear()
+        rejected.add(targets[-1])
+        inv.adjust_charge_immediate(90)
+        check(not any(data.get("option") == "Grid" for _, data in writes), "failed last target must prevent source enable")
+        rejected.clear()
+        inv.adjust_charge_immediate(90)
+        check(rows("Grid", 90), "next cycle must retry a failed sequence")
+
+        # Native reset writes 100 to its private target, never to the physical ToU targets.
+        my_predbat.set_read_only = False
+        my_predbat.set_charge_window = True
+        my_predbat.set_export_window = False
+        my_predbat.reset_inverter()
+        check(rows("Disabled", 25), "native reset must leave demand, not grid-charge to 100")
+
+        # Drive an active native plan, including the trailing discharge-stop hook and rates.
+        my_predbat.args["template"] = False
+        my_predbat.inverter_hybrid = True
+        my_predbat.set_soc_enable = True
+        my_predbat.set_reserve_enable = False
+        my_predbat.set_charge_freeze = False
+        my_predbat.set_charge_low_power = False
+        my_predbat.set_discharge_during_charge = True
+        my_predbat.inverter_needs_reset = False
+        my_predbat.fetch_inverter_data()
+        inv = my_predbat.inverters[0]
+        inv.sleep = lambda seconds: None
+        my_predbat.charge_window_best = [{"start": my_predbat.minutes_now, "end": my_predbat.minutes_now + 60, "average": 1}]
+        my_predbat.charge_limit_best = [8]
+        my_predbat.charge_limit_percent_best = [80]
+        my_predbat.execute_plan()
+        check(rows("Grid", 80), "native active charge plan must reach all six rows")
+        my_predbat.charge_window_best = []
+        my_predbat.charge_limit_best = []
+        my_predbat.execute_plan()
+        check(rows("Disabled", 25), "native demand plan must stop grid charging")
+        writes.clear()
+        my_predbat.set_read_only = True
+        my_predbat.execute_plan()
+        check(not writes, "read-only native plan must not dispatch services")
+        my_predbat.set_read_only = False
+        my_predbat.set_charge_window = False
+        my_predbat.set_export_window = False
+        my_predbat.set_soc_enable = False
+        my_predbat.execute_plan()
+        check(not writes, "Monitor must not dispatch services")
+        check(all(ha.get_state(entity) == value for entity, value in protected.items()), "all unmapped installation settings must stay untouched")
+
+        # A transport ValueError is not one of our local diagnostics. Never echo its body.
+        with patch.object(my_predbat, "call_service_wrapper", side_effect=ValueError("private-response-marker")), patch.object(inv, "log") as log, patch.object(my_predbat, "record_status") as status:
+            check(inv.call_service_template("charge_start_service", {"target_soc": 95}) is True, "a transport exception must remain handled")
+            check(status.called, "transport exception must record an error")
+            check("private-response-marker" not in str(log.call_args_list) + str(status.call_args_list), "transport response must not leak into diagnostics")
+    finally:
+        my_predbat.call_service_wrapper = saved_call
+        if saved_definition is None:
+            INVERTER_DEF.pop("DeyeSolarman", None)
+        else:
+            INVERTER_DEF["DeyeSolarman"] = saved_definition
+    print("Deye/Solarman template tests: {}".format("FAILED" if failed else "PASSED"))
+    return failed