"""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 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("\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: # the fit that counts changed values calls that slow, the readings do not (the # recorded day at 1-2 s showed it). s, m = settle.Settle(), MeterFit() for k in range(60): s.arrive(float(k), k) m.add(float(k), 0.0, 1800.0 + 100.0 * (k // 5)) check("a 1 s meter whose value changes every 5 s: not slow, though its value " "repeats", not s.slow() and m.slow()) print() if fails: print(f"{len(fails)} FAILED: " + ", ".join(fails)) sys.exit(1) print("all checks passed")