feat: validate limited discharge and full Deye command trajectories

Keep manual SOC floors and power ceilings, and replay proposed commands only against isolated states. Physical timetable repair and hold commissioning remain explicit hypotheses.

Author: arturonaredo

apps/predbat/deye_local.py

@@ -246,13 +246,15 @@ def _report(self, status, mode, reason=None):
             attrs["reason"] = reason
         self.base.dashboard_item(self.base.prefix + ".deye_local_control", state=status, attributes=attrs)
 
-    def apply(self, target_soc, freeze=False):
+    def apply(self, target_soc, freeze=False, discharge_power_w=None):
         """Apply one native intent with grid-source interlock and no global-limit writes."""
         mode = "invalid"
         if self.base.set_read_only or getattr(self.base, "set_read_only_axle", False):
             self._report("read_only", "none")
             return False
         try:
+            if type(freeze) is not bool:
+                raise ValueError("DEYE freeze must be boolean")
             target = self._number(target_soc, 0, 100, "target SOC")
             mode = "hold" if freeze else "charge" if target > 0 else "demand"
             if not self.settings.get("control", False):
@@ -263,9 +265,17 @@ def apply(self, target_soc, freeze=False):
             if self.base.manual_soc_max_keep:
                 raise ValueError("DEYE local cannot enforce a manual solar charge ceiling")
             floor = self._number(self.inverter.reserve_percent, 0, 100, "reserve")
+            floors = getattr(self.base, "all_active_keep", {})
+            if not isinstance(floors, dict):
+                raise ValueError("Invalid DEYE manual SOC floor map")
+            manual_floor = floors.get(getattr(self.base, "minutes_now", 0), 0)
+            if isinstance(manual_floor, bool) or not isinstance(manual_floor, (int, float)) or not math.isfinite(manual_floor) or not 0 <= manual_floor <= 100:
+                raise ValueError("Invalid DEYE manual SOC floor")
+            floor = max(floor, math.ceil(manual_floor - 1e-9))
             soc = 100 if freeze else max(target, floor)
             cap = min(self.settings["max_discharge_power_w"], int(round(self.inverter.battery_rate_max_discharge * MINUTE_WATT)))
-            power = 0 if freeze else cap
+            requested_power = cap if discharge_power_w is None else self._number(discharge_power_w, 0, cap, "period discharge power")
+            power = 0 if freeze else requested_power
             source = "Grid" if mode == "charge" else "Disabled"
             rows = self.settings["programs"]
             numbers = {row[field]: value for row in rows for field, value in (("power", power), ("soc", soc))}

apps/predbat/deye_local_shadow.py

@@ -13,9 +13,12 @@ class DeyeShadowState:
     def __init__(self, base, settings):
         """Copy the required observations once; never retain the real base object."""
         self.prefix = base.prefix
+        self.args = {}
         self.set_read_only = False
         self.set_read_only_axle = False
         self.manual_soc_max_keep = deepcopy(base.manual_soc_max_keep)
+        self.all_active_keep = deepcopy(getattr(base, "all_active_keep", {}))
+        self.minutes_now = getattr(base, "minutes_now", 0)
         self.timezone = base.get_arg("timezone", "Europe/London")
         self.states = {}
         self.writes = []
@@ -54,7 +57,7 @@ def dashboard_item(self, entity, state, attributes):
         self.report = {"state": state, **attributes}
 
 
-def preview_deye_local(inverter, target_soc, freeze=False):
+def preview_deye_local(inverter, target_soc, freeze=False, discharge_power_w=None):
     """Publish a non-executing trace alongside the unchanged native controller.
 
     Success means the translator completed against simulated acknowledgements,
@@ -73,6 +76,7 @@ def preview_deye_local(inverter, target_soc, freeze=False):
         "hardware_confirmed": False,
         "target_soc": target_soc,
         "freeze": bool(freeze),
+        "discharge_power_w": discharge_power_w,
         "native_inverter_type": inverter.inverter_type,
         "simulated_acknowledgements": True,
         "proposed_writes": [],
@@ -92,7 +96,7 @@ def preview_deye_local(inverter, target_soc, freeze=False):
         )
         adapter = DeyeLocal(fake_inverter)
         adapter.settings = settings
-        success = adapter.apply(target_soc, freeze)
+        success = adapter.apply(target_soc, freeze, discharge_power_w=discharge_power_w)
         state = "simulated" if success else "blocked"
         report["reason"] = simulated.report.get("reason", "Physical commissioning and executable integration remain required")
         report["mode"] = simulated.report.get("mode", "none")
@@ -110,7 +114,7 @@ def preview_deye_local(inverter, target_soc, freeze=False):
                 for row, stamp in zip(settings["programs"], ordered):
                     simulated.states[row["time"]]["state"] = stamp
                 simulated.writes.clear()
-                repaired_success = adapter.apply(target_soc, freeze)
+                repaired_success = adapter.apply(target_soc, freeze, discharge_power_w=discharge_power_w)
                 report["table_repair_hypothesis"] = {
                     "physical_table_changed": False,
                     "required_before_activation": True,
@@ -123,3 +127,58 @@ def preview_deye_local(inverter, target_soc, freeze=False):
         report["reason"] = str(exc) if isinstance(exc, ValueError) else "DEYE shadow input or simulation error"
     base.dashboard_item(base.prefix + ".deye_shadow", state=state, attributes=report)
     return report
+
+
+def check_joint_deye_commands(base, steps):
+    """Translate a whole candidate through the real Inverter entry point in memory.
+
+    A repaired timetable and a commissioned hold can be explicit hypotheses in
+    this check. They never change the actual configuration or count as fulfilled
+    activation gates. Only copied HA states are writable by the simulated base.
+    """
+    from inverter import Inverter
+
+    report = {"available": True, "passed": False, "hardware_confirmed": False, "actuator_signals_modified": False, "hypotheses": [], "checked_steps": 0}
+    try:
+        original = base.args.get("deye_shadow")
+        if not isinstance(original, dict) or original.get("control", False) is not False:
+            raise ValueError("A disabled deye_shadow mapping is required for command replay")
+        settings = DeyeLocal.validate_settings({**original, "control": True})
+        simulated = DeyeShadowState(base, settings)
+        inverter = Inverter.__new__(Inverter)
+        inverter.base = simulated
+        inverter.id = 0
+        inverter.inverter_type = "DEYE"
+        inverter.reserve_percent = int(round(base.reserve / base.soc_max * 100))
+        inverter.battery_rate_max_discharge = base.battery_rate_max_discharge
+        inverter.inv_write_and_poll_sleep = 0
+        inverter.count_register_writes = 0
+        inverter.deye_local = DeyeLocal(inverter)
+        inverter.deye_local.settings = settings
+        try:
+            inverter.deye_local.snapshot()
+        except ValueError as exc:
+            if "boundaries must be strictly increasing" not in str(exc):
+                raise
+            rows = settings["programs"]
+            times = [simulated.states[row["time"]]["state"] for row in rows]
+            minutes = [int(stamp[:2]) * 60 + int(stamp[3:5]) for stamp in times]
+            if len(set(minutes)) != 6:
+                raise ValueError("Duplicate ToU times cannot be repaired by sorting") from None
+            for row, stamp in zip(rows, [stamp for _, stamp in sorted(zip(minutes, times))]):
+                simulated.states[row["time"]]["state"] = stamp
+            report["hypotheses"].append("physical_timetable_must_be_normalised_before_activation")
+        if not settings.get("hold_verified", False):
+            settings["hold_verified"] = True
+            report["hypotheses"].append("hold_contract_still_requires_physical_commissioning")
+        for step in steps:
+            simulated.minutes_now = step["start"]
+            if not inverter.adjust_charge_immediate(step["target_soc"], freeze=step["freeze"], discharge_power_w=step["power_w"]):
+                raise ValueError(simulated.report.get("reason", "DEYE command replay did not complete"))
+            report["checked_steps"] += 1
+        report["passed"] = True
+        report["simulated_service_count"] = len(simulated.writes)
+        report["single_initial_observation_set"] = True
+    except Exception as exc:
+        report["reason"] = str(exc) if isinstance(exc, ValueError) else "DEYE command replay error"
+    return report

apps/predbat/inverter.py

@@ -3396,19 +3396,21 @@ def call_service_template(self, service, data, domain="charge", extra_data=None)
         # transient failure would silently downgrade a freeze into a stop rather than retrying it.
         return True
 
-    def adjust_charge_immediate(self, target_soc, freeze=False):
+    def adjust_charge_immediate(self, target_soc, freeze=False, discharge_power_w=None):
         """
         Adjust from charging or not charging based on passed target soc
         """
         if self.base.args.get("deye_shadow") is not None:
             try:
                 from deye_local_shadow import preview_deye_local
 
-                preview_deye_local(self, target_soc, freeze)
+                preview_deye_local(self, target_soc, freeze, discharge_power_w=discharge_power_w)
             except Exception:
                 self.log("Warn: DEYE shadow preview failed; native inverter control continues unchanged")
         if getattr(self, "deye_local", None):
-            return self.deye_local.apply(target_soc, freeze=freeze)
+            return self.deye_local.apply(target_soc, freeze=freeze, discharge_power_w=discharge_power_w)
+        if discharge_power_w is not None:
+            raise ValueError("Explicit ToU discharge power requires a DEYE local inverter")
         # A Solis with a target SoC (FB00) has its Energy Storage Control Switch driven from here on every cycle
         # that is not exporting. It stays on Backup/Reserve - Self-Use with the Battery Reserve bit, which makes the
         # reserve Predbat writes (the Reserved SOC) a real discharge floor, so the existing reserve holds work.

apps/predbat/tests/test_deye_shadow.py

@@ -5,7 +5,7 @@
 from types import SimpleNamespace
 from unittest.mock import patch
 
-from deye_local_shadow import preview_deye_local
+from deye_local_shadow import preview_deye_local, check_joint_deye_commands, DeyeShadowState
 from inverter import Inverter
 from const import MINUTE_WATT
 
@@ -136,6 +136,45 @@ def run_deye_shadow_tests(my_predbat):
                 reference = calls
             assert calls == reference and calls
             tests += 1
+
+        source = ShadowSource()
+        source.reserve = 1
+        source.soc_max = 10
+        source.battery_rate_max_discharge = 3000 / MINUTE_WATT
+        source.all_active_keep = {125: 30.25}
+        source.minutes_now = 120
+        before = deepcopy((source.states, source.args, source.all_active_keep))
+        rows = source.args["deye_shadow"]["programs"]
+        source.states[rows[2]["time"]]["state"] = "23:55:00"
+        before = deepcopy((source.states, source.args, source.all_active_keep))
+        steps = [
+            {"start": 120, "target_soc": 100, "freeze": False, "power_w": None},
+            {"start": 125, "target_soc": 0, "freeze": False, "power_w": 1500},
+            {"start": 130, "target_soc": 0, "freeze": True, "power_w": None},
+            {"start": 135, "target_soc": 0, "freeze": False, "power_w": 1800},
+        ]
+        calls = []
+        original_call = DeyeShadowState.call_service_wrapper
+
+        def record_command(simulated, service, **data):
+            """Observe every simulated command without providing real service authority."""
+            calls.append((service, dict(data)))
+            return original_call(simulated, service, **data)
+
+        with patch.object(DeyeShadowState, "call_service_wrapper", new=record_command):
+            checked = check_joint_deye_commands(source, steps)
+        assert checked["passed"] and checked["checked_steps"] == 4, checked
+        assert len(checked["hypotheses"]) == 2 and not checked["hardware_confirmed"]
+        assert before == (source.states, source.args, source.all_active_keep)
+        for row in rows:
+            assert ("number/set_value", {"entity_id": row["soc"], "value": 31}) in calls
+            assert ("number/set_value", {"entity_id": row["power"], "value": 1500}) in calls
+        tests += 1
+        for bad_power in (-1, 2001, 1500.5, float("nan"), True):
+            checked = check_joint_deye_commands(source, [{**steps[1], "power_w": bad_power}])
+            assert not checked["passed"] and checked["checked_steps"] == 0, checked
+            assert before == (source.states, source.args, source.all_active_keep)
+            tests += 1
         print("DEYE shadow: {} isolation checks passed".format(tests))
         return False
     except Exception as exc:

docs/deye-local.md

@@ -34,6 +34,14 @@ down to its target, matching `set_discharge_during_charge` on this profile.
 The charge rate must already be set and correctly modelled by the installation;
 this adapter does not modulate it or raise current limits.
 
+An explicit `discharge_power_w` request can narrow the per-period discharge
+ceiling, without changing global current limits or the EV's fixed charging rate.
+Battery-first demand then supplies up to that ceiling and the grid supplies the
+remaining load. The requested watts must be finite whole units between zero and
+the already configured ceiling. Active native manual SOC floors are preserved;
+fractional floors round upward to the hardware's whole-percent control. Neither
+a per-period ceiling nor a forecast is an instantaneous site-import guarantee.
+
 The hold contract must be commissioned on the actual inverter before setting
 `hold_verified: true`. Only then is charge freeze advertised to the planner for
 this instance. This does not change the `SK` capability declaration. If a hold is
@@ -85,6 +93,13 @@ same times sorted **only in the private copy**. This shows the proposed commands
 after a possible migration; it neither repairs the physical table nor fulfils
 the activation precondition. Duplicate or invalid times are not guessed.
 
+The joint-planning companion can also check a whole candidate through the same
+`Inverter.adjust_charge_immediate` entry point, including explicit discharge-watt
+ceilings and time-local manual SOC floors. This uses only copied observations and
+simulated acknowledgements. Any sorted timetable or commissioned hold used in the
+check is labelled as a hypothesis, never applied to the actual instance. Global
+current limits and hardware behaviour still require an operational preflight.
+
 ### Selecting the experimental controller
 
 Start from the [disabled template fragment](https://github.com/springfall2008/batpred/blob/main/templates/deye.yaml),