diff --git a/apps/predbat/execute.py b/apps/predbat/execute.py index 5e33b77d2..dd0108d19 100644 --- a/apps/predbat/execute.py +++ b/apps/predbat/execute.py @@ -21,6 +21,7 @@ from predbat_metrics import metrics from inverter import Inverter import time +import asyncio """ Execute Predbat plan @@ -35,6 +36,49 @@ class Execute: adjustment, and multi-inverter balancing. """ + def lattice_express_control(self, serial, capability, value): + """Site control plane: express a control intent and let the Lattice projection fulfil it. + + The inverter layer calls this INSTEAD of writing the inverter itself, handing up the intent + (capability + which battery + target). If the projection is enabled and can route it, the + intent is dispatched (async — best provider, with gateway/cloud fallback) and True returned. + ANY problem returns False so the caller performs its normal per-inverter write: cross-provider + routing lives here at the site layer, but control is never lost if Lattice cannot act. + """ + try: + registry = getattr(self, "components", None) + component = registry.get_component("lattice") if registry else None + projection = getattr(component, "projection", None) + if projection is None or not serial or serial == "Unknown": + return False + if not projection.would_handle(capability, "battery-system", serial): + return False + task = asyncio.get_running_loop().create_task(projection.apply(capability, "battery-system", serial, int(value))) + # Hold a strong reference (the loop only keeps a weak one — a fire-and-forget task can be + # GC'd mid-flight) and surface async failures, since the inverter has already skipped its + # own write. NOTE: a provider dropping between would_handle() and apply() means a one-cycle + # skip that self-corrects on the next ~5-min planner pass — it is logged, not silently lost. + tasks = getattr(self, "_lattice_tasks", None) + if tasks is None: + tasks = self._lattice_tasks = set() + tasks.add(task) + + def _on_done(finished, cap=capability, sn=serial): + tasks.discard(finished) + try: + result = finished.result() + if result is not None and not getattr(result, "ok", True): + self.log("Warn: Lattice: {} for {} NOT applied ({}) — will retry next cycle".format(cap, sn, getattr(result, "reason", ""))) + except Exception as exc: + self.log("Warn: Lattice: {} apply task failed for {}: {}".format(cap, sn, exc)) + + task.add_done_callback(_on_done) + self.log("Lattice: routed {} for {} = {}W via projection".format(capability, serial, int(value))) + return True + except Exception as e: + self.log("Warn: Lattice: {} handoff failed for {} ({}); using normal write".format(capability, serial, e)) + return False + def execute_plan(self): status_extra = "" # extra status text added to Predbat notifications status_hold_car = "" # car hold status text diff --git a/apps/predbat/inverter.py b/apps/predbat/inverter.py index 5b07c4fc8..dbb0d8715 100644 --- a/apps/predbat/inverter.py +++ b/apps/predbat/inverter.py @@ -189,6 +189,13 @@ def __init__(self, base, id=0, quiet=False, rest_postCommand=None, rest_getData= self.inverter_type = self.base.get_arg("inverter_type", "GE", indirect=False, index=self.id) + # Lattice node identity: producers (gateway/cloud) publish the real serial per index so the + # projection can correlate this inverter with its merged-graph node. REST "GE" overrides this + # later from rest_data; for GWMQTT/cloud it is the only serial source. + lattice_serial = self.base.get_arg("inverter_serial", default=None, indirect=False, index=self.id) + if lattice_serial: + self.serial_number = str(lattice_serial) + # Read user defined inverter type if "inverter" in self.base.args: if self.inverter_type not in INVERTER_DEF: @@ -1735,6 +1742,8 @@ def adjust_reserve(self, reserve): self.base.log("Inverter {} Current Reserve is {}% and new target is {}%".format(self.id, dp0(current_reserve), dp0(reserve))) if self.rest_data: self.rest_setReserve(reserve) + elif self._lattice_expresses("reserve_soc", reserve): + pass # intent handed to the site control plane (Lattice owns provider + fallback) else: self.write_and_poll_value("reserve", self.base.get_arg("reserve", indirect=False, index=self.id, required_unit="%"), reserve) if self.base.set_inverter_notify: @@ -1783,6 +1792,16 @@ def get_current_charge_rate(self): return current_rate + def _lattice_expresses(self, capability, value): + """Hand this control intent up to the site control plane (execute.lattice_express_control). + + Returns True if the Lattice projection took ownership of the write (best provider + fallback); + False (incl. when the host has no site control plane) so the caller does its normal write. + Cross-provider routing lives at the site layer, not in this single-inverter class. + """ + express = getattr(self.base, "lattice_express_control", None) + return bool(express and express(self.serial_number, capability, value)) + def adjust_charge_rate(self, new_rate, notify=True): """ Adjust charging rate @@ -1811,6 +1830,8 @@ def adjust_charge_rate(self, new_rate, notify=True): self.base.log("Inverter {} current charge rate is {}W and new target is {}W".format(self.id, current_rate, new_rate)) if self.rest_data: self.rest_setChargeRate(new_rate) + elif self._lattice_expresses("charge_rate", new_rate): + pass # intent handed to the site control plane (Lattice owns provider + fallback) else: if "charge_rate" in self.base.args: self.write_and_poll_value("charge_rate", self.base.get_arg("charge_rate", indirect=False, index=self.id, required_unit="W"), new_rate, fuzzy=(self.battery_rate_max_charge * MINUTE_WATT / 20), required_unit="W") @@ -1849,6 +1870,8 @@ def adjust_discharge_rate(self, new_rate, notify=True): self.base.log("Inverter {} current discharge rate is {}W and new target is {}W".format(self.id, current_rate, new_rate)) if self.rest_data: self.rest_setDischargeRate(new_rate) + elif self._lattice_expresses("discharge_rate", new_rate): + pass # intent handed to the site control plane (Lattice owns provider + fallback) else: if "discharge_rate" in self.base.args: self.write_and_poll_value("discharge_rate", self.base.get_arg("discharge_rate", indirect=False, index=self.id), new_rate, fuzzy=(self.battery_rate_max_discharge * MINUTE_WATT / 20), required_unit="W") @@ -1896,6 +1919,8 @@ def adjust_battery_target(self, soc, isCharging=False, isExporting=False): self.current_charge_limit = soc if self.rest_data: self.rest_setChargeTarget(soc) + elif self._lattice_expresses("target_soc", soc): + pass # intent handed to the site control plane (Lattice owns provider + fallback) else: self.write_and_poll_value("charge_limit", self.base.get_arg("charge_limit", indirect=False, index=self.id, required_unit="%"), soc)