From 134553dd9895d8ca24908f800ebea0bc1ae02a39 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Thu, 18 Jun 2026 09:53:53 +0100 Subject: [PATCH 1/3] feat(lattice): Solis producer + dormant control via write_cid SolisAPI.lattice_control maps reserve/target/charge/discharge to the proven write_cid (CIDs 157/ 160/7224/7226, amps at nominal_voltage), case-insensitive serial. Kept controllable=() (dormant) until live verification of voltage scaling + v1/v2 CID. Co-Authored-By: Claude Opus 4.8 (1M context) --- apps/predbat/solis.py | 36 ++++++++++++++++++++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/apps/predbat/solis.py b/apps/predbat/solis.py index c792490b2..b358083e6 100644 --- a/apps/predbat/solis.py +++ b/apps/predbat/solis.py @@ -578,6 +578,42 @@ async def decode_time_windows_v2(self, inverter_sn): self.log("Solis API: Decoded time windows v2 for {}: {}".format(inverter_sn, result)) # Debug log return result + def lattice_fragment(self): + """Export a Lattice fragment for the inverter(s) controllable via the Solis cloud API. + + Producer only: emits each inverter serial on a low-preference cloud access path so it + joins the merged site graph. Cloud control execution (lattice_control) is a follow-up. + """ + from lattice import inverter_fragment + + inverters = [{"serial": str(serial), "device_type": "solis"} for serial in getattr(self, "inverter_sn", []) or []] + # DORMANT: lattice_control below is implemented but controllable=() until live hardware + # verification. Enable order after verification: ("reserve_soc",) [safe, CID 157] then add + # charge_rate/discharge_rate (verify nominal_voltage scaling) + target_soc (verify v1/v2 CID). + return inverter_fragment(inverters, provider="solis-cloud", name="Solis Cloud", transport="https", preference=1, locality="cloud", controllable=()) + + async def lattice_control(self, node_id, capability, value): + """Execute a Lattice-resolved control write via the Solis Cloud CID API (delegates to write_cid). + + Reuses the proven write_cid path: reserve/target are SOC %, charge/discharge are converted to + amps at nominal_voltage. Returns False on unknown serial/capability so the projection falls back. + DORMANT until enabled via the fragment's controllable set (see lattice_fragment) after live + verification — current amps scaling assumes a 48V pack and target uses the v1 force-charge CID. + """ + sn = self.find_inverter_by_sn(node_id) + if not sn: + return False + iv = int(value) + if capability == "reserve_soc": + return bool(await self.write_cid(sn, SOLIS_CID_BATTERY_RESERVE_SOC, iv, field_description="reserve soc (lattice)")) + if capability == "target_soc": + return bool(await self.write_cid(sn, SOLIS_CID_BATTERY_FORCE_CHARGE_SOC, iv, field_description="target soc (lattice)")) + if capability in ("charge_rate", "discharge_rate"): + amps = int(value / self.nominal_voltage) if self.nominal_voltage else 0 + cid = SOLIS_CID_BATTERY_MAX_CHARGE_CURRENT if capability == "charge_rate" else SOLIS_CID_BATTERY_MAX_DISCHARGE_CURRENT + return bool(await self.write_cid(sn, cid, amps, field_description="{} (lattice)".format(capability))) + return False + async def reset_charge_windows_if_needed(self, inverter_sn): """ Predbat only uses 1 slot so disable the others to avoid conflicts. From 9e486ec60994e4dde1b39c1436d70dccd2542535 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Thu, 18 Jun 2026 09:53:54 +0100 Subject: [PATCH 2/3] feat(lattice): Fox producer + dormant control via schedule handler FoxAPI.lattice_fragment emits real deviceSN (device_list is dicts); lattice_control delegates to the proven write_battery_schedule_event with token-correct entity ids. Kept controllable=() (dormant) until live verification of the schedule read-modify-write cache-sync. Co-Authored-By: Claude Opus 4.8 (1M context) --- apps/predbat/fox.py | 41 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/apps/predbat/fox.py b/apps/predbat/fox.py index 57878aaa8..12cc50090 100644 --- a/apps/predbat/fox.py +++ b/apps/predbat/fox.py @@ -1618,6 +1618,47 @@ async def publish_data(self): self.dashboard_item(entity_id, state=state, attributes=attributes, app="fox") + def lattice_fragment(self): + """Export a Lattice fragment for the inverter(s) controllable via the FoxESS cloud API. + + Producer only: emits each serial on a low-preference cloud access path so it joins the + merged site graph (and gains gateway->cloud fallback where a hub also sees it). Cloud + control execution (lattice_control) is a per-brand follow-up. + """ + from lattice import inverter_fragment + + # device_list is a list of device DICTS (each carries deviceSN), NOT serial strings + serials = [d.get("deviceSN") for d in (getattr(self, "device_list", None) or []) if isinstance(d, dict) and d.get("deviceSN")] + serials = serials or list(getattr(self, "device_detail", None) or {}) # fallback: device_detail keys are serials + inverters = [{"serial": str(s), "device_type": "fox"} for s in serials] + # DORMANT: lattice_control below is implemented but controllable=() until live hardware + # verification (all four ride the schedule read-modify-write; confirm a 2nd write doesn't + # undo the first if local_schedule cache is stale). Then flip controllable to the verified set. + return inverter_fragment(inverters, provider="fox-cloud", name="FoxESS Cloud", transport="https", preference=1, locality="cloud", controllable=()) + + async def lattice_control(self, node_id, capability, value): + """Execute a Lattice-resolved control write by delegating to the proven schedule handler. + + Reuses write_battery_schedule_event (the same path HA number events use): it parses the + entity id for serial/direction/field and applies the schedule. We reconstruct a token-correct + entity id per capability rather than re-implement schedule manipulation. Returns False on + unknown serial/capability so the projection falls back. DORMANT until verified (see fragment). + """ + suffix = {"charge_rate": "charge_power", "discharge_rate": "discharge_power", "target_soc": "charge_soc", "reserve_soc": "reserve"}.get(capability) + if suffix is None: + return False + known_serials = set(getattr(self, "device_detail", None) or {}) # device_detail keys are serials + known_serials |= {d.get("deviceSN") for d in (getattr(self, "device_list", None) or []) if isinstance(d, dict)} + if not any(s and str(s).lower() == str(node_id).lower() for s in known_serials): + return False + entity_id = "number.{}_fox_{}_{}".format(self.prefix, node_id, suffix) + try: + await self.write_battery_schedule_event(entity_id, int(value)) + return True + except Exception as e: + self.log("Fox: lattice_control failed for {} {}: {}".format(node_id, capability, e)) + return False + async def write_setting_from_event(self, entity_id, value, is_number=False): """ Handle write events From 6076d6a7c01ddbaf661d6fa9b7798802620dc99b Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Thu, 18 Jun 2026 09:53:56 +0100 Subject: [PATCH 3/3] =?UTF-8?q?feat(lattice):=20Solax=20producer=20(read-o?= =?UTF-8?q?nly)=20=E2=80=94=20control=20deferred?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SolaxAPI.lattice_fragment is read-only: control is deliberately deferred because soc_target_control_mode couples (target, power, mode) in one call and node id is plant_id (never merges with a serial). Documented for the coupled-control spec work (lattice-spec#3). Co-Authored-By: Claude Opus 4.8 (1M context) --- apps/predbat/solax.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/apps/predbat/solax.py b/apps/predbat/solax.py index fd349467f..dbc6987a5 100644 --- a/apps/predbat/solax.py +++ b/apps/predbat/solax.py @@ -453,6 +453,22 @@ async def switch_event(self, entity_id, service): if "_battery_schedule_" in entity_id: await self.write_battery_schedule_event(entity_id, service) + def lattice_fragment(self): + """Export a Lattice fragment for the inverter(s) controllable via the Solax cloud API. + + Producer only: emits each plant id on a low-preference cloud access path so it joins the + merged site graph. Cloud control execution (lattice_control) is a per-brand follow-up. + """ + from lattice import inverter_fragment + + inverters = [{"serial": str(plant), "device_type": "solax"} for plant in getattr(self, "plant_list", []) or []] + # READ-ONLY (no lattice_control). Solax control is DELIBERATELY DEFERRED — unlike Solis/Fox + # there is no clean delegate: soc_target_control_mode() COUPLES (target_soc, power, mode) in + # one call, so a single-capability write must reconstruct the other params from current state; + # it also needs an 8x-retry timeout wrap, and node id = plant_id (never merges with a serial, + # so no fallback). Requires careful design + live verification before wiring — not guessed here. + return inverter_fragment(inverters, provider="solax-cloud", name="Solax Cloud", transport="https", preference=1, locality="cloud", controllable=()) + async def write_battery_schedule_event(self, entity_id, value): """ Write a battery schedule based on an event