EV-03: the charger reader keeps its own power-unit rule

test_pv.py part 6 says only the PV reader's named modules may mention its
material, so chargers.py no longer imports pv.py.

Co-Authored-By: Claude <noreply@anthropic.com>

goodwe_controller/CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ### Not in a release yet
 
+- A new *Car* card on Home shows an Alfen charger: whether a car is connected, its power and today's energy. It only reads the charger, and the car's power no longer counts in the House tile or in what the batteries plan for the night. It is off until you choose a charger in Settings ([[EV-03]]).
 - During a real peak, a battery in its maintenance charge now stays paused instead of switching between charging and giving, and the batteries give only what keeps the peak under the limit ([[MAINT-28]]).
 
 ## 0.22.0

goodwe_controller/DOCS.md

@@ -251,6 +251,10 @@ The label and description each setting shows on the Configuration tab, in tab or
 | Safety | Opposing inverters time (s) | `circulation_hold_s` | For two or more inverters: how long they may push against each other before a safety stop. Derived value, leave it unchanged. |
 | Safety | Not delivering alarm after (s) | `delivery_hold_s` | For two or more inverters: how long one may fall short of its share before an alarm, which changes nothing. Derived value, leave it unchanged. |
 | Safety | Shadow mode | `shadow_mode` | The add-on makes every decision but keeps every inverter at 0 W. 'auto' does this while control is stopped, 'on' always, 'off' never. |
+| Car charger | Charger | `ev_charger` | Which car charger the add-on reads. Off leaves everything as it is. This version only reads the charger and never changes it. |
+| Car charger | Current setting entity | `ev_current_entity` | The entity that sets the charger's current. The add-on finds the charger's other entities from its name. |
+| Car charger | Charger limit (A) | `ev_max_a` | The most current the charger may ever be asked for. Never above what the charger reports. Lower it for a weaker fuse or cable. |
+| Car charger | Phases | `ev_phases` | How many phases the charger may use: 1, 3, or 1 and 3 (auto). |
 | Advanced | Start control on boot | `auto_start` | Off by default. Leave it off until the site is commissioned, so someone presses Start after each restart. |
 | Advanced | Control gain | `gain` | How much of the remaining error control corrects each cycle. Measured on real hardware, do not raise it. |
 | Advanced | Largest change per cycle (W) | `slew_w` | How far the command may move in one cycle. 0 follows “Inverter rating” and is recommended. |
@@ -3269,6 +3273,57 @@ row is printed **always**, including when it is zero: a row that reports
 staleness only once it is stale cannot be told from one that never reports
 it.
 
+### Car charger
+
+The add-on can read your car charger and show it on **Home**, on a *Car* card. In
+this version it only reads. It never changes the charger, so it is safe to run
+beside your own automation. Later versions will add charging modes.
+
+| option | meaning |
+|---|---|
+| `ev_charger` | `off` (the default: nothing changes at all) or `alfen_ha`, an Alfen charger through its Home Assistant integration. |
+| `ev_current_entity` | The charger's current setting, for example `number.wallbox_power_connector_max_current_socket_1`. The add-on finds the charger's other entities from this name. |
+| `ev_max_a` | The most current the charger may ever be asked for. It cannot be higher than the station limit the charger reports. |
+| `ev_phases` | `1`, `3` or `auto` (1 and 3). |
+
+**What the Car card shows.** The state (Charging, Connected waiting, No car
+connected, Car is full, Charger not reachable), the power and current, and the
+energy charged today. The car's power is taken out of the **House** tile and out
+of the house profile that plans the batteries' night charge, so the batteries
+no longer buy energy for a car that charged last night. The peak limit works as
+before: the car is still part of what the grid meter sees.
+
+**When the charger does not answer.** After 30 seconds the card says *Charger
+not reachable* and one amber row appears. After 5 minutes one notification
+follows. While the car reading is older than 15 seconds the House tile shows no
+reading, and that moment is left out of the house profile, rather than guessing
+how much the car took.
+
+**Another controller.** If the charger's current changes while the add-on writes
+nothing, something else sets it, usually your own automation. The page then shows
+the grey row *Another controller changes the charger*. That is expected while
+you migrate.
+
+**Moving your own automation to the add-on.** Each step can be undone.
+
+1. Install this version and set `ev_charger` to `alfen_ha`. The add-on reads the
+   charger and writes nothing. Compare the card with your automation for a few
+   days.
+2. When a later version adds charging modes, switch your automation **off**
+   (do not delete it) in the evening with the car unplugged, then choose the
+   add-on's Manual mode with the current your helper held. Plug in and check the
+   current on the card and on the charger's display.
+3. On the next sunny day, choose Solar, with *spare sun goes first to* set as
+   your priority switch was.
+4. Keep your Home Assistant helpers for a week. To go back, set the add-on mode
+   to Off and switch the automation on. After that you can delete the helpers and
+   the automation.
+
+Never let the automation and the add-on both write the charger.
+
+**Home Assistant entities.** With MQTT the add-on adds three sensors:
+`goodwe_car_power`, `goodwe_car_state` and `goodwe_car_today_kwh`.
+
 ### Site
 
 | option | meaning |
@@ -3388,6 +3443,14 @@ battery goes on as before. Check that the add-on can reach the internet and that
 the price service token is right. The notification goes by itself when a price
 is back. It is shown only while the monthly battery cycle is switched on.
 
+### Charger not reachable
+
+The add-on has not been able to read the car charger for 5 minutes, because its
+entities are unavailable or the charger's login is off. The charger keeps its
+last setting. The add-on changes nothing on the charger in any case. Check the
+charger's integration in Home Assistant. The notification goes by itself when
+the charger answers again.
+
 ### For installers: notifications
 
 A fixed tariff beside the price service raises no notification: it is the
@@ -3414,6 +3477,7 @@ default `info` hides it.
 | Meter reading may be stuck | alarm | `goodwe_meter_frozen` | SAFETY-18 |
 | Battery cycle did not finish | warning | `goodwe_maint_failed_<name>` | MAINT-13 |
 | No prices | warning | `goodwe_no_prices` | PRICE-01 |
+| Charger not reachable | warning | `goodwe_charger_unreachable` | EV-03 |
 
 **The Details link.** Home Assistant 2026.2 and newer show add-on pages at
 `/config/app/<slug>/documentation`; 2026.5 removed the old

goodwe_controller/app/chargers.py

@@ -0,0 +1,660 @@
+"""EV-03: the car charger seam, and the Alfen driver behind Home Assistant.
+
+SHAPED LIKE `app/drivers.py`'s `Driver` (DECISIONS/EV-03-charger-seam.md): one
+actuator per site, a census key, a read, a clamped write that says whether the
+charger took it, and a hand-back. A charger is a modulated actuator with a fail
+state, not one of DEV-01's on/off loads.
+
+THIS CARD NEVER WRITES TO THE CHARGER. `write_current`, `write_phases` and
+`release` are defined so the seam is whole, and NOTHING in the controller calls
+them: Off is the only mode. test_ev03.py greps the controller for a call and
+proves a full cycle makes no service call that names a charger entity.
+
+VALUES ARE BELIEVED ONLY IN THEIR OWN UNIT (as TEL-11 does for power): power in
+a power unit, energy in a kWh unit, current in A. A reading in another unit is
+absent, never converted by a guess. ABSENCE IS NEVER ZERO.
+
+The pure half (`CarTracker`, the state words, the entity table) takes its clock
+as an argument, so the checks need no sleep.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+
+from .hass import BAD_STATES
+from .solarfc import KWH_PER
+
+OFF = "off"
+ALFEN_HA = "alfen_ha"
+#: the `ev_charger` values the options accept
+SELECTABLE = (OFF, ALFEN_HA)
+PHASES_CHOICES = ("1", "3", "auto")
+
+# config.yaml's shipped defaults, mirrored (test_defaults_pin.py checks it).
+EV_OPTION_DEFAULTS = {"ev_charger": "off", "ev_current_entity": "",
+                      "ev_max_a": 16, "ev_phases": "auto"}
+
+
+def max_a(opts: dict):
+    """`ev_max_a` as a float, or None when it is not a number."""
+    raw = opts.get("ev_max_a", EV_OPTION_DEFAULTS["ev_max_a"])
+    if raw is None or raw == "":
+        raw = EV_OPTION_DEFAULTS["ev_max_a"]
+    try:
+        return float(raw)
+    except (TypeError, ValueError):
+        return None
+
+MIN_A = 6.0                 # IEC 61851: no car charges below 6 A
+ENTITY_MAX_A = 40.0         # the Alfen current entity's own range
+
+# Reachability and the car card's timers, in seconds.
+UNREACHABLE_S = 30.0        # entities unavailable / login off this long
+NOTIFY_S = 300.0            # then one warning notification
+FULL_B_S = 60.0             # mode3 B with current allowed this long: car full
+FULL_C_S = 600.0            # mode3 C under FULL_C_W this long: car full
+FULL_C_W = 200.0
+WRITER_HOLD_S = 3600.0      # "another controller" stays shown this long
+IDLE_A = 1.0                # a phase under this draws nothing
+
+# role -> (entity domain, suffix after `<x>_`). `current` is the setpoint, the
+# one the installer names; the rest are derived by swapping its suffix.
+SETPOINT_SUFFIX = "power_connector_max_current_socket_1"
+ROLES = {
+    "current": ("number", SETPOINT_SUFFIX),
+    "readback": ("sensor", "main_active_max_current_socket_1"),
+    "state": ("sensor", "mode3_state_socket_1"),
+    "display": ("sensor", "display_state_socket_1"),
+    "power": ("sensor", "active_power_total_socket_1"),
+    "energy": ("sensor", "meter_reading_socket_1"),
+    "l1": ("sensor", "current_l1_socket_1"),
+    "l2": ("sensor", "current_l2_socket_1"),
+    "l3": ("sensor", "current_l3_socket_1"),
+    "station": ("number", "max_station_current"),
+    "station_active": ("sensor", "main_station_active_max_current_socket_1"),
+    "phases_set": ("select", "installation_max_allowed_phases"),
+    "phases_read": ("sensor", "connector_1_max_allowed_of_phases"),
+    "error": ("sensor", "display_error_number_socket_1"),
+    "login": ("binary_sensor", "https_api_login_status"),
+}
+#: a missing one of these refuses the charger (not the add-on)
+REQUIRED = ("current", "readback", "state")
+_SETPOINT_RE = re.compile(r"^number\.(?P<x>.+)_" + SETPOINT_SUFFIX + "$")
+
+_ROLE_ORDER = tuple(ROLES)
+
+
+def census_key(entity) -> str:
+    """The SAFETY-28 form: stripped, lowercase."""
+    return str(entity or "").strip().lower()
+
+
+def entities_for(setpoint) -> dict:
+    """role -> entity id, derived from the setpoint entity by its suffix.
+    Empty when the setpoint does not end the way the Alfen integration names it."""
+    m = _SETPOINT_RE.match(str(setpoint or "").strip())
+    if not m:
+        return {}
+    x = m.group("x")
+    return {role: f"{dom}.{x}_{suffix}" for role, (dom, suffix) in ROLES.items()}
+
+
+# --------------------------------------------------------------------------- #
+# What a read gives
+# --------------------------------------------------------------------------- #
+@dataclass(frozen=True)
+class ChargerReading:
+    """One read of the charger. None means "did not answer", never zero."""
+
+    available: bool = False         # the integration answers
+    state: str | None = None        # mode3: "A", "B", "C", ... or None
+    connected: bool | None = None
+    charging: bool | None = None
+    power_w: float | None = None
+    energy_kwh: float | None = None  # the meter, total increasing
+    current_a: tuple = ()           # per phase, as many as read
+    setpoint_a: float | None = None  # the entity the installer named
+    allowed_a: float | None = None  # read-back of the limit the charger applies
+    phases_allowed: int | None = None
+    station_a: float | None = None  # the charger's own station limit
+    error: str | None = None        # None: no error (or not read)
+    display: str | None = None
+    why: str = ""                   # for the log only, "" when available
+
+
+def _state_obj(obj):
+    """(text, unit, attributes) of a state object, or (None, "", {})."""
+    if not isinstance(obj, dict):
+        return None, "", {}
+    raw = str(obj.get("state", "")).strip()
+    attrs = obj.get("attributes") if isinstance(obj.get("attributes"), dict) else {}
+    unit = str(attrs.get("unit_of_measurement") or "").strip()
+    return (None if raw.lower() in BAD_STATES else raw), unit, attrs
+
+
+def _amps(obj):
+    raw, unit, _ = _state_obj(obj)
+    if raw is None:
+        return None
+    try:
+        v = float(raw)
+    except ValueError:
+        return None
+    if v != v:
+        return None
+    if unit == "A":
+        return v
+    if unit == "mA":
+        return v / 1000.0
+    return None
+
+
+_W_PER = {"mW": 0.001, "W": 1.0, "kW": 1000.0, "MW": 1000000.0}
+
+
+def _watts(obj):
+    """Power in W, believed only in a power unit (its own copy of the rule: the
+    PV channel's reader stays out of the charger's, test_pv.py part 6)."""
+    raw, unit, _ = _state_obj(obj)
+    if raw is None or unit not in _W_PER:
+        return None
+    try:
+        v = float(raw) * _W_PER[unit]
+    except ValueError:
+        return None
+    return v if v == v else None
+
+
+def _kwh(obj):
+    raw, unit, _ = _state_obj(obj)
+    if raw is None or unit not in KWH_PER:
+        return None
+    try:
+        v = float(raw) * KWH_PER[unit]
+    except ValueError:
+        return None
+    return v if v == v and v >= 0 else None
+
+
+def _phases(obj):
+    raw, _u, _ = _state_obj(obj)
+    m = re.match(r"\s*([13])\s*phase", raw or "", re.I)
+    return int(m.group(1)) if m else None
+
+
+_MODE3 = re.compile(r"^(?:STATE[_ ]?)?([A-F])$", re.I)
+
+
+def mode3_letter(text) -> str | None:
+    m = _MODE3.match(str(text or "").strip())
+    return m.group(1).upper() if m else None
+
+
+def display_class(text) -> str | None:
+    """The fallback when mode3 is missing: firmware wording, so only the words
+    the owner's automation keys on. `connected`, `charging`, `none` or None."""
+    t = str(text or "").lower()
+    if not t:
+        return None
+    if "wait vehicle" in t or "ev connected" in t or "cable connected" in t:
+        return "connected"
+    if "charging" in t:
+        return "charging"
+    if t in ("available", "ready", "idle"):
+        return "none"
+    return None
+
+
+def reading_from_states(states: dict) -> ChargerReading:
+    """A `ChargerReading` from `{role: state object or None}`. Pure."""
+    def txt(role):
+        return _state_obj(states.get(role))[0]
+
+    login, state_o = txt("login"), states.get("state")
+    # Reachable: the three entities the card cannot do without answer, and the
+    # login sensor is not off.
+    gone = [r for r in REQUIRED if txt(r) is None]
+    if gone or (login or "").lower() == "off":
+        why = ("login off" if not gone else
+               "no answer from " + ", ".join(gone))
+        return ChargerReading(available=False, why=why)
+    letter = mode3_letter(txt("state"))
+    disp = txt("display")
+    cls = display_class(disp)
+    power = _watts(states.get("power"))
+    currents = tuple(a for a in (_amps(states.get(k)) for k in ("l1", "l2", "l3"))
+                     if a is not None)
+    if letter is not None:
+        connected, charging = letter in "BCD", letter in "CD"
+    elif cls is not None:
+        connected, charging = cls in ("connected", "charging"), cls == "charging"
+    else:
+        connected = charging = None
+    err = txt("error")
+    err_none = err is None or err.strip().startswith("0")
+    return ChargerReading(
+        available=True, state=letter, connected=connected, charging=charging,
+        power_w=power, energy_kwh=_kwh(states.get("energy")),
+        current_a=currents, setpoint_a=_amps(states.get("current")),
+        allowed_a=_amps(states.get("readback")),
+        phases_allowed=_phases(states.get("phases_read")),
+        station_a=_first(_amps(states.get("station_active")),
+                         _amps(states.get("station"))),
+        error=None if err_none else err, display=disp)
+
+
+def _first(*vals):
+    return next((v for v in vals if v is not None), None)
+
+
+@dataclass(frozen=True)
+class ChargerProbe:
+    """A read-only look: which derived entities exist."""
+
+    setpoint: str = ""
+    found: tuple = ()               # ((role, entity), ...)
+    missing: tuple = ()
+    station_a: float | None = None
+    derived: bool = True            # False: the setpoint's name does not fit
+
+    @property
+    def required_missing(self) -> tuple:
+        return tuple(r for r, _e in self.missing if r in REQUIRED)
+
+    @property
+    def ok(self) -> bool:
+        return self.derived and not self.required_missing
+
+
+def probe_from(setpoint, exists) -> ChargerProbe:
+    """`exists(entity) -> state object or None`. Pure; the caller reads."""
+    ents = entities_for(setpoint)
+    if not ents:
+        return ChargerProbe(setpoint=str(setpoint or ""), derived=False)
+    found, missing = [], []
+    for role in _ROLE_ORDER:
+        (found if exists(ents[role]) is not None else missing).append(
+            (role, ents[role]))
+    st = {r: exists(e) for r, e in ents.items()}
+    return ChargerProbe(str(setpoint), tuple(found), tuple(missing),
+                        station_a=_first(_amps(st.get("station_active")),
+                                         _amps(st.get("station"))))
+
+
+# --------------------------------------------------------------------------- #
+# The seam
+# --------------------------------------------------------------------------- #
+class ChargerNotBound(RuntimeError):
+    pass
+
+
+class ChargerDriver:
+    """One charger's I/O. Constructed from the options with no Home Assistant;
+    `bind()` gives it one. Everything that talks to the world is a coroutine."""
+
+    name = ""
+    writer_key = "ev_current_entity"
+    brand = ""
+    words: dict = {}
+
+    @classmethod
+    def from_options(cls, opts: dict) -> "ChargerDriver":
+        raise NotImplementedError
+
+    def bind(self, hass) -> "ChargerDriver":
+        return self
+
+    def bind_getter(self, getter) -> "ChargerDriver":
+        return self
+
+    def bound(self) -> bool:
+        return True
+
+    def configured(self) -> bool:
+        raise NotImplementedError
+
+    def actuator_id(self) -> str:
+        raise NotImplementedError
+
+    def entities(self) -> dict:
+        return {}
+
+    async def limits(self) -> tuple:
+        """`(min_a, max_a, phases)`; None where unread."""
+        raise NotImplementedError
+
+    async def read(self) -> ChargerReading:
+        raise NotImplementedError
+
+    # --- the write side. DEFINED, UNUSED BY THE CONTROLLER IN THIS CARD. ----
+    async def write_current(self, a: float) -> bool:
+        raise NotImplementedError
+
+    async def write_phases(self, n: int) -> bool:
+        raise NotImplementedError
+
+    async def release(self, mode: str) -> bool:
+        raise NotImplementedError
+
+    async def probe(self) -> ChargerProbe:
+        raise NotImplementedError
+
+
+class AlfenHaDriver(ChargerDriver):
+    """The Alfen behind its Home Assistant integration (`alfen_wallbox`)."""
+
+    name = ALFEN_HA
+    brand = "Alfen"
+
+    def __init__(self, current_entity="", limit_a=16.0):
+        self.current_entity = str(current_entity or "").strip()
+        self.max_a = float(limit_a)
+        self._ents = entities_for(self.current_entity)
+        self._hass = None
+        self._getter = None
+
+    @classmethod
+    def from_options(cls, opts: dict) -> "AlfenHaDriver":
+        amps = max_a(opts)
+        return cls(opts.get("ev_current_entity", ""),
+                   EV_OPTION_DEFAULTS["ev_max_a"] if amps is None else amps)
+
+    @classmethod
+    def actuator_id_of(cls, opts: dict) -> str:
+        return str(opts.get("ev_current_entity") or "").strip()
+
+    def bind(self, hass):
+        self._hass, self._getter = hass, None
+        return self
+
+    def bind_getter(self, getter):
+        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 ChargerNotBound("the charger has no Home Assistant")
+        return hass
+
+    def configured(self) -> bool:
+        return bool(self._ents)
+
+    def actuator_id(self) -> str:
+        return self.current_entity
+
+    def entities(self) -> dict:
+        return dict(self._ents)
+
+    async def _states(self, roles=_ROLE_ORDER) -> dict:
+        import asyncio
+        hass = self._h()
+        got = await asyncio.gather(*(hass.state(self._ents[r]) for r in roles))
+        return dict(zip(roles, got))
+
+    async def limits(self) -> tuple:
+        st = await self._states(("station", "station_active", "phases_read"))
+        return (MIN_A, _first(_amps(st["station_active"]), _amps(st["station"])),
+                _phases(st["phases_read"]))
+
+    async def read(self) -> ChargerReading:
+        if not self.configured():
+            return ChargerReading(why="not set up")
+        return reading_from_states(await self._states())
+
+    async def probe(self) -> ChargerProbe:
+        st = await self._states()
+        by_ent = {self._ents[r]: o for r, o in st.items()}
+        return probe_from(self.current_entity, by_ent.get)
+
+    # --- the write side, UNUSED in EV-03 --------------------------------------
+    async def write_current(self, a: float) -> bool:
+        a = max(0.0, min(float(a), self.max_a, ENTITY_MAX_A))
+        return await self._h().set_number(self.current_entity, a)
+
+    async def write_phases(self, n: int) -> bool:
+        if n not in (1, 3):
+            return False
+        return await self._h().call("select", "select_option", {
+            "entity_id": self._ents["phases_set"],
+            "option": "1 Phase" if n == 1 else "3 Phases"})
+
+    async def release(self, mode: str) -> bool:
+        return await self.write_current(0.0) if mode == "solar" else True
+
+
+class UnavailableCharger(ChargerDriver):
+    """No charger: `ev_charger` is off, or the charger is refused. `why` says
+    which, for the Car card; nothing is read and nothing is written."""
+
+    name = OFF
+
+    def __init__(self, why="off", wanted=""):
+        self.why = why
+        self.wanted = wanted
+
+    @classmethod
+    def from_options(cls, opts: dict):
+        return cls()
+
+    def configured(self) -> bool:
+        return False
+
+    def actuator_id(self) -> str:
+        return ""
+
+    async def limits(self) -> tuple:
+        return None, None, None
+
+    async def read(self) -> ChargerReading:
+        return ChargerReading(why=self.why)
+
+    async def write_current(self, a):
+        return False
+
+    async def write_phases(self, n):
+        return False
+
+    async def release(self, mode):
+        return False
+
+    async def probe(self) -> ChargerProbe:
+        return ChargerProbe()
+
+
+def mode_of(opts: dict) -> str:
+    v = str((opts or {}).get("ev_charger") or OFF).strip()
+    return v if v in SELECTABLE else OFF
+
+
+def build(opts: dict) -> ChargerDriver:
+    """The driver the options name. Off, or a name this build lacks, is
+    `UnavailableCharger`; nothing here touches Home Assistant."""
+    which = mode_of(opts)
+    if which == ALFEN_HA:
+        drv = AlfenHaDriver.from_options(opts)
+        if drv.configured():
+            return drv
+        return UnavailableCharger("not set up", which)
+    return UnavailableCharger("off")
+
+
+def census_entity(opts: dict) -> str:
+    """The charger setpoint the SAFETY-28 census counts: only while a charger
+    is switched on (an off charger writes nothing)."""
+    if mode_of(opts) == OFF:
+        return ""
+    return str(opts.get("ev_current_entity") or "").strip()
+
+
+# --------------------------------------------------------------------------- #
+# The car's state, kept over time
+# --------------------------------------------------------------------------- #
+CHARGING, WAITING, NONE_, FULL, UNREACH, SETUP = (
+    "charging", "waiting", "none", "full", "unreach", "setup")
+
+
+@dataclass
+class CarTracker:
+    """What the cycle learns about the car from the readings. Pure, clocked by
+    the caller (monotonic seconds for timers, UTC datetimes for the words).
+
+    FRESHNESS (the rule applied, and the one decision EV-03 makes about a stale
+    car reading): the car's power counts only while the last GOOD reading is not
+    older than `stale_s`, the same `stale_input_s` bound the meter reading has.
+    Older than that, `fresh()` is False and the House tile and the house profile
+    say they do not know, instead of keeping the last car figure."""
+
+    stale_s: float = 15.0
+    reading: ChargerReading | None = None
+    last_ok: float | None = None
+    bad_since: float | None = None
+    bad_since_utc: object = None
+    notified: bool = False
+    state_since_utc: object = None
+    last_status: str | None = None
+    last_good: ChargerReading | None = None
+    last_good_utc: object = None
+    b_since: float | None = None
+    c_low_since: float | None = None
+    writer_at_utc: object = None
+    _prev: tuple | None = None
+    # today's energy
+    day: str | None = None
+    today_kwh: float | None = None
+    last_kwh: float | None = None
+    persist: bool = False           # set when `today` changed, for the store
+
+    def observe(self, r: ChargerReading, now: float, now_utc, day: str):
+        self.reading = r
+        usable = r.available and r.power_w is not None
+        if r.available:
+            self.bad_since = None
+            self.bad_since_utc = None
+        elif self.bad_since is None:
+            self.bad_since, self.bad_since_utc = now, now_utc
+        if usable:
+            self.last_ok = now
+            self.last_good, self.last_good_utc = r, now_utc
+        if r.available:
+            self._writer(r, now_utc)
+            self._energy(r, day)
+            self._timers(r, now)
+        self._status_edge(now, now_utc)
+
+    # -- reachability ----------------------------------------------------------
+    def unreachable(self, now: float) -> bool:
+        return self.bad_since is not None and now - self.bad_since >= UNREACHABLE_S
+
+    def notify_due(self, now: float) -> bool:
+        return (self.bad_since is not None and not self.notified
+                and now - self.bad_since >= NOTIFY_S)
+
+    def fresh(self, now: float) -> bool:
+        return self.last_ok is not None and now - self.last_ok <= self.stale_s
+
+    def car_w(self, now: float):
+        """The car's power now, or None when the reading is not fresh."""
+        if not self.fresh(now) or self.reading is None or \
+                not self.reading.available:
+            return None
+        return max(0.0, float(self.reading.power_w))
+
+    # -- the two-writer detector ----------------------------------------------
+    def _writer(self, r, now_utc):
+        cur = (r.setpoint_a, r.allowed_a)
+        prev = self._prev
+        if prev is not None:
+            for a, b in zip(prev, cur):
+                if a is not None and b is not None and a != b:
+                    self.writer_at_utc = now_utc
+        # keep the last known value where one read came back empty
+        self._prev = tuple(b if b is not None else (prev[i] if prev else None)
+                           for i, b in enumerate(cur))
+
+    def other_writer(self, now_utc) -> bool:
+        t = self.writer_at_utc
+        return t is not None and (now_utc - t).total_seconds() <= WRITER_HOLD_S
+
+    # -- today's kWh -----------------------------------------------------------
+    def load(self, saved):
+        if isinstance(saved, dict):
+            try:
+                self.day = str(saved["day"])
+                self.today_kwh = float(saved["kwh"])
+                self.last_kwh = (None if saved.get("last") is None
+                                 else float(saved["last"]))
+            except (KeyError, TypeError, ValueError):
+                self.day = self.today_kwh = self.last_kwh = None
+
+    def dump(self) -> dict:
+        return {"day": self.day, "kwh": round(self.today_kwh or 0.0, 3),
+                "last": self.last_kwh}
+
+    def _energy(self, r, day):
+        e = r.energy_kwh
+        if self.day != day:
+            self.day, self.today_kwh = day, 0.0
+            self.last_kwh = e
+            self.persist = True
+            return
+        if e is None:
+            return
+        if self.last_kwh is not None:
+            # a meter that goes back has been reset: count from its new zero
+            d = e - self.last_kwh if e >= self.last_kwh else e
+            if d:
+                self.today_kwh = (self.today_kwh or 0.0) + d
+                self.persist = True
+        self.last_kwh = e
+        if self.today_kwh is None:
+            self.today_kwh = 0.0
+
+    # -- the car's state -------------------------------------------------------
+    def _timers(self, r, now):
+        allowed = r.allowed_a is not None and r.allowed_a >= MIN_A
+        self.b_since = (self.b_since if self.b_since is not None else now) \
+            if (r.state == "B" and allowed) else None
+        low = (r.state == "C" and allowed and r.power_w is not None
+               and r.power_w < FULL_C_W)
+        self.c_low_since = (self.c_low_since if self.c_low_since is not None
+                            else now) if low else None
+
+    def status(self, now: float, configured: bool = True):
+        """One of the state words, or None before the first reading."""
+        if not configured:
+            return SETUP
+        if self.unreachable(now):
+            return UNREACH
+        r = self.reading
+        if r is None or not r.available:
+            return None if self.last_good is None else self.last_status
+        if r.state == "A" or (r.state is None and r.connected is False):
+            return NONE_
+        if r.state in ("B", "C", "D") or r.connected:
+            if self.b_since is not None and now - self.b_since >= FULL_B_S:
+                return FULL
+            if self.c_low_since is not None and \
+                    now - self.c_low_since >= FULL_C_S:
+                return FULL
+            return CHARGING if r.charging else WAITING
+        return None
+
+    def _status_edge(self, now, now_utc):
+        s = self.status(now)
+        if s != self.last_status and s is not None:
+            self.last_status, self.state_since_utc = s, now_utc
+        elif s is None and self.last_status is None:
+            pass
+
+    def phases_in_use(self) -> int | None:
+        r = self.reading
+        if r is None or not r.current_a:
+            return None
+        n = sum(1 for a in r.current_a if a >= IDLE_A)
+        return n or None

goodwe_controller/app/ctl_alarms.py

@@ -401,6 +401,29 @@ class AlarmsMixin:
             await self.hass.call("persistent_notification", "dismiss",
                                  messages.prices_back().data())
 
+    async def check_charger(self) -> None:
+        """EV-03. One warning notification when the car charger has not
+        answered for 5 minutes, taken down when it does; the WARNING log line
+        is edge-logged. READ ONLY: this raises a notification and calls no
+        service on the charger."""
+        if not self.charger.configured():
+            return
+        now = self.car_now()
+        self.car_log_edges(now)
+        if self.car.notify_due(now) and not self._car_announced:
+            self._car_announced = True
+            self.car.notified = True
+            since = brussels(self.car.bad_since_utc)
+            await self.hass.call("persistent_notification", "create",
+                                 messages.charger_unreachable(
+                                     f"{since:%H:%M}",
+                                     self.charger.brand or "Alfen").data())
+        elif self.car.bad_since is None and self._car_announced:
+            self._car_announced = False
+            self.car.notified = False
+            await self.hass.call("persistent_notification", "dismiss",
+                                 messages.charger_back().data())
+
     async def check_fixed_note(self) -> None:
         """PRICE-01 G1. First cycle only: one INFO line when `static_profile`
         sits beside `price_service` (the plan's last fallback, not a fault, so

goodwe_controller/app/ctl_cycle.py

@@ -1047,6 +1047,7 @@ class CycleMixin:
         # Not a safety check, so it is one of the cheap ones: an attribute test
         # and, on a change only, one Home Assistant call.
         await self.check_price_gap()
+        await self.check_charger()
         await self.check_fixed_note()
 
         # SAFETY-31's alarm, and its boundary is the LATCH, not a number: it

goodwe_controller/app/ctl_inputs.py

@@ -13,7 +13,9 @@ from . import freshness
 from . import lateness
 from . import messages
 from . import pv as pv_read
+from . import chargers
 from .price import brussels, price_option, provider_configured
+from .hass import OUTAGE
 from .price_chain import uses_chain
 from .static_tariff import static_status
 from .rows import tariff_plan_row
@@ -48,6 +50,10 @@ FC_HOLD_KEY = "solar_fc_hold"
 FORECAST_READ_S = 60.0
 
 
+# EV-03: the car's energy today, kept so a restart does not lose it.
+CAR_TODAY_KEY = "car_today"
+
+
 class InputsMixin:
     def _readings_late(self, standing_w):
         """SAFETY-128: update the answer gate; seconds the late verdict has
@@ -102,6 +108,89 @@ class InputsMixin:
         return "input"
 
     # -- io ------------------------------------------------------------------
+    # -- EV-03: the car charger, READ ONLY ----------------------------------------
+    @property
+    def car_enabled(self) -> bool:
+        """`ev_charger` is on, whether or not the charger could be set up."""
+        return chargers.mode_of(self.o) != chargers.OFF
+
+    def car_now(self) -> float:
+        return freshness.monotonic()
+
+    def car_term(self) -> tuple:
+        """`(car_w, unknown)`: what the House tile and the house profile take
+        out for the car. `(0.0, False)` with no charger, so a site without one
+        is exactly as before. With one, a reading older than `stale_input_s`
+        (the meter's own bound) is UNKNOWN, never the last figure: the tile
+        shows no reading and the profile leaves that moment out."""
+        if not self.charger.configured():
+            return 0.0, False
+        w = self.car.car_w(self.car_now())
+        return (0.0, True) if w is None else (w, False)
+
+    async def _read_charger(self, now_mono: float) -> None:
+        """One read of the charger, and what the tracker learns from it. It
+        never raises (telemetry has no foot in the control path) and it only
+        reads: nothing here, or anywhere in this card, writes to the charger."""
+        now_utc = datetime.now(timezone.utc)
+        day = brussels(now_utc).date().isoformat()
+        try:
+            if not self._charger_probed:
+                pr = await asyncio.wait_for(self.charger.probe(), 5.0)
+                if OUTAGE.waiting():
+                    raise RuntimeError("Home Assistant is not answering")
+                self._charger_probed = True
+                self._judge_probe(pr)
+                if not self.charger.configured():
+                    return
+            reading = await asyncio.wait_for(self.charger.read(), 5.0)
+        except Exception as err:                        # noqa: BLE001
+            _LOG.debug("The charger could not be read: %s", err)
+            reading = chargers.ChargerReading(why="no answer")
+        self.car.observe(reading, now_mono, now_utc, day)
+        if self.car.persist:
+            self.car.persist = False
+            self.store.set(CAR_TODAY_KEY, self.car.dump())
+
+    def _judge_probe(self, pr) -> None:
+        """A charger whose setpoint, read-back or state entity is missing, or
+        whose limit is above the station limit, is REFUSED: the add-on starts
+        and the Car card says so. Said once."""
+        why = None
+        if not pr.ok:
+            why = "not_found"
+            _LOG.warning("The car charger is not set up, because Home "
+                         "Assistant has no entity it needs: CHARGER_NOT_FOUND.")
+            _LOG.debug("Charger entities missing: %s.",
+                       ", ".join(e for _r, e in pr.missing) or "the name")
+        else:
+            amps = chargers.max_a(self.o)
+            if pr.station_a is not None and amps is not None \
+                    and amps > pr.station_a:
+                why = "limit"
+                _LOG.warning("The charger limit is above what the charger "
+                             "allows, so the charger is not used: "
+                             "CHARGER_LIMIT.")
+                _LOG.debug("ev_max_a %.0f A, station limit %.0f A.", amps,
+                           pr.station_a)
+        if why is not None:
+            self.charger_refused = why
+            self.charger = chargers.UnavailableCharger(why, chargers.ALFEN_HA)
+
+    def car_log_edges(self, now_mono: float) -> None:
+        """One WARNING when the charger goes unreachable, one INFO when it is
+        back. Edge-logged, never every cycle."""
+        down = self.car.unreachable(now_mono)
+        if down and not self._car_down_logged:
+            self._car_down_logged = True
+            _LOG.warning("The car charger does not answer, so the Car card has "
+                         "no reading: CHARGER_UNREACHABLE.")
+            _LOG.debug("The charger entities read unavailable, or its login "
+                       "is off, for %.0f s.", chargers.UNREACHABLE_S)
+        elif not down and self._car_down_logged and self.car.bad_since is None:
+            self._car_down_logged = False
+            _LOG.info("The car charger answers again.")
+
     @property
     def pv_roster(self) -> tuple:
         """Every configured PV production entity, or `()` (TEL-11/TEL-12). ONE
@@ -209,6 +298,10 @@ class InputsMixin:
         _reads = [self._read_unit(inv, now_mono) for inv in self.inverters]
         if self.pv_roster:
             _reads.append(self._read_pv())
+        # EV-03: the charger, read only, in the same gather. Nothing is added
+        # without a charger, so a site with none runs exactly as before.
+        if self.charger.configured():
+            _reads.append(self._read_charger(now_mono))
         # OPT-11 P4: the solar forecast, in the same gather (a slow sensor must
         # not delay control), at most once per FORECAST_READ_S.
         if self.fc_entity is not None and (

goodwe_controller/app/ctl_status.py

@@ -15,7 +15,8 @@ from .freeze import rail_unreachable, rail_w
 from .hass import OUTAGE
 from .maintenance import IDLE
 from .mqtt import (
-    announced_units, optional_keys, parse_command, unit_slugs, unit_state_key,
+    PAYLOAD_NONE, announced_units, optional_keys, parse_command, unit_slugs,
+    unit_state_key,
 )
 from . import peakguard
 from . import pv as pv_read
@@ -135,6 +136,10 @@ class StatusMixin:
         if "pv" in optional_keys(self.o):
             values["pv"] = (None if self.pv_w is None
                             else round(self.pv_w, 1))
+        # EV-03: the car, only with a charger switched on (the same predicate
+        # that announces the three sensors). Absence is `None`, never 0.
+        if "car_state" in optional_keys(self.o):
+            values.update(self.car_values())
         # ⚠️ ONLY when P1 ingestion is running: the ESP32's stale-input
         # watchdog subscribes to sensor.goodwe_p1_sample_age and forces layer-1
         # failsafe at max_age_s, so with meter_source off a published age would
@@ -161,6 +166,41 @@ class StatusMixin:
         values.update(self.mode_values())
         self.mqtt.publish(values)
 
+    # -- EV-03: the car charger ------------------------------------------------
+    def car_rows(self) -> list:
+        """The charger's status rows: amber when it is not set up or does not
+        answer, grey while something else changes it."""
+        if not self.charger.configured():
+            return [{"ok": False, "warn": True, "id": "charger_setup",
+                     "text": messages.row_charger_setup()}]
+        out = []
+        if self.car.unreachable(self.car_now()):
+            out.append({"ok": False, "warn": True, "id": "charger",
+                        "text": messages.row_charger_unreach(
+                            f"{brussels(self.car.bad_since_utc):%H:%M}")})
+        if self.car.other_writer(datetime.now(timezone.utc)):
+            out.append({"ok": True, "warn": True, "id": "charger_writer",
+                        "text": messages.row_charger_writer()})
+        return out
+
+    def car_values(self) -> dict:
+        """The MQTT values for the three car sensors."""
+        now = self.car_now()
+        state = self.car.status(now, self.charger.configured())
+        r = self.car.reading
+        down = state in (None, "unreach", "setup")
+        with homeview.in_lang("en"):
+            word = homeview.car_state_word(state)
+        return {
+            "car_state": word,
+            # `ABSENT_AS_UNKNOWN` stays the one PV key (test_pv.py): an absent
+            # car figure is HA's own "no value" payload, sent by name.
+            "car_power": (PAYLOAD_NONE if down or r is None
+                          or r.power_w is None else round(r.power_w, 1)),
+            "car_today_kwh": (PAYLOAD_NONE if self.car.today_kwh is None
+                              else round(self.car.today_kwh, 3)),
+        }
+
     # -- UI ------------------------------------------------------------------
     def checks(self, when: datetime | None = None) -> list:
         """The status rows. `when` is the reference time for anything billed by
@@ -613,6 +653,9 @@ class StatusMixin:
                 out.append({"ok": False, "warn": True, "id": "pv",
                             "text": messages.row_pv_stray(
                                 _inv_list.index(ent) + 1)})
+        # EV-03: the car charger's rows, only with a charger switched on.
+        if self.car_enabled:
+            out.extend(self.car_rows())
         return out
 
     def status_payload(self, lang: str = "en") -> dict:

goodwe_controller/app/ctl_trade.py

@@ -388,8 +388,18 @@ class TradeMixin:
                 price_now = None
         house = (None if self.grid is None or self.site_batt_w is None
                  else float(self.grid) + float(self.site_batt_w))
+        # EV-03: the car is not the house. Out of the profile that plans the
+        # batteries' night charge; with a charger whose reading is not fresh
+        # the moment is left out (None), not guessed.
+        # ONLY the profile's sample: `house` below stays what it was, since
+        # the battery decision's input is not part of this card.
+        house_sample = house
+        if house is not None:
+            _car_w, _car_unknown = self.car_term()
+            house_sample = (None if _car_unknown else max(0.0, house - _car_w)
+                            if _car_w else house)
         if self.house_profile is not None and \
-                self.house_profile.add(now, house):
+                self.house_profile.add(now, house_sample):
             self.store.set(HOUSE_PROFILE_KEY, self.house_profile.dump())
         targets = self.plan_targets()
         if targets is None:

goodwe_controller/app/homeview.py

@@ -72,16 +72,26 @@ def flow(w, *, shadow=False):
     return "resting"
 
 
-def now_block(*, grid_w, pv_w, batt_w, has_solar, shadow, settled=True):
+def now_block(*, grid_w, pv_w, batt_w, has_solar, shadow, settled=True,
+              car_w=0.0, car_unknown=False):
     """The four tiles. `house_w` needs all three readings, so without a solar
-    reading it is None rather than a number that leaves the sun out."""
+    reading it is None rather than a number that leaves the sun out.
+
+    EV-03: `car_w` is what a car charger takes, and it is not the house, so it
+    is taken out of House (the Car card shows it). `car_unknown` is a charger
+    whose reading is not fresh: House then has no figure rather than one that
+    may hold the car. Both default to "no charger", the figure as before."""
     house, disagree = None, False
     if has_solar and None not in (grid_w, pv_w, batt_w):
-        house = float(grid_w) + float(pv_w) + float(batt_w)
+        house = float(grid_w) + float(pv_w) + float(batt_w) - float(car_w or 0.0)
+        if car_unknown:
+            house = None
         # The three are read at slightly different moments, so the sum can dip
         # under zero. Inside the idle band that is 0 W; beyond it the readings
         # disagree and no house figure is shown.
-        if house < -IDLE_W:
+        if house is None:
+            pass
+        elif house < -IDLE_W:
             house, disagree = None, True
         elif house < 0.0:
             house = 0.0
@@ -1070,6 +1080,7 @@ ATTN_DOCS = {
     "not_keeping_up": messages.T_LOOP_BOUND,
     "circulating": messages.T_CIRCULATING,
     "delivery": messages.S_NOT_DELIVERING,
+    "charger": messages.T_CHARGER_UNREACH,
 }
 
 
@@ -1187,7 +1198,10 @@ def _build(c, *, level, checks, now, held=None, waiting=False):
                   "sub": banner_sub(kind)},
         "now": now_block(grid_w=c.grid, pv_w=c.pv_w, batt_w=c.site_batt_w,
                          has_solar=bool(c.pv_roster), shadow=shadow,
-                         settled=settled),
+                         settled=settled,
+                         **dict(zip(("car_w", "car_unknown"),
+                                    c.car_term()))),
+        "car": car_block(c, now),
         "batteries": rows,
         "plan": plan,
         "modes_on": bool(getattr(c, "modes_on", False)),
@@ -1208,6 +1222,85 @@ def _docs_url():
                                                        messages._site["ha"])
 
 
+# --------------------------------------------------------------------------- #
+# EV-03: the Car card. READ ONLY: Off is the only mode in this version.
+# --------------------------------------------------------------------------- #
+_CAR_KIND = {"charging": "ok", "full": "ok", "unreach": "warn",
+             "setup": "warn"}
+
+
+def car_state_word(key) -> str:
+    return _t("car.state." + (key if key else "wait"))
+
+
+def car_phases_word(n) -> str:
+    return _t("car.phase_one") if n == 1 else _t("car.phase_many", n=n)
+
+
+def _car_big(key, c, r, good):
+    if key == "charging" and r is not None and r.power_w is not None:
+        parts = [_dec(f"{r.power_w / 1000:.1f}")
+                 + " kW"]
+        amps = (r.allowed_a if r.allowed_a is not None
+                else max(r.current_a) if r.current_a else None)
+        if amps is not None:
+            parts.append(f"{amps:.0f} A")
+        n = c.car.phases_in_use()
+        if n:
+            parts.append(car_phases_word(n))
+        return " · ".join(parts)
+    if key == "unreach":
+        t = c.car.last_good_utc
+        return (_t("car.big.seen", t=hhmm(t)) if t is not None
+                else _t("car.big.no_reading"))
+    if key == "charging":                       # the reading is too old
+        return _t("car.big.no_reading")
+    return _t("car.big." + (key or "wait"))
+
+
+def _car_why(key, c, good):
+    brand = (getattr(c.charger, "brand", "") or "Alfen")
+    if key == "unreach":
+        a = None if good is None else good.setpoint_a
+        if a is None and good is not None:
+            a = good.allowed_a
+        if a is not None:
+            return _t("car.why.unreach_a", a=f"{a:.0f}", brand=brand)
+        return _t("car.why.unreach", brand=brand)
+    if key == "setup":
+        why = {"not_found": "not_found", "limit": "limit"}.get(
+            c.charger_refused or "", "name")
+        return _t("car.why." + why)
+    return _t("car.why." + (key if key else "wait"))
+
+
+def car_block(c, now):
+    """The Car card for `/status`, or None when no charger is switched on (the
+    page then draws nothing, as before)."""
+    if not getattr(c, "car_enabled", False):
+        return None
+    mono = c.car_now()
+    configured = c.charger.configured()
+    key = c.car.status(mono, configured)
+    r = c.car.reading if configured else None
+    good = c.car.last_good
+    fresh = configured and c.car.fresh(mono)
+    today = None
+    if configured and c.car.today_kwh is not None:
+        today = _t("car.today", kwh=_kwh(c.car.today_kwh))
+    return {
+        "state": key or "wait",
+        "chip": car_state_word(key),
+        "kind": _CAR_KIND.get(key, "muted"),
+        "big": _car_big(key, c, r if fresh else None, good),
+        "today": today,
+        "why": _car_why(key, c, good),
+        "mode": "off",
+        "power_w": (None if not fresh or r is None or r.power_w is None
+                    else round(float(r.power_w), 1)),
+    }
+
+
 # =========================================================================== #
 # The words. STYLE.md: plain, short, present tense. A plan line is at most
 # two sentences of at most 12 words each (UX-13 design, section 4); a title at
@@ -1715,6 +1808,13 @@ def _samples() -> list:
     out += [("plan", day_note(k, pct=70, name="Battery A", at="08:15",
                               kwh="9.2 kWh", count="6.4 kWh"))
             for k in words["day"]]
+    out += [("title", _t("car.state." + k)) for k in words["car"]["state"]]
+    out += [("plan", _t("car.big." + k, t="09:12")) for k in
+            words["car"]["big"]]
+    out += [("plan", _t("car.why." + k, a="10", brand="Alfen"))
+            for k in words["car"]["why"]]
+    out += [("plan", _t("car.today", kwh="14.2 kWh")),
+            ("plan", "11.0 kW · 16 A · " + _t("car.phase_many", n=3))]
     out += [("plan", line_no_trade(k)) for k in
             ("not_covered", "no_prices", "peak", "not_firm", "not_worth",
              "no_level")]

goodwe_controller/app/main.py

@@ -42,6 +42,7 @@ from .arbiter import (
     P_SAFETY, SET, Claim, Resolution, actuator_for, pre_resolve, resolve,
 )
 from . import allocator
+from . import chargers
 from . import dayplan
 from . import modetrade
 from . import packmode
@@ -564,6 +565,17 @@ class Controller(InputsMixin, WritesMixin, TradeMixin, CycleMixin, AlarmsMixin,
         self._charge_latch = _load_latch(
             (getattr(store, "data", None) or {}).get(CHARGE_LATCH_KEY),
             datetime.now(timezone.utc))
+        # EV-03: the car charger, READ ONLY. Off builds `UnavailableCharger`,
+        # which reads nothing, so a site without a charger is unchanged.
+        self.charger = chargers.build(opts)
+        self.charger.bind_getter(lambda: self.hass)
+        self.car = chargers.CarTracker(
+            stale_s=float(opts.get("stale_input_s", 15) or 15))
+        self.car.load((getattr(store, "data", None) or {}).get("car_today"))
+        self.charger_refused = None
+        self._charger_probed = False
+        self._car_down_logged = False
+        self._car_announced = False
         self.house_profile = None
         if self.fc_entity is not None:
             self.house_profile = dayplan.HouseProfile(

goodwe_controller/app/messages.py

@@ -186,6 +186,8 @@ T_OUTPUT_CLAMP_NL = "Vermogensvraag te hoog"
 T_PEAK_SATURATED_NL = "Belasting te hoog voor de batterij"
 T_METER_FROZEN_NL = "Meter staat mogelijk vast"
 TITLE_NL = {T_LOOP_BOUND: T_LOOP_BOUND_NL, T_OUTPUT_CLAMP: T_OUTPUT_CLAMP_NL}
+T_CHARGER_UNREACH = "Charger not reachable"
+T_CHARGER_UNREACH_NL = "Laadpaal onbereikbaar"
 T_NO_PRICES = "No prices"
 T_NO_PRICES_NL = "Prijzen ontbreken"
 
@@ -198,6 +200,7 @@ ID_NOT_DELIVERING = "goodwe_not_delivering_{slug}"
 ID_METER_FROZEN = "goodwe_meter_frozen"
 ID_MAINT_FAILED = "goodwe_maint_failed_{slug}"
 ID_NO_PRICES = "goodwe_no_prices"
+ID_CHARGER_UNREACH = "goodwe_charger_unreachable"
 # PRICE-01: an earlier build raised a card with this id. Nothing raises it now;
 # the add-on only dismisses it at start.
 OLD_FIXED_CARD_ID = "goodwe_fixed_tariff_unused"
@@ -465,6 +468,43 @@ def row_no_prices(since=None, now=None) -> str:
               "Nog geen prijzen, onderhoud wacht.")
 
 
+def charger_unreachable(since_text_, brand="Alfen") -> Notice:
+    """EV-03: the car charger has not answered for 5 minutes. Said once."""
+    en = (f"The add-on cannot reach the car charger since {since_text_}, so "
+          f"check the {brand} integration in Home Assistant.")
+    nl = (f"De add-on kan de laadpaal niet bereiken sinds {since_text_}, dus "
+          f"controleer de {brand}-integratie in Home Assistant.")
+    return Notice(ID_CHARGER_UNREACH, WARNING, T_CHARGER_UNREACH, en,
+                  body_nl=f"{T_CHARGER_UNREACH_NL}. {nl}")
+
+
+def charger_back() -> Dismiss:
+    """EV-03: takes the 'Charger not reachable' notification down."""
+    return Dismiss(ID_CHARGER_UNREACH)
+
+
+def row_charger_unreach(since_text_) -> str:
+    """EV-03: the amber row while the charger does not answer."""
+    return Bi(f"The add-on cannot reach the charger since {since_text_}.",
+              f"De add-on kan de laadpaal niet bereiken sinds {since_text_}.")
+
+
+def row_charger_writer() -> str:
+    """EV-03: the grey row while something else sets the charger's current."""
+    return Bi("Something else changes the charger, and this add-on only reads "
+              "it.",
+              "Iets anders wijzigt de laadpaal en deze add-on leest alleen.")
+
+
+def row_charger_setup() -> str:
+    """EV-03: the amber row when the charger's entities are not found."""
+    return Bi("The add-on cannot find the charger, so check its entity in "
+              "Settings.",
+              "De add-on vindt de laadpaal niet, dus controleer de entiteit "
+              "bij Instellingen.")
+
+
+
 def row_maint_waits(until_text: str) -> str:
     """MAINT-13: the status row while a failed cycle waits."""
     return Bi(f"Waits until {until_text} after a cycle that did not finish.",
@@ -1661,6 +1701,9 @@ def status_samples() -> list:
         ("row_no_prices", AMBER, row_no_prices(*_SINCE_SAMPLE)),
         ("row_no_prices", AMBER, row_no_prices(*_SINCE_OLDER)),
         ("row_no_prices", AMBER, row_no_prices()),
+        ("row_charger_unreach", AMBER, row_charger_unreach("09:12")),
+        ("row_charger_writer", GREEN, row_charger_writer()),
+        ("row_charger_setup", AMBER, row_charger_setup()),
     ]
 
 
@@ -1696,6 +1739,7 @@ def samples() -> list:
         no_prices(*_SINCE_SAMPLE),
         no_prices(*_SINCE_OLDER),
         no_prices(),
+        charger_unreachable("09:12"),
     ]
 
 
@@ -1758,6 +1802,11 @@ LABELS = {
     "solar_trust_pct": "Trust in the forecast",
     "p1_phase_import_entities": "Import entities per phase",
     "p1_phase_export_entities": "Export entities per phase",
+    # EV-03
+    "ev_charger": "Charger",
+    "ev_current_entity": "Current setting entity",
+    "ev_max_a": "Charger limit",
+    "ev_phases": "Phases",
 }
 
 
@@ -1838,6 +1887,10 @@ LABELS_NL = {
     "solar_trust_pct": "Vertrouwen in de prognose",
     "p1_phase_import_entities": "Afname-entiteiten per fase",
     "p1_phase_export_entities": "Injectie-entiteiten per fase",
+    "ev_charger": "Laadpaal",
+    "ev_current_entity": "Entiteit van de laadstroom",
+    "ev_max_a": "Laadpaalgrens",
+    "ev_phases": "Fasen",
 }
 # The inverter list shows two of them under a shorter name.
 UNIT_LABELS_NL = {"setpoint_entity": "Setpoint-entiteit",

goodwe_controller/app/mqtt.py

@@ -139,6 +139,11 @@ SENSORS = [
 # a charge-power limit, or the optimiser declining to charge.
 OPTIONAL_SENSORS = [
     ("pv", "goodwe_pv_power", "PV power", "W", "power", "measurement", None),
+    # EV-03: only with a car charger switched on. Read only: there are no
+    # command entities in this version.
+    ("car_power", "goodwe_car_power", "Car power", "W", "power", "measurement", "mdi:car-electric"),
+    ("car_state", "goodwe_car_state", "Car state", None, None, None, "mdi:ev-station"),
+    ("car_today_kwh", "goodwe_car_today_kwh", "Car energy today", "kWh", "energy", "total_increasing", "mdi:ev-station"),
 ]
 
 # ⚠️ HA's OWN "this sensor has no value" PAYLOAD, and the one place this add-on
@@ -192,7 +197,11 @@ def optional_keys(opts: dict) -> tuple:
     # `_load()` is the same decision for the same kind of reason. Measured, not
     # assumed: move this line to module scope and test_mqtt.py part 8 dies.
     from . import pv as _pv
-    return ("pv",) if _pv.entities(opts) else ()
+    from . import chargers as _ch               # EV-03: same reason
+    out = ("pv",) if _pv.entities(opts) else ()
+    if _ch.census_entity(opts):
+        out += ("car_power", "car_state", "car_today_kwh")
+    return out
 
 
 BASE = "goodwe_ctl"

goodwe_controller/app/selfconfig.py

@@ -82,7 +82,7 @@ import os
 
 import aiohttp
 
-from . import messages
+from . import chargers, messages
 from .configui import changed_keys
 from .inverter import build_roster
 from .validate import validate_startup_config, validate_unit_ranges
@@ -362,6 +362,18 @@ async def problems_and_ranges(opts: dict, hass=None, notes=None):
     unread = [unit.driver.actuator_id() for unit in units
               if unit.driver.configured() and unit.dev_min is None
               and unit.dev_max is None]
+    # EV-03: the charger limit may not exceed the station limit the charger
+    # reports. A read only; a charger that does not answer skips the check.
+    if hass is not None and chargers.mode_of(opts) != chargers.OFF:
+        drv = chargers.build(opts)
+        if drv.configured():
+            _lo, station, _ph = await drv.bind(hass).limits()
+            amps = chargers.max_a(opts)
+            if station is not None and amps is not None and amps > station:
+                problems.append(
+                    f"{messages.setting('ev_max_a')} is {amps:.0f} A, but the "
+                    f"charger allows at most {station:.0f} A. Lower it to "
+                    f"{station:.0f} A or less.")
     return problems, unread
 
 

goodwe_controller/app/static/app.css

@@ -272,3 +272,8 @@ table.fct th,table.fct td{padding:4px 3px;text-align:right;border-bottom:1px sol
 table.fct th:first-child,table.fct td:first-child{text-align:left}
 table.fct th{color:var(--muted);font-weight:600}
 .ymax{position:absolute;top:-2px;left:2px;font-size:10px;color:var(--faint);pointer-events:none}
+
+/* EV-03: the Car card */
+.car-big{font-size:20px;font-weight:700;line-height:1.3;margin:8px 0 2px;font-variant-numeric:tabular-nums}
+.car-lab{margin:12px 0 4px;color:var(--muted)}
+.car-why{display:grid;grid-template-columns:auto 1fr;gap:10px;margin-top:12px}

goodwe_controller/app/static/app.js

@@ -113,7 +113,8 @@
     const setup = S.setup_only === true;
     banner($("#home-banner"), S.level, setup ? t("setup_only.name") : S.banner, structured ? S.state.sub : setup ? t("setup_only.description") : "", S.level === "bad" ? "alert" : null);
     ["#c-now", "#c-plan", "#c-peak", "#c-attn"].forEach((id) => { $(id).hidden = !structured; });
-    if (structured) { renderNow(); renderPlan(); renderPeak(); renderAttention(); }
+    if (structured) { renderNow(); renderCar(); renderPlan(); renderPeak(); renderAttention(); }
+    else $("#c-car").hidden = true;
     renderControl(structured, setup);
     renderDetails(!structured && !setup);
     renderBatteries(force);
@@ -145,6 +146,20 @@
       h("div", { class: "small muted", style: "grid-column:1/-1" }, b.now))));
   }
 
+  /* ---------------- Home: the car (read only) ---------------- */
+  function renderCar() {
+    const c = S.car;
+    $("#c-car").hidden = !c;
+    if (!c) return;
+    const chip = $("#car-chip");
+    chip.textContent = c.chip;
+    chip.className = "chip" + (c.kind === "warn" ? " rs" : c.kind === "ok" ? " now" : "");
+    $("#car-big").textContent = c.big;
+    $("#car-today").textContent = c.today || "";
+    $("#car-today").hidden = !c.today;
+    $("#car-why").textContent = c.why;
+  }
+
   /* ---------------- Home: today's plan ---------------- */
   const LEGEND = [["cheap", "background:var(--cheap)"], ["dear", "background:repeating-linear-gradient(135deg,var(--dear) 0 3px,transparent 3px 5px)"],
                   ["grid", "background:var(--accent)"], ["give", "background:repeating-linear-gradient(135deg,var(--house) 0 3px,transparent 3px 5px)"],

goodwe_controller/app/static/index.html

@@ -39,6 +39,16 @@
     <div id="home-batts" style="margin-top:6px"></div>
   </div>
 
+  <div class="card" id="c-car" hidden>
+    <div class="row spread"><h2 data-t="car.name"></h2><span class="chip" id="car-chip"></span></div>
+    <div class="car-big" id="car-big"></div>
+    <div class="small muted" id="car-today"></div>
+    <div class="small car-lab" data-t="car.mode"></div>
+    <div class="seg" role="radiogroup" id="car-mode" data-t-label="car.mode"><button type="button" role="radio" aria-checked="true" data-t="car.mode_off"></button></div>
+    <p class="tiny muted" style="margin:6px 0 0" data-t="car.help_off"></p>
+    <div class="car-why small"><span class="muted" data-t="car.why"></span><span id="car-why"></span></div>
+  </div>
+
   <div class="card" id="c-plan">
     <div class="row spread"><h2 data-t="plan.name"></h2><span class="chip tiny muted" id="verdict"></span></div>
     <div class="small" id="plan-head"></div>

goodwe_controller/app/ui_text/en.json

@@ -4,6 +4,7 @@
     "inverters": {"name": "Inverters and batteries", "description": "The inverters the add-on controls, and your solar."},
     "tariff": {"name": "Tariff and prices", "description": "What you pay and are paid per kWh."},
     "safety": {"name": "Safety limits", "description": "The peak limit, starting and stopping, and the safety stops."},
+    "car": {"name": "Car charger", "description": "Which charger the add-on reads, and its limits."},
     "maintenance": {"name": "Maintenance", "description": "The monthly battery cycle and the log."},
     "tuning": {"name": "Control tuning", "description": "For support only, because these were measured on real hardware."}
   },
@@ -33,6 +34,8 @@
     "fallback": {"none": "None", "fixed": "One price", "daynight": "Day and night", "custom": "Other time windows"},
     "region": {"flanders": "Flanders, Fluvius Antwerpen", "own": "I enter my own figure"},
     "meter_source": {"off": "One grid power entity", "ha_dsmr": "Import and export entities", "mqtt_p1": "Meter over MQTT", "ha_signed": "One net power entity", "homewizard_local": "HomeWizard meter on the network"},
+    "ev_charger": {"off": "None", "alfen_ha": "Alfen (Home Assistant integration)"},
+    "ev_phases": {"1": "1 phase", "3": "3 phases", "auto": "1 and 3"},
     "meter_phases": {"1": "1 phase", "3": "3 phases"},
     "shadow_mode": {"auto": "Automatic", "on": "Always on", "off": "Off"},
     "price_provider": {"generic_dayahead": "Day-ahead market price", "custom_formula": "Your own price formula", "manual_csv": "A price file", "price_service": "Your own price service"},
@@ -87,6 +90,7 @@
       "loading": "Loading the forecast."
     },
     "peak": {"name": "Peak limit", "this_month": "This month's peak", "limit": "Limit", "none": "None recorded yet", "unreadable": "Cannot be read now"},
+    "car": {"name": "Car", "mode": "Mode", "mode_off": "Off", "help_off": "The add-on leaves the charger alone.", "why": "Why"},
     "attention": {"name": "Needs attention", "nothing": "Nothing to look at", "to_look": "to look at", "details": "Details", "alarm": "Alarm", "warning": "Warning", "info": "Info"},
     "control": {"name": "Control", "stop": "Stop control", "start": "Start control", "maint_start": "Start a battery cycle now", "maint_abort": "Stop the battery cycle", "docs": "Add-on documentation", "more": "More actions"},
     "details": {"name": "Details for installers", "rows": "Status rows", "checks": "Checks", "event": "Last event"},
@@ -296,6 +300,25 @@
       "waiting": "Control resumes by itself when Home Assistant answers.",
       "warning": "See Needs attention below."
     },
+    "car": {
+      "state": {"charging": "Charging", "waiting": "Connected, waiting", "none": "No car connected", "full": "Car is full", "unreach": "Charger not reachable", "setup": "Charger not set up", "wait": "No reading yet"},
+      "big": {"waiting": "Connected, not charging.", "none": "Nothing to charge.", "full": "The car is full.", "seen": "Last seen at {t}.", "no_reading": "No reading.", "setup": "Not set up.", "wait": "No reading yet."},
+      "phase_one": "1 phase",
+      "phase_many": "{n} phases",
+      "today": "{kwh} charged today.",
+      "why": {
+        "charging": "The charger is charging the car. The add-on leaves it alone.",
+        "waiting": "The car is connected and not charging. The add-on leaves the charger alone.",
+        "none": "No car is connected.",
+        "full": "The car stopped charging by itself.",
+        "unreach_a": "The charger keeps its last setting, {a} A. Check the {brand} integration in Home Assistant.",
+        "unreach": "The add-on cannot read the charger. Check the {brand} integration in Home Assistant.",
+        "not_found": "The add-on cannot find the charger. Check the current setting entity in Settings.",
+        "limit": "The charger limit is above what the charger allows. Lower it in Settings.",
+        "name": "The current setting entity does not fit the charger. Check it in Settings.",
+        "wait": "Waiting for the first reading."
+      }
+    },
     "tile": {
       "not_set_up": "Not set up",
       "no_reading": "No reading",
@@ -499,6 +522,9 @@
       "estop": "Emergency stop",
       "pv": "Solar reading",
       "solar_forecast": "Solar forecast",
+      "charger": "Charger not reachable",
+      "charger_writer": "Something else changes the charger",
+      "charger_setup": "Charger not set up",
       "none": "Inverter connection"
     },
     "months": ["January", "February", "March", "April", "May", "June", "July", "August", "September", "October", "November", "December"]

goodwe_controller/app/ui_text/nl.json

@@ -4,6 +4,7 @@
     "inverters": {"name": "Omvormers en batterijen", "description": "De omvormers die de add-on regelt, en uw zonnepanelen."},
     "tariff": {"name": "Tarief en prijzen", "description": "Wat u per kWh betaalt en terugkrijgt."},
     "safety": {"name": "Veiligheidsgrenzen", "description": "De piekgrens, starten en stoppen, en de veiligheidsstops."},
+    "car": {"name": "Laadpaal", "description": "Welke laadpaal de add-on leest, en haar grenzen."},
     "maintenance": {"name": "Onderhoud", "description": "De maandelijkse batterijcyclus en het logboek."},
     "tuning": {"name": "Afstelling van de regeling", "description": "Alleen voor support, want deze zijn gemeten op echte hardware."}
   },
@@ -33,6 +34,8 @@
     "fallback": {"none": "Geen", "fixed": "Eén prijs", "daynight": "Dag en nacht", "custom": "Andere tijdvakken"},
     "region": {"flanders": "Vlaanderen, Fluvius Antwerpen", "own": "Ik vul zelf een bedrag in"},
     "meter_source": {"off": "Eén entiteit voor netvermogen", "ha_dsmr": "Entiteiten voor afname en injectie", "mqtt_p1": "Meter via MQTT", "ha_signed": "Eén entiteit voor nettovermogen", "homewizard_local": "HomeWizard-meter op het netwerk"},
+    "ev_charger": {"off": "Geen", "alfen_ha": "Alfen (Home Assistant-integratie)"},
+    "ev_phases": {"1": "1 fase", "3": "3 fasen", "auto": "1 en 3"},
     "meter_phases": {"1": "1 fase", "3": "3 fasen"},
     "shadow_mode": {"auto": "Automatisch", "on": "Altijd aan", "off": "Uit"},
     "price_provider": {"generic_dayahead": "Day-aheadprijs van de markt", "custom_formula": "Uw eigen prijsformule", "manual_csv": "Een prijsbestand", "price_service": "Uw eigen prijsdienst"},
@@ -84,6 +87,7 @@
       "loading": "De prognose wordt geladen."
     },
     "peak": {"name": "Piekgrens", "this_month": "Piek van deze maand", "limit": "Grens", "none": "Nog niets gemeten", "unreadable": "Nu niet leesbaar"},
+    "car": {"name": "Auto", "mode": "Modus", "mode_off": "Uit", "help_off": "De add-on laat de laadpaal ongemoeid.", "why": "Waarom"},
     "attention": {"name": "Vraagt aandacht", "nothing": "Niets om na te kijken", "to_look": "na te kijken", "details": "Details", "alarm": "Alarm", "warning": "Waarschuwing", "info": "Info"},
     "control": {"name": "Regeling", "stop": "Regeling stoppen", "start": "Regeling starten", "maint_start": "Nu een batterijcyclus starten", "maint_abort": "Batterijcyclus stoppen", "docs": "Documentatie van de add-on", "more": "Meer acties"},
     "details": {"name": "Details voor installateurs", "rows": "Statusregels", "checks": "Controles", "event": "Laatste gebeurtenis"},
@@ -250,6 +254,25 @@
     "verdict": {"held": "Op 0 W gehouden", "stopped": "Gestopt", "cycle": "Vandaag maandelijkse cyclus", "trade": "Handelt op prijs", "house": "Volgt het huis"},
     "pill": {"running": "Actief", "shadow": "Schaduwmodus", "stopped": "Gestopt", "alarm": "Heeft u nu nodig", "waiting": "Wacht op Home Assistant", "warning": "Actief, met een waarschuwing"},
     "banner_sub": {"running": "De batterijen volgen het plan hieronder.", "shadow": "De plannen hieronder tonen wat de add-on zou doen.", "stopped": "Start de regeling hieronder als u klaar bent.", "alarm": "Zie Vraagt aandacht hieronder voor wat u moet doen.", "waiting": "De regeling gaat vanzelf verder zodra Home Assistant antwoordt.", "warning": "Zie Vraagt aandacht hieronder."},
+    "car": {
+      "state": {"charging": "Laadt", "waiting": "Aangesloten, wacht", "none": "Geen auto aangesloten", "full": "Auto is vol", "unreach": "Laadpaal onbereikbaar", "setup": "Laadpaal niet ingesteld", "wait": "Nog geen meting"},
+      "big": {"waiting": "Aangesloten, laadt niet.", "none": "Niets te laden.", "full": "De auto is vol.", "seen": "Laatst gezien om {t}.", "no_reading": "Geen meting.", "setup": "Niet ingesteld.", "wait": "Nog geen meting."},
+      "phase_one": "1 fase",
+      "phase_many": "{n} fasen",
+      "today": "Vandaag {kwh} geladen.",
+      "why": {
+        "charging": "De laadpaal laadt de auto. De add-on laat haar met rust.",
+        "waiting": "De auto is aangesloten en laadt niet. De add-on laat de laadpaal met rust.",
+        "none": "Er is geen auto aangesloten.",
+        "full": "De auto is zelf gestopt met laden.",
+        "unreach_a": "De laadpaal houdt haar laatste instelling, {a} A. Controleer de {brand}-integratie in Home Assistant.",
+        "unreach": "De add-on kan de laadpaal niet lezen. Controleer de {brand}-integratie in Home Assistant.",
+        "not_found": "De add-on vindt de laadpaal niet. Controleer de entiteit van de laadstroom bij Instellingen.",
+        "limit": "De laadpaalgrens is hoger dan de laadpaal toestaat. Verlaag ze bij Instellingen.",
+        "name": "De entiteit van de laadstroom past niet bij de laadpaal. Controleer ze bij Instellingen.",
+        "wait": "Wacht op de eerste meting."
+      }
+    },
     "tile": {"not_set_up": "Niet ingesteld", "no_reading": "Geen meting", "producing": "Produceert", "not_producing": "Produceert niet", "needs_solar": "Zonnemeting nodig", "not_adding_up": "Metingen kloppen niet", "using": "Verbruikt", "taking": "Neemt af van het net", "feeding": "Injecteert", "balanced": "In evenwicht", "giving": "Levert stroom", "charging": "Laadt", "resting": "Rust"},
     "now": {
       "no_reading": "Geen meting van deze batterij.",
@@ -420,6 +443,9 @@
       "estop": "Noodstop",
       "pv": "Meting van de zonnepanelen",
       "solar_forecast": "Zonneprognose",
+      "charger": "Laadpaal onbereikbaar",
+      "charger_writer": "Iets anders wijzigt de laadpaal",
+      "charger_setup": "Laadpaal niet ingesteld",
       "none": "Verbinding met de omvormer"
     },
     "months": ["januari", "februari", "maart", "april", "mei", "juni", "juli", "augustus", "september", "oktober", "november", "december"]

goodwe_controller/app/uilayout.py

@@ -77,6 +77,7 @@ _MAINT = (("maintenance_enabled", "==", True),)
 _ONE = (("inverter_count", "==", 1),)
 _SEVERAL = (("inverter_count", ">", 1),)
 _PHASES = (("meter_phases", ">", 1),)
+_CAR = (("ev_charger", "!=", "off"),)
 # One inverter stored flat has no name or capacity (uimodel._roster_encode).
 _LISTED = (("inverter_shape", "!=", "flat"),)
 
@@ -222,6 +223,17 @@ GROUPS = (
             Field("delivery_hold_s", LOCKED, "s", step=5, show=_SEVERAL),
         ), collapsed=True),
     )),
+    # EV-03: the car charger, read only. Everything after the choice shows
+    # only while a charger is switched on.
+    Group("car", (
+        Part("", (
+            Field("ev_charger", CHOICE, simple=True),
+            Field("ev_current_entity", ENTITY, kind="number", simple=True,
+                  show=_CAR),
+            Field("ev_max_a", SLIDER, "A", simple=True, show=_CAR),
+            Field("ev_phases", SEGMENTED, simple=True, show=_CAR),
+        )),
+    )),
     Group("maintenance", (
         Part("", (
             Field("maintenance_enabled", TOGGLE),

goodwe_controller/app/validate.py

@@ -332,7 +332,7 @@ import math
 import re
 from typing import NamedTuple
 
-from . import dayplan, messages, pv, solarfc
+from . import chargers, dayplan, messages, pv, solarfc
 from .arbitrage import enabled as arbitrage_on
 from .control import SLEW_RAIL_FRAC, LOOP_TICK_S, PLANT_SETTLE_S
 from . import drivers
@@ -667,6 +667,34 @@ def _labels(keys, joiner: str = "and") -> str:
     return f" {joiner} ".join(S(k) for k in keys)
 
 
+def _charger_problems(opts: dict, writers: dict) -> list[str]:
+    """EV-03: the car charger options. A charger whose entities are missing is
+    refused at run time (the Car card says so), not here: it never stops the
+    add-on. What stops it is a setpoint that an inverter also writes."""
+    out: list[str] = []
+    raw = opts.get("ev_charger", chargers.OFF)
+    if raw not in (None, "") and str(raw).strip() not in chargers.SELECTABLE:
+        out.append(f"{S('ev_charger')} is {messages.shown(raw)}. Choose "
+                   f"{', '.join(chargers.SELECTABLE)}.")
+        return out
+    if chargers.mode_of(opts) == chargers.OFF:
+        return out
+    ph = opts.get("ev_phases", chargers.EV_OPTION_DEFAULTS["ev_phases"])
+    if str(ph).strip() not in chargers.PHASES_CHOICES:
+        out.append(f"{S('ev_phases')} is {messages.shown(ph)}. Choose 1, 3 or "
+                   "auto.")
+    amps = chargers.max_a(opts)
+    if amps is None or not chargers.MIN_A <= amps <= chargers.ENTITY_MAX_A:
+        out.append(f"{S('ev_max_a')} is {messages.shown(opts.get('ev_max_a'))}"
+                   f". Set it from {chargers.MIN_A:.0f} to "
+                   f"{chargers.ENTITY_MAX_A:.0f} A.")
+    entity = chargers.census_entity(opts)
+    if entity and drivers.census_key(entity) in writers:
+        out.append(f"{S('ev_current_entity')} is {entity}, which an inverter "
+                   "also uses. Give the charger its own entity.")
+    return out
+
+
 def _roster_problems(opts: dict) -> list[str]:
     """Rule 9 (CFG-04): the `inverters:` roster, refused where it cannot mean
     what it says.
@@ -956,6 +984,12 @@ def validate_startup_config(opts: dict, dev_min=None, dev_max=None,
                 f"{actuator}: {', '.join(who)}. Give each inverter its own "
                 "setpoint entity.")
 
+    # --- EV-03: the car charger's setpoint is a writer too ---------------------
+    # Counted in the census beside the inverters', and kept apart from
+    # `actuator_writers()` so setup-only mode still reads "no inverter". An
+    # inverter's entity named as the charger is two writers on one actuator.
+    problems.extend(_charger_problems(opts, writers))
+
     # --- 9. CFG-04's roster, graded BEFORE anything reads a per-unit number --
     # ⚠️ FIRST, THOUGH IT IS THE NEWEST RULE. Rules 5 and 6 below now ask the
     # roster what one unit may be commanded, so a roster that is nonsense has to
@@ -1921,6 +1955,7 @@ def config_notes(opts: dict, *, entities: dict | None = None,
     notes: list[Note] = []
     if entities is not None:
         _entity_notes(notes, opts, entities)
+        _charger_notes(notes, opts, entities)
     cap = _number(opts, "peak_cap_w", None)
     if month_peak_w is not None and cap is not None and cap < month_peak_w:
         _advise(notes, "warn", ("peak_cap_w",),
@@ -1932,6 +1967,94 @@ def config_notes(opts: dict, *, entities: dict | None = None,
     return notes
 
 
+def charger_probe_rows(opts: dict, entities: dict | None) -> dict:
+    """EV-03: the settings panel's read-only look at the charger. `entities` is
+    Home Assistant's list by id (None: it did not answer). Nothing is written.
+    `{"ok", "found": [[role, entity]], "missing": [[role, entity]], "limit_a",
+    "state"}`; `state` is `off`, `no_name` (the setpoint is not named the way
+    the charger's integration names it), `unread`, `ok` or `missing`."""
+    if chargers.mode_of(opts) == chargers.OFF:
+        return {"ok": False, "found": [], "missing": [], "limit_a": None,
+                "state": "off"}
+    setpoint = str(opts.get("ev_current_entity") or "").strip()
+    if not chargers.entities_for(setpoint):
+        return {"ok": False, "found": [], "missing": [], "limit_a": None,
+                "state": "no_name" if setpoint else "empty"}
+    if entities is None:
+        return {"ok": False, "found": [], "missing": [], "limit_a": None,
+                "state": "unread"}
+    def state_obj(e):
+        row = entities.get(e)
+        return None if row is None else {
+            "state": row.get("state"),
+            "attributes": {"unit_of_measurement": row.get("unit")}}
+    pr = chargers.probe_from(setpoint, state_obj)
+    return {"ok": pr.ok, "found": [list(x) for x in pr.found],
+            "missing": [list(x) for x in pr.missing], "limit_a": pr.station_a,
+            "state": "ok" if pr.ok else "missing"}
+
+
+# The roles the "Found:" note names, in words (EV-03).
+_CHARGER_WORDS = (
+    ("state", "car state", "autostatus"), ("power", "power", "vermogen"),
+    ("energy", "energy", "energie"), ("readback", "read-back", "terugmelding"),
+    ("phases_read", "phases", "fasen"),
+    ("station", "the charger limit", "de laadpaalgrens"))
+
+
+def charger_probe_for(setpoint: str, entities: dict | None) -> dict:
+    """`charger_probe_rows()` for a setpoint entity typed on the page."""
+    return charger_probe_rows({"ev_charger": chargers.ALFEN_HA,
+                               "ev_current_entity": setpoint}, entities)
+
+
+def _charger_notes(notes, opts: dict, entities: dict) -> None:
+    """EV-03: a note for each charger entity Home Assistant does not have, and
+    one when the limit is above the charger's own. Never a refusal."""
+    look = charger_probe_rows(opts, entities)
+    key = ("ev_current_entity",)
+    if look["state"] == "no_name":
+        _advise(notes, "warn", key,
+                f"{S('ev_current_entity')} must be named like the Alfen "
+                "integration's current setting, so the charger is not read.",
+                nl=f"{SN('ev_current_entity')} moet heten zoals de "
+                "stroominstelling van de Alfen-integratie, dus de laadpaal "
+                "wordt niet gelezen.")
+        return
+    if look["state"] == "empty":
+        _advise(notes, "warn", key,
+                f"{S('ev_current_entity')} is empty, so the charger is not "
+                "read.",
+                nl=f"{SN('ev_current_entity')} is leeg, dus de laadpaal wordt "
+                "niet gelezen.")
+        return
+    if look["state"] not in ("ok", "missing"):
+        return
+    for role, entity in look["missing"]:
+        required = role in chargers.REQUIRED
+        _advise(notes, "warn" if required else "info", key,
+                f"Home Assistant has no entity {entity}, which the charger "
+                f"needs.",
+                nl=f"Home Assistant kent {entity} niet, die de laadpaal "
+                "nodig heeft.")
+    lim = look["limit_a"]
+    if look["ok"]:
+        have = {r for r, _e in look["found"]}
+        words = [(w_en, w_nl) for r, w_en, w_nl in _CHARGER_WORDS if r in have]
+        tail_en = f", {lim:.0f} A" if lim is not None else ""
+        _advise(notes, "ok", key,
+                "Found: " + ", ".join(w[0] for w in words) + tail_en + ".",
+                nl="Gevonden: " + ", ".join(w[1] for w in words) + tail_en
+                + ".")
+    amps = chargers.max_a(opts)
+    if lim is not None and amps is not None and amps > lim:
+        _advise(notes, "warn", ("ev_max_a",),
+                f"{S('ev_max_a')} is above the {lim:.0f} A the charger "
+                "reports, so lower it.",
+                nl=f"{SN('ev_max_a')} is hoger dan de {lim:.0f} A die de "
+                "laadpaal meldt, dus verlaag ze.")
+
+
 def _entity_notes(notes, opts: dict, entities: dict, bad=()) -> None:
     """A note for each entity these options read that Home Assistant does not
     have, or that reads in kW where watts are taken. A missing entity whose

goodwe_controller/app/web.py

@@ -51,7 +51,8 @@ from aiohttp import web
 from . import messages, price, selfconfig, uipage, uitext
 from .haentities import EntityCache
 from .homeview import priced_hours, region_of
-from .validate import config_notes, note_text, setup_checks
+from .validate import (charger_probe_for, config_notes, note_text,
+                       setup_checks)
 from .config_page import CONFIG_PAGE
 from .configui import (changed_keys, field_model, hide_secrets, keep_secrets,
                        problems_by_field, secret_keys)
@@ -474,6 +475,14 @@ def build_app(controller, selfcfg=None) -> web.Application:
             return web.json_response(await entities.reading(one))
         return web.json_response(await entities.payload(kind))
 
+    async def config_charger(request):
+        """EV-03: the settings page's read-only look at the car charger: which
+        entities Home Assistant has and which it lacks. Reads the cached entity
+        list and writes nothing, anywhere."""
+        rows, _age = await entities.get(max_age=5.0)
+        return web.json_response(charger_probe_for(
+            str(request.query.get("entity") or "")[:255], rows))
+
     app = web.Application()
     app.add_routes([
         web.get("/", index),
@@ -486,6 +495,7 @@ def build_app(controller, selfcfg=None) -> web.Application:
         web.post("/config/save", config_save),
         web.post("/config/restart", config_restart),
         web.get("/ha/entities", ha_entities),
+        web.get("/config/charger", config_charger),
     ])
     return app
 

goodwe_controller/config.yaml

@@ -872,6 +872,18 @@ options:
   # be reached by accident.
   shadow_mode: "auto"
 
+  # ==========================================================================
+  # CAR CHARGER - read only (EV-03)
+  # ==========================================================================
+  # `off` is the shipped value and means NO CHANGE AT ALL: nothing is read,
+  # shown or announced, and the House tile and the house profile are as before.
+  # `alfen_ha` reads the charger through its Home Assistant integration. THIS
+  # VERSION NEVER WRITES TO THE CHARGER; later cards add the modes.
+  ev_charger: "off"
+  ev_current_entity: ""
+  ev_max_a: 16
+  ev_phases: "auto"
+
   # ==========================================================================
   # ADVANCED - the control law and diagnostics
   # ==========================================================================
@@ -1249,6 +1261,17 @@ schema:
   # long as nobody notices.
   shadow_mode: list(auto|on|off)
 
+  # ==========================================================================
+  # CAR CHARGER - read only (EV-03)
+  # ==========================================================================
+  # `ev_current_entity` is the charger's current setpoint; the other entities
+  # are found from its name. `ev_max_a` may not exceed the charger's own station
+  # limit (the settings page and the start check read it).
+  ev_charger: list(off|alfen_ha)
+  ev_current_entity: str?
+  ev_max_a: int(6,40)
+  ev_phases: list(1|3|auto)
+
   # ==========================================================================
   # ADVANCED - the control law and diagnostics
   # ==========================================================================

goodwe_controller/test_ev03.py

@@ -0,0 +1,953 @@
+"""EV-03: the car charger seam, read only.  `python3 -B test_ev03.py`
+
+1. This card never writes to the charger: a full set of cycles with a charger
+   configured makes no write and no service call that names a charger entity,
+   and nothing in the add-on calls the seam's write methods.
+2. The seam: entities derived by suffix, a reading believes a value only in its
+   own unit, mode3 is the primary car state and the display text a fallback.
+3. Default off changes nothing: the same cycles with the new options at their
+   defaults give the same writes, status, rows and sensors as without them.
+4. Options and validator: the census, the refusals, the live limit, the probe.
+5. Unreachable: 30 s, one amber row, one notification after 5 minutes, one
+   WARNING; back again, and it is taken down.
+6. The Car card: its states, the today figure and a meter reset, EN and NL.
+7. The car is not the house: the House tile and the house profile, and what a
+   stale car reading does (not guessed).
+8. The two-writer detector.
+9. The MQTT sensors.
+Mutating a module this file grades? Read MUTATION.md first (purge __pycache__).
+"""
+
+import ast
+import asyncio
+import io
+import json
+import logging
+import os
+import sys
+from datetime import datetime, timedelta, timezone
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, os.path.join(HERE, "tools"))
+
+from app import chargers as CH, homeview, messages, mqtt, validate      # noqa: E402
+from app.main import Controller                                        # noqa: E402
+from app import freshness, selfconfig                                  # noqa: E402
+import chargersim                                                       # noqa: E402
+from chargersim import FakeHa, SimCharger                               # noqa: E402
+
+fails = []
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+class _Store:
+    def __init__(self):
+        self.data = {}
+
+    def set(self, k, v):
+        self.data[k] = v
+
+    def get(self, k, d=None):
+        return self.data.get(k, d)
+
+    def set_time(self, k, v):
+        self.data[k] = v.isoformat()
+
+    def get_time(self, k):
+        v = self.data.get(k)
+        return None if v is None else datetime.fromisoformat(v)
+
+
+class _Pub:
+    enabled = False
+
+    def __init__(self):
+        self.values = []
+
+    def publish(self, v):
+        self.values.append(dict(v))
+
+    def close(self):
+        pass
+
+
+class Site(FakeHa):
+    """The charger model plus a grid, a battery and a solar entity."""
+
+    def __init__(self, charger, grid=500.0, batt=0.0, pv=2000.0):
+        super().__init__(charger)
+        self.grid, self.batt, self.pv = grid, batt, pv
+
+    async def limits(self, _e):
+        return (-5000.0, 5000.0, 1.0)
+
+    def _all(self):
+        out = super()._all()
+        for eid, val, unit in (("sensor.grid", self.grid, "W"),
+                               ("sensor.soc", 50.0, "%"),
+                               ("sensor.batt", self.batt, "W"),
+                               ("sensor.pv", self.pv, "W")):
+            out[eid] = {"entity_id": eid, "state": str(val),
+                        "attributes": {"unit_of_measurement": unit}}
+        return out
+
+    async def number(self, entity_id, invert=False):
+        s = self._all().get(entity_id)
+        try:
+            v = float(s["state"])
+        except (TypeError, ValueError, KeyError):
+            return None
+        return -v if invert else v
+
+    async def set_number(self, entity_id, value):
+        self.sets.append((entity_id, float(value)))
+        if entity_id == "number.sp":
+            self.batt = float(value)
+        return True
+
+
+BASE = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
+        "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
+        "meter_source": "off", "max_w": 5000, "gain": 0.6, "slew_w": 500,
+        "deadband_w": 15, "target_grid_w": 0, "step_w": 10, "heartbeat_s": 10,
+        "stale_input_s": 15, "shadow_mode": "off", "peak_cap_w": 3500,
+        "pv_entities": ["sensor.pv"]}
+CAR = dict(BASE, ev_charger="alfen_ha", ev_max_a=16, ev_phases="auto",
+           ev_current_entity="number.wallbox_power_connector_max_current_socket_1")
+SET = "number.wallbox_power_connector_max_current_socket_1"
+
+
+CLOCK = [1000.0]
+freshness.monotonic = lambda: CLOCK[0]      # one clock for the whole file
+
+
+async def rin(c, t):
+    """`read_inputs` at monotonic second `t`, on the file's one clock."""
+    CLOCK[0] = t
+    await c.read_inputs(now_mono=t)
+
+
+def build(opts, site, store=None):
+    c = Controller(dict(opts), site, store or _Store(), _Pub(),
+                   stop=asyncio.Event())
+    c.auto = True
+    c.dev_min, c.dev_max = -5000.0, 5000.0
+    return c
+
+
+async def run_cycles(c, site, n=12, tick=1.0, sim=None):
+    for i in range(n):
+        if sim is not None:
+            sim.tick(tick)
+        await rin(c, 1000.0 + i * tick)
+        await c.cycle()
+
+
+def now_utc():
+    return datetime(2026, 11, 15, 9, 30, tzinfo=timezone.utc)
+
+
+# --------------------------------------------------------------------------- #
+print("1. THIS CARD NEVER WRITES TO THE CHARGER")
+# --------------------------------------------------------------------------- #
+sim = SimCharger()
+sim.plug_in()
+sim.set_current(10)
+sim.set_phases(3)
+site = Site(sim, grid=800.0)
+c = build(CAR, site)
+asyncio.run(run_cycles(c, site, n=40, sim=sim))
+print(f"    {len(site.sets)} number writes, {len(site.calls)} service calls; "
+      f"charger power {sim.power_w:.0f} W, state {sim.state}")
+check("PRECONDITION: the charger was read (the car is charging and the card "
+      "knows it)", c.car.status(CLOCK[0]) == "charging"
+      and c.car.reading.power_w > 1000)
+check("PRECONDITION: the controller did write the inverter (the cycles are "
+      "real)", any(e == "number.sp" for e, _v in site.sets))
+check("no number write names a charger entity",
+      not [s for s in site.sets if "wallbox" in s[0]])
+check("no service call names a charger entity",
+      not [x for x in site.calls if "wallbox" in json.dumps(x[2] or {})])
+check("no service call of any kind is made but a notification",
+      all(x[0] == "persistent_notification" for x in site.calls))
+check("the model's own setpoint is where it was (10 A)", sim.setpoint_a == 10)
+check("charger_writes() finds nothing", site.charger_writes() == [])
+
+
+def callers_of_writes():
+    """(file, line, name) of every call of the seam's write methods outside
+    chargers.py itself."""
+    out = []
+    for root, _d, files in os.walk(os.path.join(HERE, "app")):
+        for f in files:
+            if not f.endswith(".py") or f == "chargers.py":
+                continue
+            tree = ast.parse(io.open(os.path.join(root, f),
+                                     encoding="utf-8").read())
+            for n in ast.walk(tree):
+                if isinstance(n, ast.Call) and isinstance(n.func,
+                                                          ast.Attribute):
+                    if n.func.attr in ("write_current", "write_phases") or (
+                            n.func.attr == "release"
+                            and "charger" in ast.dump(n.func.value)):
+                        out.append((f, n.lineno, n.func.attr))
+    return out
+
+
+check(f"nothing in the add-on calls write_current, write_phases or the "
+      f"charger's release ({callers_of_writes()})", not callers_of_writes())
+check("the seam does define them (the next cards use them)",
+      all(hasattr(CH.ChargerDriver, m) for m in
+          ("write_current", "write_phases", "release")))
+_ha = FakeHa(SimCharger())
+asyncio.run(CH.AlfenHaDriver(SET, 16).bind(_ha).write_current(99))
+check("the Alfen driver's write (a later card's) is clamped to the charger "
+      "limit, 16 A and not 99", _ha.sets == [(SET, 16.0)])
+
+# --------------------------------------------------------------------------- #
+print("\n2. THE SEAM")
+# --------------------------------------------------------------------------- #
+ents = CH.entities_for(SET)
+check("the entities come from the setpoint by suffix",
+      ents["readback"] == "sensor.wallbox_main_active_max_current_socket_1"
+      and ents["state"] == "sensor.wallbox_mode3_state_socket_1"
+      and ents["login"] == "binary_sensor.wallbox_https_api_login_status"
+      and ents["station"] == "number.wallbox_max_station_current"
+      and ents["phases_set"] == "select.wallbox_installation_max_allowed_phases"
+      and CH.entities_for(SET.replace("wallbox", "garage_wallbox"))["power"]
+      == "sensor.garage_wallbox_active_power_total_socket_1")
+check("a setpoint named otherwise derives nothing",
+      CH.entities_for("number.foo") == {} and CH.entities_for("") == {})
+check("the census key is stripped and lowercase",
+      CH.census_key("  Number.X ") == "number.x")
+
+
+def st(eid, state, unit=None):
+    a = {"unit_of_measurement": unit} if unit else {}
+    return {"entity_id": eid, "state": state, "attributes": a}
+
+
+def states(**over):
+    s = SimCharger()
+    s.plug_in()
+    s.set_current(0)            # connected, no current allowed: mode3 B
+    s.tick(5)
+    d = {r: s.states()[e] for r, e in CH.entities_for(SET).items()}
+    d.update(over)
+    return d
+
+
+r = CH.reading_from_states(states())
+check("a good set of states reads: available, connected, not charging (B)",
+      r.available and r.state == "B" and r.connected and not r.charging)
+sim2 = SimCharger()
+sim2.plug_in()
+sim2.set_current(10)
+sim2.set_phases(3)
+sim2.tick(5)
+d = {r_: sim2.states()[e] for r_, e in CH.entities_for(SET).items()}
+rc = CH.reading_from_states(d)
+check("charging: C, 6.9 kW, 3 phases of 10 A, read-back 10 A, station 16 A",
+      rc.charging and abs(rc.power_w - 6900) < 1 and len(rc.current_a) == 3
+      and abs(rc.current_a[0] - 10) < 0.1 and rc.allowed_a == 10
+      and rc.station_a == 16 and rc.phases_allowed == 3)
+rk = CH.reading_from_states(dict(d, power=st("x", "6.9", "kW")))
+check("a power in kW is believed in kW (6 900 W)", abs(rk.power_w - 6900) < 1)
+rn = CH.reading_from_states(dict(d, power=st("x", "6900")))
+check("a power with no unit is NOT believed (absent, never a guess)",
+      rn.power_w is None and rn.available)
+rn = CH.reading_from_states(dict(d, power=st("x", "6900", "V")))
+check("a power in the wrong unit is absent", rn.power_w is None)
+re_ = CH.reading_from_states(dict(d, energy=st("x", "4004850", "Wh")))
+check("energy in Wh is believed in Wh (4004.85 kWh)",
+      abs(re_.energy_kwh - 4004.85) < 1e-6)
+re_ = CH.reading_from_states(dict(d, energy=st("x", "4004.85", "W")))
+check("energy in a power unit is absent", re_.energy_kwh is None)
+ra = CH.reading_from_states(dict(d, readback=st("x", "10000", "mA")))
+check("a current in mA is believed in mA (10 A)", abs(ra.allowed_a - 10) < 1e-9)
+ra = CH.reading_from_states(dict(d, readback=st("x", "10")))
+check("a current with no unit is absent", ra.allowed_a is None)
+rf = CH.reading_from_states(dict(d, state=st("x", "unknown"),
+                                 display=st("x", "Wait Vehicle Charging")))
+check("mode3 missing: the entity is not there, so the charger is not available",
+      not rf.available)
+rf = CH.reading_from_states(dict(d, state=st("x", "STATE_A"),
+                                 display=st("x", "Charging Normal")))
+check("mode3 wins over the display text (A says no car, whatever the text says)",
+      rf.state == "A" and rf.connected is False and rf.charging is False)
+for text, want in (("Wait Vehicle Charging", "connected"),
+                   ("EV Connected", "connected"),
+                   ("Cable connected", "connected"),
+                   ("Charging Normal", "charging"),
+                   ("Available", "none"), ("Booting", None)):
+    check(f"display fallback: {text!r} is {want}",
+          CH.display_class(text) == want)
+check("mode3 letters: STATE_C, C and 'State C' all read C",
+      CH.mode3_letter("STATE_C") == "C" and CH.mode3_letter("C") == "C"
+      and CH.mode3_letter("State C") == "C")
+check("login off makes the charger unavailable",
+      not CH.reading_from_states(dict(d, login=st("x", "off"))).available)
+check("an unavailable required entity makes it unavailable",
+      not CH.reading_from_states(dict(d, readback=st("x", "unavailable"))
+                                 ).available)
+check("a charger error is carried (0: No Error is none)",
+      CH.reading_from_states(d).error is None
+      and CH.reading_from_states(dict(d, error=st("x", "17: Overcurrent"))
+                                 ).error == "17: Overcurrent")
+check("a missing optional entity (energy) does not make it unavailable",
+      CH.reading_from_states(dict(d, energy=None)).available)
+check("UnavailableCharger reads nothing and writes nothing",
+      not CH.UnavailableCharger().configured()
+      and not asyncio.run(CH.UnavailableCharger().read()).available
+      and not asyncio.run(CH.UnavailableCharger().write_current(10)))
+check("build(): off and a missing entity give UnavailableCharger, alfen_ha "
+      "with its entity the Alfen driver",
+      isinstance(CH.build(BASE), CH.UnavailableCharger)
+      and isinstance(CH.build(dict(BASE, ev_charger="alfen_ha")),
+                     CH.UnavailableCharger)
+      and isinstance(CH.build(CAR), CH.AlfenHaDriver)
+      and isinstance(CH.build(dict(CAR, ev_charger="nope")),
+                     CH.UnavailableCharger))
+pr = asyncio.run(CH.build(CAR).bind(FakeHa(SimCharger())).probe())
+check("the probe finds every entity of the model", pr.ok and not pr.missing
+      and pr.station_a == 16)
+pr = asyncio.run(CH.build(CAR).bind(FakeHa(SimCharger(name="other"))).probe())
+check("the probe lists what it does not find, and refuses the charger",
+      not pr.ok and "state" in pr.required_missing)
+
+# --------------------------------------------------------------------------- #
+print("\n3. DEFAULT OFF CHANGES NOTHING")
+# --------------------------------------------------------------------------- #
+DEFAULTS = dict(BASE, ev_charger="off", ev_current_entity="", ev_max_a=16,
+                ev_phases="auto")
+
+
+def outcome(opts):
+    simx = SimCharger()
+    simx.plug_in()
+    simx.set_current(10)
+    sitex = Site(simx, grid=900.0, pv=3000.0)
+    cx = build(opts, sitex)
+    asyncio.run(run_cycles(cx, sitex, n=15, sim=simx))
+    p = cx.status_payload()
+    p.pop("events", None)
+    chk = cx.checks()
+    return {"sets": list(sitex.sets), "calls": list(sitex.calls),
+            "targets": cx.target, "payload_now": p.get("now"),
+            "car": p.get("car"),
+            "rows": [(x["id"], x["ok"], x["warn"]) for x in chk
+                     if x["id"] not in ("controller",)],
+            "attn": [a["id"] for a in p.get("attention", [])],
+            "mqtt": sorted(cx.mqtt.values[-1]) if cx.mqtt.values else None,
+            "optional": mqtt.optional_keys(opts),
+            "census": validate.actuator_writers(opts),
+            "problems": validate.validate_startup_config(opts),
+            "hp": cx.house_profile is None}
+
+
+a, b = outcome(BASE), outcome(DEFAULTS)
+check("with the defaults (charger off) the writes are the same as without "
+      "the options", a["sets"] == b["sets"] and a["calls"] == b["calls"]
+      and a["targets"] == b["targets"])
+check("...the Home `now` block, the rows and the attention list are the same",
+      a["payload_now"] == b["payload_now"] and a["rows"] == b["rows"]
+      and a["attn"] == b["attn"])
+check("...there is no Car card, no charger row and no car sensor",
+      b["car"] is None and a["car"] is None
+      and not [r_ for r_ in b["rows"] if r_[0].startswith("charger")]
+      and b["optional"] == ("pv",)
+      and not any(str(k).startswith("car") for k in (b["mqtt"] or [])))
+check("...the census and the validator say the same",
+      a["census"] == b["census"] and a["problems"] == b["problems"] == [])
+check("...the House tile is grid + solar + battery, the car taken out of "
+      "nothing", b["payload_now"]["house_w"] is not None)
+check("a Controller with the charger off holds the null charger and reads it "
+      "never",
+      isinstance(build(DEFAULTS, Site(SimCharger())).charger,
+                 CH.UnavailableCharger))
+
+# --------------------------------------------------------------------------- #
+print("\n4. OPTIONS AND VALIDATOR")
+# --------------------------------------------------------------------------- #
+check("a charger on with a good entity and 16 A is accepted",
+      validate.validate_startup_config(CAR) == [])
+check("an unknown charger is refused, naming the setting",
+      any("ev_charger" in p for p in validate.validate_startup_config(
+          dict(CAR, ev_charger="tesla"))))
+check("ev_max_a out of 6-40 A is refused (and 5 and 41)",
+      all(any("ev_max_a" in p for p in validate.validate_startup_config(
+          dict(CAR, ev_max_a=v))) for v in (5, 41, "x")))
+check("ev_phases that is not 1, 3 or auto is refused",
+      any("ev_phases" in p for p in validate.validate_startup_config(
+          dict(CAR, ev_phases="2"))))
+check("an inverter's setpoint named as the charger is refused (the census)",
+      any("ev_current_entity" in p and "number.sp" in p for p in
+          validate.validate_startup_config(
+              dict(CAR, ev_current_entity="number.sp"))))
+_inv = dict(CAR, setpoint_entity="", inverters=[
+    {"name": "a", "setpoint_entity": SET, "max_w": 2500, "bms_soc_floor": 10}])
+check("...also one that sits in the inverter list",
+      any("ev_current_entity" in p for p in
+          validate.validate_startup_config(_inv)))
+check("...and the same entity with the charger OFF is not a writer, so not "
+      "refused", not any("ev_current_entity" in p for p in
+                         validate.validate_startup_config(
+                             dict(_inv, ev_charger="off"))))
+check("the charger is not in the inverter census (setup-only still reads 'no "
+      "inverter')", SET.lower() not in validate.actuator_writers(CAR)
+      and CH.census_entity(CAR) == SET
+      and CH.census_entity(dict(CAR, ev_charger="off")) == "")
+msgs = validate.validate_startup_config(dict(CAR, ev_current_entity="number.sp"))
+check("the refusal is at most 30 words and names the label and the key",
+      all(len(p.split()) <= 30 for p in msgs)
+      and any("“Current setting entity” (ev_current_entity)" in p
+              for p in msgs))
+
+rows = {eid: {"entity_id": eid, "state": s["state"],
+              "unit": s["attributes"].get("unit_of_measurement")}
+        for eid, s in SimCharger().states().items()}
+look = validate.charger_probe_for(SET, rows)
+check("the probe lists every entity found, with the limit",
+      look["ok"] and look["state"] == "ok" and not look["missing"]
+      and look["limit_a"] == 16 and len(look["found"]) == len(CH.ROLES))
+gone = {k: v for k, v in rows.items() if "mode3" not in k}
+look = validate.charger_probe_for(SET, gone)
+check("...and what it does not find (the car state entity)",
+      not look["ok"] and ["state", "sensor.wallbox_mode3_state_socket_1"]
+      in look["missing"])
+check("...nothing is read when Home Assistant did not answer",
+      validate.charger_probe_for(SET, None)["state"] == "unread")
+check("...a setpoint named otherwise says so",
+      validate.charger_probe_for("number.foo", rows)["state"] == "no_name")
+notes = validate.config_notes(CAR, entities=gone)
+txt = [n.text for n in notes if "ev_current_entity" in n.keys]
+check("the settings notes name the missing entity",
+      any("sensor.wallbox_mode3_state_socket_1" in t for t in txt))
+notes = validate.config_notes(CAR, entities=rows)
+txt = [(n.tier, n.text) for n in notes if "ev_current_entity" in n.keys]
+check("...and say what was found when all is well",
+      any(t == "ok" and t2.startswith("Found:") and "16 A" in t2
+          for t, t2 in txt))
+nl = [validate.note_text(n, "nl") for n in notes
+      if "ev_current_entity" in n.keys]
+check("...in Dutch too", any(t.startswith("Gevonden:") for t in nl))
+lim_notes = validate.config_notes(dict(CAR, ev_max_a=20), entities=rows)
+check("an ev_max_a above the station limit is a note on the settings page",
+      any("ev_max_a" in n.keys and "16 A" in n.text for n in lim_notes))
+
+# the live refusal of the limit, on the save path
+problems, _unread = asyncio.run(selfconfig.problems_and_ranges(
+    dict(CAR, ev_max_a=20), Site(SimCharger())))
+check("saving ev_max_a 20 A with a 16 A station limit is refused, naming the "
+      "limit", any("ev_max_a" in p and "16 A" in p for p in problems))
+problems, _unread = asyncio.run(selfconfig.problems_and_ranges(
+    dict(CAR), Site(SimCharger())))
+check("...16 A is accepted", not [p for p in problems if "ev_max_a" in p])
+problems, _unread = asyncio.run(selfconfig.problems_and_ranges(
+    dict(CAR, ev_max_a=20), Site(SimCharger(name="none"))))
+check("...a charger that cannot be read skips the check (never a guess)",
+      not [p for p in problems if "ev_max_a" in p])
+
+# run-time: a missing entity refuses the CHARGER, not the add-on
+sim3 = SimCharger(name="other")
+site3 = Site(sim3)
+c3 = build(CAR, site3)
+asyncio.run(run_cycles(c3, site3, n=3, sim=sim3))
+check("a charger whose entities are missing is refused and the add-on runs on "
+      "(the inverter is still written)",
+      c3.charger_refused == "not_found" and not c3.charger.configured()
+      and any(e == "number.sp" for e, _v in site3.sets) and not c3.refused)
+p3 = c3.status_payload()
+check("the Car card says 'Charger not set up', and one amber row",
+      p3["car"]["state"] == "setup" and p3["car"]["chip"]
+      == "Charger not set up" and p3["car"]["kind"] == "warn"
+      and [(x["id"], x["warn"], x["ok"]) for x in c3.checks()
+           if x["id"].startswith("charger")] == [("charger_setup", True,
+                                                  False)])
+simL = SimCharger(station_a=10)
+siteL = Site(simL)
+cL = build(CAR, siteL)
+asyncio.run(run_cycles(cL, siteL, n=3, sim=simL))
+check("a limit above the charger's station limit refuses the charger too",
+      cL.charger_refused == "limit" and cL.status_payload()["car"]["state"]
+      == "setup")
+
+# --------------------------------------------------------------------------- #
+print("\n5. UNREACHABLE: 30 s, ONE ROW, ONE NOTIFICATION AFTER 5 MINUTES")
+# --------------------------------------------------------------------------- #
+sim = SimCharger()
+sim.plug_in()
+sim.set_current(10)
+site = Site(sim)
+logs = []
+
+
+class _Cap(logging.Handler):
+    def emit(self, rec):
+        logs.append((rec.levelname, rec.getMessage()))
+
+
+log = logging.getLogger("goodwe")
+cap = _Cap()
+log.addHandler(cap)
+log.setLevel(logging.DEBUG)
+c = build(CAR, site)
+
+
+async def at(t):
+    sim.tick(1)
+    await rin(c, t)
+    await c.cycle()
+
+
+def mine():
+    """The calls about the charger's notification (the inverter's own clamp
+    notices are not what this is grading)."""
+    return [x for x in site.calls if (x[2] or {}).get("notification_id")
+            == "goodwe_charger_unreachable"]
+
+
+async def story():
+    out = {}
+    t = 2000.0
+    for _ in range(5):
+        await at(t)
+        t += 1
+    out["ok_state"] = c.car.status(t - 1)
+    sim.go_unavailable()
+    for k in range(29):
+        await at(t)
+        t += 1
+    out["t29"] = (c.car.status(t - 1), len(mine()))
+    for k in range(3):
+        await at(t)
+        t += 1
+    out["t32"] = (c.car.status(t - 1), [x["id"] for x in c.checks()
+                                        if x["id"].startswith("charger")])
+    out["card32"] = c.status_payload()["car"]
+    out["attn32"] = [(a["id"], a["severity"]) for a in
+                     c.status_payload()["attention"]
+                     if a["id"].startswith("charger")]
+    for k in range(250):
+        await at(t)
+        t += 1
+    out["t282"] = len([x for x in mine() if x[1] == "create"])
+    for k in range(60):
+        await at(t)
+        t += 1
+    out["t342"] = len([x for x in mine() if x[1] == "create"])
+    sim.go_unavailable(False)
+    for k in range(3):
+        await at(t)
+        t += 1
+    out["back"] = (c.car.status(t - 1), [x[1] for x in mine()])
+    return out
+
+
+S = asyncio.run(story())
+log.removeHandler(cap)
+check("PRECONDITION: while it answers, the car is charging (10 A allowed)",
+      S["ok_state"] == "charging")
+check("29 s of unavailable is not yet 'not reachable' and raises nothing",
+      S["t29"][0] == "charging" and S["t29"][1] == 0)
+check("32 s: the state is 'unreach' and there is one amber row",
+      S["t32"][0] == "unreach" and S["t32"][1] == ["charger"])
+check("the card chip is amber 'Charger not reachable' with the Why that says "
+      "what the charger does meanwhile",
+      S["card32"]["chip"] == "Charger not reachable"
+      and S["card32"]["kind"] == "warn"
+      and "keeps its last setting, 10 A" in S["card32"]["why"]
+      and "Alfen" in S["card32"]["why"])
+check("Needs attention lists it as a warning",
+      S["attn32"] == [("charger", "warning")])
+check("after 4.5 minutes there is still no notification, at 5 there is "
+      "exactly one", S["t282"] == 0 and S["t342"] == 1)
+check("...and no second one follows (one notification in all)",
+      S["t342"] == 1)
+check("the notification is the 'Charger not reachable' warning",
+      any(x[0] == "persistent_notification" and x[1] == "create"
+          and x[2]["title"] == "Charger not reachable"
+          and x[2]["notification_id"] == "goodwe_charger_unreachable"
+          for x in site.calls))
+warns = [m for lv, m in logs if lv == "WARNING" and "charger" in m.lower()]
+check(f"exactly one WARNING line, with its reason code ({warns})",
+      len(warns) == 1 and "CHARGER_UNREACHABLE" in warns[0]
+      and len(warns[0].split()) <= 25)
+check("when it answers again the notification is taken down and the card "
+      "recovers", S["back"][0] == "charging"
+      and S["back"][1][-1] == "dismiss")
+check("nothing was written to the charger through all of it",
+      site.charger_writes() == [])
+
+# the 5-minute rule on its own, with a stale clock
+t = CH.CarTracker()
+bad = CH.ChargerReading(available=False, why="x")
+t.observe(bad, 0.0, now_utc(), "d")
+check("tracker: not unreachable at 29.9 s, unreachable at 30 s, notify at "
+      "300 s", not t.unreachable(29.9) and t.unreachable(30.0)
+      and not t.notify_due(299.9) and t.notify_due(300.0))
+
+# --------------------------------------------------------------------------- #
+print("\n6. THE CAR CARD")
+# --------------------------------------------------------------------------- #
+
+
+def card(setup, lang="en", **kw):
+    sim_ = SimCharger()
+    setup(sim_)
+    site_ = Site(sim_)
+    c_ = build(CAR, site_)
+    asyncio.run(run_cycles(c_, site_, n=kw.get("n", 8), sim=sim_))
+    if "after" in kw:
+        kw["after"](sim_, site_, c_)
+    return c_.status_payload(lang=lang), c_
+
+
+def charging3(s):
+    s.plug_in()
+    s.set_phases(3)
+    s.set_current(16)
+
+
+p, c6 = card(charging3)
+car = p["car"]
+check("charging: chip Charging, big line '11.0 kW · 16 A · 3 phases'",
+      car["chip"] == "Charging" and car["big"] == "11.0 kW · 16 A · 3 phases"
+      and car["kind"] == "ok")
+check("...the mode is Off only", car["mode"] == "off")
+check("...Why says the add-on leaves it alone",
+      "leaves it alone" in car["why"])
+pn, _c = card(charging3, "nl")
+check("...in Dutch: Laadt, 11,0 kW · 16 A · 3 fasen",
+      pn["car"]["chip"] == "Laadt" and pn["car"]["big"]
+      == "11,0 kW · 16 A · 3 fasen")
+p, _c = card(lambda s: (s.plug_in(), s.set_current(0)))
+check("connected with no current allowed: 'Connected, waiting'",
+      p["car"]["chip"] == "Connected, waiting"
+      and p["car"]["big"] == "Connected, not charging.")
+p, _c = card(lambda s: None)
+check("no car: 'No car connected', 'Nothing to charge.'",
+      p["car"]["chip"] == "No car connected"
+      and p["car"]["big"] == "Nothing to charge."
+      and p["car"]["why"] == "No car is connected.")
+pn, _c = card(lambda s: None, "nl")
+check("...in Dutch", pn["car"]["chip"] == "Geen auto aangesloten"
+      and pn["car"]["big"] == "Niets te laden.")
+
+
+def full_car(s):
+    s.plug_in()
+    s.set_current(10)
+    s.car_full()
+
+
+async def _full_story():
+    sim_ = SimCharger()
+    full_car(sim_)
+    site_ = Site(sim_)
+    c_ = build(CAR, site_)
+    out = []
+    t = 3000.0
+    for i in range(130):
+        sim_.tick(1)
+        await rin(c_, t)
+        await c_.cycle()
+        if i in (30, 70, 120):
+            out.append(c_.car.status(t))
+        t += 1
+    return out, c_
+
+
+(early, mid, late), cf = asyncio.run(_full_story())
+check("a car that stopped by itself (B with current allowed) reads 'Connected, "
+      "waiting' for 60 s, then 'Car is full'",
+      early == "waiting" and mid == "full" and late == "full")
+check("...the card says 'Car is full'",
+      cf.status_payload()["car"]["chip"] == "Car is full")
+
+
+def _slow_c(sim_):
+    sim_.plug_in()
+    sim_.set_current(10)
+
+
+async def _c_low():
+    sim_ = SimCharger()
+    sim_.plug_in()
+    sim_.set_current(10)
+    sim_.tick(5)
+    site_ = Site(sim_)
+    c_ = build(CAR, site_)
+    t = 5000.0
+    seen = []
+    for i in range(700):
+        sim_.state = "C"
+        sim_.power_w = 120.0          # C, under 200 W, 10 A allowed
+        await rin(c_, t)
+        if i in (300, 650):
+            seen.append(c_.car.status(t))
+        t += 1
+    return seen
+
+
+check("mode3 C under 200 W for 10 minutes with 6 A or more allowed is full "
+      "(not before)", asyncio.run(_c_low()) == ["charging", "full"])
+
+# today's energy
+t = CH.CarTracker()
+D = "2026-11-15"
+
+
+def rd(kwh):
+    return CH.ChargerReading(available=True, state="C", power_w=5000.0,
+                             energy_kwh=kwh, allowed_a=10.0, setpoint_a=10.0)
+
+
+t.observe(rd(4000.0), 1.0, now_utc(), D)
+t.observe(rd(4001.5), 2.0, now_utc(), D)
+t.observe(rd(4003.0), 3.0, now_utc(), D)
+check("today kWh counts what the meter went up since the day began (3.0)",
+      abs(t.today_kwh - 3.0) < 1e-9)
+t.observe(rd(0.4), 4.0, now_utc(), D)
+check("a meter that resets to near zero counts from its new zero (3.4), not a "
+      "negative", abs(t.today_kwh - 3.4) < 1e-9)
+t.observe(rd(1.4), 5.0, now_utc(), D)
+check("...and goes on from there (4.4)", abs(t.today_kwh - 4.4) < 1e-9)
+t.observe(rd(1.4), 6.0, now_utc(), "2026-11-16")
+check("a new day starts again at 0", t.today_kwh == 0.0)
+t2 = CH.CarTracker()
+t2.load(t.dump())
+check("it survives a restart through the store", t2.day == "2026-11-16"
+      and t2.today_kwh == 0.0 and t2.last_kwh == 1.4)
+t3 = CH.CarTracker()
+t3.observe(CH.ChargerReading(available=True, state="B", power_w=0.0,
+                             energy_kwh=None), 1.0, now_utc(), D)
+check("a charger with no energy meter has no figure (not 0)",
+      t3.today_kwh == 0.0 and t3.last_kwh is None)
+p, c6 = card(charging3)
+check("the card says how much was charged today when it knows",
+      p["car"]["today"] is not None and "charged today" in p["car"]["today"])
+check("...and the store holds it", "car_today" in c6.store.data
+      and c6.store.data["car_today"]["day"])
+
+# the states have words in both languages, and the long ones are split
+with homeview.in_lang("nl"):
+    nlw = {k: homeview.car_state_word(k) for k in
+           ("charging", "waiting", "none", "full", "unreach", "setup", None)}
+check("every state has a Dutch chip", all(nlw.values())
+      and nlw["unreach"] == "Laadpaal onbereikbaar")
+check("no entity id is on the card or in its words",
+      "wallbox" not in json.dumps(p["car"]) and "sensor." not in json.dumps(
+          p["car"]))
+
+# --------------------------------------------------------------------------- #
+print("\n7. THE CAR IS NOT THE HOUSE")
+# --------------------------------------------------------------------------- #
+nb = homeview.now_block(grid_w=7000.0, pv_w=1000.0, batt_w=-1000.0,
+                        has_solar=True, shadow=False)
+nc = homeview.now_block(grid_w=7000.0, pv_w=1000.0, batt_w=-1000.0,
+                        has_solar=True, shadow=False, car_w=6900.0)
+check("without a charger House is grid + solar + battery (7.0 kW)",
+      nb["house_w"] == 7000.0)
+check("with a charger the car's 6.9 kW is not in House (100 W)",
+      abs(nc["house_w"] - 100.0) < 1e-9)
+nu = homeview.now_block(grid_w=7000.0, pv_w=1000.0, batt_w=-1000.0,
+                        has_solar=True, shadow=False, car_unknown=True)
+check("a car that cannot be read leaves House with no figure ('No reading')",
+      nu["house_w"] is None and nu["lines"]["house"] == "No reading")
+
+sim = SimCharger()
+sim.plug_in()
+sim.set_phases(3)
+sim.set_current(10)
+site = Site(sim, grid=7500.0, batt=0.0, pv=500.0)
+c = build(CAR, site)
+asyncio.run(run_cycles(c, site, n=12, sim=sim))
+n_ = c.status_payload()["now"]
+car_w = c.car.reading.power_w
+check(f"in the real controller House excludes the {car_w:.0f} W car "
+      f"(House {n_['house_w']:.0f} W)",
+      car_w > 5000 and abs(n_["house_w"] - (site.grid + site.pv + c.site_batt_w
+                                            - car_w)) < 1.0)
+
+# the house profile's sample
+samples = []
+
+
+class _HP:
+    def add(self, now, w):
+        samples.append(w)
+        return False
+
+    def estimator(self):
+        return None
+
+
+def profile_sample(setup):
+    del samples[:]
+    sim_ = SimCharger()
+    setup(sim_)
+    site_ = Site(sim_, grid=7500.0, batt=0.0, pv=500.0)
+    opts = dict(CAR, pv_forecast_entity="sensor.fc", battery_modes_enabled=True)
+    c_ = build(opts, site_)
+    c_.house_profile = _HP()
+    asyncio.run(run_cycles(c_, site_, n=12, sim=sim_))
+    return c_, site_, list(samples)
+
+
+c_p, site_p, got = profile_sample(lambda s: (s.plug_in(), s.set_phases(3),
+                                             s.set_current(10)))
+want = site_p.grid + c_p.site_batt_w - c_p.car.reading.power_w
+check(f"the house profile's sample leaves the car out (last sample "
+      f"{got[-1]:.0f} W, want {want:.0f} W; the car is "
+      f"{c_p.car.reading.power_w:.0f} W)",
+      len(got) >= 10 and abs(got[-1] - want) < 1.0
+      and c_p.car.reading.power_w > 5000)
+c_q, site_q, got = profile_sample(lambda s: None)
+check(f"with no car the sample is the whole load, as before "
+      f"({got[-1]:.0f} W = grid + battery)",
+      abs(got[-1] - (site_q.grid + c_q.site_batt_w)) < 1.0)
+# not fresh: the last good reading is long ago and the charger now fails
+c_r, site_r, _g = profile_sample(lambda s: (s.plug_in(), s.set_phases(3),
+                                            s.set_current(10)))
+del samples[:]
+c_r.car.last_ok = 0.0
+site_r.charger.go_unavailable()
+
+
+async def _stale_cycle():
+    await rin(c_r, 1013.0)
+    await c_r.cycle()
+
+
+asyncio.run(_stale_cycle())
+check("a car reading older than stale_input_s: the House profile skips that "
+      "moment (None), it does not keep the last car figure",
+      samples and samples[-1] is None)
+mono_now = c_r.car_now()
+check("...and the House tile shows no reading, as the rule says "
+      "(freshness: the car's power counts only while the last good reading "
+      "is within stale_input_s, the meter's own bound)",
+      c_r.car.car_w(mono_now) is None and c_r.car_term() == (0.0, True)
+      and c_r.status_payload()["now"]["house_w"] is None)
+check("the stale bound is the meter's own (15 s from stale_input_s)",
+      CH.CarTracker(stale_s=15.0).stale_s == 15.0
+      and build(dict(CAR, stale_input_s=40), Site(SimCharger())
+                ).car.stale_s == 40.0)
+tt = CH.CarTracker(stale_s=15.0)
+tt.observe(rd(1.0), 100.0, now_utc(), D)
+check("tracker: fresh at 15 s, not at 15.1 s",
+      tt.car_w(115.0) == 5000.0 and tt.car_w(115.1) is None)
+
+# --------------------------------------------------------------------------- #
+print("\n8. THE TWO-WRITER DETECTOR")
+# --------------------------------------------------------------------------- #
+sim = SimCharger()
+sim.plug_in()
+sim.set_current(10)
+site = Site(sim)
+c = build(CAR, site)
+
+
+async def writer_story():
+    t = 7000.0
+    out = {}
+    for _ in range(5):
+        await rin(c, t)
+        t += 1
+    out["quiet"] = [x["id"] for x in c.checks() if x["id"] == "charger_writer"]
+    sim.set_current(12)                 # the owner's automation
+    await rin(c, t)
+    t += 1
+    out["changed"] = [(x["id"], x["ok"], x["warn"]) for x in c.checks()
+                      if x["id"] == "charger_writer"]
+    out["attn"] = [(a["id"], a["severity"], a["title"])
+                   for a in c.status_payload()["attention"]
+                   if a["id"] == "charger_writer"]
+    return out
+
+
+W = asyncio.run(writer_story())
+check("with nothing changing the charger there is no row", W["quiet"] == [])
+check("a change of the setpoint (another writer) shows the grey info row",
+      W["changed"] == [("charger_writer", True, True)])
+check("...in Needs attention as info, titled in plain words",
+      W["attn"] == [("charger_writer", "info",
+                     "Something else changes the charger")])
+check("...in Dutch too", messages.row_charger_writer().nl
+      == "Iets anders wijzigt de laadpaal en deze add-on leest alleen.")
+tw = CH.CarTracker()
+tw.observe(CH.ChargerReading(available=True, setpoint_a=10, allowed_a=10), 1,
+           now_utc(), D)
+tw.observe(CH.ChargerReading(available=True, setpoint_a=None, allowed_a=10), 2,
+           now_utc(), D)
+check("a reading that did not answer (None) is not a change",
+      tw.writer_at_utc is None)
+tw.observe(CH.ChargerReading(available=True, setpoint_a=10, allowed_a=8), 3,
+           now_utc(), D)
+check("a change of the read-back alone is another writer",
+      tw.writer_at_utc is not None)
+check("it stays for an hour and then goes",
+      tw.other_writer(now_utc() + timedelta(minutes=59))
+      and not tw.other_writer(now_utc() + timedelta(minutes=61)))
+
+# --------------------------------------------------------------------------- #
+print("\n9. THE MQTT SENSORS")
+# --------------------------------------------------------------------------- #
+check("the three car sensors are announced only with a charger on",
+      mqtt.optional_keys(CAR) == ("pv", "car_power", "car_state",
+                                  "car_today_kwh")
+      and mqtt.optional_keys(BASE) == ("pv",))
+pl = dict(mqtt.discovery_payloads(include=mqtt.optional_keys(CAR)))
+names = sorted(k for k in pl if "_car_" in k)
+check(f"discovery announces three sensors, goodwe_car_power, "
+      f"goodwe_car_state and goodwe_car_today_kwh ({names})",
+      [pl[k]["object_id"] for k in names] == [
+          "goodwe_car_power", "goodwe_car_state", "goodwe_car_today_kwh"]
+      and all(k.startswith("homeassistant/sensor/") for k in names))
+check("...and no command entity (no select, number or switch for the car)",
+      not [k for k in pl if "car" in k and not k.startswith(
+          "homeassistant/sensor/")])
+sim = SimCharger()
+sim.plug_in()
+sim.set_phases(3)
+sim.set_current(10)
+site = Site(sim)
+c = build(CAR, site)
+asyncio.run(run_cycles(c, site, n=10, sim=sim))
+c.publish()
+sent = c.mqtt.values[-1]
+check("publish() sends the three car values on their topics",
+      sent.get("car_state") == "Charging" and sent["car_power"] > 5000
+      and "car_today_kwh" in sent)
+c0 = build(BASE, Site(SimCharger()))
+c0.publish()
+check("...and none of them without a charger",
+      not [k for k in c0.mqtt.values[-1] if k.startswith("car_")])
+vals = c.car_values()
+check("published values: state text, power in W, today kWh",
+      vals["car_state"] == "Charging" and vals["car_power"] > 5000
+      and vals["car_today_kwh"] is not None)
+sim.go_unavailable()
+asyncio.run(run_cycles(c, site, n=40, sim=sim))
+vals = c.car_values()
+check("unreachable: the state says so, power and today are not zero but "
+      "absent for power (HA's None payload)",
+      vals["car_state"] == "Charger not reachable"
+      and vals["car_power"] == mqtt.PAYLOAD_NONE)
+check("an absent number goes out as HA's own 'None' payload, and the PV rule "
+      "(ABSENT_AS_UNKNOWN is one key) is not widened",
+      mqtt.ABSENT_AS_UNKNOWN == frozenset({"pv"}))
+
+print()
+if fails:
+    print(f"{len(fails)} FAILED")
+    for f_ in fails:
+        print("  -", f_)
+    sys.exit(1)
+print("all passed")

goodwe_controller/test_selfconfig.py

@@ -637,9 +637,10 @@ _routes = sorted({(r.method, r.get_info().get("path"))
                   for r in _app.router.routes() if r.method != "HEAD"})
 print("    " + "  ".join(f"{m} {p}" for m, p in _routes))
 check("the configuration surface adds five routes, UX-13 one GET and "
-      "UX-54 one more GET (the Forecast tab), and none touches the three "
+      "UX-54 one more GET (the Forecast tab), EV-03 one read-only GET (the car "
+      "charger probe), and none touches the three "
       "that were there",
-      _routes == [("GET", "/"), ("GET", "/config"), ("GET", "/config/model"),
+      _routes == [("GET", "/"), ("GET", "/config"), ("GET", "/config/charger"), ("GET", "/config/model"),
                   ("GET", "/forecast"), ("GET", "/ha/entities"),
                   ("GET", "/status"),
                   ("POST", "/action"), ("POST", "/config/restart"),

goodwe_controller/test_settings.py

@@ -80,12 +80,13 @@ for _g in LAYOUT["groups"]:
             _flat += [(_g["id"], _s) for _s in _f.get("fields", [])]
 _keys = [f["key"] for _g, f in _flat]
 FIELD = {f["key"]: f for _g, f in _flat}
-check("the six groups, in the design's order",
+check("the seven groups, in the design's order",
       [g["id"] for g in LAYOUT["groups"]]
-      == ["meter", "inverters", "tariff", "safety", "maintenance", "tuning"]
+      == ["meter", "inverters", "tariff", "safety", "car", "maintenance",
+          "tuning"]
       and [g["name"] for g in LAYOUT["groups"]]
       == ["Meter", "Inverters and batteries", "Tariff and prices",
-          "Safety limits", "Maintenance", "Control tuning"])
+          "Safety limits", "Car charger", "Maintenance", "Control tuning"])
 check(f"every field of the table is on the page once ({len(_keys)} fields)",
       sorted(_keys) == sorted(f.key for _g, _p, f in L.fields())
       and len(_keys) == len(set(_keys)))
@@ -146,7 +147,7 @@ check("a page-only field names the option keys it writes, for the grey key "
       == ["pv_entities", "pv_entity", "pv_invert_entities"]
       and "static_profile" in field(FIELD, "ui.contract")["writes"])
 check("without a Supervisor schema the page still draws, with no ranges",
-      len(uipage.layout([], {})["groups"]) == 6)
+      len(uipage.layout([], {})["groups"]) == 7)
 
 # --------------------------------------------------------------------------- #
 print("\n2. the page script names no option key")

goodwe_controller/test_translations.py

@@ -205,10 +205,11 @@ print("\n4. task ordering, and the labels that carry it")
 # prefixes appear in the file as CONTIGUOUS RUNS in the task order. A group
 # whose members are scattered renders as a jumble however good each label is.
 GROUPS = ["Hardware", "Meter", "Limits", "Tariff", "Maintenance", "Safety",
-          "Advanced"]
+          "Car charger", "Advanced"]
 NL_GROUPS = {"Hardware": "Hardware", "Meter": "Meter", "Limits": "Grenzen",
              "Tariff": "Tarief", "Maintenance": "Onderhoud",
-             "Safety": "Veiligheid", "Advanced": "Geavanceerd"}
+             "Safety": "Veiligheid", "Car charger": "Laadpaal",
+             "Advanced": "Geavanceerd"}
 _prefixes = []
 for key in ORDER:
     name = TR["en"]["configuration"][key]["name"]

goodwe_controller/tools/chargersim.py

@@ -0,0 +1,261 @@
+#!/usr/bin/env python3
+"""EV-03: a model of an Alfen car charger behind Home Assistant, for the unit
+tests and, later, for TEST-23.
+
+WHERE IT LIVES, AND WHY. The rig's hardware sims (`hwsim.py`, `plantsim.py`) are
+in the infra repo admin/goodwe under `sim/`, pinned by `.gitea/infra-pin`; there
+is no `sim/` in this repository. This card may write only to this repository, so
+the model sits beside the add-on's other tools. Moving it next to `hwsim.py` is a
+copy: it imports nothing from the add-on.
+
+WHAT IT MODELS (every entity the add-on's Alfen driver reads, with the names and
+units the leeyuentuen/alfen_wallbox integration gives, from the owner's export):
+
+  * mode3 state A (no car), B (connected, not charging), C (charging)
+  * power that RAMPS to what the car takes (`RAMP_W_S`), 1 or 3 phases
+  * the current setpoint, its read-back, the station limit (16 A at the owner's)
+  * a full car (it stops by itself: state B with the current still allowed)
+  * the integration going unavailable, and the login sensor off
+
+    from chargersim import SimCharger, FakeHa
+    c = SimCharger("wallbox"); c.plug_in(); c.set_current(10); c.tick(30)
+    ha = FakeHa(c)                       # a Home Assistant that serves its states
+
+`FakeHa` has the methods the add-on calls (`state`, `number`, `states`,
+`set_number`, `call`) and RECORDS every write, so a test can prove none happened.
+`python3 chargersim.py --selftest` runs the model's own checks. `serve(port)`
+answers the REST calls the add-on makes (`GET /api/states[/<id>]`), for a rig.
+"""
+
+import json
+import sys
+
+VOLTS = 230.0
+RAMP_W_S = 3000.0           # how fast the power follows its target
+MIN_A = 6.0
+
+
+class SimCharger:
+    def __init__(self, name="wallbox", station_a=16.0, setpoint_a=10.0,
+                 phases_allowed=1, kwh=4004.85):
+        self.x = name
+        self.station_a = float(station_a)
+        self.setpoint_a = float(setpoint_a)
+        self.readback_a = float(setpoint_a)
+        self.phases_allowed = phases_allowed
+        self.car_phases = 3           # what the car can use
+        self.state = "A"
+        self.full = False
+        self.power_w = 0.0
+        self.kwh = float(kwh)
+        self.unavailable = False
+        self.login_on = True
+        self.error = "0: No Error"
+
+    # -- the world acts on it -------------------------------------------------
+    def plug_in(self, car_phases=3):
+        self.car_phases = car_phases
+        self.full = False
+        self.state = "B"
+
+    def unplug(self):
+        self.state, self.power_w = "A", 0.0
+
+    def car_full(self):
+        self.full = True
+
+    def set_current(self, a):
+        """What an automation (or, later, this add-on) does to the setpoint."""
+        a = max(0.0, min(float(a), self.station_a))
+        self.setpoint_a = self.readback_a = a
+
+    def set_phases(self, n):
+        self.phases_allowed = 3 if n == 3 else 1
+
+    def go_unavailable(self, on=True):
+        self.unavailable = bool(on)
+
+    def login(self, on=True):
+        self.login_on = bool(on)
+
+    # -- time -----------------------------------------------------------------
+    @property
+    def phases_used(self):
+        return min(self.car_phases, self.phases_allowed)
+
+    def target_w(self):
+        if self.state == "A" or self.full or self.setpoint_a < MIN_A:
+            return 0.0
+        return self.setpoint_a * VOLTS * self.phases_used
+
+    def tick(self, seconds=1.0):
+        for _ in range(int(max(1, round(seconds)))):
+            if self.state != "A":
+                self.state = "C" if self.target_w() > 0 else "B"
+            tgt = self.target_w()
+            step = RAMP_W_S
+            if self.power_w < tgt:
+                self.power_w = min(tgt, self.power_w + step)
+            else:
+                self.power_w = max(tgt, self.power_w - step)
+            self.kwh += self.power_w / 3.6e6
+
+    # -- what Home Assistant would serve ---------------------------------------
+    def _amps(self):
+        per = (self.power_w / (VOLTS * self.phases_used)
+               if self.phases_used and self.power_w else 0.0)
+        return [per if i < self.phases_used else 0.0 for i in range(3)]
+
+    def states(self) -> dict:
+        x = self.x
+        bad = "unavailable" if self.unavailable else None
+
+        def st(eid, value, unit=None, **attrs):
+            a = dict(attrs)
+            if unit:
+                a["unit_of_measurement"] = unit
+            return eid, {"entity_id": eid,
+                         "state": bad or str(value), "attributes": a}
+        amps = self._amps()
+        rows = [
+            st(f"number.{x}_power_connector_max_current_socket_1",
+               f"{self.setpoint_a:g}", "A", min=0, max=40, step=1),
+            st(f"sensor.{x}_main_active_max_current_socket_1",
+               f"{self.readback_a:g}", "A"),
+            st(f"sensor.{x}_mode3_state_socket_1", f"STATE_{self.state}"),
+            st(f"sensor.{x}_display_state_socket_1",
+               {"A": "Available", "B": "EV Connected",
+                "C": "Charging Normal"}[self.state]),
+            st(f"sensor.{x}_active_power_total_socket_1",
+               f"{self.power_w:.0f}", "W"),
+            st(f"sensor.{x}_meter_reading_socket_1", f"{self.kwh:.2f}",
+               "kWh"),
+            st(f"sensor.{x}_current_l1_socket_1", f"{amps[0]:.1f}", "A"),
+            st(f"sensor.{x}_current_l2_socket_1", f"{amps[1]:.1f}", "A"),
+            st(f"sensor.{x}_current_l3_socket_1", f"{amps[2]:.1f}", "A"),
+            st(f"number.{x}_max_station_current", f"{self.station_a:g}", "A"),
+            st(f"sensor.{x}_main_station_active_max_current_socket_1",
+               f"{self.station_a:g}", "A"),
+            st(f"select.{x}_installation_max_allowed_phases",
+               "3 Phases" if self.phases_allowed == 3 else "1 Phase"),
+            st(f"sensor.{x}_connector_1_max_allowed_of_phases",
+               "3 Phases" if self.phases_allowed == 3 else "1 Phase"),
+            st(f"sensor.{x}_display_error_number_socket_1", self.error),
+            (f"binary_sensor.{x}_https_api_login_status",
+             {"entity_id": f"binary_sensor.{x}_https_api_login_status",
+              "state": bad or ("on" if self.login_on else "off"),
+              "attributes": {"device_class": "connectivity"}}),
+        ]
+        return dict(rows)
+
+
+class FakeHa:
+    """A Home Assistant that serves a `SimCharger` (and any `extra` states) and
+    records every service call and number write it is given."""
+
+    session = None
+    token = ""
+    read_only = False
+
+    def __init__(self, charger=None, extra=None):
+        self.charger = charger
+        self.extra = dict(extra or {})
+        self.calls = []             # (domain, service, data)
+        self.sets = []              # (entity, value)
+
+    def _all(self):
+        out = dict(self.extra)
+        if self.charger is not None:
+            out.update(self.charger.states())
+        return out
+
+    async def state(self, entity_id):
+        return self._all().get(entity_id)
+
+    async def states(self):
+        return list(self._all().values())
+
+    async def number(self, entity_id, invert=False):
+        s = self._all().get(entity_id)
+        try:
+            v = float(s["state"])
+        except (TypeError, ValueError, KeyError):
+            return None
+        return -v if invert else v
+
+    async def limits(self, _entity_id):
+        return (-5000.0, 5000.0, 1.0)
+
+    async def set_number(self, entity_id, value):
+        self.sets.append((entity_id, value))
+        return True
+
+    async def call(self, domain, service, data=None):
+        self.calls.append((domain, service, data))
+        return True
+
+    def charger_writes(self):
+        """Every write or service call that names a charger entity."""
+        x = self.charger.x if self.charger else "\0"
+        hits = [s for s in self.sets if x in s[0]]
+        hits += [c for c in self.calls if x in json.dumps(c[2] or {})]
+        return hits
+
+
+def serve(sim, port=8123):
+    """Answer `GET /api/states` and `GET /api/states/<id>` from `sim`, for a
+    rig. Reads only: any POST answers 405, so nothing can write the model."""
+    import http.server
+
+    class H(http.server.BaseHTTPRequestHandler):
+        def do_GET(self):
+            rows = sim.states()
+            path = self.path.split("?")[0].rstrip("/")
+            if path.endswith("/api/states"):
+                body = list(rows.values())
+            else:
+                body = rows.get(path.rsplit("/api/states/", 1)[-1])
+            self.send_response(200 if body is not None else 404)
+            self.send_header("Content-Type", "application/json")
+            self.end_headers()
+            self.wfile.write(json.dumps(body).encode())
+
+        def do_POST(self):
+            self.send_response(405)
+            self.end_headers()
+
+        def log_message(self, *_a):
+            pass
+
+    srv = http.server.ThreadingHTTPServer(("0.0.0.0", port), H)
+    return srv
+
+
+def _selftest():
+    c = SimCharger()
+    assert c.states()[f"sensor.wallbox_mode3_state_socket_1"]["state"] == "STATE_A"
+    c.plug_in(); c.tick(1)
+    assert c.state == "B" or c.state == "C"
+    c.set_current(10); c.set_phases(3); c.tick(5)
+    assert c.state == "C" and abs(c.power_w - 10 * 230 * 3) < 1, c.power_w
+    c.set_phases(1); c.tick(5)
+    assert abs(c.power_w - 2300) < 1
+    c.set_current(5); c.tick(5)
+    assert c.state == "B" and c.power_w == 0       # under 6 A: paused
+    c.set_current(10); c.car_full(); c.tick(5)
+    assert c.state == "B" and c.power_w == 0 and c.readback_a == 10
+    c.go_unavailable()
+    assert c.states()[f"sensor.wallbox_active_power_total_socket_1"]["state"] \
+        == "unavailable"
+    c.go_unavailable(False); c.login(False)
+    assert c.states()["binary_sensor.wallbox_https_api_login_status"]["state"] == "off"
+    print("chargersim selftest ok")
+
+
+if __name__ == "__main__":
+    if "--selftest" in sys.argv:
+        _selftest()
+    else:
+        srv = serve(SimCharger(), int(sys.argv[1]) if len(sys.argv) > 1 else 8123)
+        print("serving the charger model, GET only; Ctrl-C stops it")
+        srv.serve_forever()

goodwe_controller/tools/ux13_demo.py

@@ -36,7 +36,7 @@ sys.path.insert(0, os.path.join(HERE, "tools"))
 
 from aiohttp import web as aioweb                                # noqa: E402
 
-from app import homeview, packmode, web                          # noqa: E402
+from app import chargers, freshness, homeview, packmode, web     # noqa: E402
 from app.main import Controller                                  # noqa: E402
 from app.maintenance import DRAIN                                # noqa: E402
 from app.setuponly import SetupOnly                              # noqa: E402
@@ -44,6 +44,7 @@ from app.peakguard import PeakDecision                           # noqa: E402
 from app.price import PricePoint, PriceProduct, PriceRead        # noqa: E402
 from app.static_tariff import TariffPlan                         # noqa: E402
 import translations_lint as TL                                   # noqa: E402
+from chargersim import SimCharger                                # noqa: E402
 
 PRICES = [.24, .22, .21, .20, .21, .22, .24, .27, .26, .22, .18, .16, .15,
           .14, .16, .19, .23, .28, .33, .36, .34, .29, .26, .24]
@@ -104,6 +105,61 @@ class Hass:
         return STATES
 
 
+# EV-03: the car states. A charger model behind the demo's Home Assistant; the
+# real `Controller._read_charger()` reads it, so the card is what the add-on
+# draws, not a picture.
+CAR_STATES = ("car", "car_none", "car_waiting", "car_full", "car_unreach",
+              "car_setup")
+CAR_ENTITY = "number.wallbox_power_connector_max_current_socket_1"
+SIM = SimCharger()
+
+
+class CarHass(Hass):
+    async def state(self, e):
+        return SIM.states().get(e) or await super().state(e)
+
+    async def states(self):
+        return list(STATES) + list(SIM.states().values())
+
+
+def car_setup(state):
+    """Put the model in the state's situation."""
+    global SIM
+    SIM = SimCharger(name="other" if state == "car_setup" else "wallbox")
+    if state in ("car", "car_unreach"):
+        SIM.plug_in()
+        SIM.set_phases(3)
+        SIM.set_current(16)
+    elif state == "car_waiting":
+        SIM.plug_in()
+        SIM.set_current(0)
+    elif state == "car_full":
+        SIM.plug_in()
+        SIM.set_current(10)
+        SIM.car_full()
+    SIM.tick(12)
+
+
+async def car_feed(c, state):
+    """The cycle's own read, once a second, like the control loop."""
+    backdated = False
+    n = 0
+    while True:
+        n += 1
+        if state == "car_unreach" and n == 4:
+            SIM.go_unavailable()
+        SIM.tick(1)
+        await c._read_charger(freshness.monotonic())
+        if state == "car_unreach" and c.car.bad_since is not None \
+                and not backdated:
+            backdated = True        # 12 minutes in, so the row and the note show
+            c.car.bad_since -= 720.0
+            c.car.bad_since_utc -= timedelta(seconds=720)
+        if state == "car_full" and c.car.b_since is not None and n < 3:
+            c.car.b_since -= 120.0  # the full dwell has passed
+        await asyncio.sleep(1.0)
+
+
 KETTLE_READS = (-10, 1990, 1990, -10)
 KETTLE_COUNT = [0]
 
@@ -244,7 +300,13 @@ def controller(state):
     opts = dict(OPTS, shadow_mode="on" if state == "shadow" else "off")
     if state in P3:
         opts["battery_modes_enabled"] = True
-    c = Controller(opts, Hass(), Store(), Pub(), stop=asyncio.Event())
+    hass = Hass()
+    if state in CAR_STATES:
+        car_setup(state)
+        opts.update(ev_charger="alfen_ha", ev_current_entity=CAR_ENTITY,
+                    ev_max_a=16, ev_phases="auto")
+        hass = CarHass()
+    c = Controller(opts, hass, Store(), Pub(), stop=asyncio.Event())
     c.auto = True
     now = datetime.now(timezone.utc)
     day0, _day1 = homeview.local_day(now)
@@ -264,6 +326,12 @@ def controller(state):
             inv.last_write = 0.0
     c.grid = -10.0 if state != "shadow" else -2810.0
     c.pv_w = 3900.0
+    if state in CAR_STATES:
+        # 11 kW to the car on top of a 1.6 kW house, all from the roof
+        c.grid, c.pv_w = 0.0, (12600.0 if state == "car" else 2400.0)
+        for inv in c.inverters:
+            inv.batt = 0.0
+        c.car.stale_s = 15.0
     if state == "alarm":
         c.inverters[0].sign_wd.tripped = True
     if state in P3:
@@ -327,9 +395,17 @@ async def main(state, port, host="127.0.0.1"):
                   battery_modes_enabled=state in P3,
                   inverters=[dict(i, bms_soc_floor=10)
                              for i in OPTS["inverters"]])
-    runner = aioweb.AppRunner(web.build_app(c, Supervisor(stored)))
+    stored.update(ev_charger=c.o.get("ev_charger", "off"),
+                  ev_current_entity=c.o.get("ev_current_entity", ""),
+                  ev_max_a=c.o.get("ev_max_a", 16),
+                  ev_phases=c.o.get("ev_phases", "auto"))
+    app = web.build_app(c, Supervisor(stored))
+    runner = aioweb.AppRunner(app)
     await runner.setup()
     await aioweb.TCPSite(runner, host, port).start()
+    if state in CAR_STATES:
+        asyncio.get_running_loop().create_task(car_feed(c, state))
+        main.app = app
     print(f"serving {state} on http://127.0.0.1:{port}/", flush=True)
     await asyncio.Event().wait()
 

goodwe_controller/translations/en.yaml

@@ -283,6 +283,18 @@ configuration:
   shadow_mode:
     name: "Safety · Shadow mode"
     description: "The add-on makes every decision but keeps every inverter at 0 W. 'auto' does this while control is stopped, 'on' always, 'off' never."
+  ev_charger:
+    name: "Car charger · Charger"
+    description: "Which car charger the add-on reads. Off leaves everything as it is. This version only reads the charger and never changes it."
+  ev_current_entity:
+    name: "Car charger · Current setting entity"
+    description: "The entity that sets the charger's current. The add-on finds the charger's other entities from its name."
+  ev_max_a:
+    name: "Car charger · Charger limit (A)"
+    description: "The most current the charger may ever be asked for. Never above what the charger reports. Lower it for a weaker fuse or cable."
+  ev_phases:
+    name: "Car charger · Phases"
+    description: "How many phases the charger may use: 1, 3, or 1 and 3 (auto)."
   auto_start:
     name: "Advanced · Start control on boot"
     description: "Off by default. Leave it off until the site is commissioned, so someone presses Start after each restart."

goodwe_controller/translations/nl.yaml

@@ -264,6 +264,18 @@ configuration:
   shadow_mode:
     name: "Veiligheid · Schaduwmodus"
     description: "De add-on neemt elke beslissing maar houdt elke omvormer op 0 W. 'auto' doet dit zolang de regeling gestopt is, 'on' altijd, 'off' nooit."
+  ev_charger:
+    name: "Laadpaal · Laadpaal"
+    description: "Welke laadpaal de add-on leest. Uit laat alles zoals het is. Deze versie leest de laadpaal alleen en wijzigt niets."
+  ev_current_entity:
+    name: "Laadpaal · Entiteit van de laadstroom"
+    description: "De entiteit die de stroom van de laadpaal instelt. De add-on vindt de andere entiteiten van de laadpaal via de naam."
+  ev_max_a:
+    name: "Laadpaal · Laadpaalgrens (A)"
+    description: "De hoogste stroom die aan de laadpaal gevraagd mag worden. Nooit hoger dan wat de laadpaal meldt. Verlaag ze voor een zwakkere zekering of kabel."
+  ev_phases:
+    name: "Laadpaal · Fasen"
+    description: "Hoeveel fasen de laadpaal mag gebruiken: 1, 3, of 1 en 3 (auto)."
   auto_start:
     name: "Geavanceerd · Regeling starten bij opstarten"
     description: "Staat standaard uit. Laat het uit tot de installatie in dienst is, zodat iemand na elke herstart op Start drukt."