"""The inverter driver seam (SMA-2): how this add-on reads and steers ONE inverter. Before this, every place that read or steered an inverter talked to the GoodWe ESPHome entities itself (`hass.limits`, `hass.number`, `hass.set_number`, the SAFETY-28 census, selfconfig, validate), so a second brand meant touching all of them. Now each `Inverter` holds a `Driver`, and those places ask it. This card is a pure refactor: `GoodWeEsphomeDriver` does exactly what the code it replaced did, in the same order, with the same arguments. DECISIONS/SMA-2-driver-seam.md. THE SEAM IS NOT TCP-SHAPED. Nothing here assumes a connection, or one connection per host: a later FoxESS unit is RTU framing over a TCP socket, and its `actuator_id()` is host:port:unit_id, so two units behind one gateway are two actuators and one unit named twice is the SAFETY-28 refusal. WHAT STAYS OUTSIDE THE DRIVER, on purpose. The gate (SAFETY-28), the clamp (SAFETY-03), shadow mode and the write budget all sit above `write()` in `Controller._write_unit`, and the freshness gate sits above `read()`; a driver never sees a value those did not pass. A driver moves one number to or from one inverter and says whether it worked. """ from __future__ import annotations from dataclasses import dataclass, field from .control import PLANT_SETTLE_S from .freshness import BOUNDS # How long every write to one GoodWe inverter must have been REJECTED before # this controller stops claiming it can command it. Layer 1's own window: the # ESP32 commands 0 W after ~30 s without a refresh (`validate. # FIRMWARE_INGEST_VETO_S`). `inverter.LINK_STALE_S` is this number, re-exported. GOODWE_LINK_STALE_S = 30.0 # `driver` when an entry does not say: every install that exists today. DEFAULT_DRIVER = "goodwe_esphome" # The `driver` values the options accept. Only the GoodWe one for now; the # registry below may hold more (a test registers a fake), but this tuple is # what validate.py rule 9 lets through and what config.yaml's schema lists. SELECTABLE = (DEFAULT_DRIVER,) class DriverNotBound(RuntimeError): """A driver was asked to talk to the world before it was given a Home Assistant. A wiring bug, so it is loud rather than a quiet "no answer".""" @dataclass(frozen=True) class Probe: """What a read-only look at one inverter found. None means "did not answer".""" dev_min: float | None = None dev_max: float | None = None soc: float | None = None batt_w: float | None = None extra: dict = field(default_factory=dict) def census_key(actuator_id: str) -> str: """The form SAFETY-28 compares: stripped and lowercase, so a config cannot name one actuator twice by changing its case (`Number.X` vs `number.x`).""" return str(actuator_id or "").strip().lower() class Driver: """One inverter's I/O. Subclasses set the class attributes and the methods. Constructed once per roster entry, with no Home Assistant: `bind()` gives it one (the Controller does, and so does the settings panel, which has no Controller). Everything that talks to the world is a coroutine. """ #: the `driver` option value that selects this class name = "" #: the option key that names the actuator in an entry (the census reads it) writer_key = "" #: the brand word the per-driver sentences and device names use brand = "" #: per-driver sentences (SMA-10 and later). Empty until a caller needs one. words: dict = {} #: how long all-rejected writes last before the unit stops being live link_stale_s = GOODWE_LINK_STALE_S #: this driver's `freshness.Bound` for the battery power reading batt_bound = BOUNDS["batt"] #: the plant's settling time after a step, in s (control.PLANT_SETTLE_S) settle_s = PLANT_SETTLE_S # -- from the options ---------------------------------------------------- @classmethod def from_entry(cls, entry: dict) -> "Driver": raise NotImplementedError @classmethod def actuator_id_of(cls, entry: dict) -> str: """The actuator an options entry names, or "" for none. Pure: the SAFETY-28 census runs on options, before any driver exists.""" raise NotImplementedError # -- wiring -------------------------------------------------------------- def bind(self, hass) -> "Driver": """Give this driver its Home Assistant (or whatever it talks through).""" return self def bind_getter(self, getter) -> "Driver": """Like `bind()`, but looked up at each use: the Controller passes `lambda: self.hass`, so replacing `controller.hass` is seen at once.""" return self def bound(self) -> bool: """Has it been given what it talks through? A driver that talks through nothing the Controller owns is always bound.""" return True # -- what it is ---------------------------------------------------------- def configured(self) -> bool: """Is there any way to reach this inverter at all?""" raise NotImplementedError def actuator_id(self) -> str: """What this driver writes to. Two units with one id are one actuator.""" raise NotImplementedError def soc_source(self) -> str: """Where the pack level is read from; "" when nowhere. Status rows only.""" return "" def batt_source(self) -> str: """Where the battery power is read from; "" when nowhere.""" return "" # -- what it does -------------------------------------------------------- async def limits(self) -> tuple: """`(dev_min, dev_max)` in W the device accepts; (None, None) unread.""" raise NotImplementedError async def read(self) -> tuple: """`(soc, batt_w, extra)`. A value that did not arrive is None. `batt_w` is the RAW power, in the sign the controller uses; the freshness gate is the caller's.""" raise NotImplementedError async def write(self, w: float) -> bool: """Put `w` on the inverter. True only when it took it.""" raise NotImplementedError async def release(self) -> bool: """Stop steering this inverter. For a driver with nothing better, 0 W.""" return await self.write(0.0) async def probe(self) -> Probe: """A read-only look, usable without a Controller (panel, wizard).""" raise NotImplementedError class GoodWeEsphomeDriver(Driver): """The GoodWe behind the ESP32: three Home Assistant entities. A transcription of what `main.py` and `selfconfig.py` did inline: `limits()` is `hass.limits(setpoint)`, `read()` is `hass.number(soc)` then `hass.number(batt, invert)`, `write()` is `hass.set_number(setpoint, w)`. `release()` is `write(0)`: the board's own watchdog is what hands back. """ name = DEFAULT_DRIVER writer_key = "setpoint_entity" brand = "GoodWe" def __init__(self, setpoint_entity="", soc_entity="", batt_entity="", batt_invert=False): self.setpoint_entity = str(setpoint_entity or "") self.soc_entity = str(soc_entity or "") self.batt_entity = str(batt_entity or "") self.batt_invert = bool(batt_invert) self._hass = None self._getter = None @classmethod def from_entry(cls, entry: dict) -> "GoodWeEsphomeDriver": return cls(setpoint_entity=entry.get("setpoint_entity", ""), soc_entity=entry.get("soc_entity", ""), batt_entity=entry.get("batt_entity", ""), batt_invert=bool(entry.get("batt_invert"))) @classmethod def actuator_id_of(cls, entry: dict) -> str: return str(entry.get("setpoint_entity") or "").strip() def bind(self, hass) -> "GoodWeEsphomeDriver": self._hass, self._getter = hass, None return self def bind_getter(self, getter) -> "GoodWeEsphomeDriver": self._hass, self._getter = None, getter return self def bound(self) -> bool: return self._hass is not None or self._getter is not None def _h(self): hass = self._getter() if self._getter is not None else self._hass if hass is None: raise DriverNotBound(f"{self.brand} driver for " f"{self.setpoint_entity!r} has no Home Assistant") return hass def configured(self) -> bool: return bool(self.setpoint_entity) def actuator_id(self) -> str: return self.setpoint_entity def soc_source(self) -> str: return self.soc_entity def batt_source(self) -> str: return self.batt_entity async def limits(self) -> tuple: dev_min, dev_max, _step = await self._h().limits(self.setpoint_entity) return dev_min, dev_max async def read(self) -> tuple: hass = self._h() soc = await hass.number(self.soc_entity) raw = await hass.number(self.batt_entity, self.batt_invert) return soc, raw, {} async def write(self, w: float) -> bool: return await self._h().set_number(self.setpoint_entity, w) async def probe(self) -> Probe: dev_min, dev_max = await self.limits() soc, raw, extra = await self.read() return Probe(dev_min, dev_max, soc, raw, extra) class UnavailableDriver(Driver): """What an entry naming a driver this build does not have gets. Validation refuses such an entry before a Controller exists; this is only what `build_roster()` (which refuses nothing) builds meanwhile: a unit that is not configured, so the refusal branch names it and commands 0 W.""" name = "" # The census still counts a `setpoint_entity` on such an entry: refusing # too much is the safe direction for SAFETY-28. writer_key = "setpoint_entity" brand = "" def __init__(self, wanted=""): self.wanted = str(wanted or "") @classmethod def from_entry(cls, entry: dict) -> "UnavailableDriver": return cls(entry.get("driver")) @classmethod def actuator_id_of(cls, entry: dict) -> str: return str(entry.get("setpoint_entity") or "").strip() def configured(self) -> bool: return False def actuator_id(self) -> str: return "" async def limits(self) -> tuple: return None, None async def read(self) -> tuple: return None, None, {} async def write(self, w: float) -> bool: return False async def probe(self) -> Probe: return Probe() # name -> class. The options' `driver` value picks one. A test may add a fake; # `SELECTABLE` (not this) is what the options accept. REGISTRY = {GoodWeEsphomeDriver.name: GoodWeEsphomeDriver} def driver_name(entry) -> str: """The `driver` an options entry selects; blank or absent is the default.""" raw = entry.get("driver") if isinstance(entry, dict) else None return str(raw).strip() if raw not in (None, "") else DEFAULT_DRIVER def driver_class(entry): """The class an options entry selects, `UnavailableDriver` if none.""" return REGISTRY.get(driver_name(entry), UnavailableDriver) def build_driver(entry) -> Driver: """The driver for ONE options entry (or the flat options, for one inverter).""" entry = entry if isinstance(entry, dict) else {} return driver_class(entry).from_entry(entry)