Lattice 4/4: dormant brand adapters (Solis/Fox/Solax)

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

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

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.