"""Runnable check for the precedence arbiter. `python3 test_arbiter.py` Each assertion is a precedence question someone will eventually ask in the field: "why did it charge during a peak?", "why did it discharge while the maintenance cycle was charging?", "what happens if two things disagree?". """ import sys from app.arbiter import ( P_LOOP, P_MAINT_SHAPE, P_MAINTENANCE, P_PEAK, P_SAFETY, Claim, resolve, ) fails = [] def check(name, cond): print(f" {'ok ' if cond else 'FAIL'} {name}") if not cond: fails.append(name) print("basics") r = resolve([]) check("no claims at all means 0 W", r.target_w == 0.0 and r.winner == "failsafe") r = resolve([Claim.set("loop", P_LOOP, 900)]) check("a lone claim is honoured", r.target_w == 900) r = resolve([Claim.set("loop", P_LOOP, 900), Claim.set("maintenance", P_MAINTENANCE, 2500)]) check("higher priority set wins", r.target_w == 2500 and r.winner == "maintenance") print("limits") r = resolve([Claim.set("loop", P_LOOP, 4000), Claim.limit("supervised", P_SAFETY, lo=-2000, hi=2000, reason="max_w")]) check("safety limit binds the loop", r.target_w == 2000) check("and says what bound it", "supervised" in (r.bound_by or "")) r = resolve([Claim.set("loop", P_LOOP, 4000), Claim.limit("a", P_SAFETY, hi=3000), Claim.limit("b", P_SAFETY, hi=1500)]) check("most restrictive limit wins", r.target_w == 1500) print("the rule that matters: limits only bind claims at or below their priority") # Maintenance charge phase forbids discharging... charge_only = Claim.limit("maintenance", P_MAINT_SHAPE, hi=0, reason="charge-only") r = resolve([Claim.set("loop", P_LOOP, 900), charge_only]) check("charge-only stops the ordinary loop discharging", r.target_w == 0) # ...but peak shaving outranks it and must be able to discharge anyway. r = resolve([Claim.set("loop", P_LOOP, 900), charge_only, Claim.set("peak", P_PEAK, 2500)]) check("peak shaving is NOT bound by charge-only", r.target_w == 2500) check("peak shaving is the winner", r.winner == "peak") # Safety limits still bind everything, including peak shaving. r = resolve([Claim.set("peak", P_PEAK, 9000), charge_only, Claim.limit("device", P_SAFETY, lo=-5000, hi=5000, reason="rating")]) check("safety limits bind even the highest strategy", r.target_w == 5000) print("safety stops") r = resolve([Claim.set("loop", P_LOOP, 900), Claim.set("safety", P_SAFETY, 0, "inputs missing")]) check("a safety stop beats everything", r.target_w == 0 and r.winner == "safety") r = resolve([Claim.set("maintenance", P_MAINTENANCE, 2500), Claim.set("safety", P_SAFETY, 0, "stopped")]) check("safety stop beats maintenance too", r.target_w == 0) print("contradictions are bugs, not ties") r = resolve([Claim.set("loop", P_LOOP, 500), Claim.limit("x", P_SAFETY, lo=1000), Claim.limit("y", P_SAFETY, hi=200)]) check("impossible limits command 0 W", r.target_w == 0.0) check("and are flagged, not silently clipped", r.contradiction) check("explain() says so", "CONTRADICTION" in r.explain()) print("explainability") r = resolve([Claim.set("loop", P_LOOP, 900), charge_only]) check("explain names winner and binder", "loop" in r.explain() and "maintenance" in r.explain()) print(f" e.g. {r.explain()!r}") print("sign handling") r = resolve([Claim.set("loop", P_LOOP, -3000), Claim.limit("supervised", P_SAFETY, lo=-1000, hi=1000)]) check("charging is clamped by the low limit", r.target_w == -1000) r = resolve([Claim.set("loop", P_LOOP, -200), charge_only]) check("charge-only permits charging", r.target_w == -200) print() if fails: print(f"{len(fails)} FAILED: {', '.join(fails)}") sys.exit(1) print("all checks passed")