484 lines
21 KiB
Python
484 lines
21 KiB
Python
"""Runnable check for SAFETY-90: the peak guard goes on after a P1 meter gap.
|
|
`python3 test_safety90.py`
|
|
|
|
A gap past `meter_max_age_s` left seconds the quarter average never counted,
|
|
so the guard was blind for the rest of the quarter: a HOLD gave 0 W and the
|
|
quarter raised the month peak (SAFETY-86 rig case 7b: 3964.8 W against a
|
|
3450 W aim). Now the missed seconds count at `peakguard.gap_w()`, and so do
|
|
the seconds before the add-on's first reading, with a plain line on the
|
|
page. Drives the REAL Controller on a lagging battery (10 s), with the P1
|
|
telegrams stopped for each gap. DECISIONS/SAFETY-90-peak-guard-after-a-gap.md.
|
|
|
|
1. N=1 HOLD: gaps of 30 s, 60 s and 5 min mid-quarter land at or under the aim.
|
|
2. N=2 HOLD, B able: the same.
|
|
3. N=2 HOLD, B held at its reserve: the same, and B gives nothing (O2).
|
|
4. A gap across a quarter edge; the house rising during a gap.
|
|
5. Start-up mid-quarter: the guard runs, the page says what it assumed.
|
|
6. While the gap lasts the guard is blind as before; no gap, no change.
|
|
7. fill_gaps() and gap_w() at their boundaries.
|
|
|
|
Mutations (MUTATION.md, purged copies) are listed in the pull request.
|
|
"""
|
|
|
|
import logging
|
|
import math
|
|
import os
|
|
import sys
|
|
import types
|
|
from datetime import timedelta
|
|
|
|
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 opt11_identity as rig # noqa: E402
|
|
import safety84_identity as s84 # noqa: E402
|
|
import app.p1 as p1mod # noqa: E402
|
|
import app.tariff as tariff # noqa: E402
|
|
from app import homeview, messages, peakguard # noqa: E402
|
|
from app.maintenance import HOLD # noqa: E402
|
|
from app.p1 import make_sample, split_signed # noqa: E402
|
|
from app.store import unit_key # noqa: E402
|
|
from app.tariff import QUARTER_S, Projection # noqa: E402
|
|
|
|
fails = []
|
|
A, B = "inverter-a", "inverter-b"
|
|
AIM = 3450.0 # peak_cap_w 3500 - peak_margin_w 50
|
|
TOL = 5.0 # W over the aim a quarter may land (SAFETY-79)
|
|
CUT = 10.0
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
def store(phase_of, n):
|
|
s = rig._Store()
|
|
for u in (A, B)[:n]:
|
|
key = (lambda k: k) if n == 1 else (lambda k, u=u: unit_key(u, k))
|
|
if u in phase_of:
|
|
s.data[key("phase")] = phase_of[u]
|
|
s.data[key("phase_started")] = rig.BASE.isoformat()
|
|
else:
|
|
s.data[key("last_completed")] = rig.BASE.isoformat()
|
|
return s
|
|
|
|
|
|
def run(house, socs, dur, *, n, gaps=(), start=0.0, lag=10.0, on_cycle=None):
|
|
"""Rows and controller. A holds full; `socs` are the starting levels
|
|
(B's follows its power). `gaps`: (t0, t1), no telegram for t0 <= t < t1.
|
|
`start`: the add-on starts this many s into the run (mid-quarter). The
|
|
P1 clock is simulated, so a gap really goes stale. `c.truth` is the grid
|
|
each second, which the quarter really averages."""
|
|
clock = [0.0]
|
|
soc, last, truth = dict(socs), [start], []
|
|
|
|
def before(c, t0):
|
|
if on_cycle is not None:
|
|
on_cycle(c, t0) # the state the last cycle left
|
|
t = t0 + start
|
|
clock[0] = t
|
|
g = c.hass.grid()
|
|
truth.append((t, g))
|
|
for u in soc:
|
|
soc[u] = max(CUT, soc[u] - c.hass.power(u) * (t - last[0])
|
|
/ 3600.0 / 153.6)
|
|
last[0] = t
|
|
if any(a <= t < b for a, b in gaps):
|
|
return
|
|
imp, exp = split_signed(g)
|
|
c.p1.submit(make_sample("ha_signed", imp, exp, phases=1,
|
|
ingest_ts=rig.BASE + timedelta(seconds=t0),
|
|
ingest_mono=1e6 + t))
|
|
|
|
def power(self, u):
|
|
st = self.__dict__.setdefault("_plant", {})
|
|
a, tgt, tl = st.get(u, (0.0, 0.0, self.t))
|
|
for _ in range(int(round((self.t - tl) / 0.1))):
|
|
d = self.cmd[u] - tgt
|
|
tgt += math.copysign(min(abs(d), 80.0), d)
|
|
a += (tgt - a) * (1 - math.exp(-0.1 / lag))
|
|
st[u] = (a, tgt, self.t)
|
|
return 0.0 if self.soc(u, self.t) <= CUT and a > 0 else float(round(a))
|
|
|
|
real = (rig._Hass.power, rig.TICK, rig.BASE, p1mod.time)
|
|
rig._Hass.power, rig.TICK = power, 1.0
|
|
rig.BASE = real[2] + timedelta(seconds=start)
|
|
p1mod.time = types.SimpleNamespace(monotonic=lambda: 1e6 + clock[0])
|
|
try:
|
|
rows, c = s84.run(
|
|
n, load=house if callable(house) else (lambda t: house),
|
|
cheap=False, store=store({A: HOLD}, n), before=before, dur=dur,
|
|
soc=lambda u, t: soc[u],
|
|
extra={"maintenance_hold_min": 600})
|
|
finally:
|
|
rig._Hass.power, rig.TICK, rig.BASE, p1mod.time = real
|
|
c.truth = truth
|
|
return rows, c
|
|
|
|
|
|
def without_fix(fn):
|
|
"""release/1.0 3ed3302: nothing is filled, so a gap blinds the guard."""
|
|
real = peakguard.fill_gaps
|
|
peakguard.fill_gaps = lambda p, **kw: (p, 0.0, 0.0)
|
|
try:
|
|
return fn()
|
|
finally:
|
|
peakguard.fill_gaps = real
|
|
|
|
|
|
def quarter(c, q):
|
|
"""What quarter `q` (0 = the run's first) really averages, in W."""
|
|
return sum(max(g, 0.0) for t, g in c.truth if int(t // QUARTER_S) == q) \
|
|
/ QUARTER_S
|
|
|
|
|
|
def cmd(r):
|
|
return dict(r[2])
|
|
|
|
|
|
def mean(rows, u, lo, hi):
|
|
rs = [cmd(r)[u] for r in rows if lo <= r[0] < hi]
|
|
return round(sum(rs) / len(rs))
|
|
|
|
|
|
TABLE = []
|
|
|
|
|
|
def case(tag, n, house, socs, gaps, q=0, dur=1800.0):
|
|
"""One gap case, after and before the fix; prints and records it."""
|
|
new = run(house, socs, dur, n=n, gaps=gaps)
|
|
old = without_fix(lambda: run(house, socs, dur, n=n, gaps=gaps))
|
|
a, b = quarter(new[1], q), quarter(old[1], q)
|
|
print(f" {tag}: quarter {a:.1f} W (before the fix {b:.1f} W), aim "
|
|
f"{AIM:.0f}")
|
|
TABLE.append((tag, n, round(a, 1), round(b, 1)))
|
|
return new, old, a, b
|
|
|
|
|
|
# ============================================================================ #
|
|
print("1. N=1 HOLD, house 6500 W (5500 W for 5 min, so it can be held)")
|
|
for tag, gap, house in (("30 s", (300, 330), 6500.0),
|
|
("60 s", (300, 360), 6500.0),
|
|
("5 min", (300, 600), 5500.0)):
|
|
(rows, c), _old, a, b = case(f"N=1 gap {tag}", 1, house, {"a": 99.0},
|
|
[gap])
|
|
check(f"⭐ N=1, {tag} gap mid-quarter: the quarter lands at or under the "
|
|
f"aim ({a:.1f} W)", a <= AIM + TOL)
|
|
if tag != "30 s":
|
|
check(f"...non-vacuous: before the fix it went over ({b:.1f} W)",
|
|
b > AIM + 100)
|
|
check("...and once readings are back the battery gives at least the "
|
|
"house over the aim",
|
|
mean(rows, "a", gap[1] + 15, 900) >= house - AIM)
|
|
check("...the action log says what the gap was counted at",
|
|
any("PEAK_GUARD_GAP" in e for e in c.events))
|
|
|
|
# ============================================================================ #
|
|
print("\n2. N=2 HOLD (A holding), B able at 60 %, house 6500 W")
|
|
for tag, gap, house in (("30 s", (300, 330), 6500.0),
|
|
("60 s", (300, 360), 6500.0),
|
|
("5 min", (300, 600), 5500.0)):
|
|
(rows, c), _old, a, b = case(f"N=2 gap {tag}", 2, house,
|
|
{"a": 99.0, "b": 60.0}, [gap])
|
|
check(f"⭐ N=2, {tag} gap: the quarter lands at or under the aim "
|
|
f"({a:.1f} W)", a <= AIM + TOL)
|
|
if tag != "30 s":
|
|
check(f"...non-vacuous: before the fix it went over ({b:.1f} W)",
|
|
b > AIM + 100)
|
|
check("...B gives first: at its full 2500 W once readings are back",
|
|
mean(rows, "b", gap[1] + 30, 900) >= 2450)
|
|
|
|
# ============================================================================ #
|
|
print("\n3. N=2 HOLD, B held at 15 % (under its 20 % reserve), house 5500 W "
|
|
"(4500 W for 5 min: A alone gives 2500 W)")
|
|
for tag, gap, house in (("30 s", (300, 330), 5500.0),
|
|
("60 s", (300, 360), 5500.0),
|
|
("5 min", (300, 600), 4500.0)):
|
|
(rows, c), (orows, _oc), a, b = case(
|
|
f"N=2 B held, gap {tag}", 2, house, {"a": 99.0, "b": 15.0}, [gap])
|
|
check(f"⭐ N=2 B held, {tag} gap: the quarter lands at or under the aim "
|
|
f"({a:.1f} W)", a <= AIM + TOL)
|
|
check("⭐ O2: the held B gives nothing once readings are back; the "
|
|
"holding A gives the need",
|
|
all(cmd(r)["b"] == 0.0 for r in rows if gap[1] <= r[0] < 900)
|
|
and mean(rows, "a", gap[1] + 15, 900) >= house - AIM)
|
|
if tag != "30 s":
|
|
check("...non-vacuous: before the fix the blind guard left A at 0 W "
|
|
"and the held B was drawn at 2500 W as last resort",
|
|
mean(orows, "a", gap[1] + 30, 900) == 0
|
|
and mean(orows, "b", gap[1] + 30, 900) >= 2450)
|
|
|
|
rows, c = run(5500.0, {"a": 99.0, "b": 15.0}, 1000.0, n=2,
|
|
gaps=[(300, 360), (899, 910)])
|
|
edge = [(r[0], cmd(r)["a"], cmd(r)["b"]) for r in rows if 900 <= r[0] < 915]
|
|
print(f" a 60 s gap, then the edge's telegram 11 s late: {edge[:4]}; the "
|
|
f"second before the edge (O1, not this card): "
|
|
f"{[(r[0], cmd(r)['a'], cmd(r)['b']) for r in rows if r[0] == 899]}")
|
|
check("⭐ ...a quarter that had a gap still hands over at its edge (SAFETY-91): "
|
|
"with the edge's telegram late the held B gives nothing and A keeps "
|
|
"giving", all(b == 0.0 and a > 1500.0 for _t, a, b in edge))
|
|
|
|
# ============================================================================ #
|
|
print("\n4. a gap across a quarter edge, and a house that rises in a gap")
|
|
for n, socs in ((1, {"a": 99.0}), (2, {"a": 99.0, "b": 60.0})):
|
|
(rows, c), _o, a, b = case(f"N={n} gap across the edge", n, 6500.0, socs,
|
|
[(870, 930)], q=1)
|
|
check(f"⭐ N={n}: the quarter the readings come back in lands at or "
|
|
f"under the aim ({a:.1f} W; before the fix {b:.1f} W)",
|
|
a <= AIM + TOL and b > AIM + 100)
|
|
check("...the quarter that ended in the gap could not be steered after "
|
|
"it, and stays under the limit itself",
|
|
quarter(c, 0) <= 3500.0)
|
|
|
|
|
|
def rising(t):
|
|
return 3000.0 if t < 320 else 6500.0
|
|
|
|
|
|
for n, socs in ((1, {"a": 99.0}), (2, {"a": 99.0, "b": 15.0})):
|
|
house = rising if n == 1 else (lambda t: 3000.0 if t < 320 else 5500.0)
|
|
(rows, c), _o, a, b = case(f"N={n} house rises in a 60 s gap", n, house,
|
|
socs, [(300, 360)])
|
|
check(f"⭐ N={n}: the house goes from 3000 W to over the limit during "
|
|
f"the gap; the import seen after it is counted and the quarter "
|
|
f"lands at or under the aim ({a:.1f} W)", a <= AIM + TOL)
|
|
|
|
for n, socs in ((1, {"a": 99.0}), (2, {"a": 99.0, "b": 15.0})):
|
|
big = 6500.0 if n == 1 else 5500.0
|
|
(rows, c), _o, a, b = case(
|
|
f"N={n} house falls in a 5 min gap", n,
|
|
lambda t, big=big: big if t < 560 else 2000.0, socs, [(300, 600)])
|
|
check(f"⭐ N={n}: the house falls to 2000 W just before readings return; "
|
|
f"the import before the gap (batteries at 0 W) is counted and the "
|
|
f"quarter lands at or under the aim ({a:.1f} W; before {b:.1f} W, "
|
|
f"at N=2 by drawing the held B)",
|
|
a <= AIM + TOL and (n == 2 or b > AIM + 100))
|
|
|
|
(rows, c), _o, a, b = case(
|
|
"N=1 5 min gap at 3000 W, then 60 s gap at 6500 W next quarter", 1,
|
|
lambda t: 3000.0 if t < 900 else 6500.0, {"a": 99.0},
|
|
[(300, 600), (1200, 1260)], q=1)
|
|
check(f"⭐ ...each quarter's gap is counted on its own: a long gap at a low "
|
|
f"load does not hide a shorter one in the next quarter ({a:.1f} W)",
|
|
a <= AIM + TOL and sum("PEAK_GUARD_GAP" in e for e in c.events) == 2)
|
|
|
|
|
|
|
|
def dip(t):
|
|
return 3000.0 if 450 <= t < 700 else 6500.0
|
|
|
|
|
|
(rows, c), _o, a, b = case("N=1 two gaps, the second at a lower load", 1,
|
|
dip, {"a": 99.0}, [(300, 360), (500, 560)])
|
|
at = [float(e.split("counted at ")[1].split(" W")[0]) for e in c.events
|
|
if "PEAK_GUARD_GAP" in e]
|
|
check(f"⭐ ...two gaps in one quarter: the second, at a lower load, does not "
|
|
f"price the first one down (counted at {at}; {a:.1f} W)",
|
|
a <= AIM + TOL and len(at) == 2 and min(at) >= 6400.0)
|
|
|
|
# ============================================================================ #
|
|
print("\n5. start-up mid-quarter (the add-on starts 450 s into the quarter)")
|
|
UI = {}
|
|
|
|
|
|
def grab(c, t):
|
|
if t == 60.0:
|
|
UI["unmeasured"] = c.peak_unmeasured_s
|
|
UI["rows"] = [r for r in c.checks() if r.get("id") == "peak_guard"]
|
|
UI["en"] = homeview.month_peak(c, c.cycle_at)["line"]
|
|
homeview._LANG[0] = "nl"
|
|
try:
|
|
UI["nl"] = homeview.month_peak(c, c.cycle_at)["line"]
|
|
finally:
|
|
homeview._LANG[0] = "en"
|
|
if t == 453.0:
|
|
UI["later"] = (c.peak_unmeasured_s,
|
|
[r for r in c.checks() if r.get("id") == "peak_guard"])
|
|
|
|
|
|
def restored(t_dead, start=450.0):
|
|
"""TARIFF-01 restores the averager at start-up: 0..t_dead measured at
|
|
the aim, the held sample at t_dead (the restart's first second)."""
|
|
def fake(self, averager, now=None):
|
|
bs = int(rig.BASE.timestamp() - start)
|
|
averager.restore({"block": bs, "acc": AIM * t_dead,
|
|
"pp_acc": [AIM * t_dead], "elapsed": t_dead,
|
|
"last_t": bs + t_dead, "last_net": AIM,
|
|
"last_pp": [AIM], "phases": 1})
|
|
self.restore_state = "resumed"
|
|
return self.restore_state
|
|
return fake
|
|
|
|
|
|
def startup(house, socs, n, *, t_dead=None, on_cycle=None):
|
|
"""The add-on starts 450 s in. While it was down the batteries were at
|
|
0 W (the ESP32 watchdog), so the meter saw the house: the quarter is
|
|
that, or the restored part at the aim, plus what this run measured."""
|
|
real = tariff.PeakTracker.restore
|
|
if t_dead is not None:
|
|
tariff.PeakTracker.restore = restored(t_dead)
|
|
try:
|
|
rows, c = run(house, socs, 455.0, n=n, start=450.0,
|
|
on_cycle=on_cycle)
|
|
finally:
|
|
tariff.PeakTracker.restore = real
|
|
pre = 450.0 * house if t_dead is None else (
|
|
t_dead * AIM + (450.0 - t_dead) * house)
|
|
return rows, c, (pre + sum(max(g, 0.0) for t, g in c.truth if t < 900)) \
|
|
/ QUARTER_S
|
|
|
|
|
|
for n, socs, house in ((1, {"a": 99.0}, 5000.0),
|
|
(2, {"a": 99.0, "b": 15.0}, 4500.0)):
|
|
rows, c, land = startup(house, socs, n, on_cycle=grab if n == 1 else None)
|
|
orows, _oc, old = without_fix(lambda: startup(house, socs, n))
|
|
TABLE.append((f"N={n} start-up mid-quarter", n, round(land, 1),
|
|
round(old, 1)))
|
|
check(f"⭐ N={n}: the first 450 s (house {house:.0f} W, batteries at 0 W) "
|
|
f"count at the import seen first; the quarter lands at or under "
|
|
f"the aim ({land:.1f} W, before the fix {old:.1f} W)",
|
|
land <= AIM + TOL and (n == 2 or old > AIM + 100))
|
|
if n == 2:
|
|
check("...and the held B gives nothing (O2 at start-up); before the "
|
|
"fix it was drawn at 2500 W",
|
|
all(cmd(r)["b"] == 0.0 for r in rows if 1 <= r[0] < 450)
|
|
and mean(orows, "b", 30, 450) >= 2450)
|
|
check("...non-vacuous: before the fix A gave 0 W for the quarter",
|
|
mean(orows, "a", 30, 450) == 0)
|
|
|
|
rows, c, land = startup(6500.0, {"a": 99.0}, 1)
|
|
check(f"⭐ N=1, house 6500 W: more than the battery can make up; A gives its "
|
|
f"full 5000 W and the alarm says so ({land:.1f} W)",
|
|
mean(rows, "a", 30, 450) >= 4900
|
|
and any("PEAK_GUARD_SATURATED" in e for e in c.events))
|
|
|
|
for t_dead in (300.0, 390.0, 440.0):
|
|
rows, c, land = startup(6500.0, {"a": 99.0}, 1, t_dead=t_dead)
|
|
TABLE.append((f"N=1 restart, down {450 - t_dead:.0f} s", 1,
|
|
round(land, 1), round(without_fix(lambda: startup(
|
|
6500.0, {"a": 99.0}, 1, t_dead=t_dead))[2], 1)))
|
|
check(f"⭐ restart mid-quarter, down {450 - t_dead:.0f} s at house "
|
|
f"6500 W (the restored reading's held 30 s topped up too): the "
|
|
f"quarter lands at or under the aim ({land:.1f} W)",
|
|
land <= AIM + TOL)
|
|
|
|
rows, c = run(6500.0, {"a": 99.0}, 1350.0, n=1, start=450.0,
|
|
gaps=[(1200, 1260)])
|
|
a = quarter(c, 1)
|
|
TABLE.append(("N=1 start-up, then 60 s gap next quarter", 1, round(a, 1),
|
|
round(quarter(without_fix(lambda: run(
|
|
6500.0, {"a": 99.0}, 1350.0, n=1, start=450.0,
|
|
gaps=[(1200, 1260)]))[1], 1), 1)))
|
|
check(f"...the start-up assumption stays in its own quarter: a 60 s gap in "
|
|
f"the next one is counted as a gap and lands at or under the aim "
|
|
f"({a:.1f} W)", a <= AIM + TOL)
|
|
|
|
print(f" page: {UI.get('rows')}\n EN {UI.get('en')!r}\n"
|
|
f" NL {UI.get('nl')!r}")
|
|
check("⭐ the page says plainly that part of the period was not measured, "
|
|
"in English and Dutch",
|
|
UI.get("unmeasured", 0) > 400
|
|
and UI["rows"][0]["text"] == messages.row_peak_guard_unmeasured()
|
|
and UI["en"].startswith("Part of this 15-minute period was not "
|
|
"measured.")
|
|
and UI["nl"].startswith("Een deel van dit kwartier is niet gemeten."))
|
|
check("...and stops saying it in the next quarter",
|
|
UI["later"][0] == 0.0 and UI["later"][1][0]["text"]
|
|
!= messages.row_peak_guard_unmeasured())
|
|
|
|
# ============================================================================ #
|
|
print("\n6. while the gap lasts, and with no gap at all")
|
|
LIVE = {}
|
|
|
|
|
|
def live(c, t):
|
|
if t == 350.0:
|
|
LIVE["g"] = c.peak_guard
|
|
|
|
|
|
run(6500.0, {"a": 99.0}, 400.0, n=1, gaps=[(300, 360)], on_cycle=live)
|
|
check("⭐ 50 s into a gap the guard is blind as before (floor 0 W, no "
|
|
"charging): nothing is filled while the gap is open",
|
|
LIVE["g"].blind is not None and LIVE["g"].floor_w == 0.0)
|
|
for n, socs, house in ((1, {"a": 99.0}, 6500.0),
|
|
(2, {"a": 99.0, "b": 60.0}, 6500.0),
|
|
(2, {"a": 99.0, "b": 15.0}, 5500.0)):
|
|
rows, c = run(house, socs, 1000.0, n=n)
|
|
orows, _ = without_fix(lambda: run(house, socs, 1000.0, n=n))
|
|
check(f"⭐ N={n}, meter never drops: every row is identical with and "
|
|
f"without the fix, and nothing is filled",
|
|
rows == orows and not any("PEAK_GUARD_GAP" in e for e in c.events))
|
|
|
|
# ============================================================================ #
|
|
print("\n7. fill_gaps() and gap_w() at their boundaries")
|
|
T0 = rig.BASE
|
|
|
|
|
|
def proj(wall, covered, held=1000.0, acc_w=1000.0):
|
|
return Projection(projected_w=(acc_w * covered + held * (QUARTER_S - wall))
|
|
/ QUARTER_S, elapsed_w=acc_w, elapsed_s=wall,
|
|
covered_frac=covered / wall, complete=False,
|
|
block_start=T0, held_w=held)
|
|
|
|
|
|
def fill(p, tail, unseen=0.0, gw=4000.0, uw=3450.0):
|
|
return peakguard.fill_gaps(p, tail_s=tail, max_hold_s=30.0,
|
|
unseen_s=unseen, gap_w=gw, unseen_w=uw)
|
|
|
|
|
|
p = proj(400.0, 300.0) # 100 s uncounted
|
|
f, g, u = fill(p, 10.0)
|
|
check("90 s missed (100 s uncounted less a 10 s tail): counted at gap_w, "
|
|
"the tail left as before, no longer blind",
|
|
g == 90.0 and u == 0.0
|
|
and abs(f.projected_w - (p.projected_w + 90.0 * 4000.0 / QUARTER_S))
|
|
< 1e-9 and abs(peakguard.uncovered_s(f) - 10.0) < 1e-9
|
|
and peakguard.blindness(f, 30.0, False) is None
|
|
and peakguard.blindness(p, 30.0, False) is not None)
|
|
check("tail exactly max_hold_s: filled; one past it: the gap is open, "
|
|
"nothing filled",
|
|
fill(p, 30.0)[1] == 70.0 and fill(p, 30.001) == (p, 0.0, 0.0))
|
|
f, g, u = fill(p, 10.0, unseen=60.0)
|
|
check("unseen seconds come first and count at the aim, the rest at gap_w",
|
|
u == 60.0 and g == 30.0
|
|
and abs(f.projected_w - (p.projected_w + (60 * 3450.0 + 30 * 4000.0)
|
|
/ QUARTER_S)) < 1e-9)
|
|
check("nothing missed (or under 1 ms): the projection object is returned "
|
|
"unchanged",
|
|
fill(proj(400.0, 400.0), 0.0)[0] is not None
|
|
and fill(proj(400.0, 400.0), 0.0) == (proj(400.0, 400.0), 0.0, 0.0)
|
|
and fill(proj(400.0, 399.9995), 0.0)[1] == 0.0)
|
|
f, g, u = fill(proj(400.0, 400.0), 0.0)
|
|
f2 = peakguard.fill_gaps(proj(400.0, 400.0), tail_s=0.0, max_hold_s=30.0,
|
|
unseen_s=0.0, gap_w=4000.0, unseen_w=3450.0,
|
|
held_ws=9000.0)
|
|
check("held_ws (the gap's first 30 s topped up) is added even when nothing "
|
|
"is missing any more",
|
|
g == 0.0 and abs(f2[0].projected_w - f.projected_w - 10.0) < 1e-9)
|
|
z = peakguard.fill_gaps(Projection(0, 0, 0, 0, False, T0, 0), tail_s=0.0,
|
|
max_hold_s=30.0, unseen_s=0.0, gap_w=4000.0,
|
|
unseen_w=3450.0, held_ws=9000.0)
|
|
check("the first instant of a block (0 s elapsed) with held_ws: no division "
|
|
"by zero", abs(z[0].projected_w - 10.0) < 1e-9)
|
|
check("no open block: unchanged (blindness says so)",
|
|
fill(Projection(0, 0, 0, 0, False, None, 0), 0.0)[1:] == (0.0, 0.0))
|
|
check("gap_w(): the highest of the limit and the imports seen, None skipped",
|
|
peakguard.gap_w(3500, None, 3000.0) == 3500.0
|
|
and peakguard.gap_w(3500, 6500.0, None) == 6500.0
|
|
and peakguard.gap_w(3500, 3000.0, 4500.0) == 4500.0
|
|
and peakguard.gap_w(3500) == 3500.0)
|
|
|
|
# ============================================================================ #
|
|
print("\n case N quarter W before W aim")
|
|
for tag, n, a, b in TABLE:
|
|
print(f" {tag:<34}{n} {a:>9.1f} {b:>9.1f} {AIM:.0f}")
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED")
|
|
sys.exit(1)
|
|
print("all SAFETY-90 checks passed")
|