Fix settings writes: use FC16, correct pymodbus unit kwarg, surface errors

- Deye hybrids reject FC06; all register writes now go through FC16
- Detect device_id/slave/unit keyword from pymodbus signature (3.10+ renamed it)
- Serialize writes with polling via the coordinator lock
- Keep freshly written values for 30 s so a stale poll can't revert the UI
- Raise HomeAssistantError when the inverter refuses a write
- Fix double address-offset on reads; fix RTU-over-TCP client construction
- Number entities show unknown instead of 0 without a read-back
- Coordinator support for computed (sum) sensors

Author: rusty-bit

custom_components/deye_modbus/coordinator.py

@@ -5,11 +5,16 @@
 Unlike some vendors there is no separate "read once" settings space here, so
 every mapped register is polled each cycle. Contiguous registers are batched
 into single Modbus reads to keep the poll cheap.
+
+Writes always use function code 0x10 (Write Multiple Registers): Deye
+inverters reject or silently ignore FC06 (Write Single Register).
 """
 from __future__ import annotations
 import asyncio
+import inspect
 import logging
-from dataclasses import dataclass
+import time
+from dataclasses import dataclass, field
 from datetime import timedelta
 from typing import Any, Dict, List
 
@@ -19,11 +24,38 @@
 _LOGGER = logging.getLogger(__name__)
 
 try:
-    from pymodbus.client import AsyncModbusTcpClient, AsyncModbusSerialClient
+    from pymodbus.client import AsyncModbusTcpClient
 except Exception as exc:  # pragma: no cover
     _LOGGER.error("pymodbus import failed: %s", exc)
     raise
 
+try:  # pymodbus >= 3.7
+    from pymodbus import FramerType as _FramerType
+    _RTU_FRAMER: Any = _FramerType.RTU
+except Exception:  # noqa: BLE001
+    try:
+        from pymodbus.framer import ModbusRtuFramer as _RTU_FRAMER  # type: ignore
+    except Exception:  # noqa: BLE001
+        _RTU_FRAMER = None
+
+# How long a freshly written value is trusted over a stale poll result.
+PENDING_WRITE_SECONDS = 30
+
+
+def _unit_kw(method) -> str | None:
+    """Return the keyword pymodbus uses for the Modbus unit on this version.
+
+    pymodbus >= 3.10: device_id, 3.0-3.9: slave, 2.x: unit.
+    """
+    try:
+        params = inspect.signature(method).parameters
+    except (TypeError, ValueError):
+        return None
+    for kw in ("device_id", "slave", "unit"):
+        if kw in params:
+            return kw
+    return None
+
 
 @dataclass
 class RegisterDef:
@@ -46,6 +78,8 @@ class RegisterDef:
     icon: str | None = None
     enabled_default: bool = True             # entity_registry_enabled_default
     precision: int | None = None             # suggested_display_precision (decimals)
+    compute: str | None = None               # derived value: "sum" (no Modbus read)
+    sources: list[str] = field(default_factory=list)  # unique_ids feeding `compute`
 
 
 class DeyeModbusCoordinator(DataUpdateCoordinator[dict[str, Any]]):
@@ -74,51 +108,73 @@ def __init__(
         self._client = None
         self._lock = asyncio.Lock()
         self._addr_off = int(address_offset or 0)
+        self._computed = [r for r in registers if r.compute]
+        self._polled = [r for r in registers if not r.compute]
+        # address -> (raw word, expiry) for values we just wrote
+        self._pending: dict[int, tuple[int, float]] = {}
 
     def _addr(self, addr: int) -> int:
         return int(addr) - self._addr_off if self._addr_off else int(addr)
 
     # ------------------------------------------------------------------ client
     async def _ensure_client(self):
         if self._client is None:
+            kwargs: dict[str, Any] = {"port": self._port, "timeout": 5}
             if self._transport == "rtutcp":
-                url = f"socket://{self._host}:{self._port}"
-                params = {
-                    "method": "rtu", "port": url,
-                    "baudrate": int(self._serial_params.get("baudrate", 9600)),
-                    "bytesize": int(self._serial_params.get("bytesize", 8)),
-                    "parity": str(self._serial_params.get("parity", "N")),
-                    "stopbits": int(self._serial_params.get("stopbits", 1)),
-                    "timeout": 5,
-                }
-                self._client = AsyncModbusSerialClient(**params)
-            else:
-                try:
-                    self._client = AsyncModbusTcpClient(self._host, port=self._port, timeout=5)
-                except TypeError:
-                    self._client = AsyncModbusTcpClient(self._host, port=self._port)
+                # Modbus RTU frames tunnelled over a raw TCP socket.
+                if _RTU_FRAMER is None:
+                    raise UpdateFailed("This pymodbus version has no RTU framer")
+                kwargs["framer"] = _RTU_FRAMER
+            self._client = AsyncModbusTcpClient(self._host, **kwargs)
         if not bool(getattr(self._client, "connected", False)):
             try:
                 await self._client.connect()
             except Exception as e:  # noqa: BLE001
                 raise UpdateFailed(f"Modbus connect failed: {e}") from e
+            if not bool(getattr(self._client, "connected", False)):
+                raise UpdateFailed(f"Modbus connect to {self._host}:{self._port} failed")
         return self._client
 
+    def _unit_kwargs(self, method) -> dict[str, int]:
+        kw = _unit_kw(method)
+        return {kw: self._unit_id} if kw else {}
+
     # ------------------------------------------------------------------ polling
     async def _async_update_data(self) -> dict[str, Any]:
         async with self._lock:
             try:
                 result: dict[str, Any] = {}
-                inputs = [r for r in self._registers if r.register_type == "input"]
-                holdings = [r for r in self._registers if r.register_type != "input"]
+                inputs = [r for r in self._polled if r.register_type == "input"]
+                holdings = [r for r in self._polled if r.register_type != "input"]
                 if holdings:
                     await self._read_grouped(holdings, result, self._read_holding)
                 if inputs:
                     await self._read_grouped(inputs, result, self._read_input)
+                self._apply_computed(result)
                 return result
+            except UpdateFailed:
+                raise
             except Exception as err:  # noqa: BLE001
+                await self._reset_client()
                 raise UpdateFailed(err) from err
 
+    def _apply_computed(self, out: dict[str, Any]) -> None:
+        for r in self._computed:
+            if r.compute == "sum":
+                vals = [out.get(src) for src in r.sources]
+                vals = [v for v in vals if v is not None]
+                out[r.unique_id] = (sum(vals) * r.scale) if vals else None
+            else:
+                out[r.unique_id] = None
+
+    async def _reset_client(self) -> None:
+        if self._client is not None:
+            try:
+                self._client.close()
+            except Exception:  # noqa: BLE001
+                pass
+            self._client = None
+
     async def _read_grouped(self, regs: list[RegisterDef], out: dict[str, Any], fn):
         regs = sorted(regs, key=lambda r: r.address)
         GAP = 8          # merge registers within this many words into one read
@@ -139,7 +195,20 @@ async def _read_grouped(self, regs: list[RegisterDef], out: dict[str, Any], fn):
             await self._read_window(fn, out, start, end, acc)
 
     async def _read_window(self, fn, out, start, end, regs):
-        raw = await fn(self._addr(start), end - start)
+        raw = await fn(start, end - start)
+        if raw is None:
+            _LOGGER.debug("Read of registers %s-%s failed", start, end - 1)
+        elif fn == self._read_holding and self._pending:
+            raw = list(raw)
+            now = time.monotonic()
+            for addr, (val, expiry) in list(self._pending.items()):
+                if not start <= addr < end:
+                    continue
+                idx = addr - start
+                if raw[idx] == val or now > expiry:
+                    self._pending.pop(addr, None)   # inverter caught up (or gave up)
+                else:
+                    raw[idx] = val                  # stale poll - keep what we wrote
         for r in regs:
             out[r.unique_id] = self._decode(r, raw, start)
 
@@ -175,19 +244,10 @@ async def _call_read(self, method_name, address, count):
         client = await self._ensure_client()
         method = getattr(client, method_name)
         a = self._addr(address)
-        # pymodbus 3.x -> slave=, pymodbus 2.x -> unit=, older -> positional.
-        for kwargs in (
-            {"count": count, "slave": self._unit_id},
-            {"count": count, "unit": self._unit_id},
-            {"count": count},
-        ):
-            try:
-                return await method(a, **kwargs)
-            except TypeError:
-                continue
         try:
-            return await method(a, count)
-        except TypeError:
+            return await method(a, count=count, **self._unit_kwargs(method))
+        except Exception as err:  # noqa: BLE001
+            _LOGGER.debug("%s(%s, %s) raised %s", method_name, a, count, err)
             return None
 
     async def _read_input(self, address, count):
@@ -199,49 +259,52 @@ async def _read_holding(self, address, count):
         return None if rr is None or (getattr(rr, "isError", None) and rr.isError()) else getattr(rr, "registers", None)
 
     # ------------------------------------------------------------------ writes
-    async def write_single_register(self, address: int, value: int) -> bool:
+    async def write_multiple_registers(self, address: int, values: list[int]) -> bool:
+        """Write with FC16 - the only write function Deye inverters accept."""
+        vals = [int(v) & 0xFFFF for v in values]
         a = self._addr(address)
-        client = await self._ensure_client()
-        wire = int(value) & 0xFFFF
-        for variant in (
-            lambda: client.write_register(a, wire, slave=self._unit_id),
-            lambda: client.write_register(a, wire, self._unit_id),
-            lambda: client.write_register(address=a, value=wire, unit=self._unit_id),
-            lambda: client.write_register(a, wire),
-        ):
+        async with self._lock:
             try:
-                rr = await variant()
-                if getattr(rr, "isError", lambda: False)():
-                    continue
-                await self.async_request_refresh()
-                return True
-            except TypeError:
-                continue
-            except Exception:  # noqa: BLE001
+                client = await self._ensure_client()
+                method = client.write_registers
+                rr = await method(a, vals, **self._unit_kwargs(method))
+            except Exception as err:  # noqa: BLE001
+                _LOGGER.error("Write %s to register %s failed: %s", vals, address, err)
+                await self._reset_client()
+                return False
+            if rr is None or (getattr(rr, "isError", None) and rr.isError()):
+                _LOGGER.error("Inverter rejected write %s to register %s: %s", vals, address, rr)
                 return False
-        return False
+            expiry = time.monotonic() + PENDING_WRITE_SECONDS
+            for i, v in enumerate(vals):
+                self._pending[address + i] = (v, expiry)
+        _LOGGER.debug("Wrote %s to register %s", vals, address)
+        self._patch_data(address, vals)
+        await self.async_request_refresh()
+        return True
 
-    async def write_multiple_registers(self, address: int, values: list[int]) -> bool:
-        a = self._addr(address)
-        client = await self._ensure_client()
-        vals = [int(v) & 0xFFFF for v in values]
-        for variant in (
-            lambda: client.write_registers(a, vals, slave=self._unit_id),
-            lambda: client.write_registers(a, vals, self._unit_id),
-            lambda: client.write_registers(address=a, values=vals, unit=self._unit_id),
-            lambda: client.write_registers(a, vals),
-        ):
-            try:
-                rr = await variant()
-                if getattr(rr, "isError", lambda: False)():
-                    continue
-                await self.async_request_refresh()
-                return True
-            except TypeError:
+    async def write_single_register(self, address: int, value: int) -> bool:
+        # Deliberately FC16 with one value, not FC06.
+        return await self.write_multiple_registers(address, [value])
+
+    def _patch_data(self, address: int, vals: list[int]) -> None:
+        """Reflect a successful write in coordinator data immediately."""
+        if not self.data:
+            return
+        span = range(address, address + len(vals))
+        data = dict(self.data)
+        changed = False
+        for r in self._polled:
+            if r.register_type == "input" or r.address not in span:
                 continue
-            except Exception:  # noqa: BLE001
-                return False
-        return False
+            if r.address + r.count - 1 not in span:
+                continue
+            words = vals[r.address - address:r.address - address + r.count]
+            data[r.unique_id] = self._decode(r, words, r.address)
+            changed = True
+        if changed:
+            self._apply_computed(data)
+            self.async_set_updated_data(data)
 
     async def write_u32(self, base_address: int, value: int, word_order: str = "high_low") -> bool:
         v = int(value) & 0xFFFFFFFF
@@ -251,28 +314,25 @@ async def write_u32(self, base_address: int, value: int, word_order: str = "high
 
     async def write_coil(self, address: int, value: int) -> bool:
         a = self._addr(address)
-        client = await self._ensure_client()
-        for variant in (
-            lambda: client.write_coil(a, bool(value), slave=self._unit_id),
-            lambda: client.write_coil(a, bool(value), self._unit_id),
-            lambda: client.write_coil(address=a, value=bool(value), unit=self._unit_id),
-            lambda: client.write_coil(a, bool(value)),
-        ):
+        async with self._lock:
             try:
-                rr = await variant()
-                if getattr(rr, "isError", lambda: False)():
-                    continue
-                await self.async_request_refresh()
-                return True
-            except TypeError:
-                continue
-            except Exception:  # noqa: BLE001
+                client = await self._ensure_client()
+                method = client.write_coil
+                rr = await method(a, bool(value), **self._unit_kwargs(method))
+            except Exception as err:  # noqa: BLE001
+                _LOGGER.error("Write coil %s failed: %s", address, err)
                 return False
-        return False
+        if rr is None or (getattr(rr, "isError", None) and rr.isError()):
+            _LOGGER.error("Inverter rejected coil write %s: %s", address, rr)
+            return False
+        await self.async_request_refresh()
+        return True
 
     async def async_close(self):
         try:
             if self._client:
-                await self._client.close()
+                res = self._client.close()
+                if inspect.isawaitable(res):
+                    await res
         except Exception:  # noqa: BLE001
             pass

custom_components/deye_modbus/number.py

@@ -6,6 +6,7 @@
 from homeassistant.components.number import NumberEntity, NumberMode
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers.entity import EntityCategory
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
@@ -62,11 +63,11 @@ def __init__(self, coordinator: DeyeModbusCoordinator, entry: ConfigEntry, cfg:
         self._signed = bool(cfg.get("signed", False))
         self._read_uid: Optional[str] = cfg.get("read_unique_id")
         self._read_factor = float(cfg.get("read_factor", 1.0))
-        self._value: float = max(self._attr_native_min_value, min(self._attr_native_max_value, 0.0))
+        self._value: float | None = None
         self._sync_from_sensor()
 
     @property
-    def native_value(self) -> float:
+    def native_value(self) -> float | None:
         return self._value
 
     async def async_set_native_value(self, value: float) -> None:
@@ -77,10 +78,10 @@ async def async_set_native_value(self, value: float) -> None:
         if self._signed and raw < 0:
             raw &= 0xFFFF
         ok = await self._coordinator.write_single_register(self._address, raw)
-        if ok:
-            self._value = v
-            self.async_write_ha_state()
-            await self._coordinator.async_request_refresh()
+        if not ok:
+            raise HomeAssistantError(f"Deye inverter did not accept {self.name} = {v}")
+        self._value = v
+        self.async_write_ha_state()
 
     def _sync_from_sensor(self) -> None:
         if not self._read_uid:
@@ -131,7 +132,14 @@ def native_value(self) -> float:
     async def async_set_native_value(self, value: float) -> None:
         v = int(max(self._attr_native_min_value, min(self._attr_native_max_value, round(value))))
         ok = await self._coordinator.write_u32(self._base, v, self._order)
-        if ok:
-            self._value = float(v)
-            self.async_write_ha_state()
-            await self._coordinator.async_request_refresh()
+        if not ok:
+            raise HomeAssistantError(f"Deye inverter did not accept {self.name} = {v}")
+        self._value = float(v)
+        self.async_write_ha_state()
+
+    def _handle_coordinator_update(self) -> None:
+        rb = self._cfg.get("read_unique_id")
+        raw = (self.coordinator.data or {}).get(rb) if rb else None
+        if raw is not None:
+            self._value = float(raw)
+        self.async_write_ha_state()