fix: integrate joint planning with verified Solarman templates

Author: arturonaredo

apps/predbat/joint_site.py

@@ -97,12 +97,21 @@ class PlannerConfig:
 
     Bucketing merges search labels, never stored energy. Such merging and beam
     pruning can miss a feasible/cheaper plan, so a miss is not infeasibility proof.
+
+    `allowed_battery_modes` declares the subset of battery dispatch modes the
+    planner may emit. It must be a tuple or list of string tags drawn from
+    the supported battery-mode vocabulary (`demand`, `hold`, `charge`). The
+    default equals the full base vocabulary; a tighter subset is honoured by
+    `_validate` and the search loop. When `allow_limited_discharge` is true,
+    `limited_demand` is added on top of whatever subset is declared here.
+    The planner never returns a mode that is not in this subset.
     """
 
     time_step_minutes: int = 5
     beam_width: Optional[int] = 128
     soc_bucket_kwh: float = 0.05
     allow_limited_discharge: bool = False
+    allowed_battery_modes: Tuple[str, ...] = BATTERY_MODES
 
 
 @dataclass(frozen=True)
@@ -206,6 +215,20 @@ def _validate(slots, ev, battery, site, planner):
         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")
+    allowed = planner.allowed_battery_modes
+    if not isinstance(allowed, (tuple, list)) or isinstance(allowed, (str, bytes)):
+        raise PlannerInputError("allowed_battery_modes must be a tuple or list")
+    if not allowed:
+        raise PlannerInputError("allowed_battery_modes must not be empty")
+    seen = set()
+    for mode in allowed:
+        if type(mode) is not str:
+            raise PlannerInputError("allowed_battery_modes entries must be strings")
+        if mode not in BATTERY_MODES:
+            raise PlannerInputError("allowed_battery_modes contains unsupported mode")
+        if mode in seen:
+            raise PlannerInputError("allowed_battery_modes contains duplicate mode")
+        seen.add(mode)
     _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)")
@@ -382,13 +405,21 @@ def _resolve_action(slot, ev_on, mode, node, battery, ev, site):
     )
 
 
-def _prune(nodes, width, bucket, terminal_value):
-    """Preserve EV-progress diversity instead of pruning every energy-consuming EV path."""
+def _prune(nodes, width, bucket, terminal_value, preserve_battery=False):
+    """Preserve EV progress and optionally stored-energy resilience within the same beam bound."""
     groups = {}
     for node in nodes:
         groups.setdefault(_bucket(node.ev_kwh, bucket), []).append(node)
     for group in groups.values():
         group.sort(key=lambda n: (n.cost - terminal_value * n.battery_kwh, n.switches))
+        if preserve_battery:
+            # With no hold action, the cheapest label can have too little stored
+            # energy to serve a later fixed-rate EV start. Keep the cost leader
+            # and highest-energy label before filling the remaining cost layers.
+            high = max(group, key=lambda n: n.battery_kwh)
+            if high is not group[0]:
+                group.remove(high)
+                group.insert(1, high)
     ordered = [groups[key] for key in sorted(groups, reverse=True)]
     kept = []
     layer = 0
@@ -440,7 +471,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
+    base_modes = tuple(planner.allowed_battery_modes)
+    modes = base_modes + ((BATTERY_LIMITED_DEMAND,) if planner.allow_limited_discharge and BATTERY_LIMITED_DEMAND not in base_modes else ())
     required_end = _reachable_floor_envelope(slots, battery, site)
     deadline_idx = _deadline_slot_index(slots, ev.deadline_min)
     future = [0.0] * (len(slots) + 1)
@@ -484,7 +516,7 @@ def plan(slots: Sequence[Slot], ev: EVConfig, battery: BatteryConfig, site: Site
             return _bounded_no_plan(deadline_idx, transitions, max_frontier, "no_complete_transition_at_slot_{}".format(i))
         frontier = list(next_states.values())
         if planner.beam_width is not None and len(frontier) > planner.beam_width:
-            frontier = _prune(frontier, planner.beam_width, planner.soc_bucket_kwh, battery.terminal_value_per_kwh)
+            frontier = _prune(frontier, planner.beam_width, planner.soc_bucket_kwh, battery.terminal_value_per_kwh, preserve_battery=BATTERY_HOLD not in base_modes)
             pruned = True
         max_frontier = max(max_frontier, len(frontier))
     best = min(frontier, key=lambda n: (n.cost - battery.terminal_value_per_kwh * n.battery_kwh, n.switches))

apps/predbat/joint_site_apply.py

@@ -102,6 +102,8 @@ def execution_fingerprint(base):
     payload = {name: getattr(base, name, None) for name in names}
     payload["settings"] = base.args.get("joint_site")
     payload["deye_local"] = base.args.get("deye_local")
+    if isinstance(payload["settings"], dict) and payload["settings"].get("control_model") == "deye_solarman":
+        payload["template_controls"] = {name: base.args.get(name) for name in ("inverter", "inverter_type", "deye_entity_prefix", "charge_start_service", "charge_stop_service", "discharge_stop_service", "charge_freeze_service")}
     encoded = json.dumps(payload, sort_keys=True, allow_nan=False, default=str).encode()
     return hashlib.sha256(encoded).hexdigest()
 
@@ -118,7 +120,12 @@ def _cached(base, entity):
 
 def _control_fingerprint(inverter):
     """Detect observed control drift without claiming read-back proves hardware state."""
-    snapshot = inverter.deye_local.snapshot()
+    if getattr(inverter, "inverter_type", None) == "DeyeSolarman":
+        from joint_site_template import snapshot_template
+
+        snapshot = snapshot_template(inverter)
+    else:
+        snapshot = inverter.deye_local.snapshot()
     return hashlib.sha256(json.dumps(snapshot, sort_keys=True, allow_nan=False).encode()).hexdigest()
 
 
@@ -137,19 +144,29 @@ def _evidence(base):
     now = _clock()
     _require(all(0 <= (now - stamp).total_seconds() <= OBSERVATION_MAX_AGE_SECONDS for stamp in stamps), "physical_observation_stale_or_future")
     options = _options(base)
-    _require(options is not None and options["mode"] == "apply" and options["control_model"] == "deye_tou", "real_deye_controls_required")
+    _require(options is not None and options["mode"] == "apply" and options["control_model"] in ("deye_tou", "deye_solarman"), "real_deye_controls_required")
     _require(not _collect_blockers(base, options), "native_inputs_or_overrides_blocked")
     _require(not base.set_read_only and not getattr(base, "set_read_only_axle", False), "read_only")
     _require(not any(getattr(base, name, False) for name in ("inverter_needs_reset", "had_errors", "fatal_error")), "native_state_not_ready")
     _require(not base.get_arg("template", False), "template_mode")
     inverters = getattr(base, "inverters", None)
     _require(isinstance(inverters, list) and len(inverters) == 1, "single_inverter_required")
     inverter = inverters[0]
-    _require(getattr(inverter, "inverter_type", None) == "DEYE", "local_deye_required")
-    adapter = getattr(inverter, "deye_local", None)
-    _require(adapter is not None, "local_deye_adapter_missing")
-    _require(adapter.settings.get("control") is True and adapter.settings.get("hold_verified") is True, "deye_commissioning_required")
-    _require(getattr(inverter, "inv_support_charge_freeze", False) is True and not getattr(inverter, "in_calibration", True), "inverter_capability_or_calibration")
+    if options["control_model"] == "deye_solarman":
+        from joint_site_template import validate_template
+
+        _require(getattr(inverter, "inverter_type", None) == "DeyeSolarman", "template_inverter_required")
+        try:
+            validate_template(inverter)
+        except (ValueError, TypeError, KeyError, AttributeError):
+            raise JointBlocked("template_contract_invalid") from None
+        _require(not getattr(inverter, "in_calibration", True), "inverter_capability_or_calibration")
+    else:
+        _require(getattr(inverter, "inverter_type", None) == "DEYE", "local_deye_required")
+        adapter = getattr(inverter, "deye_local", None)
+        _require(adapter is not None, "local_deye_adapter_missing")
+        _require(adapter.settings.get("control") is True and adapter.settings.get("hold_verified") is True, "deye_commissioning_required")
+        _require(getattr(inverter, "inv_support_charge_freeze", False) is True and not getattr(inverter, "in_calibration", True), "inverter_capability_or_calibration")
     _require(not getattr(base, "manual_soc_max_keep", None), "manual_ceiling_unsupported")
     _require(base.export_limit == 0 and base.set_discharge_during_charge, "deye_dispatch_model_mismatch")
     _require(options["import_limit_w"] == getattr(base, "import_limit", options["import_limit_w"]), "import_limit_mismatch")
@@ -168,7 +185,7 @@ def _evidence(base):
     minute = (now - base.midnight_utc).total_seconds() / 60
     floor = max(base.reserve, base.all_active_keep.get(int(minute), 0) * base.soc_max / 100)
     _require(base.soc_kw >= floor - 0.002, "domestic_manual_floor")
-    adapter.snapshot()
+    _control_fingerprint(inverter)
     return options, inverter, relay, now, minute
 
 
@@ -192,6 +209,9 @@ def _intent(command, inverter):
     _require(type(command.get("target_soc")) is int and command["target_soc"] in (0, 100), "invalid_target")
     _require(type(command.get("freeze")) is bool and type(command.get("ev_on")) is bool, "invalid_command_boolean")
     _require(not command["freeze"] or command["target_soc"] == 0, "invalid_hold_target")
+    if getattr(inverter, "inverter_type", None) == "DeyeSolarman":
+        _require(command["freeze"] is False and command.get("power_w") is None, "unsupported_template_command")
+        return command["target_soc"], False, None
     cap = min(inverter.deye_local.settings["max_discharge_power_w"], inverter.battery_rate_max_discharge * MINUTE_WATT)
     power = command.get("power_w")
     _require(_finite(cap, 1), "invalid_discharge_cap")
@@ -313,7 +333,12 @@ def execute_joint_site(base):
         # No call to the adapter directly: this is the same native entry point
         # whose optional DEYE implementation owns all physical service writes.
         attempted = True
-        observed = inverter.adjust_charge_immediate(target, freeze=freeze, discharge_power_w=power)
+        if options["control_model"] == "deye_solarman":
+            from joint_site_template import dispatch_template
+
+            observed = dispatch_template(inverter, target)
+        else:
+            observed = inverter.adjust_charge_immediate(target, freeze=freeze, discharge_power_w=power)
         _require(observed is True, "inverter_command_not_observed")
         # Permission, clocks and protection inputs can change while writes settle.
         options, inverter, relay, now, minute = _evidence(base)

apps/predbat/joint_site_native.py

@@ -145,16 +145,16 @@ def _options(base):
     if mode not in ("shadow", "apply"):
         raise ValueError("joint_site mode must be off, shadow or apply")
     control_model = settings.get("control_model", "native")
-    if control_model not in ("native", "deye_tou"):
+    if control_model not in ("native", "deye_tou", "deye_solarman"):
         raise ValueError("Unsupported joint-site control model")
     if type(settings.get("allow_stale_shadow_soc", False)) is not bool:
         raise ValueError("allow_stale_shadow_soc must be boolean")
     if type(settings.get("apply_verified", False)) is not bool:
         raise ValueError("apply_verified must be boolean")
     if type(settings.get("ev_executor_lease_verified", False)) is not bool:
         raise ValueError("ev_executor_lease_verified must be boolean")
-    if mode == "apply" and (control_model != "deye_tou" or settings.get("allow_stale_shadow_soc", False)):
-        raise ValueError("Apply requires real DEYE controls and cannot use stale SOC assumptions")
+    if mode == "apply" and (control_model not in ("deye_tou", "deye_solarman") or settings.get("allow_stale_shadow_soc", False)):
+        raise ValueError("Apply requires supported real controls and cannot use stale SOC assumptions")
     for name in ("connection_observed_at_entity", "soc_observed_at_entity"):
         if settings.get(name) is not None and (not isinstance(settings[name], str) or not settings[name]):
             raise ValueError("Observation timestamp entities must be non-empty strings")
@@ -366,8 +366,26 @@ def _collect_blockers(base, options):
         blockers.append("iboost_unsupported")
     if not base.car_energy_reported_load:
         blockers.append("load_not_reported")
-    if not base.set_charge_window or not base.set_charge_freeze or base.set_charge_freeze_only or base.set_charge_low_power:
+    template_model = options and options["control_model"] == "deye_solarman"
+    if not base.set_charge_window or (not template_model and not base.set_charge_freeze) or base.set_charge_freeze_only or base.set_charge_low_power:
         blockers.append("freeze_required")
+    if template_model:
+        from joint_site_template import validate_template
+
+        inverters = getattr(base, "inverters", None)
+        if not isinstance(inverters, list) or len(inverters) != 1:
+            blockers.append("single_template_inverter_required")
+        else:
+            try:
+                validate_template(inverters[0])
+            except (ValueError, TypeError, KeyError, AttributeError):
+                blockers.append("template_contract_invalid")
+        if base.export_limit != 0 or not base.set_discharge_during_charge:
+            blockers.append("template_dispatch_model_mismatch")
+        if base.args.get("charge_freeze_service") not in (None, "", []):
+            # The native target-equals-current branch can call this even without
+            # an explicit freeze. The narrow template must not acquire that writer.
+            blockers.append("template_freeze_service_unsupported")
     if not base.car_charging_from_battery:
         blockers.append("ev_from_battery_required")
     if getattr(base, "manual_charge_times", None):
@@ -459,10 +477,18 @@ def build_candidate(base, options):
         initial_off_minutes=options["initial_off_minutes"],
     )
     site = SiteConfig(options["import_limit_w"] / 1000, base.export_limit * 60, options["load_cap"])
-    result = plan(slots, ev, battery, site, PlannerConfig(beam_width=options["beam"], allow_limited_discharge=options["control_model"] == "deye_tou"))
+    modes = ("demand", "charge") if options["control_model"] == "deye_solarman" else ("demand", "hold", "charge")
+    result = plan(slots, ev, battery, site, PlannerConfig(beam_width=options["beam"], allow_limited_discharge=options["control_model"] == "deye_tou", allowed_battery_modes=modes))
     return result, slots
 
 
+def _validate_candidate_modes(options, result):
+    """Reject unsupported solver actions before replay/staging, never reinterpret a hold as demand."""
+    if options["control_model"] == "deye_solarman":
+        if any(action.battery_mode not in ("demand", "charge") or getattr(action, "discharge_limit_kw", None) is not None for action in result.actions):
+            raise UnsupportedJointSite("unsupported_template_candidate_controls")
+
+
 def _native_windows(base, result, slots):
     """Translate modes to native charge/freeze windows without altering executable lists."""
     if len(result.actions) != len(slots) or any(action.slot_index != slot.index for action, slot in zip(result.actions, slots)):
@@ -641,7 +667,7 @@ def evaluate_joint_site(base, publish=True):
             },
             "ready_to_activate": False,
             "control_model": options["control_model"],
-            "counterfactual_controls": options["control_model"] != "native" and options["mode"] == "shadow",
+            "counterfactual_controls": options["control_model"] == "deye_tou" and options["mode"] == "shadow",
             "requested_mode": options["mode"],
             "activation_requirements": execution_requirements,
             "manual_domestic_floor_slots": len(base.all_active_keep),
@@ -694,6 +720,16 @@ def evaluate_joint_site(base, publish=True):
         _forecast_inputs(model_base, options)
         stage = "solver"
         result, slots = build_candidate(model_base, options)
+        if options["control_model"] == "deye_solarman":
+            report["supported_battery_modes"] = ["demand", "charge"]
+            report["search_beam_attempts"] = [{"beam_width": options["beam"], "solver_status": result.status}]
+            if not result.actions and options["beam"] < 512:
+                # A bounded search miss is not infeasibility. One explicit, reported
+                # retry retains the exact capabilities/inputs and stays within the cap.
+                retry_options = {**options, "beam": 512}
+                result, slots = build_candidate(model_base, retry_options)
+                report["search_beam_attempts"].append({"beam_width": 512, "solver_status": result.status})
+            _validate_candidate_modes(options, result)
         if not result.actions:
             report["status"] = "conditional_no_plan_found" if stale_hypothesis else "no_plan_found"
             report["execution_blockers"] = list(blockers)
@@ -704,12 +740,13 @@ def evaluate_joint_site(base, publish=True):
             windows = _native_windows(model_base, result, slots)
             checks = {"nominal": _replay(model_base, options, result, slots, windows, 0), "pv10": _replay(model_base, options, result, slots, windows, 1)}
             attempts = [{"forecast_basis": "nominal", "solver_status": result.status, "native_passed": all(check["passed"] for check in checks.values())}]
-            if options["control_model"] == "deye_tou" and not attempts[0]["native_passed"]:
+            if options["control_model"] in ("deye_tou", "deye_solarman") and not attempts[0]["native_passed"]:
                 # One bounded retry uses the forecast envelope and PV10 charge
                 # delivery assumption. Native nominal AND PV10 still validate
                 # the same full-rate commands; this is not charge modulation.
                 retry_options = {**options, "forecast_basis": "envelope"}
                 retry_result, retry_slots = build_candidate(model_base, retry_options)
+                _validate_candidate_modes(options, retry_result)
                 retry_checks = {}
                 if retry_result.actions:
                     retry_windows = _native_windows(model_base, retry_result, retry_slots)