Files
adminandClaude Opus 5 017b798fe6 Precedence arbiter: one rule instead of an if/else ladder
Two controllers writing one actuator is the failure this system exists to
avoid. "Exactly one writer" was true, but only as a convention held up by
careful reading - which does not survive an EV charger and a heat pump wanting
the same battery.

Strategies now return claims and arbiter.py resolves them:

  highest-priority `set` wins (none at all means 0 W), then every `limit` whose
  priority is >= that set's applies, most restrictive first; contradictory
  limits command 0 W and are flagged as the bug they are.

The second clause is the whole point. "Money outranks maintenance" used to be a
hand-written exception inside a Jinja template; it is now a consequence of the
priorities - the charge-only limit binds the loop but cannot bind a
higher-priority peak claim.

Also: maintenance shaping moved out of control.py, which is a controller again
and not a policy engine; the loop now tracks the arbiter's actual output rather
than its own last wish, so it does not jump when it regains control; and every
decision explains itself ("loop -> 0 W, limited by maintenance(charge-only)")
in the UI and the log.

19 new assertions in test_arbiter.py, each one a precedence question someone
will eventually ask in the field. Deployed to the reference site as 0.2.0 and
holding grid within a few watts of zero.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016NckgXecasQb2eSsPYNSW6
2026-08-23 02:45:32 +02:00

117 lines
4.5 KiB
Python

"""Who gets to command the inverter, and why.
Every strategy - the grid-following loop, the maintenance cycle, peak shaving,
the safety stops - stops writing the setpoint directly and instead returns a
CLAIM each cycle. This module resolves the claims into one number.
⚠️ THE REASON THIS EXISTS. Two controllers writing one actuator is the failure
this whole system is built to avoid: it is what the vendor controller did to the
reference site, and §4.4 of the project notes calls it out by name. Before this
module, "exactly one writer" was a convention held up by an if/else ladder and
careful reading. With an EV charger and a heat pump eventually wanting the same
battery, a convention is not good enough - so precedence is now a first-class
object with one rule and a printable explanation.
THE RULE, in full:
1. The highest-priority `set` claim wins. If there is no set claim at all, the
target is 0 W - fail toward inaction, never "hold the last value".
2. Every `limit` claim whose priority is >= the winning set's priority is then
applied. Most restrictive wins.
3. Contradictory limits (lo > hi) are a BUG, not a tie to break: command 0 W
and say so loudly.
Clause 2 is the interesting one. It is what makes "money outranks maintenance"
a consequence of the priorities instead of a special case somebody can forget:
the maintenance charge-only limit binds the loop, but a higher-priority peak
shaving claim simply is not bound by it.
"""
from dataclasses import dataclass, field
from math import inf
# Priorities live here and nowhere else. The moment these become magic numbers
# scattered through the strategies, the whole point of this module is lost.
P_SAFETY = 100 # stopped, inputs missing, shutting down, device limits
P_MAINTENANCE = 60 # maintenance owns the actuator outright (drain, hold)
P_PEAK = 50 # capacity-tariff peak shaving - costs real money
P_MAINT_SHAPE = 40 # maintenance shaping the loop (charge-only)
P_LOOP = 10 # ordinary grid-following
SET = "set"
LIMIT = "limit"
@dataclass(frozen=True)
class Claim:
source: str
priority: int
kind: str = SET
value: float = 0.0 # for SET
lo: float = -inf # for LIMIT
hi: float = inf # for LIMIT
reason: str = ""
@staticmethod
def set(source: str, priority: int, value: float, reason: str = "") -> "Claim":
return Claim(source, priority, SET, value=value, reason=reason)
@staticmethod
def limit(source: str, priority: int, lo: float = -inf, hi: float = inf,
reason: str = "") -> "Claim":
return Claim(source, priority, LIMIT, lo=lo, hi=hi, reason=reason)
@dataclass
class Resolution:
target_w: float
winner: str # which source set the value
bound_by: str | None = None # which limit actually changed it, if any
contradiction: bool = False
considered: list = field(default_factory=list)
def explain(self) -> str:
"""One line an operator can act on - this is the point of the module."""
if self.contradiction:
return f"CONTRADICTION between limits - commanding 0 W ({self.winner})"
if self.bound_by:
return f"{self.winner} -> {self.target_w:.0f} W, limited by {self.bound_by}"
return f"{self.winner} -> {self.target_w:.0f} W"
def resolve(claims: list) -> Resolution:
sets = sorted((c for c in claims if c.kind == SET),
key=lambda c: c.priority, reverse=True)
if sets:
winner = sets[0]
else:
# No strategy asked for anything. That is not "carry on as before" - the
# inverter holds its last command forever, so silence must mean zero.
winner = Claim.set("failsafe", P_SAFETY, 0.0, "no claim was made")
lo, hi = -inf, inf
lo_src = hi_src = None
for c in claims:
if c.kind != LIMIT or c.priority < winner.priority:
continue
if c.lo > lo:
lo, lo_src = c.lo, c
if c.hi < hi:
hi, hi_src = c.hi, c
considered = [f"{c.source}:{c.kind}" for c in claims]
if lo > hi:
return Resolution(0.0, winner.source, bound_by=None,
contradiction=True, considered=considered)
target = max(lo, min(hi, winner.value))
bound_by = None
if target != winner.value:
binder = lo_src if target == lo else hi_src
if binder is not None:
bound_by = f"{binder.source}({binder.reason})" if binder.reason else binder.source
return Resolution(float(target), winner.source, bound_by, False, considered)