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

goodwe_controller/CHANGELOG.md

@@ -1,5 +1,34 @@
 # Changelog
 
+## 0.2.0
+
+Precedence between strategies is now a first-class object instead of an if/else
+ladder, ahead of there being more than three of them.
+
+Every strategy returns a CLAIM each cycle - `set` ("I want X") or `limit` ("the
+result must stay within these bounds") - and `arbiter.py` resolves them by one
+rule:
+
+  1. Highest-priority `set` wins; no claim at all means 0 W.
+  2. Then every `limit` whose priority is >= that set's priority applies, most
+     restrictive first.
+  3. Contradictory limits are a BUG: command 0 W and say so.
+
+Clause 2 is why "money outranks maintenance" is now a consequence of the
+priorities rather than a special case in a Jinja template: the maintenance
+charge-only limit binds the loop, but will not bind a higher-priority peak
+shaving claim when one exists.
+
+- Maintenance shaping (charge-only, cheap-window floor) moved out of the control
+  law. `control.py` is once again only a controller that tracks the meter.
+- The loop computes from the ARBITER's last output, not its own last wish. If
+  something outranked it, that is what the hardware actually did, and tracking
+  anything else makes it jump when it regains control.
+- Every decision is explainable: "loop -> 0 W, limited by maintenance
+  (charge-only)" now appears in the UI and the log, instead of a bare number.
+- Safety limits (device rating, supervised max_w) bind every strategy including
+  the highest, and are still enforced a second time at the point of writing.
+
 ## 0.1.6
 
 Findings from installing this on a live system, replacing a working YAML

goodwe_controller/app/arbiter.py

@@ -0,0 +1,116 @@
+"""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)

goodwe_controller/app/control.py

@@ -44,9 +44,6 @@ def compute(
     actual_w: float,
     tuning: Tuning,
     sat_count: int = 0,
-    *,
-    charge_only: bool = False,
-    charge_floor_w: float = 0.0,
 ) -> Decision:
     """One control cycle. A cycle is one meter update (~5 s on a HomeWizard P1).
 
@@ -83,15 +80,10 @@ def compute(
     else:
         want = prev_w + tuning.gain * grid_w
 
-    # --- maintenance charge shaping ---------------------------------------
+    # ⚠️ Maintenance shaping (charge-only, cheap-window floor) used to live
-    # Applied BEFORE clamp and slew so a forced charge is still rate-limited
+    # here. It now belongs to arbiter.py as limit claims, so that precedence
-    # like any other demand.
+    # between strategies is decided in ONE place. This function is again what it
-    if charge_only:
+    # should be: a controller that knows only about tracking the meter.
-        want = min(want, 0.0)
-        reason = "charge-only"
-        if charge_floor_w > 0:
-            want = min(want, -charge_floor_w)
-            reason = "charge-floor"
 
     # --- ORDER MATTERS: clamp -> slew -> freeze ----------------------------
     # An early draft applied a floor after the clamp and let demand escape it.

goodwe_controller/app/main.py

@@ -32,6 +32,9 @@ from datetime import datetime, timezone
 
 import aiohttp
 
+from .arbiter import (
+    P_LOOP, P_MAINT_SHAPE, P_MAINTENANCE, P_SAFETY, Claim, resolve,
+)
 from .control import Tuning, compute, maintenance_charge_floor, peak_at_risk
 from .hass import HomeAssistant
 from .maintenance import IDLE, MaintConfig, Maintenance
@@ -158,17 +161,17 @@ class Controller:
 
     # -- the cycle -----------------------------------------------------------
     async def cycle(self) -> None:
+        """Gather claims, resolve precedence, write the winner.
+
+        Nothing here decides who wins - arbiter.py does, by one rule. This
+        method's only job is to state honestly what each strategy wants.
+        """
         now = datetime.now(timezone.utc)
 
         result = self.maint.tick(now, self.soc)
         for msg in result.events:
             self.log_event(msg)
 
-        # ⚠️ Fail toward inaction, and do it ACTIVELY. If the inputs are missing
-        # we command 0 rather than replaying the last value. The YAML
-        # implementation this replaces kept replaying its last setpoint when the
-        # meter died - which the hardware watchdog cannot catch, because from the
-        # ESP32's point of view Home Assistant is still talking to it.
         stale_after = float(self.o.get("stale_input_s", 15))
         if not self.inputs_ok:
             bad_for = (now - self.inputs_bad_since).total_seconds() if self.inputs_bad_since else 0.0
@@ -177,52 +180,73 @@ class Controller:
                 # going, and give it a few seconds to come back.
                 self.reason = f"inputs missing {bad_for:.0f}s"
                 return
-            if self.target != 0.0:
-                self.log_event(f"inputs missing for {bad_for:.0f}s - commanding 0 W")
-            self.target, self.reason = 0.0, "inputs-missing"
-            await self.write_setpoint(0.0)
-            return
 
-        if result.owns_setpoint:
+        claims = []
-            # drain / hold / abort: maintenance drives directly.
-            self.target = float(result.setpoint_w or 0.0)
-            self.reason = f"maintenance:{result.phase}"
-            self.sat_count = 0
-            await self.write_setpoint(self.target)
-            return
 
+        # --- safety limits: these bind EVERY strategy, always ----------------
+        # Defence in depth: write_setpoint() clamps to the device range too. The
+        # duplication is deliberate - one of them is the policy, the other is the
+        # last thing between a bug and the hardware.
+        if self.dev_min is not None and self.dev_max is not None:
+            claims.append(Claim.limit("device", P_SAFETY, self.dev_min, self.dev_max, "rating"))
+        max_w = float(self.o.get("max_w", 2000))
+        claims.append(Claim.limit("supervised", P_SAFETY, -max_w, max_w, "max_w"))
+
+        # --- safety stops ----------------------------------------------------
         if not self.auto:
-            self.target, self.reason = 0.0, "stopped"
+            claims.append(Claim.set("safety", P_SAFETY, 0.0, "stopped"))
-            await self.write_setpoint(0.0)
+        elif not self.inputs_ok:
-            return
+            if self.target != 0.0:
+                self.log_event("inputs missing - commanding 0 W")
+            claims.append(Claim.set("safety", P_SAFETY, 0.0, "inputs missing"))
 
-        charge_floor = 0.0
+        # --- maintenance -----------------------------------------------------
-        charge_only = result.charge_only
+        if result.owns_setpoint:
-        if charge_only:
+            claims.append(Claim.set("maintenance", P_MAINTENANCE,
-            # Money outranks the maintenance schedule (see control.peak_at_risk).
+                                    float(result.setpoint_w or 0.0), result.phase))
+            self.sat_count = 0
+        elif result.charge_only:
+            # Money outranks the maintenance schedule: while the quarter-hour
+            # projection is over the cap, the charge-only shaping is simply not
+            # claimed, so the loop can discharge and shave the peak. When a
+            # dedicated peak-shaving strategy arrives it will claim SET at
+            # P_PEAK and outrank this limit without any code here changing.
             if peak_at_risk(self.peak_fc, float(self.o.get("peak_cap_w", 3500))):
-                charge_only = False
+                self.log_event("peak at risk - maintenance charge shaping suspended")
-                self.log_event("peak at risk - suspending charge-only clamp")
+            else:
-            elif self.cheap:
+                hi = 0.0
-                charge_floor = maintenance_charge_floor(
+                reason = "charge-only"
-                    float(self.o.get("maintenance_charge_w", 2500)),
+                if self.cheap:
-                    self.peak_fc,
+                    floor = maintenance_charge_floor(
-                    float(self.o.get("peak_cap_w", 3500)),
+                        float(self.o.get("maintenance_charge_w", 2500)),
-                )
+                        self.peak_fc, float(self.o.get("peak_cap_w", 3500)))
+                    if floor > 0:
+                        hi, reason = -floor, "cheap-window charge"
+                claims.append(Claim.limit("maintenance", P_MAINT_SHAPE, hi=hi, reason=reason))
 
-        decision = compute(
+        # --- the grid-following loop -----------------------------------------
-            prev_w=self.target,
+        if self.auto and self.inputs_ok:
-            grid_w=self.grid,
+            # ⚠️ prev is the ARBITER's last output, not the loop's own last
-            actual_w=self.batt,
+            # wish. If something outranked the loop, that is what the hardware
-            tuning=self.tuning,
+            # actually did, and the controller must track reality or it jumps
-            sat_count=self.sat_count,
+            # the moment it regains control.
-            charge_only=charge_only,
+            decision = compute(
-            charge_floor_w=charge_floor,
+                prev_w=self.target,
-        )
+                grid_w=self.grid,
-        if decision.frozen and self.sat_count < self.tuning.saturation_cycles:
+                actual_w=self.batt,
-            self.log_event(f"saturation freeze ({self.target:.0f} W vs {self.batt:.0f} W)")
+                tuning=self.tuning,
-        self.target, self.sat_count, self.reason = (
+                sat_count=self.sat_count,
-            decision.target_w, decision.sat_count, decision.reason)
+            )
+            if decision.frozen and self.sat_count < self.tuning.saturation_cycles:
+                self.log_event(f"saturation freeze ({self.target:.0f} W vs {self.batt:.0f} W)")
+            self.sat_count = decision.sat_count
+            claims.append(Claim.set("loop", P_LOOP, decision.target_w, decision.reason))
+
+        resolution = resolve(claims)
+        if resolution.contradiction:
+            self.log_event("ARBITER CONTRADICTION - commanding 0 W, see the log")
+        self.target = resolution.target_w
+        self.reason = resolution.explain()
         await self.write_setpoint(self.target)
 
     # -- tasks ---------------------------------------------------------------

goodwe_controller/config.yaml

@@ -1,5 +1,5 @@
 name: GoodWe RS485 Controller
-version: "0.1.6"
+version: "0.2.0"
 slug: goodwe_controller
 description: >-
   Drives a GoodWe ES/BP battery inverter over RS485 by emulating its smart

goodwe_controller/test_arbiter.py

@@ -0,0 +1,99 @@
+"""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")

goodwe_controller/test_control.py

@@ -66,17 +66,8 @@ check("counter resets when tracking resumes", d4.sat_count == 0 and not d4.froze
 d = compute(prev_w=2000, grid_w=-800, actual_w=1000, tuning=t, sat_count=3)
 check("freeze still allows magnitude to fall", d.target_w < 2000)
 
-# Charge-only (maintenance charge phase)
+# Maintenance shaping (charge-only, cheap-window floor) is no longer this
-d = compute(prev_w=500, grid_w=500, actual_w=500, tuning=T, charge_only=True)
+# function's business - it is expressed as limit claims. See test_arbiter.py.
-check("charge-only never discharges", d.target_w <= 0)
-
-d = compute(prev_w=0, grid_w=0, actual_w=0, tuning=T, charge_only=True, charge_floor_w=800)
-check("charge floor pulls at least the floor", d.target_w == -800)
-
-# Charge floor must still respect slew (floor applied BEFORE slew).
-d = compute(prev_w=0, grid_w=0, actual_w=0, tuning=Tuning(slew_w=200),
-            charge_only=True, charge_floor_w=2000)
-check("charge floor is still slew-limited", d.target_w == -200)
 
 # Quantisation
 d = compute(prev_w=0, grid_w=7, actual_w=0, tuning=Tuning(deadband_w=1, step_w=10))