"""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?". ⚠️ MUTATING THE MODULE THIS FILE GRADES? PURGE `__pycache__` FIRST - or run `tools/mutation_run.sh`, which does. `python3 -B` alone does NOT save you: it stops bytecode being WRITTEN, not a stale one being READ, and a size-preserving edit reverted inside one second is invisible to CPython's staleness test. That is TEST-15, and it has already turned one live mutation into a false survivor. Read MUTATION.md before you report a survivor. """ import sys from math import inf from app.arbiter import ( ACTUATOR_SITE, CLAIM_SCOPE, P_ARBITRAGE, P_LOOP, P_MAINT_SHAPE, P_MAINTENANCE, P_PEAK, P_SAFETY, SCOPE_SITE, SCOPE_UNIT, UnitBounds, Claim, actuator_for, allocate, pre_resolve, resolve, scope_violations, site_resolve, ) fails = [] def check(name, cond): print(f" {'ok ' if cond else 'FAIL'} {name}") if not cond: fails.append(name) def _raises(fn): """True if `fn()` raises - a check that something is REFUSED, not defaulted.""" try: fn() except Exception: return True return False 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) # ============================================================================= # ARB-03: the actuator specs.md declared, and the three stages # ============================================================================= # ⚠️ EVERY CHECK ABOVE THIS LINE IS UNEDITED, AND THAT IS ITSELF THE FIRST # ACCEPTANCE CRITERION. Not one of them names an actuator, so every claim they # build is a SITE claim - which is exactly the state main.py is in at N=1 - and # they still answer what they answered before the field existed. If a line above # had to be changed to accommodate `actuator`, stage 2 would not be the shipped # arbiter and the refactor would not be pure. print() print("ARB-03 1. the field, and what its absence means") lone = Claim.set("loop", P_LOOP, 900) check("a claim built without naming an actuator carries one anyway", lone.actuator == ACTUATOR_SITE) # ⚠️ THE TYPE ITSELF, NOT ONLY THE TWO HELPERS. Measured: with only the helper # checked, emptying the DATACLASS default survives this suite untouched, because # `Claim.set()` and `Claim.limit()` pass their own default down and the field's # never runs. Anything that builds a `Claim(...)` directly - a future strategy, # a `dataclasses.replace()` - would have got the unspecified actuator that this # card exists to remove. check("...and so does one built through the dataclass directly, which is the " "constructor the two helpers do not cover", Claim("loop", P_LOOP).actuator == ACTUATOR_SITE) check("...and it is an explicit SITE, not None and not an empty string. " "'unspecified' is what the shipped code had, and 'unspecified' is what " "made every claim implicitly about the one and only actuator", isinstance(lone.actuator, str) and lone.actuator != "" and lone.actuator is not None) check("...and `source` is still the first positional argument, which is what " "test_peakguard.py part 10f's AST scan reads off main.py to pin the " "claim-source set. specs.md declares `actuator` first; the tuple position " "is not the part of that declaration that carries meaning, and moving " "`source` off zero would blind the tripwire that guards SAFETY-25", Claim.limit("device", P_SAFETY, -1, 1).source == "device") print() print("ARB-03 2. the split: sixteen sources, each classified where it is BUILT") # The classification is graded against what each claim is ABOUT, restated here # so that the table in arbiter.py and the expectation in a test are not the same # sentence read twice. test_peakguard.py part 10f pins this set against main.py's # own AST, so a ninth source cannot arrive without both files noticing. SITE_SOURCES = { "loop": "regulates grid_w, the P1 measurement - one meter, one site", "peak": "the quarter-hour average offtake AT the grid connection", "arbitrage": "buys and sells at the grid against a day-ahead price", "safety": "'stopped' is the site's switch, 'inputs missing' is the " "site's own meter - neither is about a pack", # SAFETY-31, and it is the entry most easily filed on the wrong side. "circulation": "sum(|Bi|) - |sum(Bi)| is defined OVER THE ROSTER and is " "identically zero for any one unit - there is no fact of " "the matter about which of two opposing inverters is the " "offender, so there is nothing to file it per-unit on", # SAFETY-30, and it is the entry most easily confused with `failsafe` # below, because the same three words describe both. "sign_site": "the SUM of every standing command against the meter. The " "offending number is the SITE target, which no single " "inverter owns - parking one of them would leave the rest " "pushing the same way with the guard satisfied", # SAFETY-78. "drain": "the site target may not go below a pinned maintenance " "drain, or a sibling would charge against it. Filed at N>1 " "only", # SAFETY-84. "charge": "the site target may not go above a cheap-window maintenance " "charge, or a sibling would give against it. Filed at N>1 " "only", # SAFETY-86. "hold": "the site target stays at the peak floor while a holding pack " "gives the shortfall. Filed at N>1 only", } UNIT_SOURCES = { "device": "one inverter's own reported min/max rating", "supervised": "max_w caps what may be WRITTEN to one setpoint entity", "failsafe": "per-unit evidence, per-unit remedy: since SAFETY-30 it " "grades THIS unit's command against THIS unit's OWN " "measured power (0x0518), so what it learns is about one " "device. The site-wide half moved to `sign_site`", "maintenance": "the monthly recalibration is a pack procedure", # OPT-09, and it is the entry a reader will want to check against # SAFETY-25's deferred loop-side reserve, because it has the same shape. "pack": "THIS pack's own state of charge against THIS pack's own " "`bms_soc_floor` - a fact about one battery in one cupboard, " "exactly as `device`'s rating is a fact about one box. It is " "not SAFETY-25's deferred `loop_soc_floor` reserve: no option " "is added, `peak_soc_floor` is not consulted, the SITE target " "is untouched, and main.py does not file it at all at a " "roster of one", # SAFETY-73. "refused": "THIS battery's own reading gave nothing against THIS unit's " "own standing command, so that direction is closed on THIS " "unit and its share goes to the others. Filed at N>1 only", # OPT-11, added on purpose (test_peakguard.py part 10f says why). "mode": "THIS battery's mode, chosen by its owner. A soft unit limit " "that stage 1 keeps apart from the hard ones and stage 3 " "releases when a claim above P_MODE decides the site target", } for _src, _why in sorted(SITE_SOURCES.items()): print(f" site {_src:<11} {_why}") for _src, _why in sorted(UNIT_SOURCES.items()): print(f" unit {_src:<11} {_why}") check("all SIXTEEN claim sources are classified, and only those sixteen", set(CLAIM_SCOPE) == set(SITE_SOURCES) | set(UNIT_SOURCES)) check("the nine that are about the GRID CONNECTION are site claims", {s for s, v in CLAIM_SCOPE.items() if v == SCOPE_SITE} == set(SITE_SOURCES)) check("the seven that are about a PACK, a DEVICE or one write are per-inverter", {s for s, v in CLAIM_SCOPE.items() if v == SCOPE_UNIT} == set(UNIT_SOURCES)) check("a source nobody classified cannot be routed at all - actuator_for() " "raises rather than defaulting to the site, because defaulting to the " "site is the failure this card removes", _raises(lambda: actuator_for("reserve", "inverter-a", ("inverter-a",)))) print() print("ARB-03 3. at N=1 the site IS the unit, so nothing moves") # §6.2's identity: with one inverter the fabricated grid reading and the battery # setpoint are the same number, so there is one actuator and it is both. ONE = ("inverter-a",) TWO = ("inverter-a", "inverter-b") check("with one inverter every source routes to the site, per-inverter ones " "included - that is the identity, not the classification being ignored", all(actuator_for(s, "inverter-a", ONE) == ACTUATOR_SITE for s in CLAIM_SCOPE)) check("...and with none configured at all, likewise", all(actuator_for(s, "inverter-a", ()) == ACTUATOR_SITE for s in CLAIM_SCOPE)) check("with TWO the identity breaks and the split becomes real: the seven " "per-inverter sources route to the unit this controller drives, the " "nine site sources stay on the grid connection", {s: actuator_for(s, "inverter-a", TWO) for s in CLAIM_SCOPE} == {**{s: "inverter-a" for s in UNIT_SOURCES}, **{s: ACTUATOR_SITE for s in SITE_SOURCES}}) print() print("ARB-03 4. stage 2 IS the shipped arbiter (AC 3)") # ⚠️ NOT ASSERTED, REPLAYED. Every claim list this file resolves above is run # through BOTH entry points and every field the shipped code reads off a # Resolution is compared. `resolve()` at N=1 is `site_resolve()` on the same # list, so a single edit inside stage 2 would show up in every check above as # well as here - and one that only affects the wrapper shows up only here. REPLAY = [ [], [Claim.set("loop", P_LOOP, 900)], [Claim.set("loop", P_LOOP, 900), Claim.set("maintenance", P_MAINTENANCE, 2500)], [Claim.set("loop", P_LOOP, 4000), Claim.limit("supervised", P_SAFETY, lo=-2000, hi=2000, reason="max_w")], [Claim.set("loop", P_LOOP, 4000), Claim.limit("a", P_SAFETY, hi=3000), Claim.limit("b", P_SAFETY, hi=1500)], [Claim.set("loop", P_LOOP, 900), charge_only], [Claim.set("loop", P_LOOP, 900), charge_only, Claim.set("peak", P_PEAK, 2500)], [Claim.set("peak", P_PEAK, 9000), charge_only, Claim.limit("device", P_SAFETY, lo=-5000, hi=5000, reason="rating")], [Claim.set("loop", P_LOOP, 900), Claim.set("safety", P_SAFETY, 0, "inputs missing")], [Claim.set("loop", P_LOOP, 500), Claim.limit("x", P_SAFETY, lo=1000), Claim.limit("y", P_SAFETY, hi=200)], [Claim.set("loop", P_LOOP, -3000), Claim.limit("supervised", P_SAFETY, lo=-1000, hi=1000)], [Claim.set("loop", P_LOOP, -200), charge_only], [Claim.set("arbitrage", P_ARBITRAGE, -2000, "buy"), Claim.limit("peak", P_PEAK, lo=800.0, reason="quarter peak"), Claim.limit("supervised", P_SAFETY, lo=-2000, hi=2000, reason="max_w")], ] def _shape(r): return (r.target_w, r.winner, r.bound_by, r.contradiction, list(r.considered), r.explain()) _diff = [i for i, cl in enumerate(REPLAY) if _shape(resolve(cl)) != _shape(site_resolve(cl))] print(f" replayed {len(REPLAY)} claim lists through both entry points") check("NOT ONE of them resolves differently - target, winner, binder, " "contradiction flag, the considered list and the operator line are equal " "on every one. At N=1 the three stages are the shipped arbiter with " "bookkeeping either side of it", _diff == [] and len(REPLAY) >= 12) check("...and stage 3 hands the whole target to the one actuator there is, " "which is the N=1 identity rather than an allocation", resolve([Claim.set("loop", P_LOOP, 900)]).shares == {ACTUATOR_SITE: 900.0}) check("...and says it allocated, because at N<=1 there is nothing to divide", resolve([Claim.set("loop", P_LOOP, 900)]).allocated) print() print("ARB-03 5. THE CONTRADICTION THAT CANNOT BE WRITTEN DOWN (AC 4)") # SAFETY-25 §0a, constructed. A per-inverter reserve says "this pack must not # discharge" (`hi = 0`); the peak guard says "the site must deliver at least # 1500 W" (`lo = 1500`). At N=1 those are two bounds on ONE number and they # genuinely cross - which is today's behaviour, and today's hazard, and is left # exactly as it is. Filed on TWO actuators they are bounds on two different # numbers, and there is no expression in which one is the other's counterpart. SITE_SIDE = [Claim.set("loop", P_LOOP, -1500.0, "grid"), Claim.limit("peak", P_PEAK, lo=1500.0, reason="quarter peak")] CROSSED = SITE_SIDE + [Claim.limit("maintenance", P_MAINT_SHAPE, hi=0.0, reason="charge-only")] SPLIT = SITE_SIDE + [Claim.limit("maintenance", P_MAINT_SHAPE, hi=0.0, reason="charge-only", actuator="inverter-b")] r_crossed, r_split = resolve(CROSSED), resolve(SPLIT) print(f" both on one actuator: {r_crossed.explain()}") print(f" the `hi` on inverter-b: {r_split.explain()} " f"inverter-b bounds {r_split.unit_bounds['inverter-b'].explain()}") check("filed on ONE actuator the pair still contradicts and still parks the " "battery at 0 W in the billed quarter. That is the shipped behaviour and " "this card does not change it - at N=1 the two really are bounds on one " "number", r_crossed.contradiction and r_crossed.target_w == 0.0) check("filed on TWO the site keeps its floor: the unit `hi` is not a term of " "the site's `lo > hi` test, so there is no contradiction to flag", not r_split.contradiction and r_split.target_w == 1500.0) check("...and the constraint is not lost, it is CARRIED - inverter-b's own " "interval still says 'do not discharge', for the allocator to honour when " "it divides the 1500 W", r_split.unit_bounds["inverter-b"].hi == 0.0 and r_split.unit_bounds["inverter-b"].hi_src == "maintenance") check("...and the site target is bit-for-bit what the site claims alone " "resolve to, so the unit claim contributed NOTHING to it - which is the " "'cannot be expressed' rather than 'is guarded against'", _shape(site_resolve(SITE_SIDE))[:4] == _shape(r_split)[:4]) # Swept rather than argued: no per-inverter source, at any bound, on any unit, # can reach the site's limits. _SWEEP_N = _SWEEP_BAD = 0 for _src in sorted(UNIT_SOURCES): for _unit in ("inverter-a", "inverter-b"): for _lo, _hi in ((0.0, inf), (-inf, 0.0), (2500.0, inf), (-inf, -2500.0), (900.0, 100.0)): _r = resolve(SITE_SIDE + [Claim.limit(_src, P_MAINT_SHAPE, lo=_lo, hi=_hi, actuator=_unit)]) _SWEEP_N += 1 if _r.contradiction or _r.target_w != 1500.0: _SWEEP_BAD += 1 print(f" swept {_SWEEP_N} (per-inverter source x unit x bound) filings " f"against a live site floor: {_SWEEP_BAD} disturbed the site target") check(f"NOT ONE of the {_SWEEP_N} filings contradicts the site floor or moves " f"the site target - including the last bound of each row, which is a unit " f"contradicting ITSELF (lo 900 > hi 100). A unit's own impossible pair " f"parks that unit; it does not park the site", _SWEEP_BAD == 0 and _SWEEP_N >= 40) _self_bad = resolve(SITE_SIDE + [Claim.limit("maintenance", P_MAINT_SHAPE, lo=900.0, hi=100.0, actuator="inverter-b")]) check("...and that unit says so on its own bounds, so the fault is reported " "where it was made rather than swallowed", _self_bad.unit_bounds["inverter-b"].contradiction and "CONTRADICTION" in _self_bad.unit_bounds["inverter-b"].explain()) print() print("ARB-03 6. stage 1: one unit's claims, one unit's interval") mb = pre_resolve([Claim.limit("device", P_SAFETY, -5000, 5000, "rating"), Claim.limit("supervised", P_SAFETY, -2000, 2000, "max_w"), Claim.limit("maintenance", P_MAINT_SHAPE, hi=0.0, reason="charge-only")]) print(f" a unit filing rating, max_w and charge-only: {mb.explain()}") check("the most restrictive of the unit's own limits is the interval", mb.lo == -2000 and mb.hi == 0.0 and mb.hi_src == "maintenance" and mb.lo_src == "supervised") owned = pre_resolve([Claim.set("maintenance", P_MAINTENANCE, -2500, "charge"), Claim.limit("supervised", P_SAFETY, -2000, 2000, "max_w"), Claim.limit("maintenance", P_MAINT_SHAPE, hi=0.0, reason="charge-only")]) print(f" a unit whose maintenance owns the setpoint: {owned.explain()}") check("a unit's own winning SET gates its own limits exactly as clause 2 does " "at the site - the P_MAINT_SHAPE ceiling is not applied to a " "P_MAINTENANCE drain that outranks it", owned.hi == 2000.0) check("...and the SET's value travels BESIDE the interval, not inside it. " "Folding -2500 into [-2500, -2500] would turn 'clamped to max_w' into " "'contradicts max_w', which is a different answer and a wrong one", owned.pinned_w == -2500 and owned.pinned_by == "maintenance" and owned.gate == P_MAINTENANCE and not owned.contradiction) print() print("ARB-03 7 / ALLOC-01: the seam, and what now occupies it") # ⚠️ THIS PART WAS "the seam is a seam, and NOT an allocator" AND IT GRADED THE # DECLINE. ALLOC-01 filled the seam, so the decline at N>=2 is gone - and the # checks that graded it are RE-AIMED rather than deleted, because the two halves # they were really about are both still true and both still worth a line: the # N<=1 pass-through is byte-identical, and the seam still does not BROADCAST. check("with no units the whole target goes to the site actuator", allocate(1500.0, {}) == ({ACTUATOR_SITE: 1500.0}, True)) check("with one unit the share IS the site target - §6.2's N=1 identity, " "pass-through. ⚠️ A FLOAT, UNROUNDED AND UNCLAMPED: this is the " "compatibility promise every N=1 replay in this repository depends on, " "and the general path below does neither of those things", allocate(1500.0, {"inverter-a": UnitBounds()}) == ({"inverter-a": 1500.0}, True)) check("...and it is byte-identical on a fractional target too, which is where a " "'tidy this into the general path' refactor would show up: the general " "path rounds to the 1 W register, the pass-through must not", allocate(1500.4, {"inverter-a": UnitBounds()}) == ({"inverter-a": 1500.4}, True)) _shares, _ok = allocate(1500.0, {"inverter-a": UnitBounds(), "inverter-b": UnitBounds()}) print(f" two units, ALLOC-01: shares={_shares} allocated={_ok}") check("⚠️ AND THE ONE THING THE SEAM MUST NEVER DO, WHICH IS THE WHOLE OF " "ARCH-01: it does not BROADCAST. Hand the site target to each inverter " "and each supplies it IN FULL - the gain at the meter becomes N times the " "per-unit gain, which admin/goodwe's two-master rig measures as a 4000 W " "peak-to-peak limit cycle at gain 1.0 with not one guard firing. This " "line went red on ARB-03's decline and it goes red on a broadcast, which " "is the point: it grades the OUTPUT, not the mechanism", 1500.0 not in _shares.values()) check("...it allocates instead, and the shares SUM to the site target exactly. " "That is the safety property: sum(s_i) = S is what makes the gain at the " "meter a convex combination of the per-unit gains for ANY split, so the " "weights are a comfort choice and a bad split cannot destabilise the site", _ok is True and sum(_shares.values()) == 1500 and sorted(_shares) == ["inverter-a", "inverter-b"]) check("...and an odd target still sums exactly, which is the half a per-unit " "round() gets wrong: 1501/2 is 750.5 twice, and two roundings of that are " "either 1500 or 1502. Largest remainder gives 751 and 750", sum(allocate(1501.0, {"inverter-a": UnitBounds(), "inverter-b": UnitBounds()})[0].values()) == 1501) check("⚠️ AND THE DECLINE IS NOT GONE, IT IS RE-AIMED. It is still correct when " "no safe split exists at all, and PR #40's reviewer proved that branch was " "unreachable and untested once already - removing its trigger without " "giving it a new one would rebuild that hole. Here: every inverter's " "master stale, so there is no envelope set to allocate into", allocate(1500.0, {"inverter-a": UnitBounds(), "inverter-b": UnitBounds()}, live=()) == ({}, False)) # ⚠️ AND `resolve()` HAS TO HONOUR THE ROSTER IT IS GIVEN, WHICH IS NOT THE SAME # THING AS HAVING A `units` PARAMETER. Left None the roster is DERIVED from the # claims, and that derivation is right only for a caller that files a claim for # every unit every cycle. Ignore the argument and derive it always - `if units # is None:` -> `if True:` - and a unit that filed nothing this cycle simply # CEASES TO EXIST: the site of two looks like a site of one, the seam passes the # whole grid target through to the single inverter that spoke, and ARCH-01's # fault is built with the evidence for it removed. That mutation left this whole # suite green until this check was written (PR #40 review), so it is graded on # the one observation that distinguishes the two - a SILENT unit. _silent = resolve([Claim.set("loop", P_LOOP, 900.0), Claim.limit("device", P_SAFETY, -2000, 2000, "rating", actuator="inverter-a")], units=TWO) print(f" roster of two, only inverter-a files: bounds=" f"{sorted(_silent.unit_bounds)} allocated={_silent.allocated} " f"shares={_silent.shares}") check("resolve() HONOURS the roster it is handed rather than inferring one from " "who happened to speak. inverter-b filed nothing this cycle and is still " "a unit: it gets its own interval, and the site is still a site of TWO, " "so the target is SPLIT across two inverters", sorted(_silent.unit_bounds) == ["inverter-a", "inverter-b"] and _silent.shares == {"inverter-a": 450, "inverter-b": 450}) check("...and that is the whole cost of ignoring the argument, stated as the " "number it would produce: derive the roster from the claims instead and " "the site is a site of ONE, so the seam passes ALL 900 W through to " "inverter-a - the silent inverter's absence read as a smaller site, and " "inverter-a told to do the whole site's work on its own. ⚠️ ALLOC-01 " "makes this mutation WORSE, not better: before, both answers refused, so " "only the roster listing distinguished them; now the wrong one produces a " "plausible number that is twice what inverter-a should be given", resolve([Claim.set("loop", P_LOOP, 900.0), Claim.limit("device", P_SAFETY, -2000, 2000, "rating", actuator="inverter-a")]).shares == {"inverter-a": 900.0}) check("...and left None the roster IS derived from the claims, which is the " "documented default every N=1 replay above depends on - so the parameter " "changes the answer only where the two disagree", resolve([Claim.set("loop", P_LOOP, 900.0), Claim.limit("device", P_SAFETY, -2000, 2000, "rating", actuator="inverter-a")], units=None).allocated is True) print() print("ARB-03 8. verified by breaking it: a claim routed to the wrong stage (AC 6)") # The `peak` floor, misfiled on a unit. It is the site's floor - if it lands on # one inverter's interval the SITE keeps no floor at all, and the loop is free to # go on charging through the quarter that sets the month's bill. mis_peak = SITE_SIDE[:1] + [Claim.limit("peak", P_PEAK, lo=1500.0, reason="quarter peak", actuator="inverter-b")] r_mis = resolve(mis_peak) print(f" peak floor misfiled on inverter-b: {r_mis.explain()} " f"scope_errors={r_mis.scope_errors}") # ⚠️ THE ORDERED PAIR, AND THE CHECK THAT USED TO BE HERE WAS WIDER THAN ITS # OWN CLAIM. It said `scope_violations` "names the source and the actuator it # does not belong on" and then asserted `"peak" in r_mis.scope_errors[0]`. # `arbiter.py:264` builds that sentence as # `f"{c.source}: a SITE claim filed on unit {c.actuator!r} - …"`; transpose the # two - zero bytes, no decision changes, SAFETY-38 census B row B12 - and it # reads "inverter-b: a SITE claim filed on unit 'peak'", naming the misfiled # claim as the actuator and the actuator as the claim. `"peak" in …` is STILL # TRUE under that, so the old line graded neither half of what it claimed. # The operands are read off the CLAIM this test built, not off the message. _mis_c = mis_peak[-1] check("PRECONDITION: the misfiled claim's source and actuator are DIFFERENT " "STRINGS, so a transposition genuinely produces a different sentence", _mis_c.source != _mis_c.actuator and _mis_c.source == "peak" and _mis_c.actuator == "inverter-b") check("the mis-routing is CAUGHT, and scope_violations names the source FIRST " "and the actuator it does not belong on SECOND - the ORDERED PAIR on the " "contiguous literal, because a substring test finds either name under a " "transposition that swaps their roles", len(r_mis.scope_errors) == 1 and r_mis.scope_errors[0].startswith( f"{_mis_c.source}: a SITE claim filed on unit {_mis_c.actuator!r}")) check("...and the TRANSPOSED sentence is not what it says, so the check above " "discriminates rather than decorates", f"{_mis_c.actuator}: a SITE claim filed on unit {_mis_c.source!r}" not in r_mis.scope_errors[0]) check("...and the reason it must be caught, measured rather than argued: with " "the floor on a unit the SITE has no floor, so the loop charges 1500 W " "into the quarter the guard was trying to shave", r_mis.target_w == -1500.0 and resolve(SITE_SIDE).target_w == 1500.0) check("a UNIT source filed on the site at a site that HAS named units is " "caught too - that is the direction that produces `lo > hi`", scope_violations(SITE_SIDE + [ Claim.limit("maintenance", P_MAINT_SHAPE, hi=0.0), Claim.limit("device", P_SAFETY, -1, 1, actuator="inverter-b")]) != []) check("...and at N=1, where every source legitimately routes to the site, the " "same list is clean. The check grades ROUTING, not tidiness", scope_violations(SITE_SIDE + [ Claim.limit("maintenance", P_MAINT_SHAPE, hi=0.0)]) == []) check("an unclassified ninth source is a violation wherever it is filed", scope_violations([Claim.limit("reserve", P_LOOP, hi=0.0)]) != []) check("and the sixteen, routed by actuator_for() at a two-inverter site, are " "clean - which is what says the router and the checker agree", scope_violations( [Claim.limit(s, P_SAFETY, -1, 1, actuator=actuator_for(s, "inverter-a", TWO)) for s in CLAIM_SCOPE]) == []) # --- ALLOC-01: AND NOW SOMETHING HAPPENS ABOUT IT --------------------------- # ⚠️ ARB-03 LEFT THIS DELIBERATELY INERT - `scope_violations()` logged and did # not change the resolution, and nothing graded the consequence. ALLOC-01 had to # decide, and decided the allocator CONSUMES it as a REFUSAL rather than as an # input. The reasoning is at `allocator.plan()`'s REFUSAL 1 and the short form # is: the mis-routed claim has already been partitioned by the time stage 3 # runs, so its effect is baked into S or into somebody's interval; "consuming" # it as an allocation input would mean RE-ROUTING it, which needs a guess about # which inverter it belonged to that nothing here can make. What is left is that # S itself may be wrong, and dividing a number that may be wrong is worse than # not dividing it. print(f" and stage 3's answer to it: {r_mis.allocation.explain()}") check("⚠️ the mis-routing is no longer only a log line: stage 3 REFUSES to " "allocate on a claim list it has been told is mis-routed, so main.py " "fails toward inaction and shouts instead of dividing a number that may " "be wrong. That is the same answer clause 3 already gives a contradiction", r_mis.allocated is False and r_mis.shares == {} and "mis-routed" in r_mis.allocation.refused) check("...and it is NOT consumed as an ENVELOPE, which is the tempting reading " "and the wrong one: a unit-scoped claim that landed on the site names no " "unit, so promoting it means either guessing one or applying one " "inverter's rating to a sibling that does not have it. Nothing here is " "promoted - the refusal carries no shares at all", r_mis.allocation.plans == () and r_mis.allocation.dropped == ()) check("...and a correctly routed cycle is unaffected, which is what says this " "grades ROUTING and not the presence of units: the same site claims with " "the floor where it belongs allocate normally", resolve(SITE_SIDE, units=TWO).allocated is True) print() print("OPT-11 9. the mode layer: a unit `hi` never meets the site `lo`") # A battery's mode is a SOFT unit limit at P_MODE (35). Stage 1 keeps it apart # from the hard limits; stage 3 applies it only while the site target was # decided at or below P_MODE. So the peak guard's site floor (P_PEAK) releases # every mode and SAFETY-25 section 0a's cross-inverter contradiction - 0 W in a # billed period - still cannot be written down. from app.arbiter import MODE_SOURCE, P_MODE # noqa: E402 from app.allocator import MODE_FAULT, MODE_RELEASED # noqa: E402 def _two(loop_w, *, peak_lo=None, modes=None, rating=2000.0, arb=None): """A two-inverter cycle's claims: ratings, each unit's mode, the loop, optionally the peak floor and an arbitrage set.""" cl = [] for u in TWO: cl.append(Claim.limit("supervised", P_SAFETY, -rating, rating, "max_w", actuator=u)) for u, (lo, hi) in (modes or {}).items(): cl.append(Claim.limit(MODE_SOURCE, P_MODE, lo, hi, "mode", actuator=u)) cl.append(Claim.set("loop", P_LOOP, loop_w)) if arb is not None: cl.append(Claim.set("arbitrage", P_ARBITRAGE, arb)) if peak_lo is not None: cl.append(Claim.limit("peak", P_PEAK, lo=peak_lo)) return cl check("P_MODE sits between arbitrage and the peak guard, so a trade stays " "inside each mode and the peak floor releases them all", P_ARBITRAGE < P_MODE < P_PEAK and P_MODE < P_MAINT_SHAPE) _off_b = {"inverter-b": (0.0, 0.0)} r = resolve(_two(-500.0, peak_lo=3000.0, modes=_off_b), units=TWO) print(f" peak floor 3000 W, loop asks -500 W, B off: {r.explain()}, " f"deciding {r.deciding_priority}, shares {r.shares}") check("⭐ at N=2 a mode `hi` of 0 W on B and the peak guard's site `lo` of " "3000 W do NOT contradict: no 0 W stop, the floor decides S (priority " "50), the mode is released, and the shares still sum to S", not r.contradiction and r.allocated and r.target_w == 3000.0 and r.deciding_priority == P_PEAK and sum(r.shares.values()) == 3000 and r.allocation.unmet_w == 0) _forbidden = [c if c.source != MODE_SOURCE else Claim.limit(MODE_SOURCE, P_MODE, c.lo, c.hi, actuator=ACTUATOR_SITE) for c in _two(-500.0, peak_lo=3000.0, modes=_off_b)] _rf = site_resolve([c for c in _forbidden if c.actuator == ACTUATOR_SITE]) check("...and the SAME mode filed as a SITE limit is exactly the fault the " "layer exists to keep out: stage 2 then reads `lo` 3000 > `hi` 0 and " "stops the site at 0 W (so the check above sees the difference)", _rf.contradiction and _rf.target_w == 0.0) check("...and resolve() refuses that mis-filed mode at N=2 (a UNIT claim on " "the site), rather than dividing a number that may be wrong", resolve(_forbidden, units=TWO).allocated is False) # The sweep: every mode range that holds 0 W, every loop ask, every floor. _bad = [] _n = 0 _spans = [(0.0, 0.0), (-2000.0, 0.0), (0.0, 2000.0), (-2000.0, 2000.0), (-700.0, 300.0)] for _ma in _spans: for _mb in _spans: for _loop in (-5000.0, -1200.0, 0.0, 800.0, 3900.0): for _peak in (None, 0.0, 1500.0, 3000.0, 4000.0): _n += 1 _r = resolve(_two(_loop, peak_lo=_peak, modes={"inverter-a": _ma, "inverter-b": _mb}), units=TWO) _al = _r.allocation _sh = list(_r.shares.values()) if (_r.contradiction or not _r.allocated or sum(_sh) + _al.unmet_w != _al.sum_w or (any(x > 0 for x in _sh) and any(x < 0 for x in _sh)) or (_peak is not None and _r.target_w < _peak)): _bad.append((_ma, _mb, _loop, _peak, _r.explain(), _al.explain())) for x in _bad[:5]: print(f" BAD {x}") check(f"over {_n} two-inverter cycles (5x5 mode ranges holding 0 W, 5 loop " f"asks, 5 peak floors) no cycle contradicts, refuses, loses a watt of " f"the sum, splits the sign, or goes under the peak floor", not _bad and _n == 625) check("deciding_priority: the loop alone decides at 10", resolve(_two(800.0), units=TWO).deciding_priority == P_LOOP) check("...a trade decides at 30, so the modes still apply to it", resolve(_two(800.0, arb=-1500.0), units=TWO).deciding_priority == P_ARBITRAGE) check("...a peak floor that does not bind leaves the loop deciding at 10", resolve(_two(3500.0, peak_lo=1000.0), units=TWO).deciding_priority == P_LOOP) check("...and a peak floor that binds decides at 50", resolve(_two(500.0, peak_lo=1000.0), units=TWO).deciding_priority == P_PEAK) _r35 = resolve(_two(3000.0, modes=_off_b), units=TWO) check("with the loop deciding, B's Off mode holds: B gets 0 W and A carries " "what it can, the rest reported as unmet", _r35.shares == {"inverter-a": 2000, "inverter-b": 0} and _r35.allocation.unmet_w == 1000) _ub = pre_resolve([Claim.limit("supervised", P_SAFETY, -2000, 2000), Claim.limit(MODE_SOURCE, P_MODE, 0.0, 0.0)]) check("stage 1 keeps the mode APART from the hard limits: the unit's own " "interval is still its rating and the mode travels beside it", (_ub.lo, _ub.hi, _ub.mode_lo, _ub.mode_hi, _ub.mode_priority) == (-2000, 2000, 0.0, 0.0, P_MODE) and not _ub.contradiction) _ub2 = pre_resolve([Claim.limit("supervised", P_SAFETY, -2000, 2000), Claim.limit(MODE_SOURCE, P_MODE, 500.0, 2000.0)]) check("...so a mode that does not hold 0 W cannot make the unit contradict " "itself at stage 1; stage 3 drops it as a software fault instead", not _ub2.contradiction and allocate(1000.0, {"inverter-a": _ub2, "inverter-b": UnitBounds()})[1] and resolve(_two(1000.0, modes={"inverter-a": (500.0, 2000.0)}), units=TWO).allocation.mode_faults == (("inverter-a", MODE_FAULT),)) _ubm = pre_resolve([Claim.limit("supervised", P_SAFETY, -2000, 2000), Claim.set("maintenance", P_MAINTENANCE, 2500.0, "drain"), Claim.limit(MODE_SOURCE, P_MODE, 0.0, 0.0)]) check("...and a unit's own maintenance SET (60) gates the mode out, as it " "gates every lower limit: maintenance outranks the mode", _ubm.mode_priority is None and _ubm.pinned_by == "maintenance") print() print("OPT-11 10. at N=1 the mode bounds the site target, by the same rule") _one = [Claim.limit("supervised", P_SAFETY, -5000, 5000, "max_w"), Claim.limit(MODE_SOURCE, P_MODE, 0.0, 0.0, "off")] _r1 = resolve(_one + [Claim.set("loop", P_LOOP, 1200.0)], units=ONE) check("the loop deciding: an Off battery's site target is 0 W, named as the " "mode's doing, with no contradiction", _r1.target_w == 0.0 and not _r1.contradiction and _r1.bound_by == "mode(off)" and _r1.allocated) _r1p = resolve(_one + [Claim.set("loop", P_LOOP, 1200.0), Claim.limit("peak", P_PEAK, lo=2000.0)], units=ONE) check("...a binding peak floor releases it (2000 W, no contradiction), " "which at N=1 would otherwise be `lo` 2000 > `hi` 0 and a 0 W stop", _r1p.target_w == 2000.0 and not _r1p.contradiction and _r1p.deciding_priority == P_PEAK) _r1a = resolve(_one + [Claim.set("arbitrage", P_ARBITRAGE, -3000.0)], units=ONE) check("...a trade (30) stays inside it", _r1a.target_w == 0.0) _r1f = resolve([Claim.limit(MODE_SOURCE, P_MODE, 100.0, 900.0), Claim.set("loop", P_LOOP, -1200.0)], units=ONE) check("...and a mode range without 0 W is dropped and reported, never " "obeyed: the target stays the loop's -1200 W", _r1f.target_w == -1200.0 and _r1f.mode_fault is not None) check("...and with no mode claim at all, stage 2's answer is untouched " "(the N=1 path with battery modes off)", resolve([Claim.set("loop", P_LOOP, 1200.0)], units=ONE).target_w == 1200.0) print() print("OPT-11 11. a mode gives way only as far as the peak limit needs") # Review of PR #149. The first cut released the modes only when a limit above # P_MODE bound S. A trade (30) that won above the peak floor then kept the # modes, and the units placed less than the floor: the mode beat the peak # guard (a billed peak). And a peak limit that merely capped a charge released # the modes, so an Off battery was charged from the grid. The rule now: clamp # into what the modes can place, then back into the limits above the mode. _offb = {"inverter-b": (0.0, 0.0)} def _placed(r): return sum(r.shares.values()) _ra = resolve(_two(0.0, arb=3000.0, peak_lo=2000.0, rating=1500.0, modes=_offb)[:-1] + [ Claim.set("arbitrage", P_ARBITRAGE, 3000.0), Claim.limit("peak", P_PEAK, lo=2000.0)], units=TWO) print(f" trade 3000 W over a 2000 W floor, A 1500 W, B off: {_ra.shares}") check("⭐ a trade above the peak floor with B Off: the units still place the " "floor (2000 W), A to its rating and B only the 500 W A cannot give", _placed(_ra) == 2000 and _ra.shares == {"inverter-a": 1500, "inverter-b": 500}) _r1t = resolve(_one + [Claim.set("arbitrage", P_ARBITRAGE, 3000.0), Claim.limit("peak", P_PEAK, lo=2000.0)], units=ONE) check("⭐ ...and at N=1 an Off battery under a trade gives the peak floor " "(2000 W), not 0 W", _r1t.target_w == 2000.0 and _r1t.bound_by == "peak") _rc = resolve(_two(0.0, arb=-3000.0, peak_lo=-1000.0, modes=_offb), units=TWO) check("⭐ a grid charge the peak limit caps at 1000 W does not release B's " "Off: A takes it all and B stays at 0 W", _rc.shares == {"inverter-a": -1000, "inverter-b": 0}) _r1c = resolve(_one + [Claim.set("arbitrage", P_ARBITRAGE, -3000.0), Claim.limit("peak", P_PEAK, lo=-1000.0)], units=ONE) check("...and at N=1 an Off battery is not charged from the grid under a " "capped trade", _r1c.target_w == 0.0) _rp = resolve(_two(500.0, peak_lo=1500.0, modes=_offb), units=TWO) check("a peak floor A can carry alone leaves B Off at 0 W", _rp.shares == {"inverter-a": 1500, "inverter-b": 0} and _rp.target_w == 1500.0) _rall = resolve(_two(0.0, peak_lo=3000.0, modes={"inverter-a": (0.0, 0.0), "inverter-b": (0.0, 0.0)}), units=TWO) check("every battery Off and a 3000 W peak floor: the floor is placed, " "split between them, and nothing contradicts", _placed(_rall) == 3000 and not _rall.contradiction and _rall.allocation.refused is None) import random as _rnd # noqa: E402 _g = _rnd.Random(149) _worse, _needless, _sign, _nn = [], [], [], 0 for _ in range(20000): _units = ("inverter-a", "inverter-b", "inverter-c")[:_g.choice((2, 3))] _cl = [] for _u in _units: _rt = _g.choice((1000.0, 2500.0)) _cl.append(Claim.limit("supervised", P_SAFETY, -_rt, _rt, actuator=_u)) if _g.random() < 0.2: _cl.append(Claim.limit("pack", P_SAFETY, *_g.choice(((0.0, inf), (-inf, 0.0))), actuator=_u)) _cl.append(Claim.limit(MODE_SOURCE, P_MODE, _g.choice((0.0, -_rt)), _g.choice((0.0, _rt)), actuator=_u)) _src = _g.choice(("loop", "arbitrage")) _cl.append(Claim.set(_src, P_LOOP if _src == "loop" else P_ARBITRAGE, _g.uniform(-7000.0, 7000.0))) _pk = _g.uniform(-4000.0, 6000.0) _cl.append(Claim.limit("peak", P_PEAK, lo=_pk)) _r = resolve(_cl, units=_units) _r0 = resolve([c for c in _cl if c.source != MODE_SOURCE], units=_units) _nn += 1 if _r.allocation.refused is not None: _sign.append(_cl) continue if _placed(_r) < _pk - 1 and _placed(_r0) >= _pk - 1: _worse.append(_cl) # An Off unit gives power only when the peak floor is short without it. _off = [c.actuator for c in _cl if c.source == MODE_SOURCE and c.lo == 0.0 and c.hi == 0.0] # What the units whose mode lets them give power can give, within it. _on_cap = sum(max(0.0, min(_r.unit_bounds[u].hi, _r.unit_bounds[u].mode_hi)) for u in _units if _r.unit_bounds[u].mode_hi > 0.0 and _r.unit_bounds[u].lo <= 0.0) if any(_r.shares.get(u, 0) > 0 for u in _off) and _pk <= _on_cap - 1: _needless.append(_cl) check(f"over {_nn} random N=2/3 cycles (modes, packs, a trade or the loop, a " f"peak floor): no split is refused, the placed power never falls under " f"the floor where the same cycle without modes met it, and an Off " f"battery gives power only when the others cannot carry the floor " f"({len(_sign)}, {len(_worse)}, {len(_needless)})", not _sign and not _worse and not _needless) print() if fails: print(f"{len(fails)} FAILED: {', '.join(fails)}") sys.exit(1) print("all checks passed")