admin/goodwe-addon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge release/1.0 (SAFETY-03) into TEST-06
checks / unit (pull_request) Canceled after 0s
Details

Browse Source 
Two conflicts, both predicted, both mine as the later merger.

app/main.py - #12 inserted `await self.check_output_clamp()` on the line
immediately above `await self.check_meter_frozen()`, so its hunk abuts my
comment correction. Resolved so EACH CALL CARRIES ITS OWN BOUNDARY,
because the two guards do not share one:

  - check_output_clamp() reads `self.clamp_last`, which is None until
    write_setpoint() has run, so its boundary IS the write;
  - check_meter_frozen() grades `self.target`, assigned further up, so
    its boundary is the assignment.

Letting my correction generalise across both would have been worse than
the vague comment it replaces, because it would read as deliberate. The
AST of app/main.py is identical to merged release/1.0 (+1251 bytes, all
comment).

test_freeze.py - #12 independently added `frozen_notes()` keyed on the
same id. `notes_of()` subsumes it: same id, plus the title fallback that
keeps the id's own check independently falsifiable, plus `all_notes_of()`
for claims about the whole surface.

THE THREE SURFACE-WIDE ASSERTIONS, DECIDED ONE AT A TIME. Measured on the
merged tree:

  scenario                              surface  ids             freeze
  frozen meter, railed command             2     frozen + clamp     1
  TEL-05 converged, -14.2 W                0     -                  0
  moving meter, a real 8 kW load           1     clamp              0
  uncovered transport, no unchanged_s      1     clamp              0

  - the wiring scenario STAYS SURFACE-WIDE and now asserts exactly TWO
    alarms with BOTH IDS NAMED. The answer to "is this still meant to
    raise one alarm" is no, it raises two, and a third guard has to come
    back to this line. That is the one assertion in the file that
    accounts for the controller's entire alarm output;
  - the CONVERGED case stays surface-wide and stays silent: at the
    operating point the whole add-on should have nothing to say, that is
    still true, and narrowing it would give away a real property;
  - the other two NARROW to notes_of(). The clamp alarm there is correct
    and is SAFETY-03's subject: a real 8 kW load does drive the loop past
    the rating, and an uncovered transport does not stop it. Asserting
    whole-surface silence would make this file fail whenever another
    guard correctly does its job. Each still owns the load-bearing half -
    THE FREEZE DETECTOR SAID NOTHING - and the surface is PRINTED so a
    reader sees what else fired without this file grading it.

Re-verified on the merged tree: M1 RED 4, M2 RED 5, M3 RED 3 + RED 2,
N6 RED 1, N3 green (correctly), N8 RED 4, R1 RED 2, R2 RED 1. All nine
suites green.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015VXjA3LWccN1rJsk3okyPC
This commit is contained in:
glenn schrooyen
2026-08-30 17:45:09 +02:00
co-authored by Claude Opus 5
parent d31a12b7c6 6aabb832fc
commit 0942228d65
8 changed files with 1268 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
goodwe_controller/CHANGELOG.md
+95
View File
@@ -2,6 +2,101 @@
## Unreleased
**A computed setpoint over the rating now alarms instead of being clipped in
silence (SAFETY-03).** The clamp was never the missing half - specs.md 10.3
records an unclamped integrator climbing ~130 W per 4 s past 10 kW and reporting
**14 768 W** while pinned at the 5 kW rail, and a silent clamp turns exactly that
into a flat, legal-looking 5000 W hold. `write_setpoint()` - the one path every
write takes, cycle, heartbeat, operator button and shutdown alike - now clamps to
the **validated** `max_w` and reports whether the raw value differed, and
`Controller.check_output_clamp()` raises `OUTPUT_CLAMP`: a persistent
notification (`goodwe_output_clamp`), an `ERROR`, an event row and a red status
banner. **Not a trip**: the clamped value is written and the loop keeps running,
per 11.2's "clamp *and alarm*".
**The bound is `max_w` and nothing is `min()`ed at runtime.** Section 14 requires
the loop clamp and the driver's `setpoint_limit_w` to agree, validated at startup
with a refusal on mismatch - which CFG-01 already ships. The device range is
still applied second, as defence in depth; after a successful validation it
cannot engage, so if it ever does, that is louder than the first stage and it
alarms too, naming the bound. No new config key: the guard's limit is the one
the operator already set, and its cadence is a module constant.
**`Decision` grew `i_clamped` / `clamped` booleans, and that is the load-bearing
part.** `reason` is one 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* the winding case,
so an alarm keyed on the string is silent on exactly the cycles it exists for:
mutating main.py back onto `"clamp" in decision.reason` drops the 10.3 replay
from 2 engagements to **0**. The flags are set where the clamp happens and are
never overwritten. On a default install (`integrator_max_w` following `max_w`)
the integrator bound is the one that bites and the output clamp is unreachable,
so the integrator flag is the whole signal.
**Distinct from the saturation freeze, and checked in the negative.** A clamp is
a *software* bug signal; saturation is the *hardware* failing to follow a legal
command, and its action is to freeze the integrator, not to alarm. They share no
reason code, notification id, status row or counter. `test_clamp.py` drives a
loop that saturates for 10 consecutive cycles at 1.4 % of the rating and asserts
zero clamp notifications, zero engagements, no log row and no status row - after
first asserting that the saturation freeze *really did* engage, so the silence is
an observation rather than a scenario that did nothing.
**The action log is coalesced per episode, and that was a real bug.**
`log_event()` keeps 40 rows; one row per engagement flushed the entire ring
within 40 cycles of a winding loop, and the first thing it evicted was
SAFETY-18's own `METER FROZEN?` row - this guard silencing another guard's log.
Caught by `test_freeze.py` going red, not by inspection. One row per episode now,
carrying the exact engagement count read off the counter.
**A non-finite setpoint is refused, and it was worse than passing one through.**
`nan` is loss of input (11.1 rule 1: command zero, keep commanding), not an
out-of-range value. Measured: because every comparison against `nan` is False, a
naive `max(-limit, min(limit, nan))` keeps its *first* argument and turns a `nan`
into a command **at the full rating**, then reports it as a clamp engagement.
**Review fixes (PR #12).** Three, and the first was a behavioural regression
this PR introduced: `clamp_to_rating()` **returned early** from the `max_w`
stage, so the device bound - the "applied second, defence in depth" half - was
reached only when the rating bound had NOT fired. The bigger the overshoot, the
less the guard did. At `max_w` 5000 against a +-4000 device, +9999 emitted
**+5000 W, outside the device range**, while a mere +4500 was correctly held to
+4000. That undid the pre-PR guarantee "never send a value the device will
reject outright" and undercut this module's own case for clamping to `max_w`
rather than a runtime `min()`, which rests entirely on that stage catching a
validation that did not run. Both stages now always run, the result names both
bounds when both bite (`max_w+device`), and the property is asserted over a
492-point sweep rather than on witness values - the defect was invisible to any
single-value check that happened to sit inside one stage. No reachable
production path was found for it (two validation gates and the arbiter bound sit
in front), but latent is not harmless.
Two harness defects, both of the kind where a check cannot fail for the value it
forbids: the hysteresis check issued `range(CLEAR_CYCLES - 1)` clean cycles,
which at `CLEAR_CYCLES = 1` issues **zero** and passes while printing
"0 clean cycles do NOT clear it" - green for the chatter setting the constant
exists to prevent. And the re-assert check stimulated at `REASSERT_S + 9`, green
for **any** value including 3600, i.e. the hour of silence the module promises
not to produce. Both are now written in wall-clock terms that do not move when
the constant does: **one** clean cycle must not clear the alarm, and a loop
pinned across the 900 s capacity-tariff quarter must speak at least 5 times and
on no more than 20 % of cycles. `CLEAR_CYCLES = 1`, `REASSERT_S = 3600` and
`REASSERT_S = 0` each go red.
**`test_clamp.py`** (new, assert-based, no framework) covers the clamp
arithmetic and its boundaries - exactly +-`max_w` is *not* an engagement - the
alarm's edge/re-assert/hysteresis cadence, the 10.3 runaway replayed through the
real `Controller.cycle()` against an unclamped transcription, the saturation
negative, and the direct-write path. Six author mutations reproduced red,
including "clamp kept, alarm deleted" (the historical build) and "alarm keyed on
`reason`". Two of the check's own hypotheses about how the freeze and the clamp
interact were **refused by the sweep** and rewritten to what it measured: a
frozen cycle *can* clamp, and does so exactly when the integrator reverses.
`test_freeze.py`'s notification assertions now filter by `notification_id`
rather than by service - the add-on raises more than one kind of notification,
and its railing scenarios legitimately raise this one too.
**Cheap-window maintenance charging was unreachable at the shipped defaults, and
now the config that makes it unreachable is refused (MAINT-04).** `config.yaml`
shipped `max_w: 2000` against `maintenance_charge_w: 2500`. The cheap-window
goodwe_controller/DOCS.md
+55
View File
@@ -474,6 +474,61 @@ today is operational: set `max_w` no higher than the loop actually needs (the
tuning was measured at 2000 W and a 1.78 kW load step is tracked fine there), fit
the e-stop, and watch for the alarm.
### The winding-loop alarm (the output clamp)
Every setpoint this add-on writes is clamped to `max_w` before it leaves the
process, and **when that clamp actually engages, it alarms.** The clamp on its
own is not the feature - this project already clipped, and the clipping is what
hid the defect the guard exists for: with the inverter switched off entirely an
unclamped integrator climbed ~130 W per 4 s past 10 kW and, pinned at the 5 kW
rail, reported **14 768 W**. Clipped silently, that is a flat, legal-looking
5000 W hold on every dashboard, and nothing anywhere says the loop asked for
three times it.
**What "the rating" is.** The bound is `max_w`, and only `max_w`. It is already
validated against the setpoint device's own range at startup - the add-on
*refuses to start* if `max_w` exceeds what the device accepts (and again once
ESPHome is up and the real rating is known), so by the time the loop runs the two
numbers agree. The device range is applied a second time inside the write as
defence in depth; after a successful validation it cannot bite, and if it ever
does, that engagement alarms too and names the bound.
**Clamping is not saturation, and the two never share a channel.**
| | what it means | what happens |
|---|---|---|
| **clamp** | the *computed* value exceeded the rating | a software bug signal - something upstream is winding. **Alarm.** |
| **saturation** | the inverter is not following a perfectly *legal* command | a hardware condition - freeze the integrator. **No alarm.** |
A loop can be legally saturated all day: a cold pack, a curtailed inverter, an
export limit. Routing the clamp through a saturation-shaped alarm puts the bug
signal inside the noise, which is the same silence as clipping quietly. They
share no reason code, no notification id, no status row and no counter.
**It is an alarm, not a trip.** The clamped value is written and the loop keeps
running - this is not the sign-disagreement failsafe, which commands 0 W. You
get a persistent notification (`goodwe_output_clamp`, replaced rather than
stacked), an `ERROR` in the log with the reason code `OUTPUT_CLAMP`, a row in
the event list, and a red status banner while it is engaged. Once it clears, a
warning row keeps the engagement count so the evidence does not vanish with the
condition. Consecutive engagements share one event row carrying the exact count,
because the event list keeps only 40 rows and a row per cycle would flush every
other guard's entries out of it.
**There is nothing to configure**, deliberately. The bound is `max_w`, which you
already set; giving the guard a limit of its own would be a second number that
can disagree with the first, which is exactly the failure the startup validation
exists to prevent. The alarm's cadence - first engagement, then re-asserted every
30 s while it persists, cleared after 3 clean cycles - is fixed for the same
reason `RAIL_FRAC` is: it describes the notification bus, not the plant.
WARNING: **on a default install the engagement you will see is the integrator's,
not the output's.** The control law bounds the integrator to `max_w` and *then*
derives the command from it, so the command itself can never come out over the
rating unless `integrator_max_w` has been deliberately raised above `max_w`.
Both are watched. A guard that watched only the emitted value would be silent on
every standard configuration.
### The frozen-meter alarm
A meter that has died is easy — it stops answering, the sample age climbs past
goodwe_controller/app/clamp.py
+284
View File
@@ -0,0 +1,284 @@
"""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
goodwe_controller/app/control.py
+23 -3
View File
@@ -64,6 +64,23 @@ class Decision:
# and quantisation. Carry it back in as `i_w` next cycle; that is what keeps
# it a separate quantity from the command.
i_w: float = 0.0
# SAFETY-03. Did a clamp actually engage this cycle - i.e. did the RAW
# computed value differ from the bounded one?
#
# ⚠️ THESE EXIST BECAUSE `reason` CANNOT ANSWER THAT QUESTION. `reason` is
# one string and every later stage overwrites it: 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 the winding
# case, and on the string the clamp signal is gone before anything can alarm
# on it. These two are written where the clamp happens and are never
# overwritten, so they mean exactly what they say and nothing else.
#
# `i_clamped` is the one that fires on a default install: the integrator
# bound is reached first, so `clamped` is unreachable unless
# `integrator_max_w` is deliberately set above `max_w`. Alarm on BOTH - see
# the comment at the bound below, and clamp.py.
i_clamped: bool = False
clamped: bool = False
def compute(
@@ -188,7 +205,8 @@ def compute(
# bound is not. It is what makes the worst-case unwind time finite and
# knowable instead of a function of how long the error happened to stand.
bounded = max(-limit, min(limit, i_w))
if bounded != i_w:
i_clamped = bounded != i_w
if i_clamped:
# ⚠️ SAFETY-03 (alarm whenever the loop winds into a rail) must watch
# for THIS, not for "clamped" below. At the default limit == max_w the
# integrator bound is reached first and the command derived from it can
@@ -209,7 +227,8 @@ def compute(
# --- ORDER MATTERS: clamp -> slew -> freeze ----------------------------
# An early draft applied a floor after the clamp and let demand escape it.
target = max(-tuning.max_w, min(tuning.max_w, want))
if target != want:
clamped = target != want
if clamped:
reason = "clamped"
slewed = max(prev_w - tuning.slew_w, min(prev_w + tuning.slew_w, target))
@@ -235,7 +254,8 @@ def compute(
step = max(1, int(tuning.step_w))
target = round(target / step) * step
return Decision(float(target), sat_count, frozen, reason, float(i_w))
return Decision(float(target), sat_count, frozen, reason, float(i_w),
i_clamped, clamped)
def maintenance_charge_floor(
goodwe_controller/app/main.py
+192 -16
View File
@@ -36,6 +36,10 @@ from .arbiter import (
P_LOOP, P_MAINT_SHAPE, P_MAINTENANCE, P_SAFETY, Claim, resolve,
)
from .control import Tuning, compute, maintenance_charge_floor, peak_at_risk
from .clamp import (
REASON_CODE as CLAMP_REASON, ClampResult, ClampWatchdog, clamp_to_rating,
is_commandable,
)
from .freeze import FreezeWatchdog, FreezeWatchdogConfig, rail_w
from .hass import HomeAssistant
from .maintenance import IDLE, MaintConfig, Maintenance
@@ -140,6 +144,19 @@ class Controller:
self.freeze_wd = FreezeWatchdog(
FreezeWatchdogConfig(hold_s=float(opts.get("meter_freeze_s", 300))))
# SAFETY-03. The output clamp's alarm - see clamp.py for why the clamp
# itself is the cheap half.
#
# ⚠️ NO CONFIG OF ITS OWN, on purpose. The bound it clamps to is `max_w`,
# which is already an option, already validated against the device's
# rating at startup, and already used by Tuning and the arbiter. Giving
# this a second key would be a fourth place a limit has to be kept in
# step (CFG-03 already tracks three) and, worse, would recreate exactly
# the two-numbers-that-disagree failure the §14 warning box describes.
# The anti-flood cadence is a module constant for the same reason
# freeze.RAIL_FRAC is: it describes the event bus, not the plant.
self.clamp_wd = ClampWatchdog()
# P1 ingestion (TEL-01). `meter_source: off` keeps the original
# single-entity meter_entity path, so an existing install is unchanged
# until it opts in.
@@ -171,6 +188,11 @@ class Controller:
self.last_write_ok = None
self.inputs_bad_since = None
self.dev_min = self.dev_max = None
# SAFETY-03. What the last write's clamp did, and what the loop's own
# bounds did last cycle. `loop_clamped` is a (engaged, note) pair so the
# alarm can say WHICH bound bit without re-deriving it from a string.
self.clamp_last: ClampResult | None = None
self.loop_clamped: tuple[bool, str] = (False, "")
self.events: list[str] = []
self.stopping = False
@@ -229,11 +251,37 @@ class Controller:
entity = self.o.get("setpoint_entity", "")
if not entity:
return
# Never send a value the device will reject outright. A rejected write is
# silent from the controller's point of view, and the inverter then keeps
# doing whatever it was already doing.
if self.dev_min is not None and self.dev_max is not None:
value = max(self.dev_min, min(self.dev_max, value))
# --- SAFETY-03: THE CLAMP, AND IT IS ON THE ONLY PATH TO THE WIRE -----
# Every write in this add-on comes through here - the cycle, the
# heartbeat, the operator's stop button and shutdown() - so no optimizer,
# maintenance or manual path can route around it. That placement is the
# first acceptance criterion, and it is why the guard is not in
# control.py: a guard the control law owns is a guard every other claimer
# bypasses.
#
# ⚠️ A NON-FINITE VALUE IS LOSS OF INPUT, NOT AN OUT-OF-RANGE ONE. §11.1
# rule 1: command zero and keep commanding. It must not reach the clamp
# arithmetic - see is_commandable() for what a nan does to it - and it
# must not be counted as a clamp engagement, or a transport bug lands in
# the winding-loop counter.
if not is_commandable(value):
self.log_event(f"non-finite setpoint {value!r} - commanding 0 W")
_LOG.error("refusing to write a non-finite setpoint %r - commanding "
"0 W and continuing to command. This is a bug upstream, "
"not a limit.", value)
value = 0.0
result = clamp_to_rating(value, float(self.o.get("max_w", 2000)),
self.dev_min, self.dev_max)
self.clamp_last = result
if result.engaged:
# ⚠️ EVERY engagement, never deduplicated: this is the action log
# (§15), and it is a different surface from the notification, which
# IS deduplicated. The count is what says how hard the loop pushed;
# the notification only says that it did.
self.log_event(result.describe())
value = result.applied_w
changed = (self.last_write is None or value != self.last_write)
if not (changed or force):
return
@@ -379,11 +427,29 @@ class Controller:
self.sat_count = decision.sat_count
self.i_w = decision.i_w
self.loop_w = decision.target_w
# --- SAFETY-03 ---------------------------------------------------
# ⚠️ READ THE BOOLEANS, NOT `decision.reason`. On a default install
# the integrator bound is what bites and the output clamp is
# unreachable (control.py says so), and `reason` is then overwritten
# by whichever of slew-limiting or the saturation freeze ran after
# it - so a string test is silent on the two cases a winding loop
# most often presents as. These flags are set where the clamp
# happens and nothing downstream touches them.
self.loop_clamped = (
decision.i_clamped or decision.clamped,
(f"the loop asked for {decision.i_w:+.0f} W"
+ (" (integrator bound)" if decision.i_clamped else "")
+ (" (output bound)" if decision.clamped else "")),
)
claims.append(Claim.set("loop", P_LOOP, decision.target_w, decision.reason))
else:
# Stopped or blind: no accumulation may survive the outage.
self.i_w = 0.0
self.loop_w = None
# A loop that is not running cannot be winding. Clearing this here
# rather than letting the last value stand is what makes the alarm
# de-assert when control is stopped.
self.loop_clamped = (False, "")
resolution = resolve(claims)
if resolution.contradiction:
@@ -406,24 +472,117 @@ class Controller:
self.reason = resolution.explain()
await self.write_setpoint(self.target)
# --- SAFETY-18, and it runs AFTER `self.target` is assigned ----------
# ⚠️ THE BOUNDARY IS THE ASSIGNMENT ABOVE, NOT write_setpoint(). This
# comment used to say "the command that has just gone on the wire",
# which is wrong twice over and was corrected by TEST-06's reviewer:
# `check_meter_frozen()` grades `self.target` - the arbiter's resolved
# intent - and `write_setpoint()` clamps only a local `value` and can
# skip the write entirely, so that number may never have gone anywhere.
# --- SAFETY-03, and its boundary is the WRITE ------------------------
# ⚠️ NOT THE SAME BOUNDARY AS SAFETY-18's BELOW, THOUGH THE CALLS ABUT.
# This one reads `self.clamp_last`, which is None until write_setpoint()
# has actually run, so it has to follow the write. The one below grades
# `self.target`, which is assigned further up - see check_output_clamp()
# and check_meter_frozen() for each. Two guards, two orderings, adjacent
# lines: do not "tidy" them into one rule.
await self.check_output_clamp()
# --- SAFETY-18, and its boundary is the ASSIGNMENT --------------------
# ⚠️ THE BOUNDARY IS `self.target = resolution.target_w` ABOVE, NOT
# write_setpoint(). This comment used to say "the command that has just
# gone on the wire", which is wrong twice over and was corrected by
# TEST-06's reviewer: `check_meter_frozen()` grades `self.target` - the
# arbiter's resolved intent - and `write_setpoint()` clamps only a local
# `value` and can skip the write entirely, so that number may never have
# gone anywhere.
#
# What both terms of the conjunction must describe is the SAME moment:
# the command this cycle resolved, against how long the meter that
# chose it has been standing still. Evaluated before
# the command this cycle resolved, against how long the meter that chose
# it has been standing still. Evaluated before
# `self.target = resolution.target_w` it would grade the PREVIOUS
# cycle's command against THIS cycle's meter - the one combination that
# was true of neither moment. Moving it between the assignment and
# write_setpoint() is therefore harmless; moving it above the
# assignment is not, and test_freeze.py part 5 fails on exactly that.
# write_setpoint() is therefore harmless; moving it above the assignment
# is not, and test_freeze.py part 5 fails on exactly that.
await self.check_meter_frozen()
async def check_output_clamp(self) -> None:
"""SAFETY-03. Alarm whenever a clamp actually engaged this cycle.
Runs AFTER write_setpoint() for the same reason SAFETY-18 does: the
clamp is arithmetic done inside the write, so before the write there is
nothing yet to report.
⚠️ PUBLISHING IS NOT IN THE WRITE PATH. The clamp itself is synchronous
inside write_setpoint(); this - the awaits, the notification, anything
that can block on Home Assistant - is out here. A stalled supervisor must
never be able to delay or drop a frame.
⚠️ THIS IS NOT A TRIP. No claim is filed, nothing is latched, 0 W is not
commanded and `publish()` still reports "running". §11.2's action for
this row is "clamp and alarm", and the clamped value is written and the
loop keeps running. Turning this into a trip means coming here and
changing that deliberately, not discovering it in the field.
"""
engaged_write = bool(self.clamp_last and self.clamp_last.engaged)
engaged_loop, loop_note = self.loop_clamped
engaged = engaged_write or engaged_loop
if engaged:
where = (self.clamp_last.describe() if engaged_write
else f"{CLAMP_REASON}: {loop_note}, above the "
f"{float(self.o.get('max_w', 2000)):.0f} W rating, and "
f"was clamped to {self.target:+.0f} W")
reason = (f"{where}. This is a BUG SIGNAL, not a limit being used: "
f"something upstream computed a setpoint the actuator "
f"cannot deliver.")
else:
reason = ""
fired = self.clamp_wd.note(engaged, reason, datetime.now(timezone.utc))
# --- THE ACTION LOG, AND IT IS COALESCED PER EPISODE ------------------
# Every engagement is recorded, but consecutive ones are recorded by
# UPDATING one row's count rather than inserting a row each.
#
# ⚠️ THAT IS NOT COSMETIC, AND IT WAS FOUND THE HARD WAY. `log_event()`
# keeps 40 rows; a winding loop clamps on every cycle, so a row per
# engagement flushes the whole ring within 40 cycles - and the first
# thing it evicted, in test_freeze.py, was SAFETY-18's "METER FROZEN?"
# row. A guard whose log drowns out the other guards' logs has moved the
# flood rather than removed it. The count here is `clamp_wd.engagements`,
# so nothing is lost: the row says exactly how many, and the number is
# read off the counter rather than off the number of rows.
#
# This lives here rather than only in write_setpoint() because on a
# default install the actuator-side clamp is unreachable - compute() has
# already bounded the integrator - so the only engagements that exist are
# loop-side ones. Measured: 60 cycles of the §10.3 runaway produce 0
# write-path engagements and 2 loop-side ones.
if engaged and not engaged_write:
tag = f"{CLAMP_REASON} episode {self.clamp_wd.episodes}"
row = (f"{tag}: {self.clamp_wd.engagements} engagement(s) - "
f"{reason}")
if self.events and tag in self.events[0]:
stamp = self.events[0].split(" ", 1)[0]
self.events[0] = f"{stamp} {row}"
else:
self.log_event(row)
if not fired:
return
_LOG.error(
"SAFETY-03 - %s Silently clipping this is how a winding integrator "
"reached 14 768 W on this project before: the clamp makes every "
"dashboard read a legal number while the thing producing it goes on "
"climbing. The command IS clamped and the loop is STILL RUNNING - "
"this is an alarm, not a trip. Engagements so far: %d.",
self.clamp_wd.reason, self.clamp_wd.engagements)
# A fixed notification_id, for the reason check_meter_frozen() gives -
# and a DIFFERENT one from every other alarm this add-on raises, because
# a clamp is not a frozen meter and neither is a saturation freeze.
await self.hass.call("persistent_notification", "create", {
"notification_id": "goodwe_output_clamp",
"title": "GoodWe: a computed setpoint exceeded the rating",
"message": (
f"{self.clamp_wd.reason}\n\nThe command has been clamped and "
f"the controller is STILL RUNNING. This is NOT the inverter "
f"failing to follow a legal command (that is the saturation "
f"freeze, which is normal and does not alarm) - the value asked "
f"for was itself out of range."),
})
async def check_meter_frozen(self) -> None:
"""SAFETY-18. Alarm on a meter that answers forever without changing.
@@ -704,6 +863,23 @@ class Controller:
out.append({"ok": True, "warn": False,
"text": f"setpoint {sp} (range {self.dev_min:g}…{self.dev_max:g} W)"})
# SAFETY-03. Red while a clamp is engaged, and the engagement COUNT is
# shown even after it clears - a loop that wound once and stopped is
# still a bug, and a row that vanishes with the condition takes the
# evidence with it. Reported only when there is something to report:
# unlike the freeze detector this guard has no "not covered" state, it
# runs on every write on every transport.
if self.clamp_wd.active:
out.append({"ok": False, "warn": False,
"text": "OUTPUT CLAMP ENGAGED - " + self.clamp_wd.reason
+ f" Engagements: {self.clamp_wd.engagements}."})
elif self.clamp_wd.engagements:
out.append({"ok": False, "warn": True,
"text": f"output clamp: not engaged now, but it has "
f"engaged {self.clamp_wd.engagements} time(s) in "
f"{self.clamp_wd.episodes} episode(s) since "
f"start - see the event log"})
if self.last_write_ok is False:
out.append({"ok": False, "warn": False,
"text": "last write to the inverter was REJECTED - check the log"})
goodwe_controller/test_clamp.py
+550
View File
@@ -0,0 +1,550 @@
"""Runnable check for SAFETY-03, the output clamp and its winding-loop alarm.
python3 test_clamp.py
No framework, asserts only, same house style as test_freeze.py.
Five parts, and parts 3 and 4 are the ones that carry the acceptance criteria:
1. THE CLAMP ARITHMETIC, on the shipped `clamp_to_rating()` - boundaries,
symmetry, the device second stage, and the non-finite cases. Table driven.
2. THE ALARM'S CADENCE - rising edge, re-assert, hysteresis - on the shipped
`ClampWatchdog`. Every engagement counted, notifications not flooded.
3. AC 1 + AC 3, WIRED. A computed setpoint that exceeds the rating, driven
through the REAL `Controller.cycle()`: nothing over the rating reaches the
device, and the alarm fires. Includes the historical runaway from
specs.md 10.3 replayed against an unclamped transcription, so the check
says what the clamp is actually worth in watts rather than asserting a
constant.
4. AC 2, AND IT IS THE LOAD-BEARING HALF. A loop that SATURATES WITHOUT
CLAMPING - the inverter refusing to follow a perfectly legal command - must
raise NO clamp alarm. If it does, the bug signal is inside the noise again
and the whole ticket has bought nothing.
5. NO PATH BYPASSES IT. write_setpoint() is the only route to the wire, so it
is called directly with an out-of-range value the way the heartbeat and the
operator's buttons call it.
------------------------------------------------------------------------------
WHY PART 4 ALSO ASSERTS ON `Decision`'s BOOLEANS AND NOT ON `reason`
`Decision.reason` is one string and the stages after the clamp overwrite it. The
check below drives a cycle that clamps the integrator AND is then slew limited,
and asserts that `reason` does NOT say so while `i_clamped` does. That is not a
style preference - an alarm keyed on the string is silent on exactly the cycles a
winding loop produces, and this is the check that would go red if someone
"simplified" main.py back onto it.
------------------------------------------------------------------------------
WHAT THIS CHECK DOES NOT COVER
- The firmware-side magnitude clamp (the card's "two clamps, not one"). This
add-on's `write_setpoint()` is the add-on half; the ESP32 half is DRIVER-07's
and is not in this repo's Python.
- A rig run. Nothing here needs one: the clamp is arithmetic on the write path
and part 3 drives the shipped `Controller.cycle()` against a stub Home
Assistant, which is the same surface test_freeze.py part 1-2 use. The wire-
level invariant the card asks for (`|emitted| <= rating` on every frame) is
asserted here over every value that reached `set_number()`, which is the last
thing this process controls; proving it at the inverter needs the witness log
and belongs with DRIVER-07.
"""
import asyncio
import math
import sys
from datetime import datetime, timedelta, timezone
from app.clamp import (
CLEAR_CYCLES, REASON_CODE, REASSERT_S, ClampWatchdog, clamp_to_rating,
is_commandable,
)
from app.control import Tuning, compute
from app.main import Controller
fails = []
T0 = datetime(2026, 8, 30, 9, 0, 0, tzinfo=timezone.utc)
# The live install's numbers. NOTES.md and freeze.py both say max_w: 5000 there,
# which is also the device rating, so this is the bound the alarm is written at.
LIVE_MAX_W = 5000.0
CLAMP_ID = "goodwe_output_clamp"
def check(name, cond):
print(f" {'ok ' if cond else 'FAIL'} {name}")
if not cond:
fails.append(name)
# ==============================================================================
print("1. the clamp arithmetic, on the shipped expression")
# (requested, limit, expected applied, expected engaged, expected bound)
TABLE = [
(5001.0, 5000.0, 5000.0, True, "max_w"), # +limit+1
(-5001.0, 5000.0, -5000.0, True, "max_w"), # -limit-1, the export side
(5000.0, 5000.0, 5000.0, False, ""), # exactly +limit: NOT engaged
(-5000.0, 5000.0, -5000.0, False, ""), # exactly -limit: NOT engaged
(0.0, 5000.0, 0.0, False, ""), # the failsafe value
(1234.0, 5000.0, 1234.0, False, ""),
(14768.0, 5000.0, 5000.0, True, "max_w"), # 10.3's reported runaway
]
for requested, limit, want_applied, want_engaged, want_bound in TABLE:
r = clamp_to_rating(requested, limit)
check(f"{requested:+.0f} W against a {limit:.0f} W rating -> "
f"{r.applied_w:+.0f} W, engaged={r.engaged}, bound={r.bound!r}",
r.applied_w == want_applied and r.engaged is want_engaged
and r.bound == want_bound)
# It is a MAGNITUDE clamp (3.2), so import and export must be bound identically.
# Derived over a sweep rather than asserted on two hand-picked values.
asym = [m for m in range(0, 12001, 37)
if clamp_to_rating(m, LIVE_MAX_W).applied_w
!= -clamp_to_rating(-m, LIVE_MAX_W).applied_w]
check(f"symmetric over {len(range(0, 12001, 37))} magnitudes "
f"({len(asym)} asymmetric)", not asym)
# The engagement flag has to mean "the RAW value differed", which is the words
# AC 2 uses. Re-derived from the result's own fields, not from the table.
mismatch = [m for m in range(0, 12001, 37)
for r in [clamp_to_rating(m, LIVE_MAX_W)]
if r.engaged != (r.requested_w != r.applied_w)]
check("engaged is exactly 'raw differs from applied', over the same sweep",
not mismatch)
# The device second stage. Only reachable when max_w is ABOVE the device range,
# i.e. when validate_startup_config() did not run or no longer holds.
r = clamp_to_rating(3000.0, 5000.0, dev_min=-2000.0, dev_max=2000.0)
check(f"a device narrower than max_w still bites, and says which bound "
f"({r.applied_w:+.0f} W, {r.bound!r})",
r.applied_w == 2000.0 and r.engaged and r.bound == "device")
r = clamp_to_rating(3000.0, 5000.0, dev_min=-5000.0, dev_max=5000.0)
check("...and is a no-op once the startup validation holds",
r.applied_w == 3000.0 and not r.engaged)
# ⚠️ REGRESSION, REVIEW-FOUND. THE FIRST STAGE USED TO `return` AND THE DEVICE
# STAGE WAS THEN NEVER REACHED, so the harder the overshoot the less the guard
# did: at max_w 5000 against a +-4000 device, +9999 emitted +5000 - OUTSIDE THE
# DEVICE RANGE - while a mere +4500 was correctly held to +4000. That inverts
# the guard and it was a regression against the pre-PR `write_setpoint()`, whose
# comment read "never send a value the device will reject outright".
#
# The property is asserted over a sweep, not on the two witness values, because
# the defect was invisible to every single-value check that happened to sit
# inside one stage.
DEV_LO, DEV_HI = -4000.0, 4000.0
escapes = [(m, clamp_to_rating(m, 5000.0, DEV_LO, DEV_HI).applied_w)
for m in range(-15000, 15001, 61)
for a in [clamp_to_rating(m, 5000.0, DEV_LO, DEV_HI).applied_w]
if not (DEV_LO <= a <= DEV_HI)]
big = clamp_to_rating(9999.0, 5000.0, DEV_LO, DEV_HI)
small = clamp_to_rating(4500.0, 5000.0, DEV_LO, DEV_HI)
print(f" max_w 5000 vs a device of {DEV_LO:+.0f}..{DEV_HI:+.0f}: "
f"+9999 -> {big.applied_w:+.0f} W ({big.bound}), "
f"+4500 -> {small.applied_w:+.0f} W ({small.bound})")
check(f"BOTH stages always run - {big.requested_w:+.0f} W emits "
f"{big.applied_w:+.0f} W, inside the device's {DEV_HI:+.0f} W, and names "
f"both bounds ({big.bound!r})",
big.applied_w == DEV_HI and big.bound == "max_w+device")
check(f"...and it holds over the whole range, so no overshoot escapes the "
f"device ({len(escapes)} escapes in {len(range(-15000, 15001, 61))})",
not escapes)
check("...a big overshoot is not clamped LESS than a small one",
abs(big.applied_w) <= abs(small.applied_w) + 0.0)
print(" non-finite input is loss of input, not an out-of-range value")
for bad in (None, float("nan"), float("inf"), float("-inf"), "abc", object()):
check(f"is_commandable({bad!r:.30}) is False", is_commandable(bad) is False)
for good in (0, -5000.0, 5000, 12.5):
check(f"is_commandable({good!r}) is True", is_commandable(good) is True)
# ⚠️ WHY THAT GUARD EXISTS, AND THE ANSWER IS WORSE THAN "it passes a nan
# through". MEASURED HERE, not recalled: every comparison against a nan is
# False, so `min(limit, nan)` keeps its FIRST argument and a naive clamp turns a
# nan into a command at the FULL RATING - and then reports engaged=True, because
# limit != nan, so the winding-loop counter gets a transport bug filed in it.
# The number below is computed by the same expression the guard replaces.
nan = float("nan")
naive = max(-LIVE_MAX_W, min(LIVE_MAX_W, nan))
check(f"a naive clamp turns a nan into {naive:+,.0f} W and calls it "
f"engaged={naive != nan} - hence the guard",
naive == LIVE_MAX_W and (naive != nan) is True)
# ==============================================================================
print()
print("2. the alarm's cadence: every engagement counted, no flood")
wd = ClampWatchdog()
t = T0
fired = [wd.note(True, "winding", t + timedelta(seconds=i)) for i in range(10)]
check(f"the rising edge fires ({fired[0]})", fired[0] is True)
check(f"the next 9 one-second cycles fire {sum(fired[1:])} more times "
f"(re-assert is {REASSERT_S:.0f} s)", sum(fired[1:]) == 0)
check(f"but every one of them is counted: engagements={wd.engagements}",
wd.engagements == 10)
check(f"and it is one episode, not ten: episodes={wd.episodes}",
wd.episodes == 1)
# --- THE RE-ASSERT CADENCE, PINNED IN BOTH DIRECTIONS ------------------------
# ⚠️ REVIEW-FOUND. This used to read `wd.note(True, ..., T0 + REASSERT_S + 9)`,
# which is green for ANY value of the constant - REASSERT_S = 3600 passed, and an
# hour of silence is precisely what the module promises not to do. A check whose
# stimulus is written in terms of the constant it is grading cannot grade it.
#
# The requirements are stated in wall-clock seconds and derived from what the
# alarm is FOR, so they do not move when the constant does:
#
# floor - a loop pinned on the rail must speak at least MIN_PER_QUARTER times
# inside the 900 s quarter that bills the capacity tariff. That window
# is the same one freeze.py and DOCS.md reason about.
# ceiling - it must not speak on more than MAX_FRACTION of cycles, or it is
# the per-cycle flood the split exists to remove.
QUARTER_S = 900.0
MIN_PER_QUARTER = 5
MAX_FRACTION = 0.20
pinned = ClampWatchdog()
alarms = sum(1 for i in range(int(QUARTER_S))
if pinned.note(True, "pinned", T0 + timedelta(seconds=i)))
print(f" pinned on the rail for {QUARTER_S:.0f} s at 1 s cycles: "
f"{pinned.engagements} engagements, {alarms} alarm(s) "
f"({alarms / QUARTER_S:.1%} of cycles)")
check(f"a loop pinned for a whole billing quarter speaks at least "
f"{MIN_PER_QUARTER} times (it spoke {alarms})",
alarms >= MIN_PER_QUARTER)
check(f"...and on no more than {MAX_FRACTION:.0%} of cycles (it was "
f"{alarms / QUARTER_S:.1%})", alarms <= MAX_FRACTION * QUARTER_S)
check("...while every single cycle is still counted",
pinned.engagements == int(QUARTER_S))
# --- THE HYSTERESIS, PINNED AGAINST THE CHATTER DIRECTION --------------------
# ⚠️ REVIEW-FOUND. This used to issue `range(CLEAR_CYCLES - 1)` clean cycles,
# which at CLEAR_CYCLES = 1 issues ZERO of them and passes while printing
# "0 clean cycles do NOT clear it". Green for every value >= 1 - including 1,
# which IS the chatter the constant exists to prevent. The stimulus is a literal
# now: ONE clean cycle must not clear the alarm, whatever the constant says.
wd2 = ClampWatchdog()
wd2.note(True, "winding", T0)
wd2.note(False, "", T0 + timedelta(seconds=1))
check(f"ONE clean cycle does NOT clear the alarm (active={wd2.active}) - "
f"a command dithering across the rail must not chatter",
wd2.active is True)
survived = 1
while wd2.active and survived < 50:
wd2.note(False, "", T0 + timedelta(seconds=1 + survived))
survived += 1
print(f" it took {survived} consecutive clean cycles to de-assert "
f"(CLEAR_CYCLES = {CLEAR_CYCLES})")
check(f"...and it does de-assert, after {survived} of them, rather than "
f"latching", wd2.active is False and survived == CLEAR_CYCLES)
check("...and the engagement count survives the clear, so the evidence does",
wd2.engagements == 1)
check("a clean cycle on an idle watchdog says nothing",
wd2.note(False, "", T0 + timedelta(seconds=300)) is False)
# ==============================================================================
print()
print("3. AC 1 + AC 3, wired: over the rating both clamps AND alarms")
class _Store:
data = {}
def set(self, *a):
pass
def get_time(self, *a):
return None
class _Pub:
enabled = False
def publish(self, *a, **k):
pass
def close(self):
pass
class _Hass:
"""Stub HA. `track` False models an inverter that is switched off entirely,
which is the 10.3 runaway's own setup."""
def __init__(self, grid, batt=0.0, track=True):
self.grid = grid
self.batt = batt
self.track = track
self.calls = []
self.writes = []
async def limits(self, entity):
return (-LIVE_MAX_W, LIVE_MAX_W, 1.0)
async def number(self, entity, invert=False):
return {"sensor.grid": self.grid, "sensor.soc": 50.0,
"sensor.batt": self.batt}.get(entity)
async def set_number(self, entity, value):
self.writes.append(float(value))
if self.track:
self.batt = float(value)
return True
async def call(self, domain, service, data):
self.calls.append((domain, service, data))
return True
def wired(grid, cycles=40, batt=0.0, track=True, **over):
async def go():
hass = _Hass(grid, batt=batt, track=track)
c = Controller(dict(
{"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
"soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
"meter_source": "off", "max_w": LIVE_MAX_W, "gain": 0.6,
"slew_w": 1000, "deadband_w": 15, "target_grid_w": 0,
"step_w": 10, "heartbeat_s": 10, "stale_input_s": 15},
**over), hass, _Store(), _Pub(), stop=asyncio.Event())
c.auto = True
c.dev_min, c.dev_max = -LIVE_MAX_W, LIVE_MAX_W
# The red status row exists only WHILE the clamp is engaged, and the
# hysteresis clears it a few cycles later - so it is captured live
# rather than looked for after the run, when it is correctly gone.
c.red_snapshot = []
for _ in range(cycles):
await c.read_inputs()
await c.cycle()
if c.clamp_wd.active and not c.red_snapshot:
c.red_snapshot = c.checks()
return hass, c
return asyncio.run(go())
def clamp_notes(hass):
return [d for dom, svc, d in hass.calls
if (dom, svc) == ("persistent_notification", "create")
and d.get("notification_id") == CLAMP_ID]
# THE 10.3 RUNAWAY, replayed. "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." `track=False` is the inverter switched off: it
# reports 0 W however hard it is commanded, so the error never closes.
hass_r, ctl_r = wired(6000.0, cycles=60, track=False)
emitted = [abs(w) for w in hass_r.writes]
# What the SAME error would have done to an accumulator with no bound - the
# transcription is deliberately the unclamped velocity form 10.3 describes, so
# the figure below is computed here and not quoted from the spec.
runaway, err = 0.0, 6000.0
for _ in range(60):
runaway += 0.6 * err
print(f" unclamped, the same 60 cycles reach {runaway:,.0f} W; "
f"the largest value that reached the device is {max(emitted):,.0f} W")
check(f"an unbounded accumulator really does run away here "
f"({runaway:,.0f} W >> {LIVE_MAX_W:.0f} W)", runaway > 3 * LIVE_MAX_W)
check(f"AC 1: NOT ONE of the {len(emitted)} values that reached the device "
f"exceeds the {LIVE_MAX_W:.0f} W rating",
emitted and max(emitted) <= LIVE_MAX_W)
# ⚠️ AND THE CLAMP IS NOT THE ONLY THING HOLDING THAT LINE HERE, which is worth
# saying out loud rather than letting the number above imply otherwise. Against
# an inverter reporting 0 W the saturation freeze engages after
# `saturation_cycles` and pins the command well BELOW the rail - that is why the
# largest emitted value is what it is and not 5000 W. The clamp is what makes it
# bounded; the freeze is what makes it small. A check that read
# `max(emitted) == max_w` would have been asserting the freeze, not the clamp.
check(f"...and the saturation freeze is what pins it at {max(emitted):,.0f} W "
f"rather than the rail", ctl_r.sat_count >= ctl_r.tuning.saturation_cycles)
notes_r = clamp_notes(hass_r)
print(f" the loop clamped on {ctl_r.clamp_wd.engagements} of {60} cycles, "
f"in {ctl_r.clamp_wd.episodes} episode(s), raising {len(notes_r)} "
f"notification(s)")
check("AC 3: and it ALARMED - silently clipping this is the defect",
ctl_r.clamp_wd.engagements > 0 and len(notes_r) >= 1)
check("...with its own notification_id, not shared with any other guard",
all(d["notification_id"] == CLAMP_ID for d in notes_r))
check("...one episode, re-asserted rather than one per cycle",
ctl_r.clamp_wd.episodes == 1 and len(notes_r) < ctl_r.clamp_wd.engagements)
check("...and it names itself a bug signal, not a limit in use",
bool(notes_r) and "out of range" in notes_r[0]["message"])
rows_r = [e for e in ctl_r.events if REASON_CODE in e]
check(f"the action log records every engagement, coalesced into one row per "
f"episode ({len(rows_r)} row(s), {ctl_r.clamp_wd.engagements} "
f"engagement(s))",
len(rows_r) == ctl_r.clamp_wd.episodes
and f"{ctl_r.clamp_wd.engagements} engagement(s)" in rows_r[0])
# ⚠️ REGRESSION, AND IT WAS A REAL BUG BEFORE IT WAS A CHECK. `log_event()` keeps
# 40 rows. A row per engagement flushes that ring inside 40 cycles of a winding
# loop, and the first thing it evicted was SAFETY-18's own "METER FROZEN?" row -
# i.e. this guard silencing another guard's log. Driven here with a plant that
# TRACKS the command, so the saturation freeze never engages and the integrator
# clamps on essentially every cycle.
hass_l, ctl_l = wired(9000.0, cycles=80, track=True)
rows_l = [e for e in ctl_l.events if REASON_CODE in e]
other_l = [e for e in ctl_l.events if REASON_CODE not in e]
print(f" 80 winding cycles: {ctl_l.clamp_wd.engagements} engagement(s) -> "
f"{len(rows_l)} log row(s), leaving room for {len(other_l)} other event(s)")
check(f"a long winding episode does NOT flood the 40-row event log "
f"({len(rows_l)} rows for {ctl_l.clamp_wd.engagements} engagements)",
ctl_l.clamp_wd.engagements > 40 and len(rows_l) <= ctl_l.clamp_wd.episodes)
check("...and the count in the row is the counter's, so nothing is lost",
f"{ctl_l.clamp_wd.engagements} engagement(s)" in rows_l[-1])
# The RED row exists only while engaged, and the episode above is short, so it
# is captured mid-run rather than asserted after the hysteresis has cleared it.
check("the status page goes red WHILE it is engaged",
any(not c["ok"] and not c["warn"] and "OUTPUT CLAMP ENGAGED" in c["text"]
for c in ctl_r.red_snapshot))
check("...and once clear, a warn row still says how many times it engaged",
any(c["warn"] and f"{ctl_r.clamp_wd.engagements} time(s)" in c["text"]
for c in ctl_r.checks()))
# ⚠️ NOT A TRIP. 11.2's action is "clamp and alarm"; the loop keeps running and
# keeps writing the clamped value. If a later cut changes that it must come here.
check(f"it is NOT a trip - the command stands at {ctl_r.target:+.0f} W, not 0 W",
ctl_r.target != 0.0 and ctl_r.target == hass_r.writes[-1])
check("...and the controller still reports itself running, not failsafe",
ctl_r.auto and not ctl_r.sign_wd.tripped)
# The export side, because the clamp is a magnitude clamp and a sign bug in it
# would be invisible on the import case alone.
hass_x, ctl_x = wired(-6000.0, cycles=60, track=False)
check(f"the export side clamps and alarms too "
f"({min(hass_x.writes):+,.0f} W, {len(clamp_notes(hass_x))} note(s), "
f"{ctl_x.clamp_wd.engagements} engagement(s))",
min(hass_x.writes) < 0 and abs(min(hass_x.writes)) <= LIVE_MAX_W
and clamp_notes(hass_x) and ctl_x.clamp_wd.engagements > 0)
# ==============================================================================
print()
print("4. AC 2: saturation WITHOUT clamping must NOT raise a clamp alarm")
# A legal command the inverter will not follow. The meter sits just outside the
# deadband so the loop moves a little and then rests; `batt` is pinned far from
# whatever is commanded, which is the saturation condition (|cmd - actual| >
# saturation_w) and nothing else. The integrator never approaches the rating.
hass_s, ctl_s = wired(60.0, cycles=40, batt=2500.0, track=False)
peak = max(abs(w) for w in hass_s.writes)
print(f" saturated for {ctl_s.sat_count} consecutive cycles; the command "
f"peaked at {peak:.0f} W, {peak / LIVE_MAX_W:.1%} of the rating")
# ⚠️ THE POSITIVE HALF FIRST. A negative check that saw nothing happen is not an
# observation - if the saturation freeze never engaged, "no clamp alarm" is
# vacuous and this check is worthless.
check(f"the saturation freeze REALLY DID engage (sat_count="
f"{ctl_s.sat_count} >= {ctl_s.tuning.saturation_cycles})",
ctl_s.sat_count >= ctl_s.tuning.saturation_cycles)
check("...and the event log says so, in the saturation's own words",
any("saturation freeze" in e for e in ctl_s.events))
check(f"...while the command stayed well inside the rating "
f"({peak:.0f} W < {LIVE_MAX_W:.0f} W)", peak < LIVE_MAX_W)
# Now the negative, which is only worth anything because of the three above.
check("AC 2: NO clamp notification was raised", not clamp_notes(hass_s))
check("AC 2: the clamp counted ZERO engagements", ctl_s.clamp_wd.engagements == 0)
check("AC 2: no OUTPUT_CLAMP row in the action log",
not [e for e in ctl_s.events if REASON_CODE in e])
check("AC 2: and no clamp row on the status page",
not [c for c in ctl_s.checks() if "output clamp" in c["text"].lower()])
print(" ...and the two alarms do not share a notification_id")
ids_clamp = {d["notification_id"] for d in clamp_notes(hass_r)}
ids_other = {d.get("notification_id") for dom, svc, d in hass_r.calls
if svc == "create"} - ids_clamp
check(f"clamp ids {sorted(ids_clamp)} are disjoint from every other alarm's "
f"{sorted(x for x in ids_other if x)}", not (ids_clamp & ids_other))
print(" ...and the alarm cannot be keyed on Decision.reason")
# A cycle that clamps the integrator AND is then slew limited. `reason` reports
# the LAST stage, so a string-keyed alarm is silent on it; the boolean is not.
t = Tuning(max_w=2000.0, slew_w=100.0, gain=0.6)
d = compute(prev_w=1000.0, grid_w=6000.0, actual_w=1000.0, tuning=t, i_w=1900.0)
print(f" reason={d.reason!r} i_clamped={d.i_clamped} clamped={d.clamped}")
check("the integrator DID clamp on that cycle", d.i_clamped is True)
check("...but `reason` says 'slew-limited' - the clamp was overwritten "
"downstream, so a string-keyed alarm is SILENT here",
d.reason == "slew-limited" and "clamp" not in d.reason)
# And the same for the OUTPUT clamp, on a config that lets the integrator run
# above the rail - then both flags fire and `reason` still reports neither.
t2 = Tuning(max_w=2000.0, integrator_max_w=3000.0, slew_w=100.0, gain=0.6)
d2 = compute(prev_w=1000.0, grid_w=6000.0, actual_w=1000.0, tuning=t2, i_w=2900.0)
print(f" reason={d2.reason!r} i_clamped={d2.i_clamped} clamped={d2.clamped}")
check("both clamps engaged, and `reason` reports neither",
d2.i_clamped and d2.clamped and "clamp" not in d2.reason)
# ⚠️ HOW THE FREEZE AND THE CLAMP INTERACT, MEASURED RATHER THAN ASSUMED, AND
# THE MEASUREMENT REFUSED TWO GUESSES BEFORE THIS ONE. Written first as "a
# frozen cycle never clamps" (false), then as "only the i_w == 0 reseed
# exception clamps" (also false - at i_w == 0 with a matching actual the cycle
# is not saturated at all, so it is never frozen). What the sweep actually
# shows: a frozen cycle clamps exactly when the integrator REVERSES. control.py
# permits that on purpose - "it may fall, cross zero, or reverse outright" -
# because the alternative deadlocks the loop, and a reversal is a full swing
# that can clear the bound on the far side in one step.
#
# That matters for this ticket rather than being trivia: it is the one shape
# where the saturation freeze and a clamp engagement coincide, so it is exactly
# where an alarm that shared a channel with saturation would lose the clamp.
grid_sweep = (-9000.0, -3000.0, 3000.0, 9000.0)
grid = [(pp, g) for pp in (-1900.0, -500.0, 0.0, 500.0, 1900.0)
for g in grid_sweep]
runs = {(pp, g): compute(pp, g, 0.0, t, sat_count=5, i_w=pp) for pp, g in grid}
frozen = [k for k, d in runs.items() if d.frozen]
frozen_clamps = [k for k in frozen if runs[k].i_clamped]
# The property, stated independently of the control law's own branches: the
# correction pushes AGAINST the standing integrator, and the resulting swing
# clears the bound.
reversals = [(pp, g) for pp, g in frozen
if pp * g < 0 and abs(pp + t.gain * g) > t.max_w]
print(f" {len(frozen)} of {len(grid)} sweep cells are frozen; "
f"those that also clamp: {frozen_clamps}")
check(f"a frozen cycle CAN clamp, and only by reversing "
f"({len(frozen_clamps)} of {len(frozen)} frozen cells)",
frozen_clamps and frozen_clamps == reversals)
check("...and a frozen cycle pushing the way it already leans never clamps - "
"that is the freeze doing its job",
not [k for k in frozen if runs[k].i_clamped and k[0] * k[1] > 0])
check("a healthy cycle sets neither flag",
not any([compute(0.0, 100.0, 0.0, t).i_clamped,
compute(0.0, 100.0, 0.0, t).clamped]))
# ==============================================================================
print()
print("5. no path reaches the wire around the clamp")
async def _direct():
hass = _Hass(0.0)
c = Controller({"setpoint_entity": "number.sp", "max_w": 2000,
"meter_source": "off"},
hass, _Store(), _Pub(), stop=asyncio.Event())
c.dev_min, c.dev_max = -LIVE_MAX_W, LIVE_MAX_W
# The shape the heartbeat, shutdown() and handle_action() all use.
await c.write_setpoint(9999.0, force=True)
await c.write_setpoint(-9999.0, force=True)
await c.write_setpoint(float("nan"), force=True)
await c.write_setpoint(1500.0, force=True)
return hass, c
hass_d, ctl_d = asyncio.run(_direct())
print(f" asked for 9999 / -9999 / nan / 1500 W; the device saw "
f"{hass_d.writes}")
check("a direct out-of-range write is clamped, not passed through",
hass_d.writes[:2] == [2000.0, -2000.0])
check("a nan is refused and becomes 0 W, per 11.1 rule 1",
hass_d.writes[2] == 0.0)
check("...and is NOT counted as a clamp engagement - it is a transport bug",
sum(1 for e in ctl_d.events if REASON_CODE in e) == 2)
check("an in-range value is untouched", hass_d.writes[3] == 1500.0)
check("every emitted frame satisfies |emitted| <= the rating",
all(abs(w) <= 2000.0 for w in hass_d.writes))
print()
if fails:
print(f"{len(fails)} FAILED: {', '.join(fails)}")
sys.exit(1)
print("all checks passed")
goodwe_controller/test_freeze.py
+59 -23
View File
@@ -672,6 +672,11 @@ def wired(unchanged_s, grid, cycles=60, **over):
return asyncio.run(go())
# ⚠️ SAFETY-03 (#12) arrived at the same diagnosis independently and added a
# `frozen_notes()` keyed on the id. `notes_of()` above subsumes it - same id,
# plus the title fallback that keeps the id's own check independently
# falsifiable, plus `all_notes_of()` for the claims that are about the whole
# surface rather than about this guard.
hass_f, ctl_f, trace_f = wired(3600.0, 4000.0)
notes = notes_of(hass_f) # THIS guard's
all_notes = all_notes_of(hass_f) # the whole surface
@@ -682,14 +687,19 @@ ctl_f.p1_enabled = True
rows = [c for c in ctl_f.checks() if "METER MAY BE FROZEN" in c["text"]]
print(f" frozen at +4000 W: command ran to {ctl_f.target:.0f} W, "
f"{len(notes)} notification(s), {len(rows)} red status row(s)")
# SURFACE-WIDE, deliberately. "The controller alarmed once" is a different and
# stronger claim than "the freeze guard alarmed once", and it is the one this
# row has always made. When a second guard lands (SAFETY-03's
# `goodwe_output_clamp` is the first), this assertion is where somebody decides
# whether the scenario is still meant to raise exactly one alarm in total - a
# decision, made here, rather than a claim quietly narrowed to keep it green.
check("the whole notification surface carried exactly one alarm this run",
len(all_notes) == 1)
# SURFACE-WIDE, deliberately, and the second guard has now landed. SAFETY-03
# ships `goodwe_output_clamp`, and this scenario - a runaway loop winding into
# the rail - raises it legitimately: the loop asks for more than the rating and
# is clamped. So the answer to "is this scenario still meant to raise exactly
# one alarm" is NO, it raises exactly two, and BOTH ARE NAMED here rather than
# the claim being narrowed until it went green again. A third guard has to come
# back to this line, which is the whole point of keeping one assertion in the
# file that accounts for the controller's entire alarm output.
_ids = sorted(d.get("notification_id") for d in all_notes)
print(f" the whole notification surface this run: {_ids}")
check("the whole notification surface carried exactly two alarms this run - "
"SAFETY-18's and SAFETY-03's, both named",
_ids == ["goodwe_meter_frozen", "goodwe_output_clamp"])
# THIS GUARD'S, keyed on the id.
check("...and the freeze detector is the one that raised it", len(notes) == 1)
# ⚠️ Selected by TEXT, not by id - see notes_by_text_of(). One of the two
@@ -709,24 +719,50 @@ check("the event log names it", any("METER FROZEN" in e for e in ctl_f.events))
check("and it does NOT command 0 W - this cut is an alarm only",
abs(ctl_f.target) >= rail_w(LIVE_MAX_W))
# ⚠️ THESE THREE ARE SURFACE-WIDE ON PURPOSE, and they are the three that go red
# against #12's `app/`. That is the correct signal and not a bug in them: each
# says "this scenario is quiet", which is a stronger claim than "this guard is
# quiet", and it is the claim a reader of a converged install wants. Whoever
# lands a second notifying guard has to come here and decide, per scenario,
# whether the new alarm belongs in it - `notes_of()` is one word away for the
# narrower reading, and choosing it must be a decision with a reason.
# (They also used to match `svc == "create"` in ANY domain, which was looser
# still.)
# ⚠️ THE THREE SCENARIOS THAT SAFETY-03 FORCED A DECISION ON, decided one at a
# time and not reflexively narrowed to green. Measured on the merged tree:
#
# scenario surface ids freeze
# frozen meter, railed command (above) 2 frozen + clamp 1
# TEL-05 converged, -14.2 W 0 - 0
# moving meter, a real 8 kW load 1 clamp 0
# uncovered transport, no unchanged_s 1 clamp 0
#
# CONVERGED stays SURFACE-WIDE. At the operating point this controller exists to
# hold, the correct behaviour is that the whole add-on has nothing to say - not
# merely that the freeze guard has nothing to say. Nothing about SAFETY-03
# weakens that claim, and it still holds, so narrowing it would give away a real
# property for nothing.
#
# The other two NARROW to notes_of(), and the reason is the same in both: the
# clamp alarm there is CORRECT and is SAFETY-03's subject, not this file's. A
# genuine 8 kW load really does drive the loop past the rating; an uncovered
# transport really does not stop it doing so. Asserting silence across the whole
# surface would make this file fail whenever a different guard correctly does
# its job - a claim about somebody else's behaviour wearing this file's name.
# What each still owns is the load-bearing half: THE FREEZE DETECTOR SAID
# NOTHING. The surface is printed rather than asserted, so a reader can still
# see what else fired without this file grading it.
hass_n, ctl_n, _ = wired(3600.0, -14.2)
check("wired up, TEL-05's converged case leaves the whole surface silent",
check("wired up, TEL-05's converged case leaves the WHOLE SURFACE silent - at "
"the operating point the add-on should have nothing at all to say",
not all_notes_of(hass_n))
hass_m, ctl_m, _ = wired(2.0, 8000.0)
check("wired up, a real railed load against a moving meter leaves the whole "
"surface silent",
not all_notes_of(hass_m) and abs(ctl_m.target) >= rail_w(LIVE_MAX_W))
check("an uncovered transport (no unchanged_s) leaves the whole surface silent",
not all_notes_of(wired(None, 4000.0)[0]))
print(f" real 8 kW load, moving meter: surface "
f"{sorted(d.get('notification_id') for d in all_notes_of(hass_m))}, "
f"freeze {len(notes_of(hass_m))}")
check("wired up, a real railed load against a moving meter raises nothing FROM "
"THIS GUARD (the clamp alarm there is SAFETY-03's and is correct)",
not notes_of(hass_m) and abs(ctl_m.target) >= rail_w(LIVE_MAX_W))
hass_u = wired(None, 4000.0)[0]
print(f" uncovered transport: surface "
f"{sorted(d.get('notification_id') for d in all_notes_of(hass_u))}, "
f"freeze {len(notes_of(hass_u))}")
check("an uncovered transport (no unchanged_s) raises nothing FROM THIS GUARD, "
"which is the honest report of a detector that cannot see",
not notes_of(hass_u))
# --------------------------------------------------------------------------- #
goodwe_controller/test_watchdog.py
+10
View File
@@ -274,6 +274,16 @@ from app.main import Controller # noqa: E402
class _Hass:
def __init__(self):
self.writes = []
self.calls = []
# ⚠️ ADDED FOR SAFETY-03, AND THE REASON IS WORTH READING. The e2e below
# runs a DELIBERATELY INVERTED loop (gain -0.6) into a steady 2000 W import,
# which is a loop that winds straight into the rail - so the output-clamp
# alarm fires here, and the stub not having `call` is how that was
# discovered. This scenario is now a second, independent witness that the
# clamp alarm sees a real winding loop; test_clamp.py asserts it explicitly.
async def call(self, domain, service, data):
self.calls.append((domain, service, data))
async def limits(self, entity):
return (-5000.0, 5000.0, 1.0)