TEST-10: pin max_w beside the meter_freeze_s key TEST-06 already pinned

max_w and the LIVE_MAX_W literal beside it were the two survivors left
after TEST-06 pinned meter_freeze_s: changing max_w 2000 -> 4500 in
config.yaml passed every check in test_freeze.py and test_tariff.py
GREEN, and LIVE_MAX_W = 5000.0 was a bare literal that silently
disagreed with the shipped max_w (2000) with nothing in the file
saying so. For contrast, deadband_w 15 -> 5 already went RED 2 + 1,
so the pinning technique was known to work - it just was not applied
to the key either side of it.

Three changes, each measured:

1. test_freeze.py gains a check comparing Tuning()'s code default
   (control.py, 2000.0) against SHIPPED_MAX_W, read from config.yaml
   by extending the file's existing _cfg_default() regex reader
   rather than reimplementing test_arbitrage.py's _shipped. Broken
   and confirmed red: max_w: 2000 -> 4500 in config.yaml now fails
   exactly "the code default and the shipped default agree on max_w
   too" and nothing else in either file; reverting returns both files
   to all-green. (AC 1, AC 4)

2. LIVE_MAX_W is renamed LIVE_INSTALL_MAX_W in both test_freeze.py
   and test_tariff.py, each with a comment stating plainly that 5000
   is the live install's own override, not config.yaml's shipped
   max_w (2000) - the defect was a literal that could be misread as
   the shipped default, not the number itself. (AC 2)

3. test_freeze.py states PINNED_KEYS = ("meter_freeze_s", "deadband_w",
   "max_w") in one place as the complete set of config keys this
   suite pins. Deliberately excludes maintenance_* - CFG-03 (merged
   into release/1.0 since this branch started, merged in here) gave
   those keys a separate single source of truth in app/maintenance.py
   and this ticket has no reason to duplicate that pin. Not every key
   in config.yaml is pinned here on purpose: meter_freeze_s,
   deadband_w and max_w are the three this file's mutual-exclusion
   arithmetic (section 2b) actually reasons about together. (AC 3)

Full 17-file suite run green under goodwe-addon-ci after the merge
with release/1.0 (f630ed9, CFG-03). CHANGELOG.md updated under
Unreleased.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

goodwe_controller/CHANGELOG.md

@@ -2,6 +2,28 @@
 
 ## Unreleased
 
+**`max_w` and the `LIVE_MAX_W` literal beside it were unpinned in test_freeze.py
+and test_tariff.py, the same gap TEST-06 closed one key over for
+`meter_freeze_s` (TEST-10).** Measured before the fix: `max_w: 2000 -> 4500`
+in `config.yaml` left every check in both files GREEN, and `LIVE_MAX_W = 5000.0`
+was a bare literal that silently disagreed with the shipped default (2000) with
+nothing saying so — for contrast, `deadband_w: 15 -> 5` already went RED 2 + 1,
+so the pinning technique itself was known to work, just not applied here.
+Fixed three ways. First, `test_freeze.py` gained a check comparing `Tuning()`'s
+code default (`control.py`, 2000.0) against `SHIPPED_MAX_W`, read from
+`config.yaml` by extending the file's existing `_cfg_default()` regex reader
+rather than reimplementing test_arbitrage.py's `_shipped`. Second, the literal
+itself is renamed `LIVE_INSTALL_MAX_W` in both files, with a comment stating
+plainly that 5000 is the live install's own override and not config.yaml's
+shipped `max_w` — the defect was the silent literal, not the number. Third,
+`test_freeze.py` now states `PINNED_KEYS = ("meter_freeze_s", "deadband_w",
+"max_w")` in one place as the complete set of config keys this suite pins,
+deliberately excluding `maintenance_*` (CFG-03 gave those a separate source of
+truth in `app/maintenance.py`). Verified by breaking what the new check
+watches: `max_w: 2000 -> 4500` in `config.yaml` now fails exactly
+`"the code default and the shipped default agree on max_w too"` and nothing
+else, and reverting returns the file to all-green.
+
 **Maintenance option defaults lived in three places that had already disagreed,
 and MAINT-04's tripwire was holding them apart rather than closing them
 (CFG-03).** `config.yaml` shipped `maintenance_charge_w: 2000`, `main.py` carried

goodwe_controller/test_freeze.py

@@ -96,11 +96,17 @@ fails = []
 T0 = datetime(2026, 8, 26, 7, 0, 0, tzinfo=timezone.utc)
 HERE = Path(__file__).resolve().parent
 
-# The live install's numbers, which are the ones the exposure is written at.
-LIVE_MAX_W = 5000.0
+# ⚠️ NOT config.yaml's shipped max_w. This is the LIVE INSTALL's own setting -
+# the schema maximum, 5000 - which this file uses on purpose wherever the
+# scenario under test is "a device smaller than the configured max_w" (freeze.py
+# describes exactly this install). Named LIVE_INSTALL_MAX_W, not the shorter
+# LIVE_MAX_W it used to be called, so it cannot be misread as "the current
+# shipped default" - see SHIPPED_MAX_W below for that number, and the pin two
+# sections down for what stops the two from drifting apart unnoticed (TEST-10).
+LIVE_INSTALL_MAX_W = 5000.0
 
 # --------------------------------------------------------------------------- #
-# THE SHIPPED THRESHOLD, READ FROM config.yaml (TEST-06 / AC3)
+# THE SHIPPED DEFAULTS, READ FROM config.yaml (TEST-06 / TEST-10 / AC3)
 # --------------------------------------------------------------------------- #
 # ⚠️ `meter_freeze_s: 300` used to appear in this file only as the literal 300,
 # retyped six times, so `300 -> 3600` in config.yaml left every check here
@@ -118,9 +124,18 @@ def _cfg_default(key):
     return float(found[0]) if len(found) == 1 else None
 
 
-SHIPPED_HOLD_S = _cfg_default("meter_freeze_s")
-SHIPPED_DEADBAND_W = _cfg_default("deadband_w")
-SHIPPED_MAX_W = _cfg_default("max_w")
+# THE KEYS THIS FILE PINS - one place, so the next key added to config.yaml's
+# options block can be checked against it rather than discovered unread later
+# (TEST-10 / AC3). Deliberately not every key: max_w, deadband_w and
+# meter_freeze_s are the ones this file's arithmetic actually reasons about
+# (the mutual-exclusion argument in section 2b needs all three together).
+# Everything else in config.yaml's options block is out of scope here - see
+# test_validate.py for the general "no default silently drifts" pin, and do
+# NOT add a maintenance_* key here: CFG-03 is moving that default between
+# config.yaml, main.py and maintenance.py concurrently with this ticket.
+PINNED_KEYS = ("meter_freeze_s", "deadband_w", "max_w")
+SHIPPED_HOLD_S, SHIPPED_DEADBAND_W, SHIPPED_MAX_W = (
+    _cfg_default(key) for key in PINNED_KEYS)
 
 
 def check(name, cond):
@@ -129,7 +144,7 @@ def check(name, cond):
         fails.append(name)
 
 
-def tuning(max_w=LIVE_MAX_W):
+def tuning(max_w=LIVE_INSTALL_MAX_W):
     """The shipped defaults from config.yaml, at the live install's max_w."""
     return Tuning(gain=0.6, max_w=max_w, slew_w=1000, deadband_w=15,
                   target_grid_w=-10, step_w=10, saturation_w=500,
@@ -184,34 +199,34 @@ print("SAFETY-20: the rail against the device rating")
 DEV = dict(dev_min=-4700.0, dev_max=4700.0)
 
 check("the rail is the DEVICE's rail, not max_w's, when the device is smaller",
-      rail_w(LIVE_MAX_W, **DEV) == 4700.0 * RAIL_FRAC
-      and rail_w(LIVE_MAX_W, **DEV) != LIVE_MAX_W * RAIL_FRAC)
+      rail_w(LIVE_INSTALL_MAX_W, **DEV) == 4700.0 * RAIL_FRAC
+      and rail_w(LIVE_INSTALL_MAX_W, **DEV) != LIVE_INSTALL_MAX_W * RAIL_FRAC)
 check("a command pinned at the device rating with a meter frozen for an hour "
       "FIRES - this is the ticket, and it returned False before the fix",
-      looks_frozen(3600, 4700, LIVE_MAX_W, cfg, **DEV))
+      looks_frozen(3600, 4700, LIVE_INSTALL_MAX_W, cfg, **DEV))
 check("...and the identical call without the rating is still False, so the "
       "check above is measuring the rating and not something else",
-      not looks_frozen(3600, 4700, LIVE_MAX_W, cfg))
+      not looks_frozen(3600, 4700, LIVE_INSTALL_MAX_W, cfg))
 check("the charge rail moves with it too - the runaway can be either sign",
-      looks_frozen(3600, -4700, LIVE_MAX_W, cfg, **DEV))
+      looks_frozen(3600, -4700, LIVE_INSTALL_MAX_W, cfg, **DEV))
 check("just under the DEVICE rail is still silent, so the fix widened term 2 "
       "by exactly the rating and not into a bare unchanged_s threshold",
-      not looks_frozen(3600, 4700 * RAIL_FRAC - 1, LIVE_MAX_W, cfg, **DEV))
+      not looks_frozen(3600, 4700 * RAIL_FRAC - 1, LIVE_INSTALL_MAX_W, cfg, **DEV))
 check("a device BIGGER than max_w cannot raise the rail - max_w is still a "
       "ceiling the operator set and this is a min(), not a swap",
-      rail_w(LIVE_MAX_W, dev_min=-9000.0, dev_max=9000.0)
-      == LIVE_MAX_W * RAIL_FRAC)
+      rail_w(LIVE_INSTALL_MAX_W, dev_min=-9000.0, dev_max=9000.0)
+      == LIVE_INSTALL_MAX_W * RAIL_FRAC)
 check("an unread rating (ESPHome not up) leaves the old behaviour exactly - "
       "no device clamp is applied on that path either, so max_w IS the ceiling",
-      rail_w(LIVE_MAX_W, dev_min=None, dev_max=None) == LIVE_MAX_W * RAIL_FRAC)
+      rail_w(LIVE_INSTALL_MAX_W, dev_min=None, dev_max=None) == LIVE_INSTALL_MAX_W * RAIL_FRAC)
 check("one bound alone still lowers it - hass.limits() reads min and max "
       "independently and either can come back None on its own",
-      rail_w(LIVE_MAX_W, dev_max=4000.0) == 4000.0 * RAIL_FRAC)
+      rail_w(LIVE_INSTALL_MAX_W, dev_max=4000.0) == 4000.0 * RAIL_FRAC)
 check("a non-finite rating is IGNORED, never read as a 0 W ceiling",
-      rail_w(LIVE_MAX_W, dev_min=float("nan"), dev_max=float("inf"))
-      == LIVE_MAX_W * RAIL_FRAC)
+      rail_w(LIVE_INSTALL_MAX_W, dev_min=float("nan"), dev_max=float("inf"))
+      == LIVE_INSTALL_MAX_W * RAIL_FRAC)
 check("the ceiling helper says what it measured", 
-      rail_ceiling_w(LIVE_MAX_W, **DEV) == 4700.0)
+      rail_ceiling_w(LIVE_INSTALL_MAX_W, **DEV) == 4700.0)
 
 # The degenerate end. A ceiling of 0 W does not disarm term 2, it makes it TRUE
 # OF EVERY COMMAND - including a parked 0 W one - which decays the conjunction
@@ -219,27 +234,27 @@ check("the ceiling helper says what it measured",
 # this controller's own operating point.
 check("a device that can command nothing makes the rail UNREACHABLE, with a "
       "reason for the status page",
-      bool(rail_unreachable(LIVE_MAX_W, dev_min=0.0, dev_max=0.0)))
+      bool(rail_unreachable(LIVE_INSTALL_MAX_W, dev_min=0.0, dev_max=0.0)))
 check("...and a reachable rail says so with an empty string, or the row above "
       "would be permanent",
-      rail_unreachable(LIVE_MAX_W, **DEV) == ""
-      and rail_unreachable(LIVE_MAX_W) == "")
+      rail_unreachable(LIVE_INSTALL_MAX_W, **DEV) == ""
+      and rail_unreachable(LIVE_INSTALL_MAX_W) == "")
 check("...and the predicate is SILENT there rather than degenerately true - "
       "0 W would otherwise satisfy `abs(command) >= 0`",
-      not looks_frozen(3600, 0.0, LIVE_MAX_W, cfg, dev_min=0.0, dev_max=0.0)
-      and not looks_frozen(3600, 5000, LIVE_MAX_W, cfg, dev_min=0.0,
+      not looks_frozen(3600, 0.0, LIVE_INSTALL_MAX_W, cfg, dev_min=0.0, dev_max=0.0)
+      and not looks_frozen(3600, 5000, LIVE_INSTALL_MAX_W, cfg, dev_min=0.0,
                            dev_max=0.0))
 
 # The alarm text is the third surface that carried the untrue number - it
 # outlives the page, in the log and in the HA notification.
 _wd20 = FreezeWatchdog(cfg)
-_wd20.evaluate(3600, 4700, LIVE_MAX_W, T0, **DEV)
-print(f"    at max_w {LIVE_MAX_W:.0f} on a +-4700 W device: "
+_wd20.evaluate(3600, 4700, LIVE_INSTALL_MAX_W, T0, **DEV)
+print(f"    at max_w {LIVE_INSTALL_MAX_W:.0f} on a +-4700 W device: "
       f"{_wd20.reason}")
 check("the alarm reason quotes the rail it actually used and NOT the max_w one "
       "the command could never have reached",
       f"command over {4700 * RAIL_FRAC:.0f} W" in _wd20.reason
-      and f"{LIVE_MAX_W * RAIL_FRAC:.0f} W" not in _wd20.reason)
+      and f"{LIVE_INSTALL_MAX_W * RAIL_FRAC:.0f} W" not in _wd20.reason)
 check("...and it says the device is what capped it, so the operator is not "
       "left wondering why the guard is armed somewhere other than max_w",
       "rating" in _wd20.reason and "max_w" in _wd20.reason)
@@ -277,6 +292,18 @@ check("the code default and the shipped default are the same number - "
 check("this file's `tuning()` really is config.yaml's deadband, so the "
       "mutual-exclusion arithmetic below is about the shipped law",
       tuning().deadband_w == SHIPPED_DEADBAND_W)
+# TEST-10: max_w had no such pin - `max_w: 2000 -> 4500` in config.yaml left
+# every check in this file green, the same gap TEST-06 closed for
+# meter_freeze_s one key over. `Tuning`'s own bare default (control.py) is
+# 2000.0 because that IS the shipped default; comparing the two means a
+# config.yaml edit that is not matched in control.py goes red here, the same
+# way FreezeWatchdogConfig().hold_s catches a drifted meter_freeze_s above.
+# This is deliberately NOT a check against LIVE_INSTALL_MAX_W (5000) - that
+# constant is the live install's own override and is supposed to differ from
+# the shipped default; see its definition for why.
+check("the code default and the shipped default agree on max_w too - "
+      "Tuning() is a default, config.yaml is the ship",
+      Tuning().max_w == SHIPPED_MAX_W)
 
 # The upper bound is money: config.yaml's own comment says 300 s is "still
 # inside the 900 s quarter that bills the capacity tariff". A hold that outlives
@@ -335,7 +362,7 @@ def replay(hold_s, max_w, cycle_floor_s=15.0):
     return wd, peak_cmd, peak_unchanged, fired_at
 
 
-for mw in (2000.0, LIVE_MAX_W):
+for mw in (SHIPPED_MAX_W, LIVE_INSTALL_MAX_W):
     wd_r, peak_cmd, peak_unchanged, fired = replay(SHIPPED_HOLD_S, mw)
     print(f"    max_w {mw:.0f}: worst |command| {peak_cmd:.0f} W (rail "
           f"{rail_w(mw):.0f} W), worst unchanged {peak_unchanged:.1f} s "
@@ -347,7 +374,7 @@ for mw in (2000.0, LIVE_MAX_W):
 # The margin on the OTHER term, as numbers rather than as trust. Both figures
 # are quoted in config.yaml and freeze.py, so they are re-measured every run and
 # a capture swapped underneath them cannot leave the prose saying the old thing.
-_, _, worst_unchanged, _ = replay(SHIPPED_HOLD_S, LIVE_MAX_W)
+_, _, worst_unchanged, _ = replay(SHIPPED_HOLD_S, LIVE_INSTALL_MAX_W)
 gaps, last, prev = [], rows[0]["t"], None
 for r in rows:
     if prev is None or r["pwr"] != prev:
@@ -381,7 +408,7 @@ check(f"the shipped {SHIPPED_HOLD_S:.0f} s is at least 15x the worst of the two"
 # that happens not to move.
 print("TEL-05's low-power case - a converged meter that legitimately repeats")
 tel05 = FreezeWatchdog(FreezeWatchdogConfig(hold_s=SHIPPED_HOLD_S))
-t5 = tuning(LIVE_MAX_W)
+t5 = tuning(LIVE_INSTALL_MAX_W)
 tgt, iw, sc = 0.0, 0.0, 0
 bare_would_fire = False
 tel05_cmd = 0.0
@@ -397,7 +424,7 @@ for k in range(TEL05_S):
     tel05_cmd = max(tel05_cmd, abs(tgt))
     # The REJECTED design, written out so the case can be shown to be real.
     bare_would_fire = bare_would_fire or float(k) >= SHIPPED_HOLD_S
-    tel05.evaluate(float(k), tgt, LIVE_MAX_W, T0 + timedelta(seconds=k))
+    tel05.evaluate(float(k), tgt, LIVE_INSTALL_MAX_W, T0 + timedelta(seconds=k))
 print(f"    {TEL05_S / 60:.0f} min at -14.2 W: worst |command| {tel05_cmd:.0f} W, "
       f"unchanged_s reaches {TEL05_S - 1} s, episodes {tel05.episodes}")
 check("the bare unchanged_s threshold TEL-05 rejected WOULD fire here",
@@ -405,7 +432,7 @@ check("the bare unchanged_s threshold TEL-05 rejected WOULD fire here",
 check("the conjunction does not - this is what the rail term buys",
       tel05.episodes == 0)
 check("...and the command stayed nowhere near the rail",
-      tel05_cmd < rail_w(LIVE_MAX_W))
+      tel05_cmd < rail_w(LIVE_INSTALL_MAX_W))
 
 # The structural version of the same point, re-derived rather than asserted in
 # prose: the slowest register a HEALTHY meter can have while still being seen as
@@ -421,7 +448,7 @@ print(f"    at the shipped {SHIPPED_HOLD_S:.0f} s hold, the slowest healthy "
 check("a healthy meter slow enough to look unchanged is inside the deadband",
       slowest_healthy_w < DEADBAND_W)
 _park = compute(prev_w=0.0, grid_w=-slowest_healthy_w, actual_w=0.0,
-                tuning=tuning(LIVE_MAX_W), sat_count=0, i_w=0.0)
+                tuning=tuning(LIVE_INSTALL_MAX_W), sat_count=0, i_w=0.0)
 check("...and the shipped law really does park the command there",
       _park.target_w == 0.0)
 
@@ -500,7 +527,7 @@ while True:
 '''
 
 
-async def live_run(port, mode, hold_s=10.0, seconds=16.0, max_w=LIVE_MAX_W):
+async def live_run(port, mode, hold_s=10.0, seconds=16.0, max_w=LIVE_INSTALL_MAX_W):
     """Poll a real HTTP meter through the real transport and drive the detector."""
     ingest = P1Ingest(phases=1, max_age_s=30)
     wd = FreezeWatchdog(FreezeWatchdogConfig(hold_s=hold_s))
@@ -583,7 +610,7 @@ fired2, polls2, unch2, cmd2 = with_server(
 print(f"    moving: {polls2} polls, worst unchanged {unch2:.1f} s, "
       f"worst |command| {cmd2:.0f} W, alarm at "
       f"{'never' if fired2 is None else f'{fired2:.1f} s'}")
-check("the moving meter also drove the command to the rail", cmd2 >= rail_w(LIVE_MAX_W))
+check("the moving meter also drove the command to the rail", cmd2 >= rail_w(LIVE_INSTALL_MAX_W))
 check("a REAL railed condition against a live meter stays silent", fired2 is None)
 check("...because the meter kept moving", unch2 < 10.0)
 
@@ -671,7 +698,7 @@ STALE_INPUT_S = 15.0
 def _opts(**over):
     o = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
          "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
-         "meter_source": "off", "max_w": LIVE_MAX_W, "gain": 0.6,
+         "meter_source": "off", "max_w": LIVE_INSTALL_MAX_W, "gain": 0.6,
          "slew_w": 1000, "deadband_w": SHIPPED_DEADBAND_W,
          "target_grid_w": -10, "step_w": 10, "heartbeat_s": 10,
          "stale_input_s": STALE_INPUT_S}
@@ -811,7 +838,7 @@ check("the event log names it", any("METER FROZEN" in e for e in ctl_f.events))
 # ⚠️ NOTIFY, NOT TRIP. If a later cut turns this into a trip it must come here
 # and change this expectation deliberately, not discover it in the field.
 check("and it does NOT command 0 W - this cut is an alarm only",
-      abs(ctl_f.target) >= rail_w(LIVE_MAX_W))
+      abs(ctl_f.target) >= rail_w(LIVE_INSTALL_MAX_W))
 
 # ⚠️ THE THREE SCENARIOS THAT SAFETY-03 FORCED A DECISION ON, decided one at a
 # time and not reflexively narrowed to green. Measured on the merged tree:
@@ -848,7 +875,7 @@ print(f"    real 8 kW load, moving meter: surface "
       f"freeze {len(notes_of(hass_m))}")
 check("wired up, a real railed load against a moving meter raises nothing FROM "
       "THIS GUARD (the clamp alarm there is SAFETY-03's and is correct)",
-      not notes_of(hass_m) and abs(ctl_m.target) >= rail_w(LIVE_MAX_W))
+      not notes_of(hass_m) and abs(ctl_m.target) >= rail_w(LIVE_INSTALL_MAX_W))
 
 hass_u = wired(None, 4000.0)[0]
 print(f"    uncovered transport: surface "
@@ -902,7 +929,7 @@ check("an uncovered transport (no unchanged_s) raises nothing FROM THIS GUARD, "
 print("WHERE the detector is evaluated in cycle() - the ordering AC1 protects")
 
 _railed = [i for i, (t, _, _) in enumerate(trace_f)
-           if abs(t) >= rail_w(LIVE_MAX_W)]
+           if abs(t) >= rail_w(LIVE_INSTALL_MAX_W)]
 _alarmed = [i for i, (_, _, n) in enumerate(trace_f) if n >= 1]
 _first_railed = _railed[0] if _railed else None
 _first_alarm = _alarmed[0] if _alarmed else None
@@ -972,7 +999,7 @@ print(f"    poll task killed: alarming at {_before[0]:+.0f} W "
                   for i, (t, a, n) in enumerate(_dead)))
 check("the episode really was standing before the poll task died - otherwise "
       "the release below is vacuous",
-      _before[1] and _before[2] == 1 and abs(_before[0]) >= rail_w(LIVE_MAX_W))
+      _before[1] and _before[2] == 1 and abs(_before[0]) >= rail_w(LIVE_INSTALL_MAX_W))
 check("a dead poll task drives the command to 0 W",
       bool(_zeroed) and _zeroed[0] == 0)
 check("...and on THAT SAME cycle the alarm is already clear, because the "
@@ -997,7 +1024,7 @@ check("...and it stays clear for as long as the failsafe holds it there",
 # loop re-rails, and that is a new episode which MUST speak. What must not
 # happen is a second notification stacking up beside the first.
 _re_railed = [i for i, (t, _, _) in enumerate(_revived)
-              if abs(t) >= rail_w(LIVE_MAX_W)]
+              if abs(t) >= rail_w(LIVE_INSTALL_MAX_W)]
 print(f"    transport revived: command re-rails on cycle "
       f"{_re_railed[0] if _re_railed else None}, "
       f"{_ctl_d.freeze_wd.episodes} episode(s), {len(_freeze_notes)} "
@@ -1084,7 +1111,7 @@ check("...and it quotes the hold and the rail it is armed at, derived from the "
       "config the controller was actually built with",
       bool(_rows_c)
       and f"{_ctl_c.freeze_wd.cfg.hold_s:.0f} s unchanged" in _rows_c[0]["text"]
-      and f"{rail_w(LIVE_MAX_W):.0f} W" in _rows_c[0]["text"])
+      and f"{rail_w(LIVE_INSTALL_MAX_W):.0f} W" in _rows_c[0]["text"])
 check("...as an ok row, since a covered armed guard is not a warning",
       bool(_rows_c) and _rows_c[0]["ok"] and not _rows_c[0]["warn"])
 
@@ -1120,16 +1147,16 @@ check("PRECONDITION: the device rating - not max_w - is what stopped the "
       abs(_c20.target) == 4700.0)
 check("...and that command is BELOW the max_w rail, which is why the shipped "
       "guard could never fire on this install",
-      abs(_c20.target) < rail_w(LIVE_MAX_W))
+      abs(_c20.target) < rail_w(LIVE_INSTALL_MAX_W))
 check("...and at or above the device rail, which is why it must",
-      abs(_c20.target) >= rail_w(LIVE_MAX_W, dev_min=-4700.0, dev_max=4700.0))
+      abs(_c20.target) >= rail_w(LIVE_INSTALL_MAX_W, dev_min=-4700.0, dev_max=4700.0))
 check("THE ALARM FIRES through the real cycle() - the rating reaches the "
       "predicate from the controller, not just from a unit call",
       len(_n20) == 1 and _c20.freeze_wd.active)
 check("...and the notification the operator gets quotes the capped rail, "
       "not the max_w one",
       bool(_n20)
-      and f"{rail_w(LIVE_MAX_W, dev_min=-4700.0, dev_max=4700.0):.0f} W"
+      and f"{rail_w(LIVE_INSTALL_MAX_W, dev_min=-4700.0, dev_max=4700.0):.0f} W"
       in str(_n20[0].get("message", "")))
 # p1_enabled is what a homewizard_local install has - set by hand here for the
 # same reason part 4 sets it, see freeze_rows().
@@ -1162,10 +1189,10 @@ check("a device smaller than max_w still gets exactly one freeze-detector row",
 check("...and the page does NOT print the max_w rail the command can never "
       "reach - THE DEFECT, in one assertion",
       bool(_rows_s)
-      and f"{rail_w(LIVE_MAX_W):.0f} W" not in _rows_s[0]["text"])
+      and f"{rail_w(LIVE_INSTALL_MAX_W):.0f} W" not in _rows_s[0]["text"])
 check("...it prints the rail the predicate actually uses",
       bool(_rows_s)
-      and f"{rail_w(LIVE_MAX_W, dev_min=-4700.0, dev_max=4700.0):.0f} W"
+      and f"{rail_w(LIVE_INSTALL_MAX_W, dev_min=-4700.0, dev_max=4700.0):.0f} W"
       in _rows_s[0]["text"])
 check("...and it names the device range as the reason, since the operator "
       "configured max_w and is owed the reason the guard is armed elsewhere",
@@ -1190,7 +1217,7 @@ check("...and THE RAIL THE PAGE ADVERTISES IS ONE THE WIRE CAN ACTUALLY CARRY. "
 check("...and a command at exactly that advertised rail, with the meter frozen, "
       "really does trip the shipped predicate the row is describing",
       _page_rail is not None
-      and looks_frozen(3600, _page_rail, LIVE_MAX_W, cfg,
+      and looks_frozen(3600, _page_rail, LIVE_INSTALL_MAX_W, cfg,
                        dev_min=_ctl_s.dev_min, dev_max=_ctl_s.dev_max))
 check("...as an ok row - a guard armed at a lower rail is still armed, not a "
       "warning", bool(_rows_s) and _rows_s[0]["ok"] and not _rows_s[0]["warn"])
@@ -1205,7 +1232,7 @@ check("...flagged as a warning rather than a silent ok, like the uncovered "
       "transport row it shares",
       bool(_rows_z) and not _rows_z[0]["ok"] and _rows_z[0]["warn"])
 check("...and the predicate agrees with the page rather than firing behind it",
-      not looks_frozen(3600, 0.0, LIVE_MAX_W, cfg,
+      not looks_frozen(3600, 0.0, LIVE_INSTALL_MAX_W, cfg,
                        dev_min=_ctl_z.dev_min, dev_max=_ctl_z.dev_max))
 
 

goodwe_controller/test_tariff.py

@@ -115,10 +115,16 @@ fails = []
 HERE = Path(__file__).resolve().parent
 TMP = Path(tempfile.mkdtemp(prefix="goodwe-tariff-"))
 
-# The live install's numbers: NOTES.md and freeze.py both record max_w 5000
-# there. `peak_cap_w` ships at 3500 in config.yaml and is the SOFT LIMIT AC 3
-# grades against - read out of the shipped file in part 10 rather than typed.
-LIVE_MAX_W = 5000.0
+# ⚠️ NOT config.yaml's shipped max_w (2000). NOTES.md and freeze.py both record
+# that the LIVE INSTALL runs max_w 5000, and that is the number this file uses
+# as the device's power ceiling wherever it needs one - it is named
+# LIVE_INSTALL_MAX_W, not the shorter LIVE_MAX_W, precisely so it cannot be
+# misread as "the current shipped default" (TEST-10; test_freeze.py carries the identical
+# constant, under the identical name, for the identical reason - grep it there
+# for the pin that stops config.yaml's max_w from drifting under it unnoticed).
+# `peak_cap_w` ships at 3500 in config.yaml and is the SOFT LIMIT AC 3 grades
+# against - read out of the shipped file in part 10 rather than typed.
+LIVE_INSTALL_MAX_W = 5000.0
 
 
 def check(name, cond):
@@ -298,7 +304,7 @@ CTRL_W = ctrl_closed[0].offtake_avg_w
 
 # (b) restarted at minute 10, checkpoint restored through a real SQLite file.
 st = fresh_store("restart")
-tr = PeakTracker(st, cap_w=LIVE_MAX_W)
+tr = PeakTracker(st, cap_w=LIVE_INSTALL_MAX_W)
 a1 = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
 feed(a1, 0, RESTART_AT_S)
 check("the checkpoint round-trips through SQLite", st.save_partial(a1.snapshot()))
@@ -335,7 +341,7 @@ OUTAGE_S = 200
 a4 = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
 st4 = fresh_store("outage")
 st4.save_partial(a1.snapshot())
-PeakTracker(st4, cap_w=LIVE_MAX_W).restore(a4, B + timedelta(seconds=RESTART_AT_S + OUTAGE_S))
+PeakTracker(st4, cap_w=LIVE_INSTALL_MAX_W).restore(a4, B + timedelta(seconds=RESTART_AT_S + OUTAGE_S))
 o_closed = feed(a4, RESTART_AT_S + OUTAGE_S, QUARTER_S)
 OUTAGE_UNCOVERED_S = OUTAGE_S - MAX_HOLD_S
 print(f"    a {OUTAGE_S} s outage inside the quarter, of which "
@@ -415,7 +421,7 @@ check("...so a new month starts from its OWN quarters, not from the previous "
 # The restart that straddles the boundary: a checkpoint whose open block is in
 # August, restored when `now` is already September.
 st7 = fresh_store("straddle")
-tr7 = PeakTracker(st7, cap_w=LIVE_MAX_W)
+tr7 = PeakTracker(st7, cap_w=LIVE_INSTALL_MAX_W)
 straddler = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
 for i in range(0, 400, STEP_S):
     straddler.add(sample(AUG_LATE + timedelta(seconds=i), RUN_W))
@@ -443,17 +449,17 @@ print("5. AN UNREADABLE STORE IS NOT A ZERO PEAK  (the acceptance argument)")
 
 # Three stores, three states, and the three must not answer alike.
 good = fresh_store("states")
-good_tr = PeakTracker(good, cap_w=LIVE_MAX_W)
+good_tr = PeakTracker(good, cap_w=LIVE_INSTALL_MAX_W)
 empty_r = good_tr.peak(B)
 
 gone = fresh_store("gone")
-gone_tr = PeakTracker(gone, cap_w=LIVE_MAX_W)
+gone_tr = PeakTracker(gone, cap_w=LIVE_INSTALL_MAX_W)
 gone.record_block(_CB(B, 3000.0, None))
 gone.close()                       # the database goes away under a live reader
 gone_r = gone_tr.peak(B)
 
 never = PeakStore(str(TMP / "no-such-dir" / "x.db"))
-never_tr = PeakTracker(never, cap_w=LIVE_MAX_W)
+never_tr = PeakTracker(never, cap_w=LIVE_INSTALL_MAX_W)
 never_r = never_tr.peak(B)
 
 # A store that is OPEN and whose query FAILS is a third code path - a closed
@@ -463,7 +469,7 @@ broken = fresh_store("broken-table")
 broken.record_block(_CB(B, 3000.0, None))
 broken._db.execute("DROP TABLE monthly_peak")
 broken._db.commit()
-broken_r = PeakTracker(broken, cap_w=LIVE_MAX_W).peak(B)
+broken_r = PeakTracker(broken, cap_w=LIVE_INSTALL_MAX_W).peak(B)
 
 for label, r in (("fresh store, no quarter yet", empty_r),
                  ("store that broke under a live reader", gone_r),
@@ -513,26 +519,26 @@ check("PeakReading exposes no accessor that defaults a missing peak to a "
            if a.startswith(("get", "value")) or a.endswith("_or")])
 
 # The consequence, which is the reason any of this matters.
-print(f"    headroom under a {LIVE_MAX_W:.0f} W limit: "
+print(f"    headroom under a {LIVE_INSTALL_MAX_W:.0f} W limit: "
       f"no quarter yet -> {good_tr.headroom_w(B)}, "
       f"unreadable -> {gone_tr.headroom_w(B)}")
 check("an unreadable store yields NO headroom figure - the disarming default "
       "(full headroom, guard silently off) is unreachable",
       gone_tr.headroom_w(B) is None and never_tr.headroom_w(B) is None)
-check(f"...where a genuinely empty month yields the full {LIVE_MAX_W:.0f} W, "
+check(f"...where a genuinely empty month yields the full {LIVE_INSTALL_MAX_W:.0f} W, "
       f"which is correct and is a different answer",
-      near(good_tr.headroom_w(B), LIVE_MAX_W))
+      near(good_tr.headroom_w(B), LIVE_INSTALL_MAX_W))
 check("the 80 % warning fraction is likewise None when nothing was observed",
       gone_tr.warning_frac(B) is None and good_tr.warning_frac(B) == 0.0)
 
 # And a real peak reduces headroom, so "always None" does not pass this part.
-good.record_block(_CB(B, LIVE_MAX_W * 0.9, None))
-print(f"    after one {LIVE_MAX_W * 0.9:.0f} W quarter: "
+good.record_block(_CB(B, LIVE_INSTALL_MAX_W * 0.9, None))
+print(f"    after one {LIVE_INSTALL_MAX_W * 0.9:.0f} W quarter: "
       f"headroom {good_tr.headroom_w(B):.0f} W, "
       f"warning {good_tr.warning_frac(B) * 100:.0f} % of the limit")
 check("a real observed peak gives a real headroom, so this part is not passed "
       "by a store that answers None to everything",
-      near(good_tr.headroom_w(B), LIVE_MAX_W * 0.1, 1e-6))
+      near(good_tr.headroom_w(B), LIVE_INSTALL_MAX_W * 0.1, 1e-6))
 check(f"...and it is over §9.1's 80 % warning threshold",
       good_tr.warning_frac(B) > 0.8)
 
@@ -541,7 +547,7 @@ check(f"...and it is over §9.1's 80 % warning threshold",
 corrupt_path = TMP / "corrupt.db"
 corrupt_path.write_bytes(b"this is not a database" * 64)
 corrupt = PeakStore(str(corrupt_path))
-c_r = PeakTracker(corrupt, cap_w=LIVE_MAX_W).peak(B)
+c_r = PeakTracker(corrupt, cap_w=LIVE_INSTALL_MAX_W).peak(B)
 print(f"    a corrupt database file -> {c_r.state}")
 check("a corrupt database reads NOT_OBSERVED and does not raise into the "
       "control loop (§6 fail-open)", c_r.state == NOT_OBSERVED)
@@ -746,7 +752,7 @@ under = 0
 for path in REPLAY:
     st9 = fresh_store("replay-" + path.stem)
     ing = P1Ingest(phases=1, max_age_s=MAX_HOLD_S)
-    tr9 = PeakTracker(st9, cap_w=SHIPPED_CAP_W or LIVE_MAX_W)
+    tr9 = PeakTracker(st9, cap_w=SHIPPED_CAP_W or LIVE_INSTALL_MAX_W)
     first = last = None
     with gzip.open(path, "rt", encoding="utf-8") as fh:
         for line in fh:
@@ -760,7 +766,7 @@ for path in REPLAY:
     span_h = (last - first).total_seconds() / 3600.0
     qs = [w for _, w in st9.blocks_in(sorted(months)[0])] if months else []
     peak = max(qs) if qs else 0.0
-    ok_q = sum(1 for w in qs if w <= (SHIPPED_CAP_W or LIVE_MAX_W))
+    ok_q = sum(1 for w in qs if w <= (SHIPPED_CAP_W or LIVE_INSTALL_MAX_W))
     total_q += len(qs)
     under += ok_q
     replay_days += span_h / 24.0
@@ -815,7 +821,7 @@ class _Pub:
 
 class _Hass:
     async def limits(self, entity):
-        return (-LIVE_MAX_W, LIVE_MAX_W, 1.0)
+        return (-LIVE_INSTALL_MAX_W, LIVE_INSTALL_MAX_W, 1.0)
 
     async def number(self, entity, invert=False):
         return None
@@ -831,7 +837,7 @@ def built(db_path, opts=None):
     return Controller(dict(
         {"setpoint_entity": "number.sp", "meter_source": "ha_signed",
          "p1_net_entity": "sensor.net", "meter_phases": 1,
-         "max_w": LIVE_MAX_W, "gain": 0.6, "slew_w": 1000, "deadband_w": 15,
+         "max_w": LIVE_INSTALL_MAX_W, "gain": 0.6, "slew_w": 1000, "deadband_w": 15,
          "target_grid_w": 0, "step_w": 10, "heartbeat_s": 10,
          "stale_input_s": 15}, **(opts or {})),
         _Hass(), _Store(), _Pub(), stop=asyncio.Event(),