goodwe-addon/goodwe_controller/app/freeze.py

"""SAFETY-18 - the frozen-but-fresh meter detector.

WHAT IT CATCHES, in one sentence: a meter that keeps answering 200 OK with a
document that never changes, which every freshness mechanism in this stack reads
as perfectly healthy while the loop winds to the rail and stays there.

THE FAILURE NEEDS NO FAULT ANYWHERE ELSE. A frozen-but-talking meter is
*arriving* on every poll, so:

  - `meter_max_age_s` is structurally blind. Every response resets the arrival
    stamp (p1.py, `HomeWizardLocalSource`), so the age never climbs. Age catches
    a meter that stops TALKING; this one stops CHANGING.
  - firmware layer 1 is fed. Its watchdog keys on setpoint staleness, which the
    heartbeat satisfies, and on `sensor.goodwe_p1_sample_age`, which a talking
    meter satisfies.
  - the arbiter contributes only the device rating and +-`max_w` - it bounds the
    magnitude, which is the number this fault drives us to, not away from.
  - there is no SoC guard on the grid-following loop at all. `soc_floor` belongs
    to the maintenance cycle; the BMS is the sole energy backstop.

The loop reads a constant error and integrates it. Time from freeze to the rail,
replaying the real `control.compute()` at 15 s cycles:

    meter frozen at   to rail, max_w 2000   to rail, max_w 5000
    350 W                    150 s                  360 s
    1000 W                    60 s                  135 s
    >= 3000 W                 30 s                   75 s

⚠️ THE LIVE INSTALL RUNS `max_w: 5000`, which is both the schema maximum and the
device rating, so the right-hand column is the one that describes this house.

------------------------------------------------------------------------------
THIS HAS A EURO FIGURE, AND IT IS THE REASON THE TICKET IS P1

`maintenance_charge_floor()` caps MAINTENANCE charging against `peak_cap_w`. The
grid-following loop is not capped by anything but `max_w`. A meter frozen
reading a large export drives a 5 kW unattended grid IMPORT, and >= 15 minutes
of that sets the month's billed capaciteitstarief peak: specs.md:583 records a
measured case where a single 4998 W quarter moved a monthly peak from 4832 W to
5013 W and billed the whole month at it.

⚠️ THIS CUT DOES NOT CAP THAT. It notifies. See DOCS.md, "Capacity-tariff
exposure of the grid-following loop", and SAFETY-07.

------------------------------------------------------------------------------
WHY IT IS A CONJUNCTION AND NOT AN `unchanged_s` THRESHOLD

`unchanged_s` (p1.py) already measures exactly the right thing: seconds since
the meter served a DIFFERENT measurement, fingerprinted over the power figure
and both energy registers. It is computed, it is on the status page, and TEL-05
deliberately refused to threshold it - correctly:

    this controller regulates grid power toward ~0 W, and at a converged -10 W
    the export register takes about six minutes to advance by its 1 Wh
    resolution while the power figure legitimately repeats.

So a bare `unchanged_s > N` false-trips at the exact operating point we aim for,
and `p1.py` says as much in a note that ends "whoever wires a freeze detector
must handle the low-power case first". This is that handling, and it is not a
smaller number:

    unchanged_s >= hold_s   AND   |commanded_w| is at the rail

The second term is what TEL-05's case cannot satisfy. At the converged point the
command is nowhere near +-`max_w`; on the 40 minutes of live converged behaviour
in `sim/scenarios/p1-age-2026-08-26.jsonl.gz` the meter never once leaves
+-500 W, so the loop never even approaches a 5000 W rail. And the reverse case -
a command genuinely pinned at the rail by a real 8 kW load - cannot satisfy the
FIRST term, because a real railed condition means kilowatts still crossing the
meter, and a kilowatt advances a 1 Wh register every 3.6 s.

The two terms fail for independent reasons. That is the whole design.

⚠️ AND THE SECOND HALF OF THAT WAS MEASURED ON THE RIG, WITH A RESULT WORTH
READING CAREFULLY. Driving the real transport against the rig's own hwsim under
+8000 W of house load - a genuinely railed command, meter perfectly healthy - the
POWER FIGURE DID NOT MOVE AT ALL: `active_power_w` sat at exactly 10130.0 W for
26 consecutive polls. What kept `unchanged_s` pinned at 0.0 s was the kWh
registers advancing, and nothing else.

So the register terms in `measurement_fingerprint()` are not belt-and-braces
here, they are the entire negative case. A freeze detector fingerprinted on the
power figure alone would have raised a false alarm on that rig run - a healthy
meter, a real load, a correctly railed command - which is the exact failure mode
that gets a safety guard switched off. Anyone tempted to simplify the fingerprint
should read that run first.

------------------------------------------------------------------------------
WHERE `hold_s` COMES FROM

Measured on `sim/scenarios/p1-age-2026-08-26.jsonl.gz` - 2399 s (40 min) of the
live house at the converged operating point, mean grid -14.2 W, 1194 polls:

    polls that repeated the previous reading      735 of 1193  (61.6 %)
    distinct-reading transitions                  458
    median gap between distinct readings          6.0 s
    p95 gap                                       6.0 s
    LARGEST `unchanged_s` any sample observes     8.0 s
    longest interval serving no distinct reading 15.4 s
    samples outside +-500 W                       0

The default 300 s is 37x the largest `unchanged_s` this house has been recorded
producing while converged, and 19.5x the longest interval in which it served no
distinct reading at all. The two figures differ because the capture has an 11.4 s
hole in its OWN polling; the second is quoted because it is the pessimistic one.
300 s is not fitted to either margin - it is picked as "clearly past any
plausible register-tick interval, and clearly inside the 900 s quarter that bills
the capacity tariff". test_freeze.py re-measures both every run, so this
paragraph cannot quietly stop being true.

⚠️ BE PRECISE ABOUT WHAT THAT CAPTURE PROVES. It is POWER-ONLY - the shipped
fingerprint also covers both kWh registers, so a real `unchanged_s` on that data
can only be shorter, never longer, which makes it a conservative measurement of
the converged case. But it does NOT on its own refute a bare `unchanged_s`
threshold, and saying it did would be exactly the overstatement this project
keeps having to walk back: at 8 s worst, a bare 300 s threshold is silent on it
too. The case that refutes the bare threshold is the one below, which no capture
of THIS meter contains because this meter's power figure jitters.

The floor on the other side: at the rail, the smallest residual grid flow that
still ticks a 1 Wh register inside 300 s is 12 W. Any condition that genuinely
pins this loop at +-5000 W has three orders of magnitude more than that crossing
the meter.

⚠️ AND THOSE TWO NUMBERS DO NOT MERELY HAVE MARGIN - THEY CANNOT BOTH BE TRUE.
The worry the conjunction has to answer is "a HEALTHY meter whose register is
slower than `hold_s`". At 300 s that is |grid| < 12 W. The shipped `deadband_w`
is 15 W, so anywhere the register is that slow the control law is already inside
its deadband and the command is parked - the rail term is not merely unlikely to
be true there, it is 0 W. The two terms are mutually exclusive by construction
at the defaults, not just far apart, and the check re-derives that every run.

That relationship is the one to protect. `hold_s` may be lowered only while
3600/hold_s stays under `deadband_w`; at the schema floor of 60 s that is 60 W
against a 15 W deadband, and the mutual exclusion is gone. A 60 s `meter_freeze_s`
is a supported configuration that gives up this argument and keeps only the
margin - do not set one without a capture that says the margin is enough.

⚠️ WHAT `hold_s` DOES NOT BUY IS DETECTION SPEED. The alert lands at
max(hold_s, time-to-rail) after the freeze, so at the shipped default and
`max_w: 5000` that is 300-360 s. The rail is REACHED first, from 75 s. A
notification is not a brake; see the DOCS.md note.

------------------------------------------------------------------------------
THE RAIL TERM, PRECISELY

    abs(commanded_w) >= RAIL_FRAC * max_w

`commanded_w` is the EMITTED setpoint - what the arbiter resolved and
`write_setpoint()` put on the wire - not the loop's raw wish, for the same
reason SAFETY-02 uses the emitted value: the wire carries the emitted number.

RAIL_FRAC is 0.95 and is a constant rather than an option. The command is
quantised to `step_w` (max 100), so at the rail the emitted value IS the ceiling
on every schema-legal combination; the 5 % is quantisation slack, not a tuning
knob. Making it configurable would offer a way to widen the one term that
carries the whole no-false-positive argument.

⚠️ SAFETY-20 - AND THE CEILING IS NOT `max_w`. THE 5 % USED TO BE DESCRIBED AS
"slack for a device rating that clips `max_w` slightly in `write_setpoint()`",
AND THAT SENTENCE WAS THE BUG. 5 % of slack does not absorb a device rating; it
inverts the guard. `write_setpoint()` clamps the emitted command to `max_w` AND
THEN to the setpoint device's own `[dev_min, dev_max]` range (clamp.py,
`clamp_to_rating()`), and the arbiter files the same range as a P_SAFETY limit
claim, so the largest magnitude that can ever stand on the wire is

    min(max_w, |dev_min|, |dev_max|)

not `max_w`. Measured on the shipped predicate at `max_w: 5000` against a device
of +-4700 - the exact case the ticket is named for:

    looks_frozen(unchanged_s=3600, commanded_w=4700, max_w=5000)  ->  False

The command is pinned at the hardware's rail, the meter has been frozen for an
hour, and term 2 is unsatisfiable BY CONSTRUCTION because 4700 < 4750. Any
device rating at or below `RAIL_FRAC * max_w` disarms the conjunction outright.
The status page then printed `armed (...command is over 4750 W)`, which is the
"disarmed guard that advertises itself as armed" failure validate.py rule 5 and
SAFETY-15/19 all exist to prevent - so `rail_ceiling_w()` below takes the device
range and the rail is measured against the smaller number.

WHY THE SMALLER OF THE TWO DEVICE BOUNDS, AND NOT A SIDE-AWARE PAIR. The rail is
a MAGNITUDE test against what clamp.py calls a magnitude clamp - it binds import
and export identically - so the ceiling here is a magnitude that is available in
BOTH directions. The alternative considered was a per-sign rail:
`min(max_w, dev_max)` for a positive command, `min(max_w, |dev_min|)` for a
negative one. That is arithmetically tighter and it is NOT wrong - on a device
range of -500..+5000 it would grade a -480 W charge command as railed, which the
symmetric ceiling reaches only by dropping the discharge rail to 475 W too. It
was rejected on cost, not on correctness: a per-sign rail is two rails, and two
rails need two coverage verdicts (`dev_min == 0` is a degenerate charge rail
whatever the discharge side can do) on a status row that carries one number and
in a predicate that takes one magnitude. That is a bigger change than this
defect earns, and the ONLY configuration that distinguishes the two - a device
range narrower than `max_w` on one side - is refused at startup by CFG-01 (see
the reachability note below). Between the two, the symmetric ceiling can only
ever LOWER the rail, which widens term 2 rather than disarming it; term 1
(`unchanged_s >= hold_s`) is the term carrying the no-false-positive argument,
and it is untouched. Whoever needs the tighter version should take both status
rows with it.

⚠️ AND WHEN THE CEILING IS ZERO THE RAIL IS NOT "LOW", IT IS GONE. A device that
accepts no command in one direction at all (`dev_max == 0`, or `dev_min == 0`)
gives a ceiling of 0 W, and `abs(commanded_w) >= 0` is TRUE FOR EVERY COMMAND
INCLUDING 0 W. Term 2 would not be disarmed there, it would be degenerately
true, and the detector would collapse into the bare `unchanged_s > N` threshold
TEL-05 refused - false-tripping at the converged operating point we aim for.
That case is reported as NOT COVERED, through the row the unsupported-transport
case already uses, and `looks_frozen()` returns False rather than firing.

⚠️ HOW REACHABLE IS ANY OF THIS - READ THIS BEFORE "SIMPLIFYING" IT AWAY.
`validate_startup_config()` rule 2 (CFG-01) already REFUSES TO START on
`max_w > dev_max` or `-max_w < dev_min`, and `run_control()` re-runs that gate
once ESPHome is up and the real rating is known. So on any install that reaches
the control loop with a rating it could read, `min(max_w, |dev_min|, |dev_max|)`
IS `max_w` and everything above is arithmetically dead. That is not an argument
against it - it is clamp.py's argument for its own second stage, word for word:
the case where this bites is the case where the startup gate did not run or no
longer holds (`hass.limits()` returns (None, None) whenever ESPHome is not up,
and the range checks are then skipped rather than guessed at, so a rating that
arrives late is never re-read), and a guard that quietly assumes the gate ran is
a guard that reports armed in exactly the state it cannot fire in. See AC 4's
decision, recorded in validate.py under WHAT IS DELIBERATELY *NOT* REFUSED.

⚠️ NOT GATED ON `following`, and that is deliberate. A maintenance charge
commands against the meter on purpose and can sit at the rail for hours - but it
does so while pushing kilowatts through the meter, so the FIRST term is what
excludes it, and it excludes it on physics rather than on bookkeeping. Gating on
`following` as well would additionally blind the detector to a maintenance phase
running against a meter that really has frozen, which is the same fault with a
worse actuator behind it.

------------------------------------------------------------------------------
NOTIFY, DO NOT TRIP

First cut is an alarm and nothing else: an HA persistent notification, an ERROR
in the log, an event on the status page, and a failed check row. It does NOT
claim the actuator, does NOT file a P_SAFETY claim and does NOT latch.

That is a deliberate ordering of costs, not timidity. A false trip here parks
the battery - on a house that is running this loop precisely to avoid buying
power at the wrong moment - and the operator's response to a parked battery is
to switch the guard off. A false notification costs one alert. The trip is a
separate decision that has to be earned with observed data from a real install,
and it needs an answer to a question this cut does not have one for: what should
the loop do instead, given that 0 W is itself a grid import of whatever the
house is drawing.

⚠️ TRANSPORT-CONDITIONAL. `unchanged_s` exists ONLY on `homewizard_local`,
because it is the only transport that sees the meter's own document. Under
`ha_dsmr`, `mqtt_p1`, `ha_signed` and `meter_source: off` this detector is
silent and the exposure above is uncovered. The caller passes None and gets
False; there is no half-measure available on those paths, and pretending
otherwise would be worse than the gap.
"""

from dataclasses import dataclass
from datetime import datetime

from .clamp import is_commandable

# See "THE RAIL TERM, PRECISELY" above. A constant, on purpose.
RAIL_FRAC = 0.95


@dataclass(frozen=True)
class FreezeWatchdogConfig:
    """Default, not a constant - config.yaml carries `meter_freeze_s`."""

    hold_s: float = 300.0


def rail_ceiling_w(max_w, *, dev_min=None, dev_max=None) -> float:
    """The largest command magnitude that can actually stand on the wire.

    SAFETY-20, and the whole point of the ticket: this is NOT `max_w`. The
    emitted command is bounded by `max_w` AND by the setpoint device's own
    `[dev_min, dev_max]` range - twice, in `clamp_to_rating()` and again as the
    arbiter's `device` P_SAFETY limit claim - so the reachable ceiling is the
    smallest of the three magnitudes. See "THE RAIL TERM, PRECISELY" for why it
    is the smaller of the two device bounds rather than a per-sign pair, and for
    why this is arithmetically dead on any install CFG-01 let start.

    `dev_min` / `dev_max` are `hass.limits()`'s answer and are None whenever
    ESPHome could not be reached. UNKNOWN IS NOT ZERO: a missing bound means
    this add-on applies no device clamp at all on that side (clamp.py's second
    stage is skipped unless BOTH are known), so an unknown bound must not lower
    the ceiling. `is_commandable()` is write_setpoint()'s own predicate, reused
    rather than retyped: a NaN rating must not become a 0 W ceiling.

    Returns 0.0 when the device can command nothing - see rail_unreachable().
    """
    ceiling = abs(float(max_w))
    for bound in (dev_min, dev_max):
        if is_commandable(bound):
            ceiling = min(ceiling, abs(float(bound)))
    return max(0.0, ceiling)


def rail_unreachable(max_w, *, dev_min=None, dev_max=None) -> str:
    """"" when the rail is a real threshold, else WHY it is not. SAFETY-20 AC2.

    The status page renders this as a NOT COVERED row - the same row, for the
    same reason, that an unsupported transport already gets. It is a second
    reason to use that surface, not a new mechanism: in both states the honest
    report is "this guard is not watching", and the one thing this project has
    decided repeatedly it will not ship is a page that says `armed` instead.

    ⚠️ THE UNREACHABLE CASE IS NOT "THE RAIL IS HIGH", IT IS "THE RAIL IS 0 W".
    A ceiling of zero makes `abs(commanded_w) >= 0` true for every command
    including a parked 0 W one, so term 2 is not disarmed but DEGENERATELY TRUE,
    and the conjunction decays into the bare `unchanged_s > N` threshold TEL-05
    refused - which false-trips at the converged operating point this controller
    aims for. Silence plus a NOT COVERED row is strictly better than that.
    """
    if not max_w or not is_commandable(max_w):
        return (f"max_w is {max_w!r}, so there is no ceiling to measure a rail "
                f"against")
    if rail_ceiling_w(max_w, dev_min=dev_min, dev_max=dev_max) > 0.0:
        return ""
    return (f"the setpoint device reports a range of {dev_min}...{dev_max} W, "
            f"which leaves no command magnitude available in both directions, "
            f"so the rail term would be true of every command including 0 W")


def rail_w(max_w, *, dev_min=None, dev_max=None) -> float:
    """The magnitude at or above which a command counts as railed.

    Measured against the REACHABLE ceiling, not against `max_w` - SAFETY-20.
    Callers that do not know the device range pass none and get the old
    behaviour, which is correct precisely when there is no device clamp to
    account for.
    """
    return rail_ceiling_w(max_w, dev_min=dev_min, dev_max=dev_max) * RAIL_FRAC


def looks_frozen(unchanged_s, commanded_w, max_w,
                 cfg: FreezeWatchdogConfig, *, dev_min=None, dev_max=None) -> bool:
    """The predicate, and the ONLY place the conjunction is written down.

    Public for the same reason `watchdog.disagrees()` is: the checks that grade
    it must assert on THE SHIPPED EXPRESSION. A safety predicate transcribed
    into its own test is one copy plus a future disagreement about what the
    check means, and the copy that drifts is always the one doing the grading.

    `unchanged_s`  seconds since the meter served a different measurement, or
                   None on a transport that cannot know (see the module note).
    `commanded_w`  the EMITTED setpoint standing on the wire.
    `max_w`        the configured output bound.
    `dev_min` /    the setpoint device's own reported range, or None when it
    `dev_max`      could not be read. SAFETY-20: the rail is measured against
                   `min(max_w, |dev_min|, |dev_max|)`, because that - not
                   `max_w` - is what the wire can carry.

    ⚠️ THE DEVICE BOUNDS ARE KEYWORD-ONLY AND DEFAULT TO None, WHICH IS THE
    CONSERVATIVE DIRECTION AND ALSO THE HONEST ONE. A caller that does not know
    the rating gets a rail measured against `max_w` - correct, because when this
    add-on has no rating it applies no device clamp either (clamp.py's second
    stage needs both bounds), so `max_w` really is the ceiling on that path.
    """
    if unchanged_s is None or commanded_w is None or not max_w:
        return False
    # SAFETY-20. A ceiling of 0 W would make term 2 true of EVERY command; that
    # is the degenerate case, not the armed one. Silence, and the status page
    # says NOT COVERED - see rail_unreachable().
    if rail_unreachable(max_w, dev_min=dev_min, dev_max=dev_max):
        return False
    return (float(unchanged_s) >= cfg.hold_s
            and abs(float(commanded_w))
            >= rail_w(max_w, dev_min=dev_min, dev_max=dev_max))


class FreezeWatchdog:
    """One instance per controller, evaluated once per control cycle.

    Level-triggered state with an EDGE for the alarm: `evaluate()` returns True
    on the cycle the conjunction first becomes true and False on every cycle
    after, so the caller raises one notification per episode rather than one
    every second. `active` is the level, for the status page.

    No latch, deliberately - see "NOTIFY, DO NOT TRIP". A latch exists so a trip
    cannot erase its own evidence by working; nothing here acts, so there is
    nothing to erase. When the meter starts moving again the alert clears
    itself, which is the honest report of the state.
    """

    def __init__(self, cfg: FreezeWatchdogConfig | None = None):
        self.cfg = cfg or FreezeWatchdogConfig()
        self.active = False
        self.since: datetime | None = None
        self.reason = ""
        self.episodes = 0

    def evaluate(self, unchanged_s, commanded_w, max_w, now: datetime, *,
                 dev_min=None, dev_max=None) -> bool:
        """Advance one cycle. Returns True ONLY on the rising edge.

        `now` is the caller's clock so this stays testable and replayable - the
        same discipline as `SignWatchdog.evaluate()`. `dev_min`/`dev_max` are
        the setpoint device's reported range; see looks_frozen().
        """
        if not looks_frozen(unchanged_s, commanded_w, max_w, self.cfg,
                            dev_min=dev_min, dev_max=dev_max):
            self.active = False
            self.since = None
            self.reason = ""
            return False

        # ⚠️ SAFETY-20. THE REASON QUOTES THE CEILING IT ACTUALLY USED, not
        # `max_w`. The old line said "at the 5000 W rail (... command over
        # 4750 W)" on an install whose device stopped at 4700 W, i.e. it named
        # a threshold the command could not reach - the same untruth the status
        # page was telling, printed into the log and the HA notification where
        # it would outlive the page.
        ceiling = rail_ceiling_w(max_w, dev_min=dev_min, dev_max=dev_max)
        capped = ("" if ceiling >= abs(float(max_w))
                  else f", capped from max_w {float(max_w):.0f} W by the "
                       f"setpoint device's own rating")
        self.reason = (
            f"METER_FROZEN: the meter has answered {float(unchanged_s):.0f} s "
            f"without its measurement changing while the command stands at "
            f"{float(commanded_w):+.0f} W, at the {ceiling:.0f} W rail{capped} "
            f"(limits: unchanged over {self.cfg.hold_s:.0f} s, command over "
            f"{rail_w(max_w, dev_min=dev_min, dev_max=dev_max):.0f} W)")
        if self.active:
            return False
        self.active = True
        self.since = now
        self.episodes += 1
        return True