goodwe-addon/goodwe_controller/app/arbiter.py

"""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)
# OPT-04. §10.1 priority 3 - price arbitrage. ⚠️ ITS POSITION IS THE SPEC, NOT A
# preference: P_PEAK (50) is ABOVE it so clause 2 lets the capacity-tariff floor
# clamp an arbitrage charge, which is §9.1's "give peak shaving priority over
# arbitrage whenever the two conflict" expressed as arithmetic rather than as a
# rule somebody has to remember. P_MAINT_SHAPE (40) is above it so a speculative
# spread cannot override the pack's balancing cycle. P_LOOP (10) is below it
# because trading against the meter is the entire point - the grid-following
# loop would otherwise cancel every charge the moment it created import.
P_ARBITRAGE = 30      # price arbitrage - trades against the meter on purpose
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)