adminandClaude Opus 5.5
3c0f2b9669
- Engage (and seed from a stale stop) below REFUSED_BELOW_W, the probe's own 100 W answer, not below saturation_w: a battery derated to 100-499 W was held at 0 W ~93 % of the time. - A reading of 100 W or more in the held direction ends the refusal, probe or not: at a 1 s meter SETTLE_CYCLES is 12 s, shorter than a healthy reading's lag, and that held a healthy step at 0 W until the next probe. - test_refusal part 5b and a seed test; DOCS states the N>1 limit; the DECISIONS record carries the traced cause of the replay's extra trips. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
98 lines
4.8 KiB
Python
98 lines
4.8 KiB
Python
#!/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()
|