goodwe-addon/goodwe_controller/app/peakguard.py

"""SAFETY-07 - the quarter-peak projection guard: one arbiter `lo` limit at
`P_PEAK` from tariff.py's measured month peak. Pure: no I/O, clock or state.
Why a limit, the floor's derivation, blindness, the SoC reserve and the
options: DECISIONS/SAFETY-07-peak-guard.md."""

from dataclasses import dataclass, replace

from . import messages
from .arbiter import ACTUATOR_SITE, P_PEAK, Claim
from .control import peak_at_risk
from .tariff import QUARTER_S

# Why the projection may not be graded; the status page and log say which.
BLIND_NO_BLOCK = "no open 15-minute period, the meter has given no reading yet"
BLIND_COVERAGE = "meter gap"
BLIND_FROZEN = "the meter is answering without changing"

# Where the ceiling came from. Only an OBSERVED reading may relax it.
CEILING_CAP = "peak_cap_w"
CEILING_MONTH_PEAK = "this month's peak"

# SAFETY-24. Why a forced discharge may not be sized.
FLOOR_SOC_UNREADABLE = ("the battery level was not read this cycle, and "
                        "unread is not full")
# The words are "the peak reserve", not "the SoC floor": `peak_soc_floor`
# bounds this guard only (SAFETY-25).
FLOOR_AT_SOC_FLOOR = "the battery level is at or below the peak reserve"

SOURCE = "peak"

# Mirror of config.yaml's peak options, pinned to the file by
# test_peakguard.py. Not a read: `decide()` takes them as arguments.
PEAK_OPTION_DEFAULTS = {
    # Percent a forced discharge may not go below. 20 is derived from the
    # reference pack, 300 Ah at 51.2 V with DoD 90 - 13.8 kWh usable (a test
    # pins this wording). See DECISIONS/SAFETY-07-peak-guard.md.
    "peak_soc_floor": 20,
    # The pack's own cut-out in percent; rule 8 of validate.py refuses a
    # `peak_soc_floor` at or below it.
    "bms_soc_floor": 10,
    # SAFETY-79. Watts the guard aims under `peak_cap_w`.
    "peak_margin_w": 50,
}
# SAFETY-79. Range of `peak_margin_w` in config.yaml; test_peakguard.py pins it.
PEAK_MARGIN_RANGE_W = (0, 500)


@dataclass(frozen=True)
class PeakDecision:
    """What the guard concluded this cycle. No `floor_w_or(default)`: turning "not
    measured" into a number is the defect `PeakReading` removes."""

    engaged: bool
    floor_w: float | None            # the `lo` the claim carries; None = no claim
    ceiling_w: float
    ceiling_from: str
    blind: str | None = None
    projected_w: float | None = None
    excess_w: float = 0.0
    wanted_w: float | None = None    # the floor before `achievable_w` clamped it
    target_w: float | None = None    # SAFETY-79: what the guard aims at
    saturated: bool = False
    # SAFETY-24: a forced discharge was wanted and the peak reserve refused it;
    # excludes `saturated`. SAFETY-87: at N>=2 the others may be asked up to
    # `able_w`.
    floor_limited: bool = False
    soc: float | None = None
    soc_floor: float | None = None
    soc_floor_reason: str | None = None
    reason: str = ""
    # SAFETY-87, N>=2: batteries at or under their reserve, and what the others
    # can give.
    held: tuple = ()
    able_w: float | None = None
    # SAFETY-126: power is wanted but the quarter still ends within
    # ALARM_MISS_W of the aim with the batteries at 0 W (`misses_aim()`). Set
    # by main.py.
    within_aim: bool = False

    def explain(self) -> str:
        """Idle: the status row's plain words (UX-08). Engaged: `reason`, the
        technical record with every figure, for the log and the action log."""
        if self.floor_limited:
            return self.reason
        if not self.engaged:
            if self.projected_w is None:
                return messages.row_peak_guard_quiet()
            return messages.row_peak_guard_idle(
                self.projected_w, self.ceiling_w,
                month_peak=self.ceiling_from == CEILING_MONTH_PEAK)
        return self.reason


def blind_kind(blind: str) -> str:
    """Which of the three kinds of blindness, as a key of
    `messages.BLIND_WORDS`, for the status row."""
    if blind == BLIND_FROZEN:
        return "frozen"
    if blind == BLIND_NO_BLOCK:
        return "no_block"
    return "gap"


def quarter_ceiling_w(cap_w, reading) -> tuple[float, str]:
    """The average this quarter may not exceed, and where it came from.
    Only an OBSERVED reading raises it; a failed read keeps the cap (stricter)."""
    cap = float(cap_w)
    if (reading is not None and getattr(reading, "is_observed", False)
            and reading.peak_w is not None and float(reading.peak_w) > cap):
        return float(reading.peak_w), CEILING_MONTH_PEAK
    return cap, CEILING_CAP


def quarter_target_w(cap_w, margin_w, reading) -> float:
    """SAFETY-79. The average the guard AIMS at: `margin_w` under
    `quarter_ceiling_w()`. Reports and the bill keep the real limit (`ceiling_w`)."""
    return quarter_ceiling_w(cap_w, reading)[0] - float(margin_w)


def excess_w(projection, ceiling_w) -> float:
    """Watts that must come off the meter now to land the quarter on the ceiling:
    `900 * (projected - ceiling) / remaining`; 0.0 when `remaining <= 0`."""
    remaining = QUARTER_S - float(projection.elapsed_s)
    if remaining <= 0.0:
        return 0.0
    return QUARTER_S * (float(projection.projected_w) - float(ceiling_w)) / remaining


def uncovered_s(projection) -> float:
    """Seconds of the open block the meter did not cover (`covered_frac` of `elapsed_s`)."""
    return max(0.0, float(projection.elapsed_s)
               * (1.0 - float(projection.covered_frac)))


def blindness(projection, max_hold_s, frozen: bool) -> str | None:
    """Why this projection may not be graded, or None. Checked first: a fresh
    averager's all-zero `Projection` sits under every ceiling."""
    if frozen:
        return BLIND_FROZEN
    if projection.block_start is None:
        return BLIND_NO_BLOCK
    gap = uncovered_s(projection)
    if max_hold_s is not None and gap > float(max_hold_s):
        return (f"{BLIND_COVERAGE}, {gap:.0f} s of this 15-minute period not "
                f"covered, past the {float(max_hold_s):.0f} s a reading is "
                f"held")
    return None


# SAFETY-90. Uncounted seconds below this are float noise, never a gap.
GAP_TOL_S = 1e-3


def gap_w(ceiling_w, *seen_w) -> float:
    """SAFETY-90: the watts a meter gap is counted at. The highest of the
    limit and `seen_w`, the import with the batteries at 0 W (grid + their
    discharge) just before and just after the gap; None entries are skipped.
    DECISIONS/SAFETY-90-peak-guard-after-a-gap.md."""
    return max([float(ceiling_w)] + [float(w) for w in seen_w if w is not None])


def fill_gaps(projection, *, tail_s, max_hold_s, unseen_s, gap_w, unseen_w,
              held_ws=0.0):
    """SAFETY-90: `projection` with the meter-gap seconds the averager did not
    count filled at `gap_w` (those before the first reading, `unseen_s`, at
    `unseen_w`), plus `held_ws`. Returns `(projection, gap s, unseen s)`.
    `tail_s`: up to `max_hold_s` of the held reading is left as before; past
    that nothing is filled. DECISIONS/SAFETY-90-peak-guard-after-a-gap.md."""
    if projection.block_start is None or max_hold_s is None:
        return projection, 0.0, 0.0
    wall = float(projection.elapsed_s)
    tail = min(max(0.0, float(tail_s)), wall)
    if tail > float(max_hold_s):
        return projection, 0.0, 0.0
    covered = wall - uncovered_s(projection)
    missing = wall - covered - tail
    if missing <= GAP_TOL_S:
        missing = 0.0
        if held_ws <= 0.0:
            return projection, 0.0, 0.0
    unseen = min(missing, max(0.0, float(unseen_s)))
    gap = missing - unseen
    add_ws = gap * float(gap_w) + unseen * float(unseen_w) + float(held_ws)
    done = covered + missing
    return (replace(
        projection,
        projected_w=float(projection.projected_w) + add_ws / QUARTER_S,
        elapsed_w=((float(projection.elapsed_w) * covered + add_ws) / done
                   if done > 0.0 else float(projection.elapsed_w)),
        covered_frac=(done / wall if wall > 0.0
                      else float(projection.covered_frac))), gap, unseen)


def discharge_permitted(soc, soc_floor) -> str | None:
    """SAFETY-24. May the guard size a forced discharge? None if so, else the reason.
    `None` for `soc` or `soc_floor` is NOT permission, and neither has a default.
    `<=`: a pack exactly on the floor has spent what the guard may spend."""
    if soc is None:
        return FLOOR_SOC_UNREADABLE
    if soc_floor is None:
        return ("no peak guard SoC floor is configured, so there is nothing to "
                "bound a forced discharge with")
    if float(soc) <= float(soc_floor):
        return FLOOR_AT_SOC_FLOOR
    return None


@dataclass(frozen=True)
class Pack:
    """SAFETY-32. One pack's numbers for the floor question.
    `bms_floor` is its own cut-out percent; None when the roster states none."""

    name: str
    soc: float | None
    bms_floor: float | None


def pack_floor_pct(pack, soc_floor):
    """The percent THIS pack may not be spent below, or None if unknown:
    `max(site EMS floor, its own cut-out)`. An unstated cut-out is None, not 0
    (`discharge_permitted()` reads None as no permission). The reduction is
    written out because it must not depend on validate.py rule 8."""
    if soc_floor is None or pack.bms_floor is None:
        return None
    return max(float(soc_floor), float(pack.bms_floor))


def pack_permissions(packs, soc_floor) -> list[tuple[str, str | None]]:
    """SAFETY-32. Per pack, in roster order: None if it may be spent, else why not.
    Unlike the `min()` over SoC, it keeps which pack and which floor."""
    return [(p.name, discharge_permitted(p.soc, pack_floor_pct(p, soc_floor)))
            for p in packs]


def binding_soc_floor(packs, soc_floor):
    """The one floor `decide()` grades `pack_floor_soc` against: the MAXIMUM of the
    packs' floors, so never more permissive than grading each pack. None if any
    floor is unknown. At N=1 it is `soc_floor` (test_peakguard.py part 12)."""
    if not packs:
        return None if soc_floor is None else float(soc_floor)
    floors = [pack_floor_pct(p, soc_floor) for p in packs]
    if any(f is None for f in floors):
        return None
    return max(floors)


def reserve_split(packs, soc_floor) -> tuple[list, tuple]:
    """SAFETY-87: `(packs the guard may spend, names of those it may not)`,
    each pack graded on its own floor by `pack_permissions()`."""
    perms = pack_permissions(packs, soc_floor)
    return ([p for p, (_n, why) in zip(packs, perms) if why is None],
            tuple(n for n, why in perms if why is not None))


def place_on_able(decision, held, *, able_w, soc, soc_floor) -> PeakDecision:
    """SAFETY-87: `decision` with the batteries in `held` (at or under their
    own reserve) giving nothing for the peak. `able_w`, `soc` and `soc_floor`
    describe the others, as `decide()`'s arguments do for the roster. A
    floor-limited decision is re-taken on them: they are asked for the need
    up to `able_w`, and it stays floor-limited only if that is not enough.
    `soc` and `soc_floor` keep the roster's verdict, as `decide()` echoed it.
    DECISIONS/SAFETY-87-guard-uses-able-batteries.md."""
    held = tuple(held)
    if not decision.floor_limited or decision.wanted_w is None \
            or discharge_permitted(soc, soc_floor) is not None:
        return replace(decision, held=held)
    wanted = float(decision.wanted_w)
    floor = min(wanted, float(able_w))
    head = (f"this 15-minute period heads for {decision.projected_w:.0f} W "
            f"against a {decision.ceiling_w:.0f} W limit "
            f"({decision.ceiling_from}), aiming at {decision.target_w:.0f} W. "
            f"At the peak reserve, so giving nothing for the peak: "
            f"{messages.names(held)}. The others are asked for {floor:+.0f} W")
    if floor >= wanted:
        return replace(decision, floor_limited=False, floor_w=floor,
                       soc_floor_reason=None, held=held, able_w=float(able_w),
                       reason=f"PEAK_GUARD: {head}")
    return replace(decision, floor_w=floor, held=held, able_w=float(able_w),
                   reason=(f"PEAK_GUARD_FLOOR_LIMITED: {head}, all they can "
                           f"give, but {wanted:+.0f} W was needed, so this "
                           f"period may go over the limit"))


def decide(projection, *, batt_w, cap_w, margin_w, reading, achievable_w, soc,
           soc_floor, max_hold_s=None, frozen: bool = False) -> PeakDecision:
    """The whole guard. Pure: no clock, no I/O, no state carried between calls.

    `batt_w` measured battery power, positive = discharge; None invents no floor.
    `achievable_w` largest discharge the arbiter could grant; the floor is
    clamped to it so the claim never produces `lo > hi` (resolves to 0 W).
    `soc` percent; None is floor-limited. `soc_floor`, `margin_w`: required."""
    ceiling, from_where = quarter_ceiling_w(cap_w, reading)
    target = quarter_target_w(cap_w, margin_w, reading)
    aim = "" if target == ceiling else f", aiming at {target:.0f} W"
    ceil_top = abs(float(achievable_w))

    blind = blindness(projection, max_hold_s, frozen)
    if blind is not None:
        # 0.0: nothing trustworthy to size a floor from; forbid charging, which
        # costs no pack energy.
        return PeakDecision(
            engaged=True, floor_w=0.0, ceiling_w=ceiling, ceiling_from=from_where,
            target_w=target, blind=blind, projected_w=None,
            reason=(f"PEAK_GUARD_BLIND: {blind}. The battery may not charge "
                    f"while this 15-minute period cannot be measured, so it "
                    f"is asked for no charging, floor 0 W"))

    projected = float(projection.projected_w)
    if batt_w is None:
        # main.py's P_SAFETY stop already owns a missing input; file nothing.
        return PeakDecision(
            engaged=False, floor_w=None, ceiling_w=ceiling,
            ceiling_from=from_where, target_w=target, projected_w=projected,
            reason="no battery measurement to size a floor from")
    if QUARTER_S - float(projection.elapsed_s) <= 0.0:
        # A block with no seconds left cannot be influenced (`excess_w` answers
        # 0.0).
        return PeakDecision(
            engaged=False, floor_w=None, ceiling_w=ceiling,
            ceiling_from=from_where, target_w=target, projected_w=projected,
            reason="the quarter has no seconds left to influence")

    ex = excess_w(projection, target)
    wanted = float(batt_w) + ex
    if wanted <= 0.0:
        # The test is on the floor (`batt_w + excess`), not on `excess > 0`:
        # excess falls to zero once the battery is removing it, and releasing
        # there cycled the guard 17 times in one quarter (2026-08-20).
        # DECISIONS/SAFETY-07-peak-guard.md.
        return PeakDecision(
            engaged=False, floor_w=None, ceiling_w=ceiling,
            ceiling_from=from_where, target_w=target, projected_w=projected,
            excess_w=ex,
            wanted_w=wanted,
            reason="the load is under the level this quarter can afford")

    # SAFETY-24, only here: `wanted > 0`, so a forced discharge is really asked
    # for.
    denied = discharge_permitted(soc, soc_floor)
    if denied is not None:
        # 0.0, never negative: forbids charging only. It bounds the guard, not
        # the loop.
        return PeakDecision(
            engaged=True, floor_w=0.0, ceiling_w=ceiling,
            ceiling_from=from_where, target_w=target, projected_w=projected,
            excess_w=ex,
            wanted_w=wanted, floor_limited=True, soc=soc, soc_floor=soc_floor,
            soc_floor_reason=denied,
            reason=(
                f"PEAK_GUARD_FLOOR_LIMITED: this 15-minute period heads for "
                f"{projected:.0f} W against a {ceiling:.0f} W limit "
                f"({from_where}){aim}, and {wanted:+.0f} W from the battery "
                f"would hold it. But {denied}"
                + ("" if soc is None else f", at {float(soc):.0f} %")
                + ("" if soc is None or soc_floor is None
                   else f" against {float(soc_floor):.0f} %")
                + ", so no forced discharge and no charging, and this period "
                  "will go over the limit"))

    floor = min(wanted, ceil_top)
    saturated = floor < wanted
    return PeakDecision(
        engaged=True, floor_w=floor, ceiling_w=ceiling, ceiling_from=from_where,
        target_w=target, blind=None, projected_w=projected, excess_w=ex,
        wanted_w=wanted, saturated=saturated, soc=soc, soc_floor=soc_floor,
        reason=(
            f"PEAK_GUARD: this 15-minute period heads for {projected:.0f} W "
            f"against a {ceiling:.0f} W limit ({from_where}){aim}, with "
            f"{QUARTER_S - projection.elapsed_s:.0f} s left, so {ex:.0f} W must "
            f"come off the meter. The battery measured {float(batt_w):+.0f} W "
            f"and is asked for {floor:+.0f} W"
            + (f", the most the inverters give, but {wanted:+.0f} W was needed"
               if saturated else "")))


# SAFETY-81. A quarter must end this many W over the real limit (and for
# ALARM_HOLD_S, so a hob cycling does not count) before "Load too high" fires.
ALARM_MISS_W = 10.0
ALARM_HOLD_S = 10.0


def load_too_high(projection, *, batt_w, ceiling_w, achievable_w) -> bool:
    """SAFETY-81. May "Load too high for battery" fire? Only if, with the battery at
    full power, the quarter still ends over the real limit by `ALARM_MISS_W`.
    No `decide()` gain; `batt_w` is the lower of two cycles' readings.
    DECISIONS/SAFETY-81-load-too-high.md."""
    remaining = QUARTER_S - float(projection.elapsed_s)
    if batt_w is None or remaining <= 0.0:
        return False
    rest = float(projection.held_w) + float(batt_w) - float(achievable_w)
    done_ws = (QUARTER_S * float(projection.projected_w)
               - max(float(projection.held_w), 0.0) * remaining)
    ends_w = (done_ws + max(rest, 0.0) * remaining) / QUARTER_S
    return ends_w - float(ceiling_w) > ALARM_MISS_W


def misses_aim(projection, *, batt_w, target_w) -> bool:
    """SAFETY-126. With the batteries at 0 W from now on, does the quarter end more
    than `ALARM_MISS_W` over the aim?
    DECISIONS/SAFETY-126-landing-error-is-not-an-overrun.md."""
    return load_too_high(projection, batt_w=batt_w, ceiling_w=target_w,
                         achievable_w=0.0)


def charge_headroom_w(decision) -> float | None:
    """SAFETY-45. The hardest this SITE may charge without putting the billed
    quarter over the ceiling, in W; None when nothing measured it (not unlimited).

    It is `wanted_w` read the other way, independent of the battery, so sizing a
    trade to it cannot oscillate. 0.0 when blind or holding. A SIZING, NOT A
    CLAIM: a negative `lo` would cross the maintenance shaping's negative `hi`
    (0 W, battery parked). DECISIONS/SAFETY-45-sizing-the-trade.md section 2b."""
    if decision is None:
        return None
    if decision.blind:
        return 0.0
    if decision.wanted_w is None:
        return None
    return max(0.0, -float(decision.wanted_w))


def charge_room_w(decision, projection) -> float | None:
    """SAFETY-125: `charge_headroom_w()` less the unread seconds counted at the
    last reading, so a late reading is not room for a grid charge.
    DECISIONS/SAFETY-125-grid-charge-under-the-month-peak.md."""
    room = charge_headroom_w(decision)
    if not room or projection is None or projection.block_start is None:
        return room
    left = QUARTER_S - float(projection.elapsed_s)
    if left <= 0.0:
        return room
    late_ws = max(0.0, float(projection.held_w)) * uncovered_s(projection)
    return max(0.0, room - late_ws / left)


def shaping_suspended(decision, peak_forecast_w, cap_w) -> bool:
    """Must the maintenance charge-only shaping stand down this cycle?
    Crossing limits resolve to 0 W, so the rule is suppression, not priority.
    Both terms are needed: the forecast entity is empty on a default install,
    where the measured guard fires. Dropping either parks the battery in the
    billed quarter (test_peakguard.py part 7). The one place this conjunction is
    written, public so the test grades the shipped expression: do not inline it.
    DECISIONS/SAFETY-07-peak-guard.md."""
    if peak_at_risk(peak_forecast_w, float(cap_w)):
        return True
    return decision is not None and decision.engaged


def claim(decision: PeakDecision) -> Claim | None:
    """The decision as one arbiter claim (`lo` only, `hi` stays `+inf`), or None."""
    if not decision.engaged or decision.floor_w is None:
        return None
    # Blind, floor-limited and a genuine zero floor all file `lo = 0 W`; each
    # gets its own reason. On the SITE actuator: the billed quantity is one per
    # site.
    return Claim.limit(SOURCE, P_PEAK, lo=float(decision.floor_w),
                       reason=("blind" if decision.blind
                               else "SoC floor" if decision.floor_limited
                               and not decision.floor_w
                               else "quarter peak"),
                       actuator=ACTUATOR_SITE)