goodwe-addon/goodwe_controller/app/clamp.py

"""SAFETY-03 - the output clamp, and the alarm that makes it audible.

WHAT IT CATCHES, in one sentence: a control law that computes a setpoint the
actuator cannot deliver, which every layer below it will happily clip to
something legal-looking while the thing that produced it goes on winding.

THE CLAMP IS NOT THE DELIVERABLE. specs.md 10.3 is explicit that this project
already clipped, and that the clipping is what hid the defect:

    with the inverter switched off entirely, an unclamped integrator climbed
    ~130 W per 4 s to over 10 kW and, pinned at the 5 kW rail, reported
    14 768 W

A clamp alone turns that into a silent, perfectly flat 5000 W hold. Every
dashboard reads "commanded 5000 W", which is a legal number, and nothing
anywhere says the loop asked for three times it. 11.2's watchdog row says
"Clamp AND ALARM - the loop is winding"; the second verb is this module.

------------------------------------------------------------------------------
WHAT "THE RATING" IS HERE, AND WHY IT IS ONE NUMBER AND NOT A MIN()

The card names two bounds and warns against choosing between them:

  - `DriverCapabilities.setpoint_limit_w` (3.2), the inverter's own magnitude
    rating. In this add-on that arrives as the setpoint entity's `min`/`max`
    from `hass.limits()` - `Controller.dev_min` / `dev_max`.
  - `optimizer.loop.clamp_w` (14), the configured loop clamp. In this add-on
    that is the `max_w` option, which `Tuning.max_w` and the arbiter's
    `supervised` limit both already use.

14 says these two must AGREE, validated at startup with a refusal on
mismatch - and that gate already ships: `validate_startup_config()` refuses
`max_w > dev_max` and `-max_w < dev_min` (validate.py, problem 1), and
run_control() re-runs it once ESPHome is up and the real rating is known.

So this guard clamps to `max_w`, THE VALIDATED VALUE, and does not `min()` it
with the device rating at runtime. That is the card's instruction and it is also
the only reading that keeps `max_w` meaningful: a silent runtime `min()` is
precisely the "helper bounded to +-500 W under a loop clamped to 1500 W" failure
the 14 warning box is about.

WARNING - THE DEVICE BOUND IS STILL APPLIED, SECOND, AND IT IS NOT REDUNDANT.
After a successful validation `+-max_w` lies inside `[dev_min, dev_max]`, so the
second stage is arithmetically a no-op and can never engage. If it DOES engage,
the validation did not run or no longer holds - `hass.limits()` returns
(None, None) whenever ESPHome is not up yet, and the range checks are skipped
rather than guessed at - and that is a louder fault than the first stage, not a
quieter one. It therefore alarms too, through the same channel, with the bound
named.

⚠️ SAFETY-20 UPDATED THE SECOND HALF OF THAT SENTENCE. It used to read "this is
NOT SAFETY-20; nothing here changes what `freeze.py` measures its rail against",
and that was true of this module and false of the system: this stage is exactly
what clips the emitted command below `RAIL_FRAC * max_w`, which is how a device
rating disarmed the freeze detector while the page said `armed`. `freeze.py` now
takes `dev_min`/`dev_max` and measures its rail against
`min(max_w, |dev_min|, |dev_max|)`. THE CLAMP ITSELF IS UNCHANGED - it still
clamps to `max_w` first and to the device range second, for the reasons above,
and it still does not `min()` the two. What changed is only that the detector
downstream now knows this stage exists.

------------------------------------------------------------------------------
WHERE THE WIND ACTUALLY SHOWS UP - READ THIS BEFORE "SIMPLIFYING" THE CALLER

`control.compute()` bounds the integrator to `integrator_max_w or max_w` and
THEN derives the command from it, so on a default install (`integrator_max_w`
unset, i.e. following `max_w`) the command can never exceed `max_w` and the
actuator-side clamp below IS UNREACHABLE. control.py says so in as many words:

    SAFETY-03 ... must watch for THIS, not for "clamped" below.

An alarm wired only to this module's `ClampResult.engaged` would therefore be
silent on the exact configuration the whole ticket is about. The winding is
visible one layer up, at the integrator bound, and that is why `Decision` now
carries `i_clamped` / `clamped` BOOLEANS and the caller ORs them in here.

AND WHY BOOLEANS RATHER THAN `Decision.reason`. `reason` is a single string that
later stages overwrite: a cycle that clamps the integrator and is then slew
limited reports "slew-limited", and one that clamps while saturated reports
"saturated-freeze". Both are exactly the winding-into-a-rail case, and on the
string the clamp signal is destroyed before anything can alarm on it. The two
flags are set where the clamp happens and are never overwritten. A check in
test_clamp.py drives that precise combination.

------------------------------------------------------------------------------
DISTINCT FROM SATURATION - THIS IS ACCEPTANCE CRITERION 2, NOT A NICETY

    clamp      the COMPUTED value exceeded the rating. A software bug signal:
               something upstream is winding. Alarm, every engagement.
    saturation |command - inverter AC power| > `saturation_w` for
               `saturation_cycles`. The HARDWARE cannot follow a perfectly legal
               command. Action is "freeze the integrator" (11.2), not an alarm.

A loop can be legally saturated all day - a cold pack, a curtailed inverter, an
export limit - so a saturation-shaped alarm is one an operator learns to ignore.
Routing the clamp through it puts the bug signal inside the noise, which is the
same silence 10.3 is complaining about. They share no reason code, no
notification id, no status row and no counter.

------------------------------------------------------------------------------
ANTI-FLOOD, WITHOUT GOING SILENT

A genuinely winding loop clamps on EVERY cycle for as long as it winds. One HA
notification per cycle is how an operator learns to dismiss notifications
unread, and this add-on already made that argument once (see the fixed
`notification_id` note in main.check_meter_frozen).

So the two surfaces are split, deliberately:

  - EVERY engagement is recorded, never deduplicated, and that record is what
    says how hard the loop is pushing. 15's action log is this add-on's
    `Controller.events`, and it gets a row from `write_setpoint()` on every
    engaged WRITE; `engagements` here counts every engaged CYCLE. The two layers
    are counted separately on purpose - the heartbeat re-writes an
    already-clamped value between cycles, so a per-write count would inflate and
    a per-cycle log would miss an out-of-cycle write.
  - THE NOTIFICATION fires on the rising edge and then RE-ASSERTS every
    `REASSERT_S`, so a loop pinned on the rail for an hour does not go quiet.
  - It DE-ASSERTS only after `CLEAR_CYCLES` consecutive clean cycles, which is
    the 10.2 hysteresis: a command sitting exactly on the rail dithers in and
    out of engagement by one quantisation step, and without this the alarm
    chatters.

`REASSERT_S` and `CLEAR_CYCLES` are module constants, not options, for the same
reason `freeze.RAIL_FRAC` is: they describe the event bus, not the plant, and
every knob 10.3 asked to keep exposed is a LOOP TUNING knob and is untouched by
this module. No new config key is added anywhere - CFG-03 already has three
places a default has to be kept in step and this ticket does not earn a fourth.

------------------------------------------------------------------------------
NOT A TRIP

Clamping does not stop the loop and does not command 0 W. The clamped value is
written and control continues, which is what 11.2 asks for ("Clamp and alarm")
and what distinguishes this from SAFETY-01/02, whose action is 0 W. It follows
that this must NOT drive the failsafe status - `publish()` keeps reporting
"running", because the controller is.
"""

import math
from dataclasses import dataclass
from datetime import datetime

# The reason code, distinct from any saturation-freeze code by construction:
# nothing else in this add-on uses it. See "DISTINCT FROM SATURATION".
REASON_CODE = "OUTPUT_CLAMP"

# Re-assert the notification this often while the clamp stays engaged.
#
# ⚠️ RAISED 30 s -> 150 s on 2026-09-05, and the reason is a REGIME CHANGE, not a
# change of mind about the alarm. 30 s was the planning cadence (2.3), chosen
# when a loop-side engagement was a rare event: this module's own note records
# "60 cycles of the §10.3 runaway produce 2 loop-side engagements". On the live
# install it produced 48 IN 25 MINUTES of ordinary operation, because `max_w` is
# 5000 W where the loop's tuning was derived at a 1500 W clamp (SAFETY-39) - so
# the integrator reaches its bound constantly and the notification re-asserted
# every 30 s, all day. An alarm that speaks that often about the normal case is
# how an operator learns to ignore the one that matters, which is the exact
# failure this module's docstring is written against.
#
# ⚠️ 150 s IS NOT A COMFORT CHOICE, IT IS THE REQUIREMENT'S OWN EDGE, WITH ONE
# STEP OF MARGIN. test_clamp.py part 2 pins the cadence in WALL-CLOCK terms
# rather than against this constant - a loop pinned on the rail must speak at
# least MIN_PER_QUARTER = 5 times inside the 900 s quarter that bills the
# capacity tariff. That caps this constant at 900/5 = 180 s. At 180 s the check
# passes with exactly 5 of a required 5, so any later change to the firing logic
# tips it red; 150 s gives 6. The floor is the thing that must not move.
#
# ⚠️ THIS IS A STOP-GAP AND SHOULD BE REVISITED. It reduces how often the alarm
# repeats; it does not make the alarm correct. SAFETY-40 is the wording and the
# i-clamped/clamped split, and SAFETY-39 is the tuning that would stop the loop
# railing in the first place. Neither is done by this line.
REASSERT_S = 150.0

# Consecutive clean cycles before the alarm de-asserts (10.2 hysteresis).
CLEAR_CYCLES = 3


@dataclass(frozen=True)
class ClampResult:
    """What the guard did, with the RAW value preserved.

    AC 2 turns on "the raw computed value differed from the clamped one", so the
    raw value has to survive the call - a guard that returns only the applied
    number cannot be asked the question the acceptance criterion asks.
    """

    requested_w: float
    applied_w: float
    limit_w: float
    engaged: bool
    bound: str = ""          # "" | "max_w" | "device" - which stage bit

    def describe(self) -> str:
        return (f"{REASON_CODE}: the computed setpoint {self.requested_w:+.0f} W "
                f"was clamped to {self.applied_w:+.0f} W by the {self.bound} "
                f"bound (rating {self.limit_w:.0f} W) - something upstream is "
                f"winding")


def is_commandable(value) -> bool:
    """False for None, NaN and the infinities.

    A non-finite setpoint is LOSS OF INPUT, not an out-of-range value, and
    11.1 rule 1 says loss of input means command zero and keep commanding. It
    must not reach the clamp arithmetic: `min(limit, nan)` returns nan on
    CPython, so a nan would sail straight through a naive clamp and then report
    `engaged` True on the `applied != requested` test - because nan != nan - i.e.
    a transport bug filed in the winding-loop counter, with a nan on the wire.
    """
    if value is None:
        return False
    try:
        return math.isfinite(float(value))
    except (TypeError, ValueError):
        return False


def clamp_to_rating(requested_w, limit_w, dev_min=None, dev_max=None) -> ClampResult:
    """Bound a computed setpoint to the validated rating. The ONLY clamp.

    `limit_w`  the validated magnitude bound - `max_w`. Symmetric, because
               3.2 calls `setpoint_limit_w` a MAGNITUDE clamp: it binds import
               (+) and export (-) identically.
    `dev_min` / `dev_max`  the device's own asymmetric range, applied second and
               expected never to bite. See the module note.

    Public, and the predicate lives here rather than in the test, for the reason
    `freeze.looks_frozen()` gives: a safety expression transcribed into its own
    check is one copy plus a future disagreement about what the check means.

    EXACTLY +-`limit_w` IS NOT AN ENGAGEMENT. The bound is inclusive; a loop
    resting on the rail is doing what it was told, and alarming there would fire
    on every healthy full-power hour.
    """
    requested = float(requested_w)
    limit = abs(float(limit_w))

    # ⚠️ BOTH STAGES ALWAYS RUN. THIS USED TO `return` FROM THE FIRST ONE, AND
    # THAT WAS A REGRESSION AGAINST THE CODE IT REPLACED. `write_setpoint()`
    # previously clamped to the device range unconditionally, under the comment
    # "never send a value the device will reject outright"; short-circuiting here
    # meant the device bound was applied only when the rating bound had NOT
    # fired, i.e. THE BIGGER THE OVERSHOOT THE LESS THE GUARD DID. Measured on
    # the shipped function at max_w 5000 against a device of +-4000:
    #
    #     +9999 -> +5000  bound='max_w'   <- outside the device range
    #     +4500 -> +4000  bound='device'
    #
    # It also undercut this module's own argument for clamping to `max_w` rather
    # than a runtime `min()`, which rests entirely on this stage catching the
    # case where the startup validation did not run.
    applied = max(-limit, min(limit, requested))
    bound = "max_w" if applied != requested else ""

    # Second stage. Arithmetically dead after a successful startup validation -
    # and therefore worth alarming on precisely BECAUSE it fired.
    if dev_min is not None and dev_max is not None:
        bounded = max(float(dev_min), min(float(dev_max), applied))
        if bounded != applied:
            bound = "device" if not bound else "max_w+device"
            applied = bounded

    return ClampResult(requested, applied, limit, applied != requested, bound)


class ClampWatchdog:
    """One instance per controller. Counts every engagement, alarms sparsely.

    `note()` is called once per write with whether the clamp bit; the caller ORs
    in the loop's own `i_clamped` / `clamped` flags, which is where the winding
    is actually visible on a default install (see the module note).
    """

    def __init__(self):
        self.active = False
        self.engagements = 0      # every engagement, never deduplicated
        self.episodes = 0         # rising edges
        self.since: datetime | None = None
        self.last_alarm: datetime | None = None
        self.reason = ""
        self.clean_cycles = 0

    def note(self, engaged: bool, reason: str, now: datetime) -> bool:
        """Advance one cycle. True when the caller should raise the alarm.

        `now` is the caller's clock, the same discipline as
        `SignWatchdog.evaluate()` and `FreezeWatchdog.evaluate()`.
        """
        if engaged:
            self.engagements += 1
            self.clean_cycles = 0
            self.reason = reason
            if not self.active:
                self.active = True
                self.since = now
                self.episodes += 1
                self.last_alarm = now
                return True
            # Still engaged: re-assert on the slow cadence, not every cycle.
            if (self.last_alarm is None
                    or (now - self.last_alarm).total_seconds() >= REASSERT_S):
                self.last_alarm = now
                return True
            return False

        if not self.active:
            return False
        self.clean_cycles += 1
        if self.clean_cycles >= CLEAR_CYCLES:
            self.active = False
            self.since = None
            self.last_alarm = None
            self.reason = ""
            self.clean_cycles = 0
        return False