goodwe-agent
0f9c119bcf
F1 the on-change case is tested as slow, F2 override and direction checks, F3 the 3.75 s rule in the texts, F4 the quarter figures and the live note. No behaviour change. Co-Authored-By: Claude <noreply@anthropic.com>
264 lines
9.8 KiB
Python
264 lines
9.8 KiB
Python
"""Runnable check for SAFETY-136: a calmer loop on a slow meter.
|
|
`python3 test_safety136.py`
|
|
|
|
With a 5 s meter and Home Assistant readings 10-15 s late the loop answers the
|
|
same error again at each cycle and overshoots: with an oven, a kettle and hob,
|
|
or a step the export was 3-10 times that of a meter with no lag. After a move
|
|
of 150 W or more `app/settle.py` now runs the loop at half its gain (at most
|
|
2 cycles) until the meter shows half of it. It is off for a meter under 4 s.
|
|
DECISIONS/SAFETY-136-slow-meter-settle.md.
|
|
|
|
The REAL Controller on TEST-23's plant (tools/safety128_rig.py). "Off" is the
|
|
same run with `Settle.moved` doing nothing: the tree before this card.
|
|
|
|
1. `Settle` alone.
|
|
2. 5 s meter, readings 10 s late: oven, kettle and hob, a step down and a step
|
|
up export clearly less than with it off.
|
|
3. 5 s meter, no lag: the same houses are no worse than a small margin.
|
|
4. 1 s and 2 s meters: the commands are those of the run with it off, and the
|
|
loop never waits.
|
|
5. The slow meter is told by its observed interval, at 4 s and 5 s.
|
|
|
|
Mutations (MUTATION.md, purged copies) are listed in the pull request.
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
import sys
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, HERE)
|
|
sys.path.insert(0, os.path.join(HERE, "tools"))
|
|
logging.disable(logging.WARNING)
|
|
|
|
import safety128_rig as R # noqa: E402
|
|
from app.control import Tuning # noqa: E402
|
|
from named_absence import nth, present # noqa: E402
|
|
from app import settle # noqa: E402
|
|
from app.lateness import MeterFit # noqa: E402
|
|
|
|
fails = []
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
def off(fn):
|
|
"""`fn()` with the settle wait off: the tree before this card."""
|
|
real = settle.Settle.moved
|
|
settle.Settle.moved = lambda self, *a, **k: None
|
|
try:
|
|
return fn()
|
|
finally:
|
|
settle.Settle.moved = real
|
|
|
|
|
|
def oven(t):
|
|
return 1800.0 + (2000.0 if (t >= 300.0 and (t - 300.0) % 105 < 45) else 0.0)
|
|
|
|
|
|
def kettle(t):
|
|
w = 600.0
|
|
if t % 400 < 120:
|
|
w += 2200
|
|
if t >= 900 and t % 90 < 30:
|
|
w += 1500
|
|
return w
|
|
|
|
|
|
def step(t):
|
|
return 1800.0 if t < 320.0 else 300.0
|
|
|
|
|
|
def stepup(t):
|
|
return 300.0 if t < 320.0 else 2500.0
|
|
|
|
|
|
# name: (house, noise, run length, lag end)
|
|
HOUSES = {"oven": (oven, 80.0, 900.0, 900.0),
|
|
"kettle": (kettle, 80.0, 1500.0, 1500.0),
|
|
"step": (step, 50.0, 700.0, 600.0),
|
|
"stepup": (stepup, 50.0, 700.0, 600.0)}
|
|
ON = 300.0
|
|
|
|
|
|
def wh(rows, t0, t1, sign):
|
|
"""Export (sign -1) or import (sign +1) energy in Wh over [t0, t1)."""
|
|
return sum((-min(r[1], 0.0) if sign < 0 else max(r[1], 0.0))
|
|
for r in rows if t0 <= r[0] < t1) / 3600.0
|
|
|
|
|
|
def run(name, lag, p1_s=5.0, units=("a", "b"), dur=None):
|
|
house, noise, length, lag_off = HOUSES[name]
|
|
r = R.Rig(house=house, noise=noise, p1_s=p1_s, units=units, lag_s=lag,
|
|
lag_on=ON, lag_off=lag_off)
|
|
waits = []
|
|
r.run(dur or length + 20.0,
|
|
lambda rig, c, t: waits.append(1) if c.settle.pend else None)
|
|
return r, len(waits)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("1. Settle alone")
|
|
T = Tuning()
|
|
s = settle.Settle()
|
|
check("no move: the tuning is untouched", s.tuning(T, 500.0, True) is T)
|
|
s.moved(500.0, 0.0, 100.0, True)
|
|
check("a move under 150 W is not waited for", s.pend is None)
|
|
s.moved(500.0, 0.0, 600.0, False)
|
|
check("a meter that is not slow is not waited for", s.pend is None)
|
|
s.moved(500.0, 0.0, 600.0, True)
|
|
check("a move of 600 W is waited for: half the gain",
|
|
s.tuning(T, 500.0, True).gain == T.gain * 0.5)
|
|
check("the meter fell 200 W of 600 W: still half the gain",
|
|
s.tuning(T, 300.0, True).gain == T.gain * 0.5)
|
|
check("two cycles waited: the third has the whole gain",
|
|
s.tuning(T, 300.0, True) is T and s.pend is None)
|
|
s.moved(500.0, 0.0, 600.0, True)
|
|
check("the meter went the other way by half the move: the wait stands",
|
|
s.tuning(T, 800.0, True).gain == T.gain * 0.5)
|
|
s.reset()
|
|
s.moved(500.0, 0.0, -600.0, True)
|
|
check("a move down, the meter rose half the move: the whole gain",
|
|
s.tuning(T, 800.0, True) is T)
|
|
s.reset()
|
|
s.moved(500.0, 0.0, 600.0, True)
|
|
check("the meter fell half the move: the whole gain at once",
|
|
s.tuning(T, 200.0, True) is T and s.pend is None)
|
|
s.moved(500.0, 0.0, -600.0, True)
|
|
check("a move down is waited for too",
|
|
s.tuning(T, 500.0, True).gain == T.gain * 0.5)
|
|
s.reset()
|
|
check("reset ends the wait", s.pend is None)
|
|
s.moved(500.0, 0.0, 600.0, True)
|
|
check("a meter that turns out not to be slow ends the wait",
|
|
s.tuning(T, 500.0, False) is T and s.pend is None)
|
|
check("the numbers are pinned",
|
|
(settle.MOVE_W, settle.SHOWN, settle.WAIT_SHARE, settle.WAIT_CYCLES)
|
|
== (150.0, 0.5, 0.5, 2))
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n2. 5 s meter, readings 10 s late")
|
|
tot_on = tot_off = 0.0
|
|
for name, (_h, _n, length, lag_off) in HOUSES.items():
|
|
a, waited = run(name, 10.0)
|
|
b, _w = off(lambda: run(name, 10.0))
|
|
ea, eb = wh(a.rows, ON, lag_off, -1), wh(b.rows, ON, lag_off, -1)
|
|
tot_on, tot_off = tot_on + ea, tot_off + eb
|
|
print(f" {name}: export {ea:.1f} Wh, off {eb:.1f} Wh, waited {waited} s")
|
|
check(f"{name}: waits at some point", waited > 0)
|
|
check(f"{name}: exports at most 75 % of the run with it off",
|
|
ea <= 0.75 * eb)
|
|
print(f" sum: {tot_on:.1f} Wh, off {tot_off:.1f} Wh")
|
|
check("the four together export at most 65 % of the run with it off",
|
|
tot_on <= 0.65 * tot_off)
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n2b. Another writer overrides the loop: the wait ends")
|
|
halved = []
|
|
real_tuning = settle.Settle.tuning
|
|
|
|
|
|
def watch_tuning(self, tuning, grid_w, slow):
|
|
r = real_tuning(self, tuning, grid_w, slow)
|
|
halved.append(r is not tuning)
|
|
return r
|
|
|
|
|
|
settle.Settle.tuning = watch_tuning
|
|
try:
|
|
mark = []
|
|
|
|
def bump(rig, c, t):
|
|
# right after a cycle that started a wait, another writer moves the
|
|
# command, so the next cycle sees `target != loop_w`
|
|
if not mark and c.settle.pend and t > 330.0:
|
|
mark.append(len(halved))
|
|
c.target = c.target + 300.0
|
|
|
|
r = R.Rig(house=oven, noise=80.0, p1_s=5.0, lag_s=10.0, lag_on=ON,
|
|
lag_off=900.0)
|
|
r.run(500.0, bump)
|
|
finally:
|
|
settle.Settle.tuning = real_tuning
|
|
nxt = nth(halved, mark[0] if mark else 10 ** 6)
|
|
check("the cycle after an override has the whole gain",
|
|
bool(mark) and present(nxt) and nxt is False)
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n3. 5 s meter, no lag")
|
|
for name in ("oven", "kettle", "step", "stepup"):
|
|
_h, _n, length, lag_off = HOUSES[name]
|
|
a, _w = run(name, 0.0)
|
|
b, _w = off(lambda: run(name, 0.0))
|
|
ea, eb = wh(a.rows, ON, lag_off, -1), wh(b.rows, ON, lag_off, -1)
|
|
ia, ib = wh(a.rows, ON, lag_off, 1), wh(b.rows, ON, lag_off, 1)
|
|
print(f" {name}: export {ea:.1f} Wh (off {eb:.1f}), import {ia:.1f} Wh "
|
|
f"(off {ib:.1f})")
|
|
check(f"{name}: export at most 6 Wh over the run with it off",
|
|
ea <= eb + 6.0)
|
|
check(f"{name}: import at most 1 Wh over the run with it off",
|
|
ia <= ib + 1.0)
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n4. 1 s and 2 s meters")
|
|
for p1 in (1.0, 2.0):
|
|
for lag in (0.0, 10.0):
|
|
for units in (("a", "b"), ("a",)):
|
|
a, waited = run("oven", lag, p1_s=p1, units=units, dur=700.0)
|
|
b, _w = off(lambda: run("oven", lag, p1_s=p1, units=units,
|
|
dur=700.0))
|
|
same = ([(r[0], r[1], r[2]) for r in a.rows]
|
|
== [(r[0], r[1], r[2]) for r in b.rows])
|
|
check(f"{p1:.0f} s meter, {lag:.0f} s late, N={len(units)}: "
|
|
f"same commands, never waits", same and waited == 0)
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n5. The slow meter is told by its observed interval")
|
|
s = settle.Settle()
|
|
for k in range(30):
|
|
s.arrive(float(k), k)
|
|
check("a reading each second is not a slow meter", not s.slow())
|
|
s = settle.Settle()
|
|
for k in range(30):
|
|
s.arrive(5.0 * k, k)
|
|
check("a reading every 5 s is a slow meter", s.slow())
|
|
s = settle.Settle()
|
|
for k in range(4):
|
|
s.arrive(5.0 * k, k)
|
|
check("fewer than 5 readings in a minute tell nothing", not s.slow())
|
|
s = settle.Settle()
|
|
for k in range(30):
|
|
s.arrive(5.0 * k, 0 if k else 1)
|
|
check("the same count again is no new reading", not s.slow())
|
|
for p1, expect in ((1.0, False), (2.0, False), (3.0, False), (4.0, True),
|
|
(5.0, True), (10.0, True)):
|
|
r = R.Rig(house=1800.0, p1_s=p1)
|
|
seen = []
|
|
r.run(120.0, lambda rig, c, t: seen.append(c.settle.slow()))
|
|
last = nth(seen, -1)
|
|
check(f"{p1:.0f} s meter: slow is {expect}",
|
|
present(last) and last is expect)
|
|
# A 1 s meter on a house that changes every 5 s repeats its value 4 times in 5.
|
|
# Where every reading arrives (the rig, a polled meter) it is not slow; where
|
|
# Home Assistant sends a reading only when the value changes (ha_signed) the
|
|
# same meter is slow (T3 of the review; DECISIONS).
|
|
s, m, h = settle.Settle(), MeterFit(), settle.Settle()
|
|
for k in range(60):
|
|
s.arrive(float(k), k)
|
|
h.arrive(float(k), k // 5)
|
|
m.add(float(k), 0.0, 1800.0 + 100.0 * (k // 5))
|
|
check("every reading arrives, value changes every 5 s: not slow (the value "
|
|
"fit says slow)", not s.slow() and m.slow())
|
|
check("the same meter through Home Assistant on change (a reading per "
|
|
"change): slow", h.slow())
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED: " + ", ".join(fails))
|
|
sys.exit(1)
|
|
print("all checks passed")
|