"""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 cap_floor(decision, cap_w): """SAFETY-128: the floor no higher than `cap_w` (the command standing while the readings have not answered it), or `decision` unchanged when None. DECISIONS/SAFETY-128-late-readings.md.""" if (cap_w is None or decision is None or decision.floor_w is None or decision.floor_w <= cap_w): return decision return replace( decision, floor_w=float(cap_w), saturated=False, reason=decision.reason + f", held at {float(cap_w):+.0f} W until the " "readings answer the last power request") 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)