"""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. This is NOT SAFETY-20; nothing here changes what `freeze.py` measures its rail against. ------------------------------------------------------------------------------ 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. 30 s is # the planning cadence (2.3), i.e. slow enough not to flood and fast enough # that a pinned loop is never quiet for a whole billing quarter. REASSERT_S = 30.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