goodwe-addon/goodwe_controller/app/drivers.py

"""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 = ""
    #: SMA-10. Does `release()` give the inverter BACK to something that steers
    #: it by itself (SMA energy management)? False: `release()` is `write(0)`,
    #: the add-on goes on holding 0 W, and the Controller never calls it.
    hands_back = False
    #: per-driver sentences. A driver that hands back says whose control the
    #: unit returns to: `{"back": "SMA control", "back_nl": "de sturing van SMA"}`
    #: ("goes back to SMA control"). Missing words fall back to a plain phrase.
    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.

        SMA-10. Only a driver with `hands_back = True` has this called by the
        Controller, and only when the add-on goes away (Stop, shutdown, restart,
        task death) or sits in shadow mode. After it the Controller writes
        nothing more to the unit until control starts again. A safety stop is
        never a release: that is `write(0)`."""
        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)` and `hands_back` is False: the board's own watchdog
    is what hands back, so after Stop the unit keeps its 0 W write and heartbeat.
    """

    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)