admin/goodwe-addon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

UX-11: safety log lines give one plain sentence and the code, the figures go to DEBUG
checks / unit (pull_request) Successful in 9m3s
Details
infra-pin / fresh (pull_request) Successful in 2s
Details

Browse Source 
- Every WARNING/ERROR line that reported a safety reason (direction, delivery,
  stuck meter, stale reading, peak guard, circulation, over-rating) now ends in
  the reason's code (messages.reason_code()). The reason itself, figures and
  limits included, is on a DEBUG line right after it and in the action log.
- The reasons and state.json are unchanged, so an old state file loads and
  shows the same way: every saved reason opens with its code.
- test_messages part 6 checks every WARNING+ line at 25 words as logged; nine
  non-safety families are listed with a reason and capped. The byte pin in
  test_watchdog now pins the meaning, in order.
- The meter logs one WARNING per quiet episode, and INFO when back, so the red
  row's "the add-on log shows why" holds. The Why line past the hold names
  the missing reading. Action log: "25 engagements", "Shadow mode off: ...".

No trigger, latch, threshold, log level, notification id, cadence, row colour
or banner order changed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LyR88rXETvwhTgx6yHE39p
This commit is contained in:
glenn schrooyen
2026-09-27 13:34:30 +02:00
co-authored by Claude Opus 5.5
parent c4032ebc21
commit 29837dfaa7
10 changed files with 425 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
goodwe_controller/test_messages.py
+249 -25
View File
@@ -877,6 +877,8 @@ DATA = {
"main.py:messages.when(brussels(self.tariff_at))",
"main.py:messages.duration(TARIFF_MAX_AGE_S)",
"main.py:messages.count(len(targets), 'inverter')",
"main.py:messages.seconds(self.p1.published_age_s)",
"main.py:messages.seconds(self.p1.max_age_s)",
# maintenance.py, mqtt.py, p1.py
"maintenance.py:phase", "mqtt.py:err", "p1.py:err", "p1.py:self.host",
"p1.py:self.topic", "p1.py:poll_s", "p1.py:ingest.max_age_s",
@@ -1047,20 +1049,35 @@ def _static(spec):
return [t for t in out if len(words(t)) > 1]
# UX-11: a WARNING+ line quotes a safety reason by its code only. The reason,
# with its figures, is on a DEBUG line beside it and in the action log.
SAFETY_REASONS = {
"sign": _sign_texts, "circulation": _circ_texts,
"delivery": _delivery_texts, "clamp": _clamp_texts, "peak": _peak_texts,
"freeze": _freeze_texts, "stale": _stale_texts,
}
def _coded(maker):
"""The codes a WARNING+ line shows for the reasons `maker` builds, plus
the fallback for a restored stop with no reason saved."""
return lambda: ([messages.reason_code(t) for t in maker()]
+ [messages.reason_code("")])
TEXT = {
"main.py:first.sign_wd.reason": _sign_texts,
"main.py:inv.sign_wd.reason": _sign_texts,
"main.py:inv.sign_wd.reason or '(reason not recorded)'": _sign_texts,
"main.py:self.site_sign_wd.reason": _sign_texts,
"main.py:self.site_sign_wd.reason or '(reason not recorded)'": _sign_texts,
"main.py:self.circ_wd.reason": _circ_texts,
"main.py:self.circ_wd.reason or '(reason not recorded)'": _circ_texts,
"main.py:inv.delivery_wd.reason": _delivery_texts,
"main.py:inv.delivery_wd.reason or '(reason not recorded)'": _delivery_texts,
"main.py:self.clamp_wd.reason": _clamp_texts,
"main.py:g.reason": _peak_texts,
"main.py:self.freeze_wd.reason": _freeze_texts,
"main.py:inv.batt_fresh.reason": _stale_texts,
"main.py:messages.reason_code(first.sign_wd.reason)": _coded(_sign_texts),
"main.py:messages.reason_code(inv.sign_wd.reason)": _coded(_sign_texts),
"main.py:messages.reason_code(self.site_sign_wd.reason)":
_coded(_sign_texts),
"main.py:messages.reason_code(self.circ_wd.reason)": _coded(_circ_texts),
"main.py:messages.reason_code(inv.delivery_wd.reason)":
_coded(_delivery_texts),
"main.py:messages.reason_code(self.clamp_wd.reason)": _coded(_clamp_texts),
"main.py:messages.reason_code(g.reason)": _coded(_peak_texts),
"main.py:messages.reason_code(self.freeze_wd.reason)":
_coded(_freeze_texts),
"main.py:messages.reason_code(inv.batt_fresh.reason)": _coded(_stale_texts),
"main.py:rep.explain()": _migration_texts,
"store.py:rep.explain()": _migration_texts,
"main.py:plan.words": _tariff_texts,
@@ -1086,6 +1103,26 @@ TEXT = {
}
# UX-11: a WARNING+ line, as logged with its built text filled in, is at most
# 25 words (STYLE.md). These families are not safety reasons and were not
# rewritten by UX-11. Each is listed with why, the list may only shrink
# (LONG_LINE_CAP), and a listed key that no longer runs long fails.
LONG_LINE_OK = {
"main.py:plan.words": "tariff plan words, TARIFF-11 pins them, follow-up",
"static_tariff.py:read.reason": "price feed reasons, follow-up",
"main.py:rep.explain()": "state migration report, names saved keys",
"store.py:rep.explain()": "state migration report, names saved keys",
"main.py:refusal": "write refusals, a software fault only",
"main.py:' / '.join(resolution.scope_errors)":
"claim scope errors, a software fault only",
"maintenance.py:msg": "maintenance gate reasons, follow-up",
"price_forecast.py:gate.reason": "trading gate reasons, follow-up",
"price_history.py:self.last_error": "price history errors, follow-up",
}
LONG_LINE_CAP = 9
LINE_WORDS = 25
def reason_problems(text):
"""STYLE.md's red-alarm shape, for a text a log line quotes."""
out = []
@@ -1133,10 +1170,12 @@ def _fill(template, fmt_node, key_part, sample):
return template # %-style: filled by the caller
def scan_log_lines(files, text=None, data=None):
def scan_log_lines(files, text=None, data=None, long_ok=None, long_seen=None):
"""-> (unclassified keys, stale keys, [(key, sample, problems)])."""
text = TEXT if text is None else text
data = DATA if data is None else data
long_ok = LONG_LINE_OK if long_ok is None else long_ok
long_seen = set() if long_seen is None else long_seen
seen, bad = set(), []
for fname, tree in files:
for key, call in dynamic_parts(tree, fname):
@@ -1162,6 +1201,11 @@ def scan_log_lines(files, text=None, data=None):
line = PLACEHOLDER.sub("X", line).replace("\0", sample)
p += [f"as logged: {x}" for x in line_problems(line)
+ logged_shape(line) if x not in p]
if len(words(line)) > LINE_WORDS:
long_seen.add(key)
if key not in long_ok:
p.append(f"as logged: over {LINE_WORDS} words "
f"({len(words(line))})")
if p:
bad.append((key, sample, p))
unknown = sorted(seen - set(data) - set(text))
@@ -1171,14 +1215,79 @@ def scan_log_lines(files, text=None, data=None):
_FILES = [(f.name, ast.parse(f.read_text(encoding="utf-8")))
for f in sorted(APP.glob("*.py"))]
_unknown, _stale, _bad = scan_log_lines(_FILES)
_long_seen = set()
_unknown, _stale, _bad = scan_log_lines(_FILES, long_seen=_long_seen)
for k, sample, p in _bad:
print(f" BAD {k}: {'; '.join(p)} :: {sample[:110]!r}")
print(f" BAD {k}: {'; '.join(p)} :: {sample!r}")
check(f"every dynamic part of a WARNING+ line is classified as data or as a "
f"built text (unclassified: {_unknown})", not _unknown)
check(f"...and no classification is left over from a line that is gone "
f"(stale: {_stale})", not _stale)
_n_texts = sum(len(m()) for m in TEXT.values() if m is not None)
check(f"UX-11: no WARNING+ line runs over {LINE_WORDS} words as logged, except "
f"the {len(LONG_LINE_OK)} listed families, and none of them is a safety "
f"reason (cap {LONG_LINE_CAP})",
len(LONG_LINE_OK) <= LONG_LINE_CAP
and not any("reason_code" in k for k in LONG_LINE_OK))
check(f"...and every listed family still runs long, so the list only shrinks "
f"(no longer long: {sorted(set(LONG_LINE_OK) - _long_seen)})",
set(LONG_LINE_OK) <= _long_seen)
def _code_lines_without_evidence(files):
"""WARNING+ calls quoting `messages.reason_code(X)` whose next statement
is not a DEBUG call that logs X itself (UX-11: nothing is lost)."""
out, n = [], 0
for fname, tree in files:
for node in ast.walk(tree):
for body in (getattr(node, f, None) for f in ("body", "orelse")):
if not isinstance(body, list):
continue
for i, st in enumerate(body):
call = st.value if isinstance(st, ast.Expr) else None
if not (isinstance(call, ast.Call)
and LOG_CALL.search(ast.unparse(call.func))):
continue
coded = [a.args[0] for a in call.args
if isinstance(a, ast.Call) and a.args
and ast.unparse(a.func) == "messages.reason_code"]
if not coded:
continue
n += 1
nxt = body[i + 1] if i + 1 < len(body) else None
dbg = (nxt.value if isinstance(nxt, ast.Expr)
and isinstance(nxt.value, ast.Call)
and ast.unparse(nxt.value.func).endswith(".debug")
else None)
want = ast.unparse(coded[0])
if dbg is None or not any(
ast.unparse(a) == want
or ast.unparse(a).startswith(want + " or ")
for a in dbg.args):
out.append(f"{fname}:{call.lineno}")
return n, out
_n_coded, _no_evidence = _code_lines_without_evidence(_FILES)
check(f"every WARNING+ line that quotes a reason code ({_n_coded}) has the "
f"reason itself on a DEBUG line right after it (missing: {_no_evidence})",
_n_coded >= 14 and not _no_evidence)
_n_m, _m_bad = _code_lines_without_evidence([("x.py", ast.parse(
"_LOG.error('Stop: %s.', messages.reason_code(wd.reason))\n"
"_LOG.debug('Something else.')\n"))])
check("...and a line whose DEBUG evidence was dropped is caught",
_n_m == 1 and _m_bad == ["x.py:1"])
_evidence_bad = [(n, t, reason_problems(t)) for n, m in SAFETY_REASONS.items()
for t in m() if reason_problems(t)]
check(f"the safety reasons behind those lines, on the DEBUG line and in the "
f"action log, keep the red-alarm shape ({_evidence_bad})",
not _evidence_bad)
_codes = {n: [messages.reason_code(t) for t in m()]
for n, m in SAFETY_REASONS.items()}
check(f"...and every one carries a code the line can quote ({_codes})",
all(re.fullmatch(r"[A-Z]+(?:_[A-Z]+)+", c)
and c not in (messages.NO_REASON_SAVED, messages.NO_REASON_CODE)
for cs in _codes.values() for c in cs))
check(f"every built text, run for real, obeys the rule, and so does the line "
f"that quotes it ({_n_texts} texts, {len(_bad)} bad)", not _bad)
_counts = [f"{f}:{n.lineno}" for f, t in _FILES for n in ast.walk(t)
@@ -1208,29 +1317,43 @@ _clamp_lines = {
"error, write": (_clamp_line("Power request too high"), _w1),
"warning, loop": (_clamp_line("The battery gives less than asked"), _at),
}
_clamp_shown = {k: (f % (r, messages.times(1)) if f else None)
_clamp_shown = {k: (f % (messages.reason_code(r), messages.times(1))
if f else None)
for k, (f, r) in _clamp_lines.items()}
for k, v in _clamp_shown.items():
print(f" {k}: {v!r} ({len(words(v or ''))} words)")
check("the clamp lines for one inverter are at most 3 sentences and 30 words, "
"and keep OUTPUT_CLAMP and the numbers",
all(v and sentences(v) <= 3 and len(words(v)) <= 30
check("UX-11: the clamp lines are at most 25 words and keep OUTPUT_CLAMP and "
"the count, and the numbers are in the reason the DEBUG line beside "
"them logs",
all(v and len(words(v)) <= LINE_WORDS
and "OUTPUT_CLAMP" in v and "1 time so far" in v
for v in _clamp_shown.values())
and all(x in _clamp_shown["error, loop"]
for x in ("+9000 W", "+5000 W", "+2000 W", "software fault")))
and "+9000" not in v for v in _clamp_shown.values())
and "software fault" in _clamp_shown["error, loop"]
and all(x in _over for x in ("+9000 W", "+5000 W", "+2000 W")))
# Mutations: the scan catches what the 2d scan missed.
_OLD_CLAMP = ("OUTPUT_CLAMP: the loop computed +2148 W, over the 2000 W rating, "
"and it was clamped to +2000 W; the loop's command this cycle was "
"+440 W. This is a BUG SIGNAL, not a limit being used: something "
"upstream computed a setpoint the actuator cannot deliver.")
_mut_text = dict(TEXT, **{"main.py:self.clamp_wd.reason": lambda: [_OLD_CLAMP]})
_mut_text = dict(TEXT, **{"main.py:messages.reason_code(self.clamp_wd.reason)":
lambda: [_OLD_CLAMP]})
_u, _s, _b = scan_log_lines([f for f in _FILES if f[0] == "main.py"],
text=_mut_text)
check("an injected bad reason (the 2d clamp text) is caught, on both clamp "
f"lines ({sorted({k for k, _t, _p in _b if 'clamp' in k})})",
[k for k, _t, _p in _b].count("main.py:self.clamp_wd.reason") == 2)
[k for k, _t, _p in _b].count(
"main.py:messages.reason_code(self.clamp_wd.reason)") == 2)
# UX-11: the line that quoted the whole STALE_READING reason (60 words on
# TEST-23) is caught by the length rule alone.
_old_stale = dict(TEXT, **{"main.py:messages.reason_code(inv.batt_fresh.reason)":
_stale_texts})
_u, _s, _b = scan_log_lines([f for f in _FILES if f[0] == "main.py"],
text=_old_stale)
check("...and the UX-10 stale line, which quoted the whole reason, is caught "
f"for its length ({[p for k, _t, p in _b if 'batt_fresh' in k][:1]})",
any("batt_fresh" in k and any(f"over {LINE_WORDS} words" in x for x in p)
for k, _t, p in _b))
_new_site = ast.parse("_LOG.warning('Something broke: %s', self.new_wd.reason)")
_u, _s, _b = scan_log_lines([("main.py", _new_site)])
check(f"...and a new line quoting a text nobody classified is caught ({_u})",
@@ -1487,6 +1610,9 @@ class _H:
async def call(self, *a, **k):
return True
async def set_number(self, *a, **k):
return True
class _S:
data = {}
@@ -1549,6 +1675,104 @@ _c.why_words = "Keeping the grid near zero: 0 W"
check("...and once a cycle decides again, the line is that decision",
_c.why_words == "Keeping the grid near zero: 0 W")
# UX-11 item 3: past the hold, the Why line says which reading is missing,
# not "A reading is missing: 0 W". Driven through cycle() for real.
_c.inputs_bad_since = _dt.now(_tz.utc) - _td(seconds=20)
asyncio.run(_c.cycle())
check(f"past the hold, the Why line names the missing reading "
f"({_c.why_words!r})",
_c.why_words == "No usable meter reading for over 15 seconds: 0 W")
# UX-11 item 2: the red meter row says the log shows why. A meter that goes
# quiet logs one WARNING per episode, not one per cycle, and INFO when back.
_c.p1_enabled = True
_c.p1 = _p1.P1Ingest(max_age_s=30.0)
def _meter_at(age_s):
import time as _time
_c.p1.submit(_p1.make_sample(_p1.SOURCE_HA, 500.0, 0.0, phases=1,
ingest_mono=_time.monotonic() - age_s))
def _quiet_episodes():
for age in (1, 40, 45, 50, 1, 2, 60, 65):
_meter_at(age)
asyncio.run(_c.read_inputs())
_mq = [r for r in _captured("goodwe", _quiet_episodes)
if "meter" in r.getMessage().lower()]
_mq_warn = [r.getMessage() for r in _mq if r.levelno == logging.WARNING]
_mq_info = [r.getMessage() for r in _mq if r.levelno == logging.INFO]
check(f"a meter that goes quiet logs one WARNING per episode, not per cycle, "
f"and one INFO when readings return ({_mq_warn}, {_mq_info})",
len(_mq_warn) == 2 and _mq_info == ["Meter readings are back."]
and _mq_warn[0].startswith("No meter reading for 40 seconds, over the "
"limit of 30 seconds")
and len(words(_mq_warn[0])) <= LINE_WORDS)
# UX-11 item 1: a state file saved before UX-11 holds the long reasons, and
# they load unchanged. The ERROR line at start shows the code, and the saved
# text goes on the DEBUG line beside it. Old and new files look the same,
# because the saved reason's format did not change and it opens with its code.
_OLD = {
"sign": "SIGN_DISAGREEMENT: commanded -1000 W against a meter reading "
"+1500 W for 180 s continuously (limits: both over 300 W, held "
"over 180 s)",
"delivery": "UNIT_NOT_DELIVERING: inverter-a is standing at +2000 W and "
"delivering +0 W - 2000 W of its share (100 % of it) is not "
"arriving - while inverter-b is not in that state. Nothing "
"this add-on reads explains it - held for 121 s continuously "
"(limit: over 120 s)",
"circ": "CIRCULATING_POWER: 5800 W measured going round between the "
"inverters for 61 s, peaking at 6000 W, over the limits of 250 W "
"and 60 s. The meter cannot see it, because it nets to zero there",
}
class _OldStore(_S):
data = {"unit/inverter-a/sign_tripped": True,
"unit/inverter-a/sign_reason": _OLD["sign"],
"unit/inverter-a/delivery_tripped": True,
"unit/inverter-a/delivery_reason": _OLD["delivery"],
"circ_tripped": True, "circ_reason": _OLD["circ"],
"site_sign_tripped": True, "site_sign_reason": ""}
_restart = []
_rs = _captured("goodwe", lambda: _restart.append(Controller(
{"setpoint_entity": "number.sp", "meter_source": "off", "max_w": 2000},
_H(), _OldStore(), _P(), stop=asyncio.Event())))
_rs_err = [r.getMessage() for r in _rs if r.levelno == logging.ERROR]
_rs_dbg = [r.getMessage() for r in _rs if r.levelno == logging.DEBUG]
for _m in _rs_err:
print(f" {_m!r} ({len(words(_m))} words)")
check("an old state file's long reasons load unchanged, so the stops stay on",
_restart and _restart[0].inverters[0].sign_wd.tripped
and _restart[0].inverters[0].sign_wd.reason == _OLD["sign"]
and _restart[0].inverters[0].delivery_wd.reason == _OLD["delivery"]
and _restart[0].circ_wd.reason == _OLD["circ"])
check("...each ERROR line at start is one plain sentence with the code, at "
f"most {LINE_WORDS} words, and a stop saved without a reason says so",
len(_rs_err) == 4
and all(len(words(m)) <= LINE_WORDS and "(limits" not in m
for m in _rs_err)
and any(m.endswith(": SIGN_DISAGREEMENT.") for m in _rs_err)
and any(m.endswith(": UNIT_NOT_DELIVERING.") for m in _rs_err)
and any(m.endswith(": CIRCULATING_POWER.") for m in _rs_err)
and any(m.endswith(f": {messages.NO_REASON_SAVED}.") for m in _rs_err))
check("...and the saved text, figures and all, is on a DEBUG line beside it",
all(any(t in m for m in _rs_dbg) for t in _OLD.values()))
# UX-11 item 4: the action log's clamp count and the shadow-off entry.
_main_src = (APP / "main.py").read_text(encoding="utf-8")
check("the action log counts 'engagements' with a real plural and the "
"shadow-off entry is plain, not 'engagement(s)' or 'SHADOW MODE OFF -'",
"engagement(s)" not in _main_src and "SHADOW MODE OFF" not in _main_src
and messages.count(25, "engagement") == "25 engagements"
and messages.count(1, "engagement") == "1 engagement")
print()
if fails:
print(f"{len(fails)} FAILED")