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

Merge pull request 'SAFETY-52: a battery that refuses is asked for 0 W, with a 200 W probe every 15 min' (#141) from SAFETY-52-refused-hold into release/1.0
infra-pin / fresh (push) Successful in 1s
Details

Browse Source
This commit was merged in pull request #141.
This commit is contained in:
admin
2026-09-27 18:21:21 +02:00
parent d1f7e3336e 3c0f2b9669
commit 652097abde
11 changed files with 894 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
goodwe_controller/DECISIONS/SAFETY-52-a-battery-that-refuses.md
+110
View File
@@ -0,0 +1,110 @@
# SAFETY-52 - a battery that refuses is asked for 0 W
**Decided 2026-09-27 by the owner, on the phase A plan (option D3). All four
questions answered yes.**
## The defect
At the BMS cut-out the battery gives ~0 W whatever it is asked. The saturation
freeze (`control.py`, `frozen`) then holds the request where it stood, and its
SAFETY-13 exit cannot fire, because a reading of ~0 against a large ask lies
inside the interval a limit can produce. The request stood at +5000 W for
11.87 h on 2026-09-06 and at +1940 W for 3 h 55 m on the bus on 2026-09-19. It
moved again only when the grid error changed sign. SAFETY-48's lead cap only
changes where `i_w` parks (one slew step ahead instead of at the rail). When
the request moved after the reading went flat, the stale-reading gate won
instead and ran SAFETY-59's ~70 s trip cycle.
## What was decided
1. **A refused direction is held at 0 W**, reason `battery refuses`. It engages
in `compute()` when the freeze has held `SETTLE_CYCLES` and the reading is
below `REFUSED_BELOW_W` (100 W, the probe's own answer) against an ask beyond
`saturation_w`. The drop to 0 skips
the slew limit (it only removes an ask nobody meets). The direction is
carried like `sat_count` (`refused=`, `Decision.refused`).
2. **A 200 W probe every 15 minutes, 34 s long** (`PROBE_W`, `PROBE_EVERY_S`,
`PROBE_S`; twice `BATT_OBLIGED_S`). A reading of at least 100 W in the held
direction ends the refusal, during a probe or not. 200 W is below `freshness.MOVE_W` (250 W), so a probe cannot arm
the gate. The clock is main.py's `now_mono`; `compute()` stays clockless.
3. **A gate trip while the loop is frozen on a ~0 W reading is a refusal** for
control (`Controller._refusal_from_stale`). This is what stops the ~70 s
cycle. The log keeps SAFETY-56's two-explanation wording.
4. **No SoC in the loop.** SAFETY-25 Option A stands. Nothing reads SoC.
Other exits: the loop wants the other direction (charge is accepted at the
floor), control stopped or started, another writer on the wire (the arbiter
reseed). A gap in the readings keeps the refusal, because the gate's own stop is
what seeds it.
**One inverter only.** At N>1, OPT-09's `pack` limit already closes an empty
battery's discharge end and moves its share. There, a one-unit gate trip must
not refuse the site. `compute()` is passed `refused=None` there, and that path
is release/1.0's law unchanged (the `i_w=None` equivalence sweep in
`test_control.py` still holds).
## Evidence
`test_refusal.py` drives the real Controller and gate (the 09-06 and 09-19
shapes, recovery with no sign change, the probe cadence, every exit).
`tools/replay_refusal.py` replays the 16-day capture, same model as
`replay_recorded_loop.py`, and should be read for deltas:
| plant | law | gate trips | request >1 kW above delivery | longest held | import / export kWh | worst 15-min |
|---|---|---|---|---|---|---|
| fitted | old | 85 | 113.3 h | 16.57 h | 112.0 / 35.2 | 3434 W |
| fitted | new | 130 (133 after review) | 11.4 h | 0.21 h | 112.3 / 35.4 | 3456 W |
| offset 0 W | old | 1010 | 102.2 h | 16.58 h | 112.0 / 35.2 | 3434 W |
| offset 0 W | new | 85 | 6.1 h | 0.33 h | 111.9 / 35.1 | 3434 W |
**Traced in review (#141).** The +22 W worst quarter is the old law's worst
quarter too. There the old law stands at +2060 W on a pack giving 0 W, and the
new law asks 0 W, which the fitted plant answers with its -22 W offset: 22 W
charged from the grid. The fit above the floor reads `batt = cmd - 28 W`, so a
real inverter asked 0 W may do the same. Of the 71 new-law trips with no
old-law trip within 2 min, 59 come 85-97 s after an answered probe. The offset has lifted the modelled pack
just above its floor, the probe is answered, control ramps, the pack empties
again mid-ramp, and the gate trips, which re-seeds the hold. The other 12
follow an other-direction exit the same way. On a site this is one
`STALE_READING` ERROR line per answered probe at most, and only on a battery
that answers a probe and then cuts out again. Catching the re-cut before the
gate needs a faster engage than `SETTLE_CYCLES`, which accuses a healthy slow
battery, so it is left to SAFETY-58/59. After the review changes below the
fitted-plant row reads 133 trips (the delivery exit now also catches answers
that arrive after the probe window); the other figures are unchanged. Rejected alternatives and their figures are in
the phase A plan: bounding the integrator by demand made the worst quarter
+541 W worse or cycled the gate 3368 times, and an SoC envelope needs the
configured floor to be right and needs SoC.
## Review changes (#141)
- **"Gives nothing" is below 100 W, not below 500 W.** At `saturation_w` a
battery derated to 100-499 W (cold, a tapering battery) was held at 0 W for
~93 % of the time: 300 W derated gave 86 Wh in 4 h against 1 200 Wh on the
old law. The capture has 0.11 h in that band, so the replay does not move.
- **Any delivery ends the hold, not only an answered probe.** With
`meter_poll_s: 1` the 12 cycles last 12 s, less than a healthy battery's
reading lag (up to `BATT_OBLIGED_S`, 17 s). That engaged on a healthy step and
held it for 15 min. Now the late reading ends it within seconds (2 s held in
`test_refusal.py` part 5b).
- **N>1 is stated as a limit in DOCS.** The pack limit there covers an empty
battery only when its cut-out level is configured right.
## Not covered
A battery that gives a steady part of the ask (e.g. a 500 W keep-alive) is a
genuine limit (SAFETY-13 case a) and still holds its request.
## What SAFETY-58 and SAFETY-59 must take into account
- **SAFETY-58** (the first read after a rest trips at once): a probe never arms
the gate (200 < 250 W), but the refusal seeding reads the trip. If 58 delays
the first verdict, the seed arrives later. The 09-19 shape then needs the
compute-side engagement (12 cycles) to win, which it does only while the
request stands still. Keep `test_refusal.py` part 4 green.
- **SAFETY-59** (DOCS says a ~70 s pattern proves a failed register): with this
change a refusing battery no longer produces the pattern at one inverter. The
DOCS text should say so, and that the `battery refuses` reason is the
discriminator the add-on now has. At N>1 the old pattern can still occur.
- If either changes `MOVE_W`, `PROBE_W` must stay below it.
`test_refusal.py` part 2 checks the inequality.
goodwe_controller/DOCS.md
+22 -1
View File
@@ -526,7 +526,7 @@ accumulator is doing:
| `Loop integrator lead` | W | that value minus **the command it was computed against**, i.e. the command the previous cycle left standing |
| `Loop reason` | — | the law's own word for this cycle |
`Loop reason` takes one of eight values:
`Loop reason` takes one of ten values:
| value | what happened |
|---|---|
@@ -538,6 +538,27 @@ accumulator is doing:
| `saturated-freeze` | the inverter is not following, so the command's magnitude may fall but not rise |
| `freeze-released` | the freeze had run out of plant-shaped explanations and the command was let move again |
| `not-running` | control is stopped **and** shadow mode is off, so the loop did not run |
| `battery refuses` | the battery gave less than 100 W for 12 cycles (a minute at a 5-second meter) while asked for more than 500 W, so the add-on asks 0 W in that direction (see below) |
| `battery probe` | the add-on asks 200 W to test whether a refusing battery gives power again |
**When the battery refuses.** At its cut-out level the battery gives nothing,
whatever it is asked. The add-on then asks 0 W instead of repeating its last
request, and the `Why` row reads *The battery refuses to give power: 0 W*. Every
15 minutes it asks 200 W for about half a minute. Control resumes as soon as
the battery gives 100 W or more, when the grid needs charging instead, and when
control is stopped or started. 200 W is below the 250 W the stale-reading check
needs, so a test never trips it. A battery that gives some power, even 100 W,
is not refusing and keeps its request.
⚠️ **One inverter only.** With several inverters this hold is off. There, the
battery level limit moves an empty battery's share to the others, but only if
its “Battery cut-out level” matches where the battery really stops. A battery
that refuses at another level, or for another reason (cold, a battery fault),
can still hold its last request until the grid starts to export.
*For installers:* SAFETY-52. The event log reads `BATTERY REFUSES …` and
`BATTERY REFUSES ENDED: …`, and the INFO line ends in `BATTERY_REFUSES`.
`DECISIONS/SAFETY-52-a-battery-that-refuses.md` has the replay figures.
⚠️ **`Loop reason` is not the `Why` row on the status page, and the two will
often disagree.** `Why` is the *arbiter's* answer in words — *Keeping the grid
goodwe_controller/app/control.py
+58 -1
View File
@@ -265,6 +265,34 @@ _SAT_COUNT_CAP = SETTLE_CYCLES + 1 # 13
# `slew_limit_w()` and therefore with the rail.
SLEW_LEAD_STEPS = 1
# ---------------------------------------------------------------------------
# SAFETY-52. A BATTERY THAT REFUSES IS ASKED FOR 0 W, NOT HELD AT ITS LAST ASK.
#
# At the BMS cut-out the pack delivers ~0 W whatever it is asked, so the
# saturation freeze below holds the command where it stood (the recorded
# 09-06 case: +5000 W for 11.87 h) until the grid error changes sign. The
# freeze's own exit cannot help: a reading of ~0 against a large ask is inside
# the interval a limit can produce. So a refused DIRECTION is held at 0 W,
# carried between cycles like `sat_count`, and left again when the loop wants
# the other direction, when a small probe is answered, or when the caller
# resets it (stop, start, another writer). The record is DECISIONS/SAFETY-52.
#
# PROBE_W is below freshness.MOVE_W (250 W) on purpose: a probe that small can
# never arm the stale-reading gate, so it cannot start the gate's ~70 s cycle.
# test_refusal.py pins the inequality. The cadence and length are the owner's
# 2026-09-27 answer (every 15 minutes) and freshness.BATT_OBLIGED_S twice over
# (17 s is how long a healthy reading needs to show a command).
REFUSED = "battery refuses"
PROBING = "battery probe"
PROBE_W = 200.0
PROBE_ANSWER_W = PROBE_W / 2
# "Gives nothing" is below the probe's own answer, not below `saturation_w`: a
# pack derated to 100-499 W is a limit, and at 500 W it kept ~7 % of its power
# (review of #141). The capture's refusals read -4 W.
REFUSED_BELOW_W = PROBE_ANSWER_W
PROBE_EVERY_S = 900.0
PROBE_S = 34.0
@dataclass(frozen=True)
class Tuning:
@@ -346,6 +374,9 @@ class Decision:
i_bound_w: float = 0.0
requested_w: float = 0.0
clamped_w: float = 0.0
# SAFETY-52. The refused direction to carry back in (+1 discharge, -1
# charge, 0 none), or None when the caller does not carry it.
refused: int | None = None
def slew_limit_w(tuning: Tuning) -> float:
@@ -442,6 +473,8 @@ def compute(
tuning: Tuning,
sat_count: int = 0,
i_w: float | None = None,
refused: int | None = None,
probe: bool = False,
) -> Decision:
"""One control cycle. A cycle is one meter update (~5 s on a HomeWizard P1).
@@ -449,6 +482,9 @@ def compute(
`grid_w` net grid power, + = importing
`actual_w` what the inverter reports it is doing, + = discharging
`i_w` the carried integrator, or None to seed it from `prev_w`
`refused` SAFETY-52's carried refused direction, or None to leave it out
(the `i_w=None` equivalence path does)
`probe` the caller's clock says a refused direction may try PROBE_W now
"""
reason = "tracking"
@@ -739,6 +775,27 @@ def compute(
target = min(target, prev_w) if prev_w > 0 else max(target, prev_w)
reason = "saturated-freeze"
# --- SAFETY-52: a refused direction is held at 0 W ----------------------
# Engages only where the freeze has held for SETTLE_CYCLES and the unit
# delivers ~nothing: a reading below REFUSED_BELOW_W against an ask beyond
# `saturation_w` is a refusal, not a slow plant. The drop to 0 skips the slew
# limit because it only removes an ask the battery is not meeting.
if refused is not None:
if (refused == 0 and frozen and sat_count >= SETTLE_CYCLES
and abs(prev_w) > tuning.saturation_w
and abs(actual_w) < REFUSED_BELOW_W):
refused = 1 if prev_w > 0 else -1
if refused and refused * i_w < 0:
refused = 0 # the loop wants the other way
elif refused and refused * actual_w >= PROBE_ANSWER_W:
refused = 0 # it delivers: a probe answered,
# or a slow reading arriving
elif refused:
wants = refused * i_w > 0
i_w = 0.0
target = refused * PROBE_W if probe and wants else 0.0
reason = PROBING if probe and wants else REFUSED
# --- quantise ----------------------------------------------------------
# 10 W. The register is 1 W and the inverter reports at 1 W, but its
# response lands on a coarser ladder (~17.6 W measured at ~900 W, consistent
@@ -749,7 +806,7 @@ def compute(
return Decision(float(target), sat_count, frozen, reason, float(i_w),
i_clamped, clamped, float(i_bound_w), float(want),
float(clamped_w))
float(clamped_w), refused)
def maintenance_charge_floor(
goodwe_controller/app/main.py
+94
View File
@@ -39,6 +39,7 @@ from .arbiter import (
P_LOOP, P_MAINT_SHAPE, P_MAINTENANCE, P_PEAK, P_SAFETY,
Claim, Resolution, actuator_for, resolve,
)
from . import control as _ctl
from .control import Tuning, compute, maintenance_charge_floor, peak_at_risk
from .delivery import (
DELIVERY_OPTION_DEFAULTS, DeliveryConfig, UnitDelivery,
@@ -241,6 +242,10 @@ def why_key(winner: str, claims) -> str:
disagreement"). A failsafe with no claim of its own is the arbiter's "no
claim was made", or main.py's refusal to split the power ("unsplit").
"""
if winner == "loop":
# SAFETY-52: a loop holding a refused direction says so.
return next((c.reason for c in claims if c.source == "loop"
and c.reason in (_ctl.REFUSED, _ctl.PROBING)), winner)
if winner in ("safety", "failsafe"):
return next((c.reason for c in claims
if c.source == winner and c.kind == "set"),
@@ -925,6 +930,12 @@ class Controller:
self.target = 0.0
self.sat_count = 0
self.i_w = 0.0 # the loop's integrator, carried between cycles
# SAFETY-52. The direction the battery refuses (+1 / -1 / 0), carried
# like `i_w`, and when the refusal (or its last probe) started.
self.batt_refused = 0
self.batt_refused_at = None
self._now_mono = None
self._refusal_auto = False
# TEL-10. What the last cycle's loop did, kept for publication. Seeded
# to the SAME values the "stopped or blind" branch of `cycle()` writes,
# because before the first cycle the loop has not run - which is the
@@ -1058,6 +1069,69 @@ class Controller:
self.loop_pass_mono = None
# -- helpers -------------------------------------------------------------
# --- SAFETY-52: a battery that refuses is asked for 0 W ------------------
def _refusal_probe(self) -> bool:
"""True while a refused direction may try `PROBE_W` (every 15 min)."""
if (not self.batt_refused or self.batt_refused_at is None
or self._now_mono is None):
return False
held = self._now_mono - self.batt_refused_at
if held >= _ctl.PROBE_EVERY_S + _ctl.PROBE_S:
self.batt_refused_at = self._now_mono
return False
return held >= _ctl.PROBE_EVERY_S
def _refusal_start(self, sign: int, evidence: str) -> None:
self.batt_refused = sign
self.batt_refused_at = self._now_mono
self.log_event(f"BATTERY REFUSES {evidence}",
hint=messages.hint("battery_refuses"))
_LOG.info("The battery refuses to give power, so the add-on asks 0 W "
"and tries %.0f W every %.0f minutes: BATTERY_REFUSES.",
_ctl.PROBE_W, _ctl.PROBE_EVERY_S / 60)
_LOG.debug("Evidence: BATTERY_REFUSES: %s.", evidence)
def _refusal_end(self, why: str) -> None:
if not self.batt_refused:
return
self.batt_refused, self.batt_refused_at = 0, None
self.log_event(f"BATTERY REFUSES ENDED: {why}",
hint=messages.hint("battery_gives"))
_LOG.info("The battery refusal ends, control asks for power again: %s.",
why)
def _refusal_update(self, decision, asked_w, batt_w, probe) -> None:
"""Follow `compute()`'s carried refused direction, and say so."""
now = decision.refused or 0
if now and not self.batt_refused:
self._refusal_start(now, f"asked {asked_w:+.0f} W, the battery "
f"gave {batt_w:+.0f} W for "
f"{decision.sat_count} cycles")
elif self.batt_refused and not now:
gave = self.batt_refused * batt_w >= _ctl.PROBE_ANSWER_W
self._refusal_end("the probe was answered" if gave and probe
else "the battery gives power" if gave
else "the grid now asks the other direction")
def _refusal_from_stale(self, inv, now_mono) -> None:
"""A stale-reading stop while the loop is frozen on ~0 W is a refusal.
Only at one inverter: with several, one empty battery must not stop the
site (OPT-09's pack limit moves its share to the others).
"""
t = self.tuning
standing = inv.standing_w()
if (len(self.inverters) == 1 and (self.auto or self.shadow)
and not self.batt_refused
and self.sat_count >= t.saturation_cycles
and inv.batt_fresh.held is not None
and abs(inv.batt_fresh.held) < _ctl.REFUSED_BELOW_W
and abs(standing) > t.saturation_w):
self._now_mono = now_mono
self._refusal_start(1 if standing > 0 else -1,
f"stale reading {inv.batt_fresh.held:+.0f} W "
f"while asked {standing:+.0f} W")
def log_event(self, msg: str, hint: str | None = None) -> None:
"""`msg` goes on the action log, the add-on's own technical record.
`hint` is `messages.hint()`'s plain words for the same event, and is
@@ -1840,6 +1914,7 @@ class Controller:
"""
o = self.o
now_mono = freshness.monotonic() if now_mono is None else float(now_mono)
self._now_mono = now_mono
if self.p1_enabled:
# ⚠️ P1 is the only authoritative measurement of what the utility
# sees (§5.1). When it is stale this is None, which falls into the
@@ -2003,6 +2078,7 @@ class Controller:
if inv.batt_fresh.stale and not was_stale:
self.log_event(f"STALE {inv.name}: " + inv.batt_fresh.reason,
hint=messages.hint("battery_stuck", unit=inv.name))
self._refusal_from_stale(inv, now_mono)
# ⚠️ SAFETY-57 REWROTE THE LAST SENTENCE OF THIS LINE AND NOTHING
# ELSE IN IT. The escape clause used to read "It clears itself the
# moment the register reports a different value", full stop - and
@@ -3340,6 +3416,12 @@ class Controller:
# loop should be. It is safe because the loop's `prev_w` is
# `self.target`, which the end of this method re-seeds to what is
# actually standing on the wire.
# SAFETY-52: a refusal learned while control ran does not survive a
# stop or a start (shadow mode keeps the loop running when stopped).
if bool(self.auto) != self._refusal_auto:
self._refusal_auto = bool(self.auto)
self._refusal_end("control started" if self.auto
else "control stopped")
if (self.auto or self.shadow) and self.inputs_ok:
# ⚠️ prev is the ARBITER's last output, not the loop's own last
# wish. If something outranked the loop, that is what the hardware
@@ -3357,6 +3439,13 @@ class Controller:
# stale period as power.
if self.loop_w is None or self.target != self.loop_w:
self.i_w = self.target
# SAFETY-52: another writer overrode the loop, so a refusal learned
# against the loop's own ask no longer describes the wire. A gap
# (`loop_w` None: blind or stopped) keeps it: the stale-reading stop
# is what seeds it, and dropping it there restarts the ~70 s cycle.
if self.loop_w is not None and self.target != self.loop_w:
self._refusal_end("another writer")
probe = self._refusal_probe()
# ⚠️ `actual_w` IS THE SITE SUM AND THAT IS DEV-02's OTHER HALF.
# The anti-windup test inside compute() is
# `abs(prev_w - actual_w) > saturation_w`, and `prev_w` is the SITE
@@ -3385,6 +3474,10 @@ class Controller:
tuning=self.tuning,
sat_count=self.sat_count,
i_w=self.i_w,
# SAFETY-52 at one inverter only: with several, OPT-09's pack
# limit already closes an empty battery's discharge end.
refused=self.batt_refused if len(self.inverters) == 1 else None,
probe=probe,
)
if decision.frozen and self.sat_count < self.tuning.saturation_cycles:
self.log_event(f"saturation freeze ({self.target:.0f} W vs "
@@ -3393,6 +3486,7 @@ class Controller:
self.sat_count = decision.sat_count
self.i_w = decision.i_w
self.loop_w = decision.target_w
self._refusal_update(decision, loop_prev, site_batt, probe)
# --- TEL-10 ------------------------------------------------------
# ⚠️ THE INTEGRATOR HAS NEVER LEFT THIS PROCESS, AND THAT IS WHAT
# THE CARD IS. `self.i_w` is carried between cycles and published
goodwe_controller/app/messages.py
+5
View File
@@ -802,6 +802,9 @@ WHY_WINNER = {
# failsafe winner. test_messages.py part 7 finds every one in the code.
"sign disagreement": "Safety stop, power went the wrong way",
"unsplit": "The power could not be split, all at 0 W",
# SAFETY-52: the loop's two words for a battery that gives nothing.
"battery refuses": "The battery refuses to give power",
"battery probe": "Testing if the battery gives power again",
}
WHY_LIMIT = {
"peak(quarter peak)": "the peak limit",
@@ -852,6 +855,8 @@ HINTS = {
"stop_restored": "{unit}: an earlier safety stop is back, it stays at 0 W",
"battery_stuck": "{unit}: battery power reading stopped changing",
"battery_back": "{unit}: battery power reading is used again",
"battery_refuses": "The battery refuses to give power, the add-on asks 0 W",
"battery_gives": "The battery refusal ended, control asks for power again",
"bad_value": "The add-on made an invalid power value, inverters set to 0 W",
"write_blocked": "{unit}: a power request was blocked, it is set to 0 W",
"all_blocked": "A power request was blocked, inverters set to 0 W",
goodwe_controller/test_clamp.py
+12 -3
View File
@@ -497,8 +497,13 @@ check(f"AC 1: NOT ONE of the {len(emitted)} values that reached the device "
# 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.
# SAFETY-52 re-baseline: an inverter reporting 0 W for SETTLE_CYCLES is a
# battery that refuses, so the command then drops to 0 W. The freeze still
# engaged first, and the event log says so.
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)
f"rather than the rail",
any("saturation freeze" in e for e in ctl_r.events)
and max(emitted) < LIVE_MAX_W)
notes_r = clamp_notes(hass_r)
print(f" the loop clamped on {ctl_r.clamp_wd.engagements} of {60} cycles, "
@@ -580,8 +585,12 @@ check("UX-07 item 1: during a bound episode an earlier over-rating count is "
# ⚠️ 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])
# SAFETY-52: the loop itself now asks 0 W of an inverter that gives nothing,
# with its own reason. That is the loop's decision, not a stop.
check(f"it is NOT a trip - the command stands at {ctl_r.target:+.0f} W because "
f"the loop says '{ctl_r.loop_reason}', not because of a stop",
ctl_r.loop_reason == "battery refuses"
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)
goodwe_controller/test_delivery.py
+8 -3
View File
@@ -1147,8 +1147,10 @@ async def _one():
"soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
"meter_source": "off", "max_w": 2000, "gain": 0.6,
# ⚠️ THE SAME 1 s AS THE TWO-INVERTER FIXTURE, AND THAT IS WHAT
# MAKES THIS SECTION NON-VACUOUS. The run below is 2 s of wall clock
# against a 1 s window, so a hold is not what stops this firing -
# MAKES THIS SECTION NON-VACUOUS. The run below is 1.1 s of wall
# clock against a 1 s window (SAFETY-52 shortened it from 2 s, and
# "the window was never even opened" below is what still makes it
# non-vacuous), so a hold is not what stops this firing -
# the reduction is. With a shipped-length hold "nothing was
# reported" would be true of a fixture that simply ran out of time.
"bms_soc_floor": _FLOOR_PCT, "delivery_hold_s": 1,
@@ -1158,7 +1160,10 @@ async def _one():
c.inverters[0].dev_min, c.inverters[0].dev_max = -2000.0, 2000.0
c.auto = True
hass.ctrl = c
await _drive(c, n=20, dt=0.1)
# SAFETY-52: from SETTLE_CYCLES of delivering nothing the loop asks 0 W
# (the battery refuses), so the run stops just before that. 11 cycles is
# still 1.1 s against the 1 s hold.
await _drive(c, n=11, dt=0.1)
return c, hass, store
goodwe_controller/test_messages.py
+6
View File
@@ -1399,6 +1399,7 @@ import logging # noqa: E402
from app import hass as _hass, p1 as _p1, validate as _validate, web as _web # noqa: E402
from app.arbiter import Claim # noqa: E402
from app.main import Controller, why_key # noqa: E402
from app import control as _control # noqa: E402
from app import main as _main # noqa: E402
@@ -1585,6 +1586,11 @@ for _f in sorted(APP.glob("*.py")):
_keys = {why_key(s, [Claim.set(s, 0, 0.0, r)]) for s, r in _SOURCES}
_keys |= {why_key("failsafe", []),
why_key("failsafe", [Claim.set("loop", 0, 1.0, "tracking")])}
# SAFETY-52: the loop's claim reason is computed, so the scan cannot see it.
_keys |= {why_key("loop", [Claim.set("loop", 0, 0.0, _r)])
for _r in (_control.REFUSED, _control.PROBING)}
check("...including the loop's two battery-refuses reasons, which get words of "
"their own", {_control.REFUSED, _control.PROBING} <= _keys)
check(f"the scan found the claims ({len(_SOURCES)}), including the latched "
f"per-inverter stop the 2d test saw as 'Other'",
("failsafe", "sign disagreement") in _SOURCES and len(_SOURCES) >= 8)
goodwe_controller/test_price.py
+7 -3
View File
@@ -4412,12 +4412,16 @@ check(f"AC, AND THE WHOLE OF DEFECT 4: 'trading is off' and 'trading is on "
f"`off (arbitrage_enabled)` for both - naming an option that is `true` "
f"in the options blob",
_T11_OFF[0]["Arbitrage"] != _T11_ON_HELD[0]["Arbitrage"])
# ⚠️ THE "Why" ROW OF THE SAME RENDER, NOT A SECOND READ OF `why_words`. Since
# UX-10 the waiting sentence counts its seconds when it is read, so reading the
# property again here, after the other renders above, could cross a second
# ("1 second" in the row, "2 seconds" now) - CI run 953 on PR #141 did.
check(f"...and the armed-but-held row carries the CONTROL LOOP'S OWN REASON "
f"({_T11_ON_HELD[1].why_words!r}), so the operator is told which input is "
f"({_T11_ON_HELD[0]['Why']!r}), so the operator is told which input is "
f"missing rather than being sent to look at a switch that is already on. "
f"The reason comes from main.py, not from this file",
_T11_ON_HELD[1].why_words
and _T11_ON_HELD[1].why_words in _T11_ON_HELD[0]["Arbitrage"])
_T11_ON_HELD[0]["Why"].startswith("A reading is missing")
and _T11_ON_HELD[0]["Why"] in _T11_ON_HELD[0]["Arbitrage"])
check(f"...and a cycle that DID reach the step renders the strategy's own "
f"sentence - {_T11_ON_OK[1].arb.explain()[:50]!r} - rather than a "
f"re-derivation of it, which is the guarantee OPT-04 put on this row and "
goodwe_controller/test_refusal.py
+475
View File
@@ -0,0 +1,475 @@
"""Runnable check for SAFETY-52. `python3 test_refusal.py`
A battery that refuses to give power (the BMS cut-out) used to leave the
command standing at its last ask for hours: +5000 W for 11.87 h on 2026-09-06,
+1940 W for 3 h 55 m on the bus on 2026-09-19. Now a refused direction is held
at 0 W, a 200 W probe asks every 15 minutes, and the refusal ends on charge, on
an answered probe, on stop/start and when another writer takes the wire.
DECISIONS/SAFETY-52 has the record and the replay figures.
Mutating the modules this file grades? Purge `__pycache__` first, or run
`tools/mutation_run.sh`, which does - see MUTATION.md.
Parts 1-5 drive the REAL `Controller` (read_inputs + cycle, the real
SAFETY-37/57 gate) against a scripted battery whose reading stands at exactly
-4 W for any request >= -4 W at the cut-out, and follows otherwise. Time is the
`now_mono` handed to `read_inputs()`, so 12 hours run in seconds. `OLD` runs the
same fixture with `refused=None` forced into `compute()` - release/1.0's law -
so every "new" figure is shown beside the one it replaces. Part 4 also runs
`compute()` alone, as the bus saw 09-19 before the gate existed.
Mutations, each on a copy of this tree purged of `__pycache__`, graded with
`python3 -B` over test_refusal, test_control, test_clamp, test_telemetry and
test_messages (MUTATION.md). Reds are failing checks, on this tree:
# mutation reds
-- ------------------------------------------------------- ---------------
R1 control.py: the engage `if` -> `if False and ...` 8 here + 1 clamp
(release/1.0's behaviour)
R2 control.py: the exit on the other direction removed 1
R3 control.py: the exit on an answered probe removed 2
R4 control.py: `PROBE_W = 300.0`, above the gate's 250 W 4
R5 control.py: the hold keeps the old target, not 0 W 4
R6 main.py: `_refusal_from_stale()` returns at once 1
R7 main.py: another writer no longer ends the refusal 1
R8 main.py: stop/start no longer ends the refusal 1
R9 main.py: the probe window never opens 4
R10 main.py: seed from a stale stop below `saturation_w` again, 1
not below REFUSED_BELOW_W (review of #141). The same mutation
in control.py's engage clause SURVIVES, and that is expected:
the delivery exit (R11) ends such a refusal in the same cycle.
R11 control.py: only an answered PROBE ends the refusal 2
(`and probe` back in the delivery exit; review of #141)
"""
import asyncio
import sys
from datetime import datetime, timedelta, timezone
from app import control, freshness
from app.control import PROBE_EVERY_S, PROBE_S, PROBE_W, REFUSED, PROBING
from app.main import Controller, why_key
from app.arbiter import Claim
from app import messages
fails = []
def check(name, cond):
print(f" {'ok ' if cond else 'FAIL'} {name}")
if not cond:
fails.append(name)
LIVE = {"max_w": 5000.0, "gain": 0.6, "slew_w": 400.0, "target_grid_w": -10.0,
"deadband_w": 15, "step_w": 10}
TICK = 5.0 # the meter's measured median tick (replay_recorded_loop)
CUT_W = -4.0 # what the register reads at the cut-out, 2026-09-06
H = 3600.0
BASE = datetime(2026, 9, 6, 19, 30, tzinfo=timezone.utc)
class _Store:
def __init__(self):
self.data = {}
def set(self, *a):
pass
def get_time(self, *a):
return None
class _Pub:
enabled = False
def publish(self, values):
pass
def close(self):
pass
class _Hass:
"""House load, and a battery that follows unless `cut(t)` says it refuses.
At the cut-out the reading is exactly CUT_W for any request >= CUT_W (a flat
register, which is what arms or does not arm the stale-reading gate). A
charge request is followed: the pack accepts charge at the floor.
`cap(t)` limits discharge while not cut (the 09-06 pack gave 4570 W at 10 %).
"""
def __init__(self, load, cut, cap=lambda t: 1e9):
self.load, self.cut, self.cap = load, cut, cap
self.t = 0.0
self.cmd = 0.0
self.writes = []
def batt(self):
if self.cut(self.t) and self.cmd >= CUT_W:
return CUT_W
return float(round(min(self.cmd, self.cap(self.t))))
async def limits(self, entity):
return (-LIVE["max_w"], LIVE["max_w"], 1.0)
async def number(self, entity, invert=False):
b = self.batt()
return {"sensor.grid": self.load(self.t) - b, "sensor.soc": 9.0,
"sensor.batt": b}.get(entity)
async def set_number(self, entity, value):
self.writes.append((self.t, float(value)))
self.cmd = float(value)
return True
async def call(self, *a):
return True
def run(load, cut, dur, cap=lambda t: 1e9, old=False, events=None):
"""Real cycles every TICK seconds of `now_mono`. `events[t] = fn(c)` pokes
the controller (stop, start, another writer) at that instant.
Returns [(t, grid, batt, setpoint, i_w, loop_reason, batt_refused)], the
number of gate trips, and the controller."""
async def go():
hass = _Hass(load, cut, cap)
c = Controller(dict(
{"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
"soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
"meter_source": "off", "heartbeat_s": 10, "stale_input_s": 15},
**LIVE), hass, _Store(), _Pub(), stop=asyncio.Event())
c.auto = True
c.dev_min, c.dev_max = -LIVE["max_w"], LIVE["max_w"]
import app.main as m
real, real_dt = m.compute, m.datetime
clock = [0.0]
class _DT(real_dt):
"""`cycle()`'s wall clock follows `now_mono`, so the 15 s input hold
and the gate's release run in simulated seconds too."""
@classmethod
def now(cls, tz=None):
return BASE + timedelta(seconds=clock[0])
m.datetime = _DT
if old:
m.compute = lambda *a, **k: real(*a, **dict(k, refused=None))
rows, trips, was = [], 0, False
try:
t = 0.0
while t < dur:
hass.t = clock[0] = t
for et, fn in list((events or {}).items()):
if et <= t:
fn(c)
del events[et]
await c.read_inputs(now_mono=t)
stale = c.inverters[0].batt_fresh.stale
trips += stale and not was
was = stale
await c.cycle()
rows.append((t, hass.load(t) - hass.batt(), hass.batt(),
hass.cmd, c.i_w, c.loop_reason, c.batt_refused))
t += TICK
finally:
m.compute, m.datetime = real, real_dt
return rows, trips, c
return asyncio.run(go())
def at(rows, t):
return min(rows, key=lambda r: abs(r[0] - t))
def longest_hold_h(rows, floor_w=250.0, until=None):
"""Longest run of one identical request at or above `floor_w`, in hours."""
best = cur = 0.0
last = None
for r in rows:
if until is not None and r[0] >= until:
break
cur = cur + TICK if (r[3] == last and abs(r[3]) >= floor_w) else 0.0
last = r[3]
best = max(best, cur)
return best / H
# ==============================================================================
print("1. the 09-06 shape: a load step takes the request to 5000 W, the battery "
"cuts out, the load ends")
# ==============================================================================
def load906(t):
return 1020 if t < 60 else 4800 if t < 162 else 940 if t < 11.9 * H else -600
def cut906(t):
return t >= 132
def cap906(t):
return 4570.0
old906, otrips906, _ = run(load906, cut906, 12.2 * H, cap906, old=True)
new906, ntrips906, c906 = run(load906, cut906, 12.2 * H, cap906)
for name, rows, tr in (("old", old906, otrips906), ("new", new906, ntrips906)):
print(f" {name}: request at 2 min {at(rows, 120)[3]:+.0f} W, at 6 h "
f"{at(rows, 6 * H)[3]:+.0f} W (i_w {at(rows, 6 * H)[4]:+.0f}), longest "
f"hold {longest_hold_h(rows):.2f} h, gate trips {tr}")
check("PRECONDITION: the old law holds +5000 W for the whole night on this "
"fixture, as it did on 2026-09-06 - the fixture reproduces the card",
at(old906, 6 * H)[3] == 5000.0 and longest_hold_h(old906) > 11.5)
check("the request reaches the rail before the cut, so the new law starts from "
"the same place", at(new906, 120)[3] == 5000.0)
t0 = next((r[0] for r in new906 if r[5] == REFUSED), None)
check(f"after the cut the battery-refuses hold takes the request to 0 W within "
f"about a minute (at t={t0})", t0 is not None and 132 < t0 <= 132 + 90
and at(new906, t0)[3] == 0.0)
check("...and it stays refused overnight, with the integrator at 0, and no "
"request of 250 W or more stands for longer than one probe interval",
at(new906, 6 * H)[6] == 1 and at(new906, 6 * H)[4] == 0.0
and longest_hold_h(new906, until=11.9 * H) < PROBE_EVERY_S / H)
check(f"...and the stale-reading gate does not cycle: {ntrips906} trips in "
f"12 hours (the old law's latch also kept it quiet: {otrips906})",
ntrips906 == 0)
morning = [r for r in new906 if r[0] >= 11.9 * H + 60]
check("when solar makes the grid export, the refusal ends and the battery "
"charges - the exit on the other direction",
morning and morning[-1][6] == 0 and morning[-1][3] < -400)
# ==============================================================================
print("2. the probe: 200 W every 15 minutes, below the gate's 250 W")
# ==============================================================================
probes = []
for r in new906:
if r[3] == PROBE_W and (not probes or r[0] - probes[-1][1] > TICK):
probes.append([r[0], r[0]])
elif r[3] == PROBE_W:
probes[-1][1] = r[0]
starts = [p[0] for p in probes]
gaps = [b - a for a, b in zip(starts, starts[1:])]
print(f" {len(probes)} probes, first at {starts[0] / 60 if starts else 0:.1f} "
f"min, gaps {sorted(set(round(g) for g in gaps))} s, lengths "
f"{sorted(set(round(p[1] - p[0] + TICK) for p in probes))} s")
check("the probe asks exactly PROBE_W, which is below the stale-reading gate's "
"threshold, so a probe can never arm it",
PROBE_W < freshness.MOVE_W and all(r[3] in (0.0, PROBE_W)
for r in new906 if r[6] == 1))
check("there is a probe every 15 minutes through the night (at least 40 in "
"11.8 hours), each about 34 s long",
len(probes) >= 40 and all(PROBE_EVERY_S <= g <= PROBE_EVERY_S + PROBE_S
+ 2 * TICK for g in gaps)
and all(PROBE_S - TICK <= p[1] - p[0] + TICK <= PROBE_S + TICK
for p in probes))
check("...and a probe reads as a probe on the loop's own reason row",
any(r[5] == PROBING for r in new906))
# ==============================================================================
print("3. recovery with no sign change: the battery gives power again at 2 h")
# ==============================================================================
def cut_rec(t):
return 132 <= t < 2 * H
oldr, _, _ = run(load906, cut_rec, 2.6 * H, cap906, old=True)
newr, _, crec = run(load906, cut_rec, 2.6 * H, cap906)
after_o = [r for r in oldr if 2 * H <= r[0] < 2 * H + 600]
after_n = [r for r in newr if 2 * H <= r[0] < 2.6 * H]
exp_o = sum(max(0.0, -r[1]) * TICK for r in after_o) / 3600
exp_n = sum(max(0.0, -r[1]) * TICK for r in after_n) / 3600
print(f" old: worst grid {min(r[1] for r in after_o):+.0f} W, export "
f"{exp_o:.0f} Wh; new: worst grid {min(r[1] for r in after_n):+.0f} W, "
f"export {exp_n:.0f} Wh")
check("PRECONDITION: the old law answers a recovering battery with its "
"standing +5000 W and exports kilowatts into a 940 W house",
min(r[1] for r in after_o) < -3000)
check("the new law never exports more than the probe while it waits, and the "
"answered probe ends the refusal so control resumes",
min(r[1] for r in after_n) > -PROBE_W - 50
and after_n[-1][6] == 0 and abs(after_n[-1][1] - (-10)) < 100)
check("...and the event log says the probe was answered",
any("BATTERY REFUSES ENDED: the probe was answered" in e
for e in crec.events))
# ==============================================================================
print("4. the 09-19 shape: a stale-reading stop on a frozen ~0 W reading "
"is a refusal")
# ==============================================================================
# The request steps +1540 -> +1940 and more as the battery cuts out mid-ramp:
# the request moves after the reading went flat, so the gate trips before the
# hold's 12 cycles - the ~70 s cycle SAFETY-59 measured. The trip seeds the hold.
def load919(t):
return 1530 if t < 60 else 2400 if t < 1800 else 900 if t < 4 * H else -600
def cut919(t):
return t >= 65
def run919(old):
return run(load919, cut919, 4.2 * H, old=old)
old919, otr919, _ = run919(True)
new919, ntr919, c919 = run919(False)
print(f" old: gate trips {otr919}, longest hold {longest_hold_h(old919):.2f} h;"
f" new: gate trips {ntr919}, longest hold {longest_hold_h(new919):.2f} h")
check("PRECONDITION: on the old law the gate trips over and over all night "
"(the SAFETY-59 cycle) or the request stands for hours",
otr919 > 20 or longest_hold_h(old919) > 1.0)
check("the new law stops after at most two trips and holds 0 W, and the event "
"log says the battery refuses",
ntr919 <= 2 and at(new919, 2 * H)[3] in (0.0, PROBE_W)
and any("BATTERY REFUSES" in e for e in c919.events))
# The same shape through `compute()` alone, as the bus saw it on 2026-09-19
# before the gate existed: +1540 -> +1940, then a battery giving nothing.
T = control.Tuning(**LIVE)
res = {}
for name, carry in (("old", None), ("new", 0)):
# sat_count 2: the +1540 and +1940 cycles were already short (the plan)
prev, i_w, sat, ref, cmds = 1940.0, 2340.0, 2, carry, []
for _ in range(2000): # ~2.8 h at 5 s
d = control.compute(prev, 1950.0, CUT_W, T, sat, i_w, refused=ref)
prev, i_w, sat, ref = d.target_w, d.i_w, d.sat_count, d.refused
cmds.append(prev)
res[name] = cmds
print(f" compute() alone: old stands at {res['old'][-1]:+.0f} W "
f"({sum(v == 1940.0 for v in res['old'])} of 2000 cycles at +1940 W), "
f"new at {res['new'][-1]:+.0f} W from cycle "
f"{res['new'].index(0.0) if 0.0 in res['new'] else None}")
check("PRECONDITION: without the gate the old law parks at +1940 W, the 7 065 "
"identical writes the bus recorded", set(res["old"]) == {1940.0})
check("...and the new law holds 0 W within SETTLE_CYCLES cycles",
res["new"].index(0.0) < control.SETTLE_CYCLES
and set(res["new"][control.SETTLE_CYCLES:]) == {0.0})
# ==============================================================================
print("5. the other exits: stop, start, another writer")
# ==============================================================================
def stop(c):
c.auto = False
def start(c):
c.auto = True
rows, _, c = run(load906, cut906, 1200, cap906, events={600.0: stop,
700.0: start})
check("stopping control ends the refusal, and starting again begins from "
"nothing refused", at(rows, 590)[6] == 1 and at(rows, 650)[6] == 0
and at(rows, 705)[6] == 0)
def other_writer(c):
c.loop_w = 1234.0 # the arbiter gave the wire something the loop did not ask
rows, _, c = run(load906, cut906, 900, cap906, events={600.0: other_writer})
check("another writer on the wire ends the refusal (the arbiter reseed)",
at(rows, 590)[6] == 1 and at(rows, 605)[6] == 0)
# ==============================================================================
print("5b. review of #141: a pack that gives SOME power, and a reading that is "
"only late, are not a refusal")
# ==============================================================================
# A pack derated to 100-499 W (cold, BMS taper) gives what it can against a
# larger ask. Engaging below `saturation_w` (500 W) held it at 0 W for ~93 % of
# the time; "gives nothing" is below the probe's own answer.
# `compute()` alone: through the Controller the stale-reading gate trips on the
# flat 300 W first, and that fixture then passes whatever this threshold is.
def derated(carry, cap_w=300.0, n=1800):
prev, i_w, sat, ref, gave, held = 0.0, 0.0, 0, carry, 0.0, 0
b = 0.0
for _ in range(n): # 2.5 h at 5 s
d = control.compute(prev, 2000.0 - b, b, T, sat, i_w, refused=ref)
prev, i_w, sat, ref = d.target_w, d.i_w, d.sat_count, d.refused
held += bool(ref)
b = max(0.0, min(prev, cap_w)) # one tick late
gave += b
return gave / n, held
T = control.Tuning(**LIVE)
gave_old, _ = derated(None)
gave_new, held_new = derated(0)
print(f" pack derated to 300 W: old law {gave_old:.0f} W average, new "
f"{gave_new:.0f} W, held as refusing on {held_new} cycles")
check("a pack that gives a steady 300 W against a 2000 W ask is never held at "
"0 W as refusing, and keeps giving what the old law got from it",
held_new == 0 and gave_new == gave_old and gave_old > 290.0)
def lagging(tick, dead_s=6.0, publish_s=10.0, phase_s=5.0, dur=600.0):
"""`compute()` on a HEALTHY pack whose reading lags: it follows the command
of `dead_s` ago, published every `publish_s`. A 1 s meter (`meter_poll_s`
1) makes SETTLE_CYCLES only 12 s, shorter than that lag."""
prev, i_w, sat, ref, hist, reg, pub = 0.0, 0.0, 0, 0, [], 0.0, -phase_s
held = engaged = 0
t = 0.0
while t < dur:
then = [c for tc, c in hist if tc <= t - dead_s]
b = then[-1] if then else 0.0
if t - pub >= publish_s:
reg, pub = b, t
d = control.compute(prev, (3000.0 if t >= 10 else 0.0) - b, reg, T,
sat, i_w, refused=ref)
engaged += bool(d.refused) and not ref
prev, i_w, sat, ref = d.target_w, d.i_w, d.sat_count, d.refused
hist.append((t, prev))
held += tick * bool(ref)
t += tick
return engaged, held
lag1 = lagging(1.0)
print(f" 1 s meter, healthy pack with a late reading: engaged {lag1[0]} "
f"time(s), held {lag1[1]:.0f} s")
check("a healthy pack whose reading is merely late is let go as soon as the "
"reading shows it delivering - not held until the next probe",
lag1[1] <= 10.0)
async def _seed(held_w):
"""`_refusal_from_stale()` on a unit standing at +2000 W, frozen."""
c = Controller(dict(
{"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
"soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
"meter_source": "off"}, **LIVE),
_Hass(lambda t: 0.0, lambda t: False), _Store(), _Pub(),
stop=asyncio.Event())
c.auto, c.sat_count = True, c.tuning.saturation_cycles
inv = c.inverters[0]
inv.last_write, inv.batt_fresh.held = 2000.0, held_w
c._refusal_from_stale(inv, 100.0)
return c.batt_refused
check("a stale-reading stop on a register standing at 300 W is not a refusal "
"(it gives something), and one at -4 W is",
asyncio.run(_seed(300.0)) == 0 and asyncio.run(_seed(CUT_W)) == 1)
d = control.compute(0.0, 1500.0, 1400.0, T, 0, 0.0, refused=1, probe=False)
check("...i.e. a held direction that reads 100 W or more ends the refusal "
"without waiting for a probe", d.refused == 0)
# ==============================================================================
print("6. words")
# ==============================================================================
w = messages.why(why_key("loop", [Claim.set("loop", 0, 0.0, REFUSED)]), 0.0)
wp = messages.why(why_key("loop", [Claim.set("loop", 0, PROBE_W, PROBING)]),
PROBE_W)
print(f" {w!r} / {wp!r}")
check("the page's Why line says the battery refuses, not 'Keeping the grid "
"near zero: 0 W'", w == "The battery refuses to give power: 0 W")
check("...and a probe says what it is", wp.startswith("Testing if the battery"))
check("...and an ordinary loop cycle still reads as the loop",
why_key("loop", [Claim.set("loop", 0, 10.0, "tracking")]) == "loop")
print()
if fails:
print(f"{len(fails)} FAILED: " + ", ".join(fails))
sys.exit(1)
print("all checks passed")
goodwe_controller/tools/replay_refusal.py
+97
View File
@@ -0,0 +1,97 @@
#!/usr/bin/env python3
"""SAFETY-52's evidence: the 16-day capture replayed with and without the
battery-refuses hold. `python3 tools/replay_refusal.py`
Same disturbance, fitted plant and pack floor as `replay_recorded_loop.py`,
plus main.py's reseed, 15 s input hold and 0 W rules and the REAL SAFETY-37/57
gate on a register published every 10 s. `old` is `compute()` with
`refused=None`, which is release/1.0's law unchanged; `new` carries the refusal,
runs the probe on main.py's clock and seeds a refusal from a gate trip.
Modelled watt-hours: read the deltas, not the absolutes (see that tool).
"""
import math
import pathlib
import sys
ADDON = pathlib.Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ADDON))
sys.path.insert(0, str(ADDON / "tools"))
import replay_recorded_loop as R # noqa: E402
from app import freshness, control as NEW # noqa: E402
def go(which, rows, fit, publish_s=10.0, stale_input_s=15.0):
law = NEW
tun = law.Tuning(**R.LIVE)
t0, g0, b0, sp0, soc0 = rows[0]
pl = R.Plant(*fit[:4], soc0); pl.seed(t0, b0, sp0)
st = dict(target=sp0, i_w=sp0, sat=0, loop_w=sp0, ref=0, anchor=None)
fr = freshness.Freshness(freshness.BOUNDS["batt"]); reg, reg_t = None, -1e18
bad, trips, was = None, 0, False
M = dict(und=0.0, imp=0.0, exp=0.0, q={}, run=0.0, mx=0.0, last=None, refh=0.0, probes=0)
for i in range(1, len(rows)):
t, gr, ba, sp, soc = rows[i]
dt = t - rows[i - 1][0]
if dt > R.TICK_CAP_S:
pl = R.Plant(*fit[:4], soc); pl.seed(t, ba, sp)
st.update(target=sp, i_w=sp, sat=0, loop_w=sp, ref=0, anchor=None)
fr = freshness.Freshness(freshness.BOUNDS["batt"]); reg_t = -1e18; bad = None
continue
b = pl.step(t, dt, st["target"]); grid = gr + ba - b
if t - reg_t >= publish_s:
reg, reg_t = float(round(b)), t
fr.submit(reg, st["target"], t); batt = fr.reading()
if fr.stale and not was:
trips += 1
if (which == "new" and not st["ref"] and st["sat"] >= tun.saturation_cycles
and abs(reg) < NEW.REFUSED_BELOW_W and abs(st["target"]) > tun.saturation_w):
st["ref"], st["anchor"] = (1 if st["target"] > 0 else -1), t
was = fr.stale
if batt is None:
bad = t if bad is None else bad
if t - bad >= stale_input_s:
st.update(target=0.0, i_w=0.0, loop_w=None)
else:
bad = None
if st["loop_w"] is None or st["target"] != st["loop_w"]:
st["i_w"] = st["target"]
if which == "old":
d = law.compute(st["target"], grid, batt, tun, st["sat"], st["i_w"],
refused=None)
else:
probe = False
if st["ref"] and st["anchor"] is not None:
h = t - st["anchor"]
if h >= NEW.PROBE_EVERY_S + NEW.PROBE_S: st["anchor"] = t
else: probe = h >= NEW.PROBE_EVERY_S
d = law.compute(st["target"], grid, batt, tun, st["sat"], st["i_w"], refused=st["ref"], probe=probe)
if d.refused and not st["ref"]: st["anchor"] = t
if not d.refused: st["anchor"] = None
if probe and d.target_w == NEW.PROBE_W and M["last"] != NEW.PROBE_W: M["probes"] += 1
st["ref"] = d.refused
st.update(sat=d.sat_count, i_w=d.i_w, loop_w=d.target_w, target=d.target_w)
tg = st["target"]
if tg - b > 1000: M["und"] += dt / 3600
if st["ref"]: M["refh"] += dt / 3600
M["imp"] += max(0, grid) * dt / 3.6e6; M["exp"] += max(0, -grid) * dt / 3.6e6
k = math.floor(t / 900); a = M["q"].setdefault(k, [0.0, 0.0]); a[0] += grid * dt; a[1] += dt
M["run"] = M["run"] + dt if (M["last"] == tg and abs(tg) >= 250) else 0.0
M["mx"] = max(M["mx"], M["run"]); M["last"] = tg
worst = max(max(0.0, s / d) for s, d in M["q"].values() if d > 600)
return trips, M, worst
def main():
rows = R.stream()
fit = R.fit_plant(rows, stride=7)
print(f"fit {[round(x, 2) for x in fit]}, {len(rows)} meter instants")
for label, f in (("fitted plant", fit),
("plant offset 0 W", (fit[0], fit[1], fit[2], 0.0, fit[4]))):
for which in ("old", "new"):
tr, M, w = go(which, rows, f)
print(f"{label:17s} {which}: gate trips {tr:4d} | request >1 kW above "
f"delivery {M['und']:6.1f} h | longest held >=250 W "
f"{M['mx'] / 3600:5.2f} h | import {M['imp']:6.1f} kWh export "
f"{M['exp']:5.1f} kWh | worst 15-min {w:5.0f} W | refused "
f"{M['refh']:5.1f} h, probes {M['probes']}", flush=True)
if __name__ == "__main__":
main()