test: integrate bounded power-sharing and Deye command checks

Sources: PR springfall2008#5498 at 091716c and PR springfall2008#5497 at 68682c7. Conditional stale-SOC scenarios retain their freshness blocker. All controls remain shadow-only in the local rollout.

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/joint_site.py

@@ -10,14 +10,15 @@
 proof of global optimality. Expected availability never establishes connection.
 """
 
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 import math
 from typing import Optional, Sequence, Tuple
 
 STATUS_FEASIBLE = "feasible"
 STATUS_BOUNDED_NO_PLAN = "bounded_no_plan"
 BATTERY_DEMAND, BATTERY_HOLD, BATTERY_CHARGE = "demand", "hold", "charge"
 BATTERY_MODES = (BATTERY_DEMAND, BATTERY_HOLD, BATTERY_CHARGE)
+BATTERY_LIMITED_DEMAND = "limited_demand"
 EV_OFF, EV_ON = "off", "on"
 EPSILON = 1e-9
 
@@ -101,6 +102,7 @@ class PlannerConfig:
     time_step_minutes: int = 5
     beam_width: Optional[int] = 128
     soc_bucket_kwh: float = 0.05
+    allow_limited_discharge: bool = False
 
 
 @dataclass(frozen=True)
@@ -147,6 +149,7 @@ class SlotAction:
     slot_cost: float
     ev_on_minutes: float
     segments: Tuple[FlowSegment, ...]
+    discharge_limit_kw: Optional[float] = None
 
 
 @dataclass(frozen=True)
@@ -201,6 +204,8 @@ def _validate(slots, ev, battery, site, planner):
         raise PlannerInputError("Unsupported planning step")
     if planner.beam_width is not None and (type(planner.beam_width) is not int or not 1 <= planner.beam_width <= 4096):
         raise PlannerInputError("beam_width must be an integer in 1..4096 or None")
+    if type(planner.allow_limited_discharge) is not bool:
+        raise PlannerInputError("allow_limited_discharge must be boolean")
     _require_finite("soc_bucket_kwh", planner.soc_bucket_kwh, 1e-6)
     if not slots or len(slots) > 576 or (planner.beam_width is None and len(slots) > 12):
         raise PlannerInputError("Horizon must have 1..576 slots (at most 12 for exhaustive search)")
@@ -312,18 +317,28 @@ def _segment(duration, house_kw, ev_kw, pv_kw, soc, mode, battery, ev, site):
 def _resolve_action(slot, ev_on, mode, node, battery, ev, site):
     """Compute a complete physical step, never a partially served household/EV load."""
     on_minutes = 0.0
+    discharge_limit = None
     if ev_on:
         if not _ev_is_available(ev, slot.index) or slot.start_min >= ev.deadline_min or node.ev_kwh >= ev.stop_soc_kwh - EPSILON:
             return None
         if not node.on and (node.off_minutes + EPSILON < ev.min_off_minutes or node.battery_kwh + EPSILON < battery.capacity_kwh * ev.restart_battery_percent / 100):
             return None
         on_minutes = min(slot.duration_min, ev.deadline_min - slot.start_min, (ev.stop_soc_kwh - node.ev_kwh) / ev.efficiency / ev.rate_kw * 60)
+    flow_mode = mode
+    if mode == BATTERY_LIMITED_DEMAND:
+        # Still battery-first, but under a reproducible per-period power ceiling.
+        # The fixed ceiling remains in place after a partly completed EV step;
+        # it is not a fictitious continuously modulating grid-first controller.
+        net_kw = (slot.ac_load_kwh - slot.pv_kwh) * 60 / slot.duration_min + (ev.rate_kw if on_minutes else 0)
+        discharge_limit = min(battery.max_discharge_kw, max(0, math.ceil((net_kw - site.import_cap_kw) * 1000 + 2) / 1000))
+        battery = replace(battery, max_discharge_kw=discharge_limit)
+        flow_mode = BATTERY_DEMAND
     segments = []
     soc = node.battery_kwh
     for duration, ev_kw in ((on_minutes, ev.rate_kw), (slot.duration_min - on_minutes, 0)):
         if duration <= 1e-10:
             continue
-        result = _segment(duration, slot.ac_load_kwh * 60 / slot.duration_min, ev_kw, slot.pv_kwh * 60 / slot.duration_min, soc, mode, battery, ev, site)
+        result = _segment(duration, slot.ac_load_kwh * 60 / slot.duration_min, ev_kw, slot.pv_kwh * 60 / slot.duration_min, soc, flow_mode, battery, ev, site)
         if result is None:
             return None
         parts, soc = result
@@ -363,6 +378,7 @@ def _resolve_action(slot, ev_on, mode, node, battery, ev, site):
         cost,
         on_minutes,
         tuple(segments),
+        discharge_limit,
     )
 
 
@@ -396,6 +412,25 @@ def _bounded_no_plan(deadline_idx, transitions, max_frontier, reason):
     return PlanResult(STATUS_BOUNDED_NO_PLAN, None, (), (), (), (), (), (), diagnostic, deadline_idx)
 
 
+def _reachable_floor_envelope(slots, battery, site):
+    """Optimistic charging bound that preserves time-local floors during pruning.
+
+    Assume the EV is off and any grid/PV headroom can charge the battery, even
+    where an actual discrete action could not use all that headroom. This only
+    rejects states unable to reach a future floor under the most favourable
+    charging assumption; it does not claim a feasible continuation exists.
+    """
+    required_end = [0.0] * len(slots)
+    required = battery.end_min_kwh
+    for slot in reversed(slots):
+        required = max(required, slot.battery_min_kwh, battery.reserve_kwh)
+        required_end[slot.index] = required
+        headroom = max(0, site.import_cap_kw + (slot.pv_kwh - slot.ac_load_kwh) * 60 / slot.duration_min)
+        gain = min(battery.max_charge_kw, headroom) * slot.duration_min / 60 * battery.charge_efficiency
+        required = max(slot.battery_min_kwh, battery.reserve_kwh, required - gain)
+    return required_end
+
+
 def plan(slots: Sequence[Slot], ev: EVConfig, battery: BatteryConfig, site: SiteConfig, planner: Optional[PlannerConfig] = None) -> PlanResult:
     """Jointly search all eligible prices, EV decisions and executable-mode physics.
 
@@ -405,6 +440,8 @@ def plan(slots: Sequence[Slot], ev: EVConfig, battery: BatteryConfig, site: Site
     """
     planner = planner if planner is not None else PlannerConfig()
     _validate(slots, ev, battery, site, planner)
+    modes = BATTERY_MODES + (BATTERY_LIMITED_DEMAND,) if planner.allow_limited_discharge else BATTERY_MODES
+    required_end = _reachable_floor_envelope(slots, battery, site)
     deadline_idx = _deadline_slot_index(slots, ev.deadline_min)
     future = [0.0] * (len(slots) + 1)
     for i in range(len(slots) - 1, -1, -1):
@@ -420,12 +457,14 @@ def plan(slots: Sequence[Slot], ev: EVConfig, battery: BatteryConfig, site: Site
     for i, slot in enumerate(slots):
         next_states = {}
         for node in frontier:
-            for mode in BATTERY_MODES:
+            for mode in modes:
                 for on in (False, True):
                     transitions += 1
                     action = _resolve_action(slot, on, mode, node, battery, ev, site)
                     if action is None:
                         continue
+                    if action.battery_soc_kwh_end < required_end[i] - EPSILON:
+                        continue
                     if action.ev_soc_kwh_end + future[i + 1] < ev.stop_soc_kwh - EPSILON:
                         continue
                     if i == len(slots) - 1 and action.battery_soc_kwh_end < battery.end_min_kwh - EPSILON:
@@ -464,7 +503,7 @@ def plan(slots: Sequence[Slot], ev: EVConfig, battery: BatteryConfig, site: Site
         ev_stop_target_met=True,
         battery_end_min_met=True,
         ev_on_slot_count=sum(a.ev_action == EV_ON for a in actions),
-        mode_counts=tuple((mode, sum(a.battery_mode == mode for a in actions)) for mode in BATTERY_MODES),
+        mode_counts=tuple((mode, sum(a.battery_mode == mode for a in actions)) for mode in modes),
         notes=notes,
         transitions_evaluated=transitions,
         max_frontier_size=max_frontier,