adminandClaude Opus 5.5
e0b578756f
Part 6 checked the quoted reason alone, so title + reason + tails logged as 4+ sentences. It now also checks the whole line (at most 3 sentences, none over 30 words); 25 lines failed and are rewritten (one-sentence wrappers, two-sentence circulation and peak reasons, STALE_READING's fixed limits on a DEBUG line). The clamp lines for one inverter are pinned at 30 words with OUTPUT_CLAMP and the numbers kept. Home Assistant's bare '400: Bad Request' while it loads becomes one plain line; still WARNING, throttled, a failed write. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1557 lines
70 KiB
Python
1557 lines
70 KiB
Python
"""UX-08: user-facing text follows STYLE.md. RUN: python3 -B test_messages.py
|
|
|
|
1. Every notification in `app/messages.py` obeys the rule. The test renders
|
|
each one with realistic values and checks the title length, the number of
|
|
sentences and words for its severity, the banned terms, the entity ids and
|
|
the Details line.
|
|
2. No notification is worded anywhere else. An AST scan of `app/` fails any
|
|
`persistent_notification` call whose payload is not built by
|
|
`app/messages.py`.
|
|
3. The status page. Every row and banner in `app/messages.py` obeys the rule
|
|
for its colour: a row one sentence of at most 15 words, a red row up to 3
|
|
sentences and 30 words, a banner at most 12 words.
|
|
THE RATCHET for the kinds not rewritten yet. Every status-row and banner
|
|
literal left in `app/`, every validator refusal, every WARNING+ log line and every option
|
|
name and description in `translations/en.yaml` must either obey its limit
|
|
or appear in `ux08_allowlist.txt`. A new string that breaks the rule and is
|
|
not listed goes red. So does a listed entry that no longer matches a
|
|
failing string, so the list can only shrink: remove the line when you
|
|
rewrite the string.
|
|
4. DOCS.md has a section for every notification, under "## Notifications".
|
|
5. Mutations: each rule is shown to catch a spoiled input (habit 1 of
|
|
MUTATION.md).
|
|
|
|
Run `python3 -B test_messages.py --violations` to print what part 3 sees.
|
|
"""
|
|
|
|
import ast
|
|
import hashlib
|
|
import os
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
sys.path.insert(0, str(HERE / "tools"))
|
|
|
|
from app import messages # noqa: E402
|
|
import translations_lint as TL # noqa: E402
|
|
|
|
fails = []
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
# --- the rule (STYLE.md sections 1 and 4) --------------------------------------
|
|
BANNED = re.compile(
|
|
r"\b(?:(?:SAFETY|TARIFF|TEL|CFG|FCST|OPT|UX|HW|FW|WITNESS|PKG|ARCH)-\d+|"
|
|
r"integrator|anti-?windup|windup|clamp(?:ed|s)?|i_clamped|saturat\w*|"
|
|
r"BUG SIGNAL|latch(?:ed)?|arbiter|allocator|registers?|0x[0-9A-Fa-f]+|"
|
|
r"layer [123]|heartbeat|watchdog|telegrams?|ACKs?|CRC|OTA|Modbus|planner|"
|
|
r"capaciteitstarief|journalctl|ssh|grep|"
|
|
r"\d+\.\d+\.\d+\.\d+|[\w-]+\.(?:py|md|yaml|json|db))\b|§\s?\d", re.I)
|
|
ENTITY = re.compile(r"\b(?:sensor|number|binary_sensor|input_\w+|switch|select)"
|
|
r"\.[a-z0-9_*]+")
|
|
SNAKE = re.compile(r"\b[a-z0-9]+(?:_[a-z0-9]+)+\b")
|
|
SHOUT = re.compile(r"\b[A-Z]{2,}[A-Z0-9]*(?:[\s-]+[A-Z]{2,}[A-Z0-9]*)+\b")
|
|
PLACEHOLDER = re.compile(r"\{[^{}]*\}|%[-+ #0-9.]*[sdrfgx%]")
|
|
|
|
|
|
def words(text):
|
|
return [w for w in re.findall(r"\S+", PLACEHOLDER.sub("X", text))
|
|
if re.search(r"[A-Za-z0-9]", w)]
|
|
|
|
|
|
def sentences(text):
|
|
"""Sentence count: a stop followed by a space and a capital, or the end."""
|
|
t = PLACEHOLDER.sub("X", text.strip())
|
|
t = re.sub(r"\b(e\.g|i\.e|etc)\.", "eg", t)
|
|
return len([s for s in re.split(r"[.!?](?:\s+|$)", t) if s.strip()])
|
|
|
|
|
|
def title_problems(title):
|
|
out = []
|
|
if len(words(title)) > 5:
|
|
out.append(f"title over 5 words ({len(words(title))})")
|
|
if title.lower().startswith("goodwe"):
|
|
out.append("title carries a GoodWe prefix")
|
|
if title.rstrip().endswith("."):
|
|
out.append("title ends in a full stop")
|
|
return out + text_problems(title, keys=False, entities=False)
|
|
|
|
|
|
def text_problems(text, *, keys, entities):
|
|
t = PLACEHOLDER.sub("X", text)
|
|
out = []
|
|
hit = sorted({m.group(0) for m in BANNED.finditer(t)})
|
|
if hit:
|
|
out.append("banned: " + ", ".join(hit[:4]))
|
|
if not entities and ENTITY.search(t):
|
|
out.append("entity id: " + ENTITY.search(t).group(0))
|
|
if not keys and SNAKE.search(ENTITY.sub("", t)):
|
|
out.append("option key: " + SNAKE.search(ENTITY.sub("", t)).group(0))
|
|
if SHOUT.search(t):
|
|
out.append("capitals: " + SHOUT.search(t).group(0))
|
|
return out
|
|
|
|
|
|
def body_problems(body, severity):
|
|
out = []
|
|
n_s, n_w = sentences(body), len(words(body))
|
|
if severity == messages.ALARM:
|
|
if n_s > 3:
|
|
out.append(f"alarm body over 3 sentences ({n_s})")
|
|
if n_w > 70:
|
|
out.append(f"alarm body over 70 words ({n_w})")
|
|
long = [s for s in re.split(r"(?<=[.!?])\s+", body) if len(words(s)) > 30]
|
|
if long:
|
|
out.append(f"sentence over 30 words ({len(words(long[0]))})")
|
|
else:
|
|
if n_s > 1:
|
|
out.append(f"{severity} body over 1 sentence ({n_s})")
|
|
if n_w > 25:
|
|
out.append(f"{severity} body over 25 words ({n_w})")
|
|
if ";" in body:
|
|
out.append("semicolon: split it into sentences or drop a clause")
|
|
return out + text_problems(body, keys=False, entities=False)
|
|
|
|
|
|
def notice_problems(n):
|
|
out = []
|
|
if n.severity not in (messages.ALARM, messages.WARNING, messages.INFO,
|
|
messages.RESOLVED):
|
|
out.append(f"unknown severity {n.severity!r}")
|
|
out += title_problems(n.title) + body_problems(n.body, n.severity)
|
|
if not ends_in_details(n.data()["message"], n.docs):
|
|
out.append("message does not end in its Details line")
|
|
return out
|
|
|
|
|
|
def ends_in_details(message, docs):
|
|
return message.endswith("\n\n" + messages.details(docs))
|
|
|
|
|
|
# ============================================================================== #
|
|
print("1. every notification obeys the rule")
|
|
SAMPLES = messages.samples()
|
|
for n in SAMPLES:
|
|
p = notice_problems(n)
|
|
check(f"{n.severity:8} {n.title!r}: {len(words(n.body))} words, "
|
|
f"{sentences(n.body)} sentence(s){' - ' + '; '.join(p) if p else ''}",
|
|
not p)
|
|
ids = {v for k, v in vars(messages).items() if k.startswith("ID_")}
|
|
seen = {re.sub(r"inverter_b$", "{slug}", n.id) for n in SAMPLES}
|
|
check(f"the samples cover every notification id in the catalogue "
|
|
f"(missing: {sorted(ids - seen)})", ids <= seen)
|
|
check("the payload carries exactly the three keys Home Assistant accepts",
|
|
all(set(n.data()) == {"notification_id", "title", "message"}
|
|
for n in SAMPLES))
|
|
# The Details link (UX-09). Home Assistant 2026.2+ serves add-on pages at
|
|
# /config/app/<slug>/<tab> and 2026.5 dropped /hassio/addon/<slug>/<tab>, so
|
|
# the path follows the version the Supervisor reports; unknown means current.
|
|
_LINE = ("Details: [add-on documentation](/config/app/local_goodwe_controller/"
|
|
"documentation), section “Battery is not keeping up” under "
|
|
"Notifications.")
|
|
os.environ["HOSTNAME"] = "3f2a9c1b7d4e" # a container id, as on TEST-23
|
|
messages.set_site("local_goodwe_controller", "2026.9.1")
|
|
check("with the Supervisor's slug the Details line is exactly the current-HA "
|
|
"link plus the section by name, even when the hostname is a container id",
|
|
SAMPLES[0].data()["message"].endswith("\n\n" + _LINE))
|
|
_link = {v: messages.docs_url("local_goodwe_controller", v)
|
|
for v in ("2026.1.3", "2026.2.0", "2026.4.2", "2026.10.0b1",
|
|
"2025.12.5", None, "rubbish")}
|
|
check(f"Home Assistant 2026.1 and older get the old /hassio path, 2026.2 and "
|
|
f"newer and an unknown version the /config/app path ({_link})",
|
|
_link == {
|
|
"2026.1.3": "/hassio/addon/local_goodwe_controller/documentation",
|
|
"2025.12.5": "/hassio/addon/local_goodwe_controller/documentation",
|
|
"2026.2.0": "/config/app/local_goodwe_controller/documentation",
|
|
"2026.4.2": "/config/app/local_goodwe_controller/documentation",
|
|
"2026.10.0b1": "/config/app/local_goodwe_controller/documentation",
|
|
None: "/config/app/local_goodwe_controller/documentation",
|
|
"rubbish": "/config/app/local_goodwe_controller/documentation"})
|
|
messages.set_site("../../etc", "2026.9.1")
|
|
check("a slug that is not this add-on's shape is not put in a link",
|
|
"](" not in SAMPLES[0].data()["message"])
|
|
messages.set_site(None, None)
|
|
os.environ["HOSTNAME"] = "abcd1234-goodwe-controller"
|
|
check("without the Supervisor's answer the slug comes from a Supervisor-style "
|
|
"hostname",
|
|
"(/config/app/abcd1234_goodwe_controller/documentation)"
|
|
in SAMPLES[0].data()["message"])
|
|
os.environ["HOSTNAME"] = "some-laptop"
|
|
check("...and anywhere else it names the section without a guessed link",
|
|
SAMPLES[0].data()["message"].endswith(
|
|
"\n\nDetails: add-on documentation, section “Battery is not keeping "
|
|
"up” under Notifications."))
|
|
del os.environ["HOSTNAME"]
|
|
|
|
# Where the slug and the version come from: the Supervisor, read once at start.
|
|
import asyncio # noqa: E402
|
|
from app.main import docs_link_facts # noqa: E402
|
|
|
|
|
|
class _Resp:
|
|
def __init__(self, body):
|
|
self.body = body
|
|
|
|
async def __aenter__(self):
|
|
if isinstance(self.body, Exception):
|
|
raise self.body
|
|
return self
|
|
|
|
async def __aexit__(self, *a):
|
|
return False
|
|
|
|
async def json(self, content_type=None):
|
|
return self.body
|
|
|
|
|
|
class _Session:
|
|
def __init__(self, answers):
|
|
self.answers, self.asked, self.timeouts = answers, [], []
|
|
|
|
def get(self, url, **kw):
|
|
self.asked.append((url, kw.get("headers", {}).get("Authorization")))
|
|
self.timeouts.append(kw.get("timeout"))
|
|
return _Resp(self.answers[url])
|
|
|
|
|
|
_ok = _Session({
|
|
"http://supervisor/addons/self/info": {"result": "ok", "data": {
|
|
"slug": "local_goodwe_controller", "version": "0.12.0"}},
|
|
"http://supervisor/info": {"result": "ok", "data": {
|
|
"homeassistant": "2026.9.1", "supervisor": "2026.08.0"}}})
|
|
_down = _Session({"http://supervisor/addons/self/info": OSError("refused"),
|
|
"http://supervisor/info": ValueError("not json")})
|
|
check("at start the add-on asks the Supervisor for its slug and HA's version, "
|
|
"with its own token",
|
|
asyncio.run(docs_link_facts(_ok, "tok")) == {
|
|
"slug": "local_goodwe_controller", "ha_version": "2026.9.1"}
|
|
and all(a == "Bearer tok" for _u, a in _ok.asked))
|
|
check("...and a Supervisor that does not answer costs the link, never the "
|
|
"start",
|
|
asyncio.run(docs_link_facts(_down, "tok")) == {
|
|
"slug": None, "ha_version": None})
|
|
check("...and each question has a short timeout, so a Supervisor that hangs "
|
|
f"delays control start by seconds, not minutes ({_ok.timeouts})",
|
|
len(_ok.timeouts) == 2
|
|
and all(isinstance(t, (int, float)) and 0 < t <= 3 for t in _ok.timeouts))
|
|
|
|
# Shadow mode (UX-09): nothing the add-on decides reaches the inverter, so no
|
|
# text may say a command was sent or that the battery is giving power.
|
|
_c1 = [messages.Clamp(None, 7412, 5000, "rating")]
|
|
_c2 = [messages.Clamp("inverter-a", -3000, -2500, "device"),
|
|
messages.Clamp("inverter-b", 6000, 4600, "max_w+device")]
|
|
_live = [messages.output_clamp(c).body for c in (_c1, _c2, ())]
|
|
_shad = [messages.output_clamp(c, shadow=True).body for c in (_c1, _c2, ())]
|
|
check("live, the over-rating alarm says what was sent and that control runs",
|
|
" and sent 5 000 W" in _live[0] and " and sent 2 500 W" in _live[1]
|
|
and all("Control is still running" in b or "control is still running"
|
|
in b for b in _live))
|
|
check("in shadow mode it says what would have been sent, and that the "
|
|
"inverters stay at 0 W",
|
|
"would have sent 5 000 W" in _shad[0]
|
|
and "would have sent 2 500 W" in _shad[1]
|
|
and "would have been cut back" in _shad[2]
|
|
and all(" sent" not in b.replace("would have sent", "") for b in _shad)
|
|
and all("0 W" in b and "hadow mode" in b for b in _shad))
|
|
_row_live = messages.row_power_too_high(_c1, 3)
|
|
_row_shad = messages.row_power_too_high(_c1, 3, shadow=True)
|
|
check("the over-rating row says 'sent' live and 'would send' in shadow mode",
|
|
"sent 5 000 W" in _row_live and "would" not in _row_live
|
|
and "would send 5 000 W" in _row_shad and " sent" not in _row_shad
|
|
and "shadow mode" in _row_shad)
|
|
_sat = messages.peak_saturated(5000, shadow=True).body
|
|
check("in shadow mode the peak alarm does not claim the battery is giving "
|
|
"power, and keeps its action",
|
|
"giving all it can" not in _sat and "gives nothing" in _sat
|
|
and "Switch off large loads" in _sat
|
|
and "giving all it can" in messages.peak_saturated(5000).body)
|
|
check("shadow mode changes no notification's id or severity",
|
|
all(messages.output_clamp(c, shadow=True).id == messages.ID_OUTPUT_CLAMP
|
|
and messages.output_clamp(c, shadow=True).severity == messages.ALARM
|
|
for c in (_c1, _c2, ()))
|
|
and messages.peak_saturated(1, shadow=True).id == messages.ID_PEAK_SATURATED
|
|
and messages.peak_saturated(1, shadow=True).severity == messages.ALARM)
|
|
check("the output-clamp alarm quotes asked before sent",
|
|
messages.output_clamp([messages.Clamp(None, 7412, 5000, "rating")]).body
|
|
.index("7 412 W") < messages.output_clamp(
|
|
[messages.Clamp(None, 7412, 5000, "rating")]).body.index("5 000 W"))
|
|
|
|
|
|
# ============================================================================== #
|
|
print("\n2. no notification is worded outside app/messages.py")
|
|
APP = HERE / "app"
|
|
|
|
|
|
# UX-08 2d: a refusal is counted as it is SHOWN. A placeholder that names a
|
|
# setting is expanded to the label and key `messages.setting()` prints, so
|
|
# `{S('max_w')}` counts as “Inverter rating” (max_w), three words, not one.
|
|
SETTING_CALLS = ("S", "_S", "setting", "messages.setting")
|
|
CONSTANTS = {} # module-level NAME = "text", per file, for keys
|
|
|
|
|
|
def _key(node):
|
|
if isinstance(node, ast.Constant) and isinstance(node.value, str):
|
|
return node.value
|
|
if isinstance(node, ast.Name):
|
|
return CONSTANTS.get(node.id)
|
|
return None
|
|
|
|
|
|
def expand_setting(node):
|
|
"""The shown text of a placeholder that names a setting, else None."""
|
|
if isinstance(node, ast.Call) and node.args:
|
|
name = ast.unparse(node.func)
|
|
if name in SETTING_CALLS:
|
|
key = _key(node.args[0])
|
|
if key not in messages.LABELS:
|
|
return None
|
|
unit = node.args[1] if len(node.args) > 1 else None
|
|
plain = unit is None or (isinstance(unit, ast.Constant)
|
|
and not unit.value)
|
|
return messages.setting(key, "" if plain else "inverter-b")
|
|
if name == "_labels" and isinstance(node.args[0], (ast.List, ast.Tuple)):
|
|
keys = [_key(e) for e in node.args[0].elts]
|
|
if all(k in messages.LABELS for k in keys):
|
|
return " and ".join(messages.setting(k) for k in keys)
|
|
if (isinstance(node, ast.Subscript)
|
|
and ast.unparse(node.value) == "messages.LABELS"):
|
|
key = _key(node.slice)
|
|
return messages.LABELS.get(key) if key else None
|
|
return None
|
|
|
|
|
|
def render(node, expand=False):
|
|
"""A string expression as text, placeholders as {expr}; None otherwise.
|
|
`expand` shows setting placeholders as the page shows them."""
|
|
if isinstance(node, ast.Constant) and isinstance(node.value, str):
|
|
return node.value
|
|
if isinstance(node, ast.JoinedStr):
|
|
out = []
|
|
for v in node.values:
|
|
if isinstance(v, ast.Constant):
|
|
out.append(v.value)
|
|
continue
|
|
shown = expand_setting(v.value) if expand else None
|
|
out.append(shown if shown is not None
|
|
else "{" + ast.unparse(v.value) + "}")
|
|
return "".join(out)
|
|
if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add):
|
|
a, b = render(node.left, expand), render(node.right, expand)
|
|
if a is None and b is None:
|
|
return None
|
|
return (a if a is not None else "{…}") + (b if b is not None else "{…}")
|
|
if isinstance(node, ast.IfExp):
|
|
a, b = render(node.body, expand), render(node.orelse, expand)
|
|
if a is None and b is None:
|
|
return None
|
|
return (a or "") + " / " + (b or "")
|
|
return None
|
|
|
|
|
|
def notification_calls(tree):
|
|
for node in ast.walk(tree):
|
|
if (isinstance(node, ast.Call) and node.args
|
|
and ast.unparse(node.func).endswith("call")
|
|
and render(node.args[0]) == "persistent_notification"):
|
|
yield node
|
|
|
|
|
|
bad_calls, n_calls = [], 0
|
|
for f in sorted(APP.glob("*.py")):
|
|
for call in notification_calls(ast.parse(f.read_text(encoding="utf-8"))):
|
|
n_calls += 1
|
|
payload = call.args[2] if len(call.args) > 2 else None
|
|
ok = (isinstance(payload, ast.Call)
|
|
and isinstance(payload.func, ast.Attribute)
|
|
and payload.func.attr == "data"
|
|
and "messages." in ast.unparse(payload.func.value))
|
|
if not ok:
|
|
bad_calls.append(f"{f.name}:{call.lineno}")
|
|
check(f"the scan found the add-on's notification calls ({n_calls})", n_calls >= 7)
|
|
check(f"every one is built by app/messages.py (not: {bad_calls})", not bad_calls)
|
|
|
|
|
|
# ============================================================================== #
|
|
print("\n3. the ratchet: other user-facing text obeys its limit or is listed")
|
|
|
|
LIMITS = { # kind: (words, keys ok, entity ids ok)
|
|
"row": (15, False, False),
|
|
"banner": (12, False, False),
|
|
"refusal": (30, True, True),
|
|
"log": (25, True, True),
|
|
"option-name": (6, False, False),
|
|
"option-desc": (30, True, True),
|
|
}
|
|
LOG_CALL = re.compile(r"(?:^|\.)(?:_LOG|log|logger|LOG)\.(warning|error|critical|exception)$")
|
|
|
|
|
|
def strings_in(node, expand=False):
|
|
"""Top-level string expressions inside `node` (not their parts)."""
|
|
t = render(node, expand)
|
|
if t is not None:
|
|
yield node, t
|
|
return
|
|
for child in ast.iter_child_nodes(node):
|
|
yield from strings_in(child, expand)
|
|
|
|
|
|
def _module_constants(tree):
|
|
"""Module-level `NAME = "text"` assignments."""
|
|
return {t.id: n.value.value for n in tree.body
|
|
if isinstance(n, ast.Assign) and isinstance(n.value, ast.Constant)
|
|
and isinstance(n.value.value, str)
|
|
for t in n.targets if isinstance(t, ast.Name)}
|
|
|
|
|
|
_MODCONST = {f.stem: _module_constants(ast.parse(f.read_text(encoding="utf-8")))
|
|
for f in sorted(APP.glob("*.py"))}
|
|
|
|
|
|
def collect():
|
|
"""(kind, where, text) for every user-facing literal of a ratcheted kind."""
|
|
out = []
|
|
for f in sorted(APP.glob("*.py")):
|
|
if f.name == "messages.py":
|
|
continue
|
|
tree = ast.parse(f.read_text(encoding="utf-8"))
|
|
# The file's own constants, and those it imports from another app
|
|
# module (`from .validate import TAXES_PROFILE_OPTION`), so a key
|
|
# named by a constant from elsewhere is expanded too.
|
|
CONSTANTS.clear()
|
|
CONSTANTS.update(_module_constants(tree))
|
|
for n in tree.body:
|
|
if isinstance(n, ast.ImportFrom) and n.level == 1 and n.module:
|
|
theirs = _MODCONST.get(n.module, {})
|
|
CONSTANTS.update({a.asname or a.name: theirs[a.name]
|
|
for a in n.names if a.name in theirs})
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Dict):
|
|
for k, v in zip(node.keys, node.values):
|
|
key = render(k) if k is not None else None
|
|
if key in ("text", "banner"):
|
|
for n, t in strings_in(v):
|
|
out.append(("row" if key == "text" else "banner",
|
|
f"{f.name}:{n.lineno}", t))
|
|
elif isinstance(node, ast.Assign) and f.name == "main.py":
|
|
tgt = ast.unparse(node.targets[0]).replace(" ", "").strip("()")
|
|
if tgt in ("level,banner", "banner"):
|
|
for n, t in strings_in(node.value):
|
|
if len(words(t)) > 1:
|
|
out.append(("banner", f"{f.name}:{n.lineno}", t))
|
|
elif isinstance(node, ast.Call):
|
|
name = ast.unparse(node.func)
|
|
if LOG_CALL.search(name) or name.endswith("._moan"):
|
|
arg = node.args[1 if name.endswith("._moan") else 0] \
|
|
if node.args else None
|
|
if arg is not None:
|
|
for n, t in strings_in(arg):
|
|
out.append(("log", f"{f.name}:{n.lineno}", t))
|
|
elif re.search(r"problems\.(append|extend)$", name):
|
|
for a in node.args:
|
|
for n, t in strings_in(a, expand=True):
|
|
if len(words(t)) > 1:
|
|
out.append(("refusal", f"{f.name}:{n.lineno}", t))
|
|
# UX-08 2d: the Dutch option texts are held to the same length.
|
|
for lang in ("en", "nl"):
|
|
tr = TL.parse_translation_file(HERE / "translations" / f"{lang}.yaml")
|
|
|
|
def walk(conf, path):
|
|
for key, v in (conf or {}).items():
|
|
if not isinstance(v, dict):
|
|
continue
|
|
if isinstance(v.get("name"), str):
|
|
out.append(("option-name", f"{lang}.yaml:{path}{key}",
|
|
v["name"].split("·")[-1].strip()))
|
|
if isinstance(v.get("description"), str):
|
|
out.append(("option-desc", f"{lang}.yaml:{path}{key}",
|
|
v["description"]))
|
|
if isinstance(v.get("fields"), dict):
|
|
walk(v["fields"], f"{path}{key}.")
|
|
walk(tr.get("configuration"), "")
|
|
return out
|
|
|
|
|
|
def problems_of(kind, text):
|
|
lim, keys, ents = LIMITS[kind]
|
|
p = []
|
|
if len(words(text)) > lim:
|
|
p.append(f"over {lim} words ({len(words(text))})")
|
|
if ";" in text:
|
|
p.append("semicolon")
|
|
return p + text_problems(text, keys=keys, entities=ents)
|
|
|
|
|
|
def key_of(kind, text):
|
|
return f"{kind} {hashlib.sha1(text.encode('utf-8')).hexdigest()[:12]}"
|
|
|
|
|
|
ALLOW_FILE = HERE / "ux08_allowlist.txt"
|
|
# The allowlist may never grow. The test demands the list be exactly this long,
|
|
# so deleting lines means lowering this number in the same change, and adding
|
|
# one means raising it where a reviewer sees it.
|
|
ALLOWLIST_MAX = 0
|
|
|
|
|
|
def read_allowlist(path):
|
|
keys = set()
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
|
line = line.split("#", 1)[0].strip()
|
|
if line:
|
|
keys.add(" ".join(line.split()[:2]))
|
|
return keys
|
|
|
|
|
|
found = collect()
|
|
violating = {}
|
|
for kind, where, text in found:
|
|
p = problems_of(kind, text)
|
|
if p:
|
|
violating.setdefault(key_of(kind, text), (kind, where, text, p))
|
|
if "--violations" in sys.argv:
|
|
for k, (kind, where, text, p) in sorted(violating.items(),
|
|
key=lambda kv: kv[1][:2]):
|
|
print(f"{k} # {where} {'; '.join(p)} :: {text[:70]!r}")
|
|
sys.exit(0)
|
|
|
|
STATUS = messages.status_samples()
|
|
|
|
|
|
def row_problems(colour, text):
|
|
"""STYLE.md section 1 for a status row (by colour) or a banner (None)."""
|
|
n_s, n_w = sentences(text), len(words(text))
|
|
out = []
|
|
if colour is None:
|
|
if n_w > 12:
|
|
out.append(f"banner over 12 words ({n_w})")
|
|
elif colour == messages.RED:
|
|
if n_s > 3:
|
|
out.append(f"red row over 3 sentences ({n_s})")
|
|
if n_w > 30:
|
|
out.append(f"red row over 30 words ({n_w})")
|
|
else:
|
|
if n_s > 1:
|
|
out.append(f"{colour} row over 1 sentence ({n_s})")
|
|
if n_w > 15:
|
|
out.append(f"{colour} row over 15 words ({n_w})")
|
|
if ";" in text:
|
|
out.append("semicolon: split it into sentences or drop a clause")
|
|
return out + text_problems(text, keys=False, entities=False)
|
|
|
|
|
|
for fn, colour, text in STATUS:
|
|
p = row_problems(colour, text)
|
|
check(f"{colour or 'banner':6} {text!r}{' - ' + '; '.join(p) if p else ''}",
|
|
not p)
|
|
_catalogue = {k for k, v in vars(messages).items()
|
|
if callable(v) and k.startswith(("row_", "banner_"))}
|
|
_missing = sorted(_catalogue - {fn for fn, _c, _t in STATUS})
|
|
check(f"the samples cover every row and banner in the catalogue "
|
|
f"(missing: {_missing})", not _missing)
|
|
# A red row with no banner of its own heads the page as "Not ready - <its first
|
|
# sentence>". The three trips have their own branch in status_payload(), and
|
|
# main.py gives the other two a "banner" key (both are sampled above).
|
|
_own_banner = {"row_direction_trip", "row_site_direction_trip",
|
|
"row_circulating_trip", "row_peak_reserve", "row_power_too_high"}
|
|
_long = [t for fn, c, t in STATUS if c == messages.RED and fn not in _own_banner
|
|
and len(words(messages.banner_not_ready({"text": t}))) > 12]
|
|
# The hint under the buttons (UX-08 2d): the last event in plain words, one
|
|
# per key of messages.HINTS, held to the rule for a green row.
|
|
for key, text in messages.hint_samples():
|
|
p = row_problems(messages.GREEN, text)
|
|
check(f"hint {text!r}{' - ' + '; '.join(p) if p else ''}", not p)
|
|
_calls, _bare = 0, []
|
|
for f in sorted(APP.glob("*.py")):
|
|
for node in ast.walk(ast.parse(f.read_text(encoding="utf-8"))):
|
|
if (isinstance(node, ast.Call)
|
|
and ast.unparse(node.func).endswith("log_event")):
|
|
_calls += 1
|
|
h = next((k.value for k in node.keywords if k.arg == "hint"), None)
|
|
if not (isinstance(h, ast.Call)
|
|
and ast.unparse(h.func) in ("messages.hint", "hint")):
|
|
_bare.append(f"{f.name}:{node.lineno}")
|
|
check(f"every event the add-on records gives the page its plain words, from "
|
|
f"messages.hint() ({_calls} calls, without: {_bare})",
|
|
_calls >= 40 and not _bare)
|
|
_raw = ("OUTPUT_CLAMP", "BUG SIGNAL", "FAILSAFE", "SIGN_DISAGREEMENT",
|
|
"ALLOCATOR", "UNMET", "ARBITER", "PEAK GUARD", "PEAK_GUARD")
|
|
check("no hint carries the raw event codes the page used to show",
|
|
not [t for _k, t in messages.hint_samples()
|
|
if any(r in t for r in _raw)])
|
|
|
|
check(f"every other red row makes a 'Not ready' banner of at most 12 words "
|
|
f"(too long: {_long})", not _long)
|
|
|
|
# A latch row must say how it clears: that is the one action it exists for.
|
|
_clears = {
|
|
"row_direction_trip": "A restart does not clear this",
|
|
"row_site_direction_trip": "Only starting control clears this",
|
|
"row_circulating_trip": "Only starting control clears this",
|
|
"row_not_delivering": "only stopping and starting control clears this",
|
|
}
|
|
_unclear = [fn for fn, _c, t in STATUS if fn in _clears and _clears[fn] not in t]
|
|
check(f"every latch row says how it clears (missing: {_unclear})",
|
|
not _unclear and {fn for fn, _c, _t in STATUS} >= set(_clears))
|
|
|
|
allow = read_allowlist(ALLOW_FILE)
|
|
by_kind = {}
|
|
for kind, *_ in found:
|
|
by_kind[kind] = by_kind.get(kind, 0) + 1
|
|
for _fn, colour, _t in STATUS:
|
|
kind = "row" if colour else "banner"
|
|
by_kind[kind] = by_kind.get(kind, 0) + 1
|
|
check(f"the scan found every ratcheted kind ({by_kind})",
|
|
all(by_kind.get(k, 0) >= 5 for k in LIMITS))
|
|
new = sorted(k for k in violating if k not in allow)
|
|
for k in new:
|
|
kind, where, text, p = violating[k]
|
|
print(f" NEW {where}: {'; '.join(p)} :: {text[:90]!r}\n"
|
|
f" (rewrite it to STYLE.md; listing it needs a reason)")
|
|
check(f"no new string breaks the rule ({len(new)} new)", not new)
|
|
stale = sorted(k for k in allow if k not in violating)
|
|
for k in stale:
|
|
print(f" STALE {k} - that string now obeys the rule or is gone: "
|
|
f"delete its line from ux08_allowlist.txt")
|
|
check(f"no allowlist entry is stale, so the list only shrinks "
|
|
f"({len(stale)} stale)", not stale)
|
|
check(f"the allowlist is exactly its cap ({len(allow)} entries, cap "
|
|
f"{ALLOWLIST_MAX}): lower the cap when you delete lines",
|
|
len(allow) == ALLOWLIST_MAX)
|
|
|
|
# A refusal names a setting by its Configuration-tab label (UX-08 2c). The
|
|
# catalogue's labels must be the en.yaml names without group prefix and unit.
|
|
_TR = TL.parse_translation_file(HERE / "translations" / "en.yaml")["configuration"]
|
|
|
|
|
|
def _label_stems(key):
|
|
names = [c[key]["name"] for c in (_TR, _TR["inverters"]["fields"])
|
|
if isinstance(c.get(key), dict)]
|
|
out = set()
|
|
for n in names:
|
|
n = n.split("·")[-1].strip()
|
|
while re.search(r"\s*\([^()]*\)$", n):
|
|
n = re.sub(r"\s*\([^()]*\)$", "", n)
|
|
out.add(n)
|
|
return out
|
|
|
|
|
|
_drift = sorted(f"{k}: {v!r}" for k, v in messages.LABELS.items()
|
|
if v not in _label_stems(k))
|
|
check(f"every label a refusal quotes is the setting's name in en.yaml "
|
|
f"(drifted: {_drift})", not _drift)
|
|
check("...and a refusal quotes the label and then the key, so the config "
|
|
"panel still files it under that setting",
|
|
messages.setting("max_w", "inverter-b")
|
|
== "“Inverter rating” for inverter-b (inverters.max_w)")
|
|
# UX-08 2d: the refusals as the validator SHOWS them. The scan above expands
|
|
# setting labels in a template, but a reason that carries a parser's error
|
|
# (`{e}`) is only as long as that error, so the long ones found by the 2c test
|
|
# are driven for real here and held to the same 30 words.
|
|
from app import custom_formula as _cf # noqa: E402
|
|
from app.validate import validate_startup_config # noqa: E402
|
|
|
|
_BASE = {"setpoint_entity": "number.goodwe_setpoint", "max_w": 2000,
|
|
"soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
|
|
"meter_source": "off", "maintenance_charge_w": 2000,
|
|
"maintenance_discharge_w": 2000}
|
|
_CARD = _cf._LEGS["price_formula"].example.replace("15.55", "15.05")
|
|
_TABLE = ('{"periods":[{"name":"day","from":"07:00","to":"22:00","days":'
|
|
'"mon-fri","eur_per_kwh":0.1},{"name":"night","from":"22:00","to":'
|
|
'"07:00","days":"mon-fri","eur_per_kwh":0.08},{"name":"night",'
|
|
'"from":"00:00","to":"24:00","days":"sat,sun","eur_per_kwh":0.08}]}')
|
|
_ROSTER = {k: v for k, v in _BASE.items()
|
|
if k not in ("setpoint_entity", "soc_entity", "batt_entity",
|
|
"max_w")}
|
|
_ROSTER["inverters"] = [
|
|
dict(name=n, setpoint_entity=f"number.sp_{n[-1]}", max_w=w,
|
|
soc_entity=f"sensor.soc_{n[-1]}", batt_entity=f"sensor.batt_{n[-1]}",
|
|
bms_soc_floor=10)
|
|
for n, w in (("inverter-a", 2000), ("inverter-b", 200))]
|
|
_DRIVEN = {
|
|
"formula card": dict(_BASE, price_provider="custom_formula",
|
|
price_formula=_CARD),
|
|
"formula not valid": dict(_BASE, price_provider="custom_formula",
|
|
price_formula='{"index": "epex_be"} x'),
|
|
"injection card": dict(_BASE, price_export_formula=_CARD),
|
|
"roster thresholds": dict(_ROSTER, maintenance_charge_w=100,
|
|
maintenance_discharge_w=100),
|
|
"levies table and flat": dict(_BASE, price_taxes_profile=_TABLE,
|
|
price_taxes_eur_kwh=0.1198),
|
|
"feed and entities": dict(_BASE, price_provider="generic_dayahead",
|
|
price_now_entity="sensor.now",
|
|
price_avg_entity="sensor.avg"),
|
|
"fixed tariff and entities": dict(_BASE, static_profile=_TABLE,
|
|
price_now_entity="sensor.now",
|
|
price_avg_entity="sensor.avg"),
|
|
"fixed tariff not valid": dict(_BASE, static_profile='{"periods": []} x'),
|
|
}
|
|
_LEAKS = ("c_eur_per_kwh", "vat_pct", "its own card", "line 1 column",
|
|
"None", "JSON", "one you do not use")
|
|
_shown, _long, _leaky = [], [], []
|
|
for _name, _o in _DRIVEN.items():
|
|
_ps = validate_startup_config(_o, -5000, 5000)
|
|
_shown.append((_name, len(_ps)))
|
|
_long += [f"{_name}: {len(words(t))} words: {t}" for t in _ps
|
|
if len(words(t)) > 30]
|
|
_leaky += [f"{_name}: {t}" for t in _ps if any(x in t for x in _LEAKS)]
|
|
check(f"every driven case is refused ({_shown})",
|
|
all(n >= 1 for _n, n in _shown)
|
|
and dict(_shown)["roster thresholds"] >= 2)
|
|
check(f"...and every refusal, as SHOWN, is at most 30 words (too long: "
|
|
f"{_long})", not _long)
|
|
check(f"...and shows no parser text, file units or Python's None (found: "
|
|
f"{_leaky})", not _leaky)
|
|
|
|
# The config panel files a reason beside every setting whose key it names, but
|
|
# `inverters.max_w` names a setting of one inverter in the list, not the flat
|
|
# `max_w` (UX-08 2d, from the 2c test).
|
|
from app.configui import problems_by_field # noqa: E402
|
|
_by, _un = problems_by_field(
|
|
["“Inverter rating” for inverter-b (inverters.max_w) is 200 W.",
|
|
"“Inverter rating” (max_w) is 0."], ["max_w", "inverters"])
|
|
check("a roster refusal shows beside `inverters` only, and a flat one beside "
|
|
f"`max_w` ({ {k: len(v) for k, v in _by.items()} })",
|
|
_by.get("inverters") == [
|
|
"“Inverter rating” for inverter-b (inverters.max_w) is 200 W."]
|
|
and _by.get("max_w") == ["“Inverter rating” (max_w) is 0."]
|
|
and not _un)
|
|
|
|
print(f"CI: UX-08 allowlist {len(allow)} entries "
|
|
+ ", ".join(f"{k} {sum(1 for x in allow if x.startswith(k + ' '))}"
|
|
for k in LIMITS))
|
|
|
|
|
|
# ============================================================================== #
|
|
print("\n4. DOCS.md holds the detail of every notification")
|
|
DOCS = (HERE / "DOCS.md").read_text(encoding="utf-8")
|
|
part = DOCS.split("\n## Notifications\n", 1)
|
|
check("DOCS.md has a '## Notifications' section", len(part) == 2)
|
|
section = part[-1].split("\n## ", 1)[0] if len(part) == 2 else ""
|
|
heads = set(re.findall(r"^### (.+)$", section, re.M))
|
|
missing = sorted({n.docs for n in SAMPLES} - heads)
|
|
check(f"...with a heading for every Details line (missing: {missing})", not missing)
|
|
check("DOCS.md explains the status page rows under "
|
|
f"'## {messages.DOCS_STATUS}'", f"\n## {messages.DOCS_STATUS}\n" in DOCS)
|
|
|
|
|
|
# ============================================================================== #
|
|
print("\n5. mutations: each rule catches a spoiled input")
|
|
N = messages.Notice
|
|
check("a six-word title is refused",
|
|
bool(title_problems("The meter reading may be stuck")))
|
|
check("a GoodWe prefix is refused", bool(title_problems("GoodWe: meter stuck")))
|
|
check("a second sentence in a warning is refused",
|
|
bool(body_problems("The battery lags. Check it.", messages.WARNING)))
|
|
check("a semicolon run-on is refused, in an alarm too",
|
|
bool(body_problems("It stopped; check it.", messages.ALARM)))
|
|
check("a fourth sentence in an alarm is refused",
|
|
bool(body_problems("One. Two. Three. Four.", messages.ALARM)))
|
|
check("an internal name is refused", bool(body_problems(
|
|
"The integrator reached its bound.", messages.WARNING)))
|
|
check("a card id is refused", bool(body_problems(
|
|
"See SAFETY-25.", messages.WARNING)))
|
|
check("shouting is refused", bool(body_problems(
|
|
"Control is STILL RUNNING.", messages.WARNING)))
|
|
check("an entity id in an alarm is refused", bool(body_problems(
|
|
"Check sensor.battery_power now.", messages.ALARM)))
|
|
check("...but allowed in a refusal", not problems_of(
|
|
"refusal", "Check sensor.battery_power in Developer Tools."))
|
|
check("a semicolon in a refusal or an option text is refused",
|
|
bool(problems_of("refusal", "It is wrong; fix it."))
|
|
and bool(problems_of("option-desc", "It is wrong; fix it.")))
|
|
check("a label that drifted from en.yaml is caught",
|
|
"Inverter power" not in _label_stems("max_w"))
|
|
check("an option key in an alarm is refused", bool(body_problems(
|
|
"Lower peak_soc_floor.", messages.ALARM)))
|
|
check("a message that lost its Details line is refused",
|
|
ends_in_details(N("x", messages.WARNING, "Meter stuck", "It is.")
|
|
.data()["message"], "Meter stuck")
|
|
and not ends_in_details("It is.", "Meter stuck"))
|
|
_fake = " x = {'text': 'THE METER IS FROZEN AND NOTHING WORKS AT ALL'}\n"
|
|
_n = [t for node in ast.walk(ast.parse(_fake.strip()))
|
|
if isinstance(node, ast.Dict) for _, t in strings_in(node.values[0])]
|
|
check("a second sentence in an amber row is refused",
|
|
bool(row_problems(messages.AMBER, "Meter stuck. Check it.")))
|
|
check("a fourth sentence in a red row is refused",
|
|
bool(row_problems(messages.RED, "One. Two. Three. Four.")))
|
|
check("a 16-word green row is refused",
|
|
bool(row_problems(messages.GREEN, " ".join(["word"] * 16))))
|
|
check("a 13-word banner is refused",
|
|
bool(row_problems(None, " ".join(["word"] * 13))))
|
|
check("an entity id in a status row is refused",
|
|
bool(row_problems(messages.GREEN, "Battery level: sensor.soc = 54")))
|
|
check("a semicolon in a status row is refused",
|
|
bool(row_problems(messages.RED, "It stopped; check it.")))
|
|
check("a new shouting status row is seen by the scan and is not listed",
|
|
_n and problems_of("row", _n[0])
|
|
and key_of("row", _n[0]) not in allow)
|
|
_bad = ast.parse("await self.hass.call('persistent_notification', 'create', "
|
|
"{'title': 'x', 'message': 'y'})", mode="exec")
|
|
check("a hand-written notification payload is seen by part 2",
|
|
any(not (isinstance(c.args[2], ast.Call)) for c in notification_calls(_bad)))
|
|
|
|
|
|
# ============================================================================== #
|
|
print("\n6. log lines as they are built at run time (UX-10)")
|
|
# Part 3 reads a WARNING+ line's own literal and counts each placeholder as one
|
|
# word. So a reason text built elsewhere and passed in as `%s` was never seen:
|
|
# the 2d test found "…clamped to +2000 W; … This is a BUG SIGNAL … Cut back 1
|
|
# times" in the log while part 3 was green, because `loop_clamp_note()` built
|
|
# it. This part closes that gap in two steps.
|
|
# a. Every dynamic part of every WARNING+ line in `app/` is classified here,
|
|
# as data (a number, a name, a path, an exception's own text) or as a text
|
|
# the add-on builds. A new dynamic part that is in neither list fails, so a
|
|
# new builder cannot slip past.
|
|
# b. Every built text is produced by the real code and checked. It is the
|
|
# evidence behind an alarm, so it gets the red-alarm shape: at most 3
|
|
# sentences of at most 30 words, 70 in all. It keeps its reason code. The
|
|
# whole line, as logged, may carry no semicolon, banned term, capitals,
|
|
# "1 times", "(s)" or a doubled full stop.
|
|
from datetime import datetime as _dt, timedelta as _td, timezone as _tz # noqa: E402
|
|
|
|
from app import clamp as _clamp, freshness as _fresh, peakguard as _pg # noqa: E402
|
|
from app import store as _store # noqa: E402
|
|
from app.circulation import CirculationConfig, CirculationWatchdog # noqa: E402
|
|
from app.control import Tuning, compute # noqa: E402
|
|
from app.delivery import (DeliveryConfig, DeliveryWatchdog, UnitDelivery, # noqa: E402
|
|
delinquent)
|
|
from app.freeze import FreezeWatchdog, FreezeWatchdogConfig # noqa: E402
|
|
from app.price import PricePoint, PriceProduct, PriceRead # noqa: E402
|
|
from app.static_tariff import TariffPlan # noqa: E402
|
|
from app.tariff import Projection # noqa: E402
|
|
from app.watchdog import (AGAINST_METER, AGAINST_OWN_POWER, SignWatchdog, # noqa: E402
|
|
SignWatchdogConfig)
|
|
|
|
_T0 = _dt(2026, 9, 27, 9, 0, tzinfo=_tz.utc)
|
|
PLURAL = re.compile(r"\b1 [a-z]{2,}s\b(?<!\b1 [a-z]ss)|\w\(s\)")
|
|
COUNT_PLURAL = re.compile(r"%d [a-z]{2,}s\b")
|
|
|
|
|
|
def dynamic_parts(tree, fname):
|
|
"""(key, call) for every dynamic part of a WARNING+ log call in `tree`."""
|
|
for node in ast.walk(tree):
|
|
if not isinstance(node, ast.Call):
|
|
continue
|
|
name = ast.unparse(node.func)
|
|
moan = name.endswith("._moan")
|
|
if not (LOG_CALL.search(name) or moan):
|
|
continue
|
|
i = 1 if moan else 0
|
|
if len(node.args) <= i:
|
|
continue
|
|
fmt = node.args[i]
|
|
parts = [ast.unparse(a) for a in node.args[i + 1:]]
|
|
if isinstance(fmt, ast.JoinedStr):
|
|
parts += [ast.unparse(v.value) for v in fmt.values
|
|
if isinstance(v, ast.FormattedValue)]
|
|
elif not (isinstance(fmt, ast.Constant) and isinstance(fmt.value, str)):
|
|
parts.append(ast.unparse(fmt))
|
|
for part in parts:
|
|
yield f"{fname}:{part}", node
|
|
|
|
|
|
# a. DATA: shown as it is, never worded by the add-on. An exception's text is
|
|
# the library's (hass.py words the Home Assistant ones, UX-10 item 3).
|
|
DATA = {
|
|
# hass.py: the _moan() wrapper and what its callers pass
|
|
"hass.py:msg", "hass.py:*args", "hass.py:entity_id", "hass.py:err",
|
|
"hass.py:obj", "hass.py:raw", "hass.py:domain", "hass.py:service",
|
|
"hass.py:resp.status", "hass.py:said", "hass.py:entity", "hass.py:what",
|
|
# main.py
|
|
"main.py:'cancelled' if task.cancelled() else repr(exc) if exc else 'returned'",
|
|
"main.py:EXIT_CONTROL_DIED", "main.py:first.name", "main.py:OPTIONS_PATH",
|
|
"main.py:err", "main.py:', '.join(sorted(stranded))",
|
|
"main.py:sorted(on_roster)", "main.py:inv.name",
|
|
"main.py:inv.batt_fresh.bound.release_s", "main.py:inv.batt_fresh.bound.move_w",
|
|
"main.py:inv.batt_fresh.bound.unchanged_s", "main.py:inv.batt_fresh.held",
|
|
"main.py:value", "main.py:float(value)",
|
|
"main.py:entity", "main.py:WRITE_BUDGET_S",
|
|
"main.py:_e",
|
|
"main.py:messages.times(self.clamp_wd.engagements)",
|
|
"main.py:messages.when(brussels(self.tariff_at))",
|
|
"main.py:messages.duration(TARIFF_MAX_AGE_S)",
|
|
"main.py:messages.count(len(targets), 'inverter')",
|
|
# 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",
|
|
"p1.py:backoff", "p1.py:self.url", "p1.py:str(err) or type(err).__name__",
|
|
"p1.py:self.poll_s",
|
|
# price.py
|
|
"price.py:doc.get('deprecated')", "price.py:len(dropped)",
|
|
"price.py:messages.count(len(times), 'time')", "price.py:plugin_id",
|
|
"price.py:override",
|
|
"price.py:', '.join((str(i) for i in dropped[:5])) + (' and more' if len(dropped) > 5 else '')",
|
|
"price.py:*(d.isoformat() for d in brussels_date_range(since, until))",
|
|
"price.py:pid",
|
|
"price.py:'/'.join(sorted((c.value for c in caps.sources - {Component.ENERGY})))",
|
|
"price.py:' and '.join(ignored) + (' is' if len(ignored) == 1 else ' are')",
|
|
"price.py:getattr(cls, 'id', cls)", "price.py:e", "price.py:type(e).__name__",
|
|
# price_history.py, selfconfig.py, static_tariff.py, store.py, tariff.py
|
|
"price_history.py:self.path", "price_history.py:day_words(want)",
|
|
"price_history.py:messages.count(len(points), 'price')",
|
|
"price_history.py:err",
|
|
"selfconfig.py:messages.count(len(problems), 'problem')",
|
|
"selfconfig.py:', '.join(changed) or 'none this page also edited'",
|
|
"static_tariff.py:e", "store.py:err", "store.py:key", "store.py:raw",
|
|
"tariff.py:self.path", "tariff.py:err", "tariff.py:start", "tariff.py:month",
|
|
# validate.py: option values and keys (the refusals are part 3's)
|
|
"validate.py:integ_w", "validate.py:max_w", "validate.py:disch_w",
|
|
"validate.py:maint_where", "validate.py:maint_cap", "validate.py:pid",
|
|
"validate.py:i", "validate.py:TAXES_PROFILE_OPTION", "validate.py:negative",
|
|
"validate.py:EXPORT_FORMULA_OPTION", "validate.py:xspec.passthrough",
|
|
"validate.py:xspec.offset_eur_kwh", "validate.py:'=' * 72",
|
|
"validate.py:'-' * 72",
|
|
"web.py:err",
|
|
}
|
|
|
|
|
|
# b. BUILT TEXT: each producer, run for real. A reason that is only ever
|
|
# stored and restored (a saved safety stop) is produced by the same builder.
|
|
def _sign(against, cmd=1500.0, got=-1400.0):
|
|
wd = SignWatchdog(SignWatchdogConfig(min_w=250.0, hold_s=60.0),
|
|
against=against)
|
|
for s in (0, 61):
|
|
wd.evaluate(cmd, got, True, _T0 + _td(seconds=s))
|
|
return wd.reason
|
|
|
|
|
|
def _sign_texts():
|
|
return [_sign(AGAINST_METER), _sign(AGAINST_OWN_POWER, -1100.0, 2000.0)]
|
|
|
|
|
|
def _circ_texts():
|
|
wd = CirculationWatchdog(CirculationConfig(min_w=250.0, hold_s=60.0))
|
|
for s, w in ((0, 5200.0), (30, 6000.0), (61, 5800.0)):
|
|
wd.evaluate(w, _T0 + _td(seconds=s))
|
|
return [wd.reason]
|
|
|
|
|
|
def _delivery_texts():
|
|
units = [UnitDelivery("inverter-a", 1000.0, 990.0),
|
|
UnitDelivery("inverter-b", 2000.0, 0.0)]
|
|
sentence = delinquent(units, 200.0)["inverter-b"]
|
|
wd = DeliveryWatchdog(DeliveryConfig(hold_s=120.0))
|
|
for s in (0, 121):
|
|
wd.evaluate(sentence, _T0 + _td(seconds=s))
|
|
return [wd.reason]
|
|
|
|
|
|
def _freeze_texts():
|
|
out = []
|
|
for dev in (None, 4700.0):
|
|
wd = FreezeWatchdog(FreezeWatchdogConfig(hold_s=600.0))
|
|
wd.evaluate(3600.0, 4480.0 if dev else 5000.0, 5000.0, _T0,
|
|
dev_min=-dev if dev else None, dev_max=dev)
|
|
out.append(wd.reason)
|
|
return out
|
|
|
|
|
|
def _stale_texts():
|
|
one = _fresh.Freshness(_fresh.BOUNDS["batt"])
|
|
for t in range(0, 40):
|
|
one.submit(-1200.0, -1200.0 if t < 1 else 1300.0, float(t))
|
|
two = _fresh.Freshness(_fresh.BOUNDS["batt"])
|
|
for t in range(0, 40):
|
|
two.submit(-1200.0, -1200.0 if t < 1 else 1300.0, float(t))
|
|
for t in range(40, 80):
|
|
two.submit(-1200.0, 0.0, float(t))
|
|
for t in range(80, 130):
|
|
two.submit(-1200.0, 1300.0, float(t))
|
|
return [one.reason, two.reason]
|
|
|
|
|
|
def _clamp_texts():
|
|
over = compute(prev_w=1000.0, grid_w=9000.0, actual_w=1000.0,
|
|
tuning=Tuning(max_w=5000.0, slew_w=1000.0, gain=0.6,
|
|
integrator_max_w=9000.0), i_w=8000.0)
|
|
at = compute(prev_w=1000.0, grid_w=9000.0, actual_w=1000.0,
|
|
tuning=Tuning(max_w=5000.0, slew_w=1000.0, gain=0.6),
|
|
i_w=4500.0)
|
|
write = [(n, _clamp.clamp_to_rating(w, 5000.0, -2500.0, 2500.0))
|
|
for n, w in (("inverter-a", 3000.0), ("inverter-b", -7412.0))]
|
|
# One inverter is not named (main.py passes None), several are.
|
|
return [_clamp.loop_clamp_note(over), _clamp.loop_clamp_note(at),
|
|
_clamp.write_clamp_note([(None, write[0][1])]),
|
|
_clamp.write_clamp_note(write[:1]), _clamp.write_clamp_note(write)]
|
|
|
|
|
|
def _peak_texts():
|
|
proj = Projection(projected_w=4338.0, elapsed_w=4338.0, elapsed_s=799.0,
|
|
covered_frac=1.0, complete=False, block_start=_T0,
|
|
held_w=0.0)
|
|
kw = dict(cap_w=3500.0, reading=None, soc_floor=30.0, max_hold_s=30.0)
|
|
sat = _pg.decide(proj, batt_w=1500.0, achievable_w=5000.0, soc=80.0, **kw)
|
|
low = _pg.decide(proj, batt_w=1500.0, achievable_w=5000.0, soc=20.0, **kw)
|
|
unread = _pg.decide(proj, batt_w=1500.0, achievable_w=5000.0, soc=None, **kw)
|
|
return [sat.reason, low.reason, unread.reason]
|
|
|
|
|
|
def _migration_texts():
|
|
out = []
|
|
for moved, kept, lost, wrong in (
|
|
(2, 0, 0, 0), (2, 1, 1, 0), (2, 0, 0, 1), (0, 2, 0, 0),
|
|
(0, 2, 1, 0)):
|
|
rep = _store.MigrationReport()
|
|
rep.ran = moved > 0 or lost > 0
|
|
rep.from_version = 1
|
|
rep.roster = ["inverter-a"]
|
|
rep.moved = {f"unit/inverter-a/k{i}": True for i in range(moved)}
|
|
rep.kept = [f"unit/inverter-a/old{i}" for i in range(kept)]
|
|
for i in range(lost):
|
|
note = f"unit/inverter-a/k{i} was written as True and reads back as None"
|
|
rep.problems.append(note)
|
|
rep.lost.append(note)
|
|
for i in range(wrong):
|
|
note = (f"unit/garage/k{i} was migrated into namespace 'garage', "
|
|
f"which is not in the inverter list")
|
|
rep.problems.append(note)
|
|
rep.misdirected.append(note)
|
|
out.append(rep.explain())
|
|
return out
|
|
|
|
|
|
def _tariff_texts():
|
|
hours = [_T0 + _td(minutes=15 * i) for i in range(8)]
|
|
pts = [PricePoint(h, 0.2, i >= 4) for i, h in enumerate(hours)]
|
|
read = PriceRead.series(pts, PriceProduct.DAY_AHEAD, "x", _T0)
|
|
return [TariffPlan(read, None, "provider").words,
|
|
TariffPlan(PriceRead.no_data("the price feed timed out"), None,
|
|
"none").words]
|
|
|
|
|
|
def _static(spec):
|
|
"""Templates for a family of builders that is not one function: the first
|
|
argument of every call to `spec`, or every value assigned to `spec`, in
|
|
the named file (placeholders count as one word, as in part 3)."""
|
|
fname, kind, name = spec
|
|
out = []
|
|
for f in ([APP / fname] if fname != "*" else sorted(APP.glob("*.py"))):
|
|
tree = ast.parse(f.read_text(encoding="utf-8"))
|
|
for node in ast.walk(tree):
|
|
if kind == "call" and isinstance(node, ast.Call) and node.args \
|
|
and ast.unparse(node.func).split(".")[-1] == name:
|
|
out += [t for _n, t in strings_in(node.args[0])]
|
|
elif kind == "assign" and isinstance(node, ast.Assign) \
|
|
and ast.unparse(node.targets[0]) == name:
|
|
out += [t for _n, t in strings_in(node.value)]
|
|
elif kind == "return" and isinstance(node, ast.FunctionDef) \
|
|
and node.name == name:
|
|
out += [t for r in ast.walk(node) if isinstance(r, ast.Return)
|
|
and r.value is not None for _n, t in strings_in(r.value)]
|
|
return [t for t in out if len(words(t)) > 1]
|
|
|
|
|
|
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:rep.explain()": _migration_texts,
|
|
"store.py:rep.explain()": _migration_texts,
|
|
"main.py:plan.words": _tariff_texts,
|
|
# Families: every builder of the text, as a template.
|
|
"main.py:refusal": lambda: (_static(("main.py", "return", "write_refusal"))
|
|
+ _static(("allocator.py", "return", "authorises"))),
|
|
"main.py:' / '.join(resolution.scope_errors)":
|
|
lambda: _static(("arbiter.py", "return", "scope_violations"))
|
|
+ _static(("arbiter.py", "call", "append")),
|
|
"price_forecast.py:gate.reason": lambda: _static(("*", "call", "block")),
|
|
"static_tariff.py:read.reason": lambda: _static(("*", "call", "no_data")),
|
|
"price_history.py:self.last_error":
|
|
lambda: _static(("price_history.py", "assign", "self.last_error")),
|
|
"price_history.py:bad": lambda: _static(("price_history.py", "return",
|
|
"_parse_points")),
|
|
"maintenance.py:msg": lambda: _static(("maintenance.py", "assign", "msg")),
|
|
"validate.py:aged": lambda: _static(("*", "return", "stale_card")),
|
|
"validate.py:xaged": lambda: _static(("*", "return", "stale_card")),
|
|
# Refusals: part 3 holds them to the refusal rule already.
|
|
"selfconfig.py:problems[0]": None,
|
|
"validate.py:problem": None,
|
|
"p1.py:problem": None,
|
|
}
|
|
|
|
|
|
def reason_problems(text):
|
|
"""STYLE.md's red-alarm shape, for a text a log line quotes."""
|
|
out = []
|
|
n_s, n_w = sentences(text), len(words(text))
|
|
if n_s > 3:
|
|
out.append(f"over 3 sentences ({n_s})")
|
|
if n_w > 70:
|
|
out.append(f"over 70 words ({n_w})")
|
|
long = [x for x in re.split(r"(?<=[.!?])\s+", text) if len(words(x)) > 30]
|
|
if long:
|
|
out.append(f"a sentence over 30 words ({len(words(long[0]))})")
|
|
return out + line_problems(text)
|
|
|
|
|
|
def logged_shape(line):
|
|
"""UX-10 review: the line as logged, wrapper and reason together, has the
|
|
red-alarm shape too. The reason alone passing is not enough: the 2d-style
|
|
"title: reason. tail. tail." line read as 4 sentences with a 3-sentence
|
|
reason inside it."""
|
|
out = []
|
|
if sentences(line) > 3:
|
|
out.append(f"over 3 sentences ({sentences(line)})")
|
|
long = [x for x in re.split(r"(?<=[.!?])\s+", line) if len(words(x)) > 30]
|
|
if long:
|
|
out.append(f"a sentence over 30 words ({len(words(long[0]))})")
|
|
return out
|
|
|
|
|
|
def line_problems(text):
|
|
"""What no WARNING+ line may carry, whatever its length."""
|
|
out = []
|
|
if ";" in text:
|
|
out.append("semicolon")
|
|
if re.search(r"(?<!\.)\.\.(?!\.)|\.{4,}|\.:|:\s*\.", text):
|
|
out.append("a doubled full stop")
|
|
if PLURAL.search(text):
|
|
out.append("plural: " + PLURAL.search(text).group(0))
|
|
return out + text_problems(text, keys=True, entities=True)
|
|
|
|
|
|
def _fill(template, fmt_node, key_part, sample):
|
|
"""The line as logged: `sample` in the text slot, `X` in the others."""
|
|
if isinstance(fmt_node, ast.JoinedStr):
|
|
return template.replace("{" + key_part + "}", sample)
|
|
return template # %-style: filled by the caller
|
|
|
|
|
|
def scan_log_lines(files, text=None, data=None):
|
|
"""-> (unclassified keys, stale keys, [(key, sample, problems)])."""
|
|
text = TEXT if text is None else text
|
|
data = DATA if data is None else data
|
|
seen, bad = set(), []
|
|
for fname, tree in files:
|
|
for key, call in dynamic_parts(tree, fname):
|
|
seen.add(key)
|
|
if key in data or key not in text:
|
|
continue
|
|
maker = text[key]
|
|
if maker is None:
|
|
continue
|
|
i = 1 if ast.unparse(call.func).endswith("._moan") else 0
|
|
fmt = call.args[i]
|
|
template = render(fmt) or ""
|
|
args = [ast.unparse(a) for a in call.args[i + 1:]]
|
|
part = key.split(":", 1)[1]
|
|
for sample in maker():
|
|
p = reason_problems(sample)
|
|
if isinstance(fmt, ast.JoinedStr):
|
|
line = template.replace("{" + part + "}", sample)
|
|
else:
|
|
slots = iter(["%s" if a != part else "\0" for a in args])
|
|
line = re.sub(r"%[-+ #0-9.]*[sdrfgx]",
|
|
lambda m: next(slots, m.group(0)), template)
|
|
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 p:
|
|
bad.append((key, sample, p))
|
|
unknown = sorted(seen - set(data) - set(text))
|
|
stale = sorted((set(data) | set(text)) - seen)
|
|
return unknown, stale, bad
|
|
|
|
|
|
_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)
|
|
for k, sample, p in _bad:
|
|
print(f" BAD {k}: {'; '.join(p)} :: {sample[:110]!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"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)
|
|
if isinstance(n, ast.Call) and (LOG_CALL.search(ast.unparse(n.func))
|
|
or ast.unparse(n.func).endswith("._moan"))
|
|
for a in n.args[:2] if isinstance(a, ast.Constant)
|
|
and isinstance(a.value, str) and COUNT_PLURAL.search(a.value)]
|
|
check(f"no WARNING+ line puts a plural noun after a bare count, which reads "
|
|
f"'1 times' (use messages.times() and friends; found: {_counts})",
|
|
not _counts)
|
|
|
|
# UX-10 review: the two clamp lines the owner quoted, as logged for one
|
|
# inverter, fit the red shape the tight way: at most 3 sentences and 30 words
|
|
# in all, code and numbers kept. The real literals from main.py, filled with
|
|
# the real notes.
|
|
def _clamp_line(opening):
|
|
for n in ast.walk(dict(_FILES)["main.py"]):
|
|
if isinstance(n, ast.Call) and LOG_CALL.search(ast.unparse(n.func)) \
|
|
and n.args and (render(n.args[0]) or "").startswith(opening):
|
|
return n.args[0].value
|
|
return None
|
|
|
|
|
|
_over, _at, _w1, _w1n, _w2 = _clamp_texts()
|
|
_clamp_lines = {
|
|
"error, loop": (_clamp_line("Power request too high"), _over),
|
|
"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)
|
|
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
|
|
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")))
|
|
|
|
# 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]})
|
|
_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)
|
|
_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})",
|
|
_u == ["main.py:self.new_wd.reason"])
|
|
check("...and '1 times', '(s)', a semicolon and a doubled full stop in a "
|
|
"logged line are caught",
|
|
all(line_problems(t) for t in ("Cut back 1 times so far.",
|
|
"2 malformed point row(s) dropped.",
|
|
"It failed; retrying.",
|
|
"Cut back to +2000 W.. Only this.",
|
|
"index 0, 1, .... Those are left out.",
|
|
"moved to one part: . Problems")))
|
|
check("...and so is the 2d-style line, a 2-sentence reason inside a title and "
|
|
"two tails, which reads as 4 sentences although the reason passes",
|
|
logged_shape("Power request too high: OUTPUT_CLAMP: cut to +5000 W. The "
|
|
"command was +1000 W. Control is still running. Cut back "
|
|
"1 time so far.")
|
|
and not reason_problems("OUTPUT_CLAMP: cut to +5000 W. The command was "
|
|
"+1000 W"))
|
|
check("...while '1 s' and '1 time' are fine",
|
|
not line_problems("Held for 1 s, cut back 1 time so far."))
|
|
# INFO lines are not held to the length rule, but the 2d test found "§8.3",
|
|
# ISO timestamps and "keys ARE" in them. Their literals may carry none.
|
|
INFO_JARGON = re.compile(r"§\s?\d|isoformat|\b(?:ARE|NOT|AND)\b")
|
|
_info_bad = [f"{f}:{n.lineno}" for f, t in _FILES for n in ast.walk(t)
|
|
if isinstance(n, ast.Call) and n.args
|
|
and re.search(r"(?:^|\.)(?:_LOG|log|logger|LOG)\.info$",
|
|
ast.unparse(n.func))
|
|
for _n, s in strings_in(n.args[0])
|
|
if INFO_JARGON.search(ast.unparse(n.args[0])) or SHOUT.search(s)]
|
|
check(f"no INFO line carries a section number, an ISO timestamp or capitals "
|
|
f"(found: {_info_bad})", not _info_bad)
|
|
check("...and one that did is caught",
|
|
bool(INFO_JARGON.search('"§8.3 price history: %s"'))
|
|
and bool(SHOUT.search("the flat keys ARE NOT that pack's history")))
|
|
|
|
|
|
# ============================================================================== #
|
|
print("\n7. log levels, log noise and the display bugs (UX-10)")
|
|
import logging # noqa: E402
|
|
|
|
from app import hass as _hass, p1 as _p1, validate as _validate, web as _web # noqa: E402
|
|
from app.arbiter import Claim # noqa: E402
|
|
from app.main import Controller, why_key # noqa: E402
|
|
|
|
|
|
class _Rec(logging.Handler):
|
|
def __init__(self):
|
|
super().__init__(logging.DEBUG)
|
|
self.records = []
|
|
|
|
def emit(self, record):
|
|
self.records.append(record)
|
|
|
|
|
|
def _captured(logger_name, fn):
|
|
rec, lg = _Rec(), logging.getLogger(logger_name)
|
|
old = lg.level
|
|
lg.addHandler(rec)
|
|
lg.setLevel(logging.DEBUG)
|
|
try:
|
|
fn()
|
|
finally:
|
|
lg.removeHandler(rec)
|
|
lg.setLevel(old)
|
|
return rec.records
|
|
|
|
|
|
def _level_of(fname, words_in_line):
|
|
"""The level of the log call in `fname` whose literal holds the words."""
|
|
tree = ast.parse((APP / fname).read_text(encoding="utf-8"))
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Call) and node.args \
|
|
and re.search(r"_LOG\.\w+$", ast.unparse(node.func)):
|
|
text = render(node.args[0]) or ""
|
|
if words_in_line in text:
|
|
return ast.unparse(node.func).rsplit(".", 1)[-1]
|
|
return None
|
|
|
|
|
|
check("the shadow line of every cycle is DEBUG, not INFO in capitals "
|
|
f"({_level_of('main.py', 'Shadow mode: would send')})",
|
|
_level_of("main.py", "Shadow mode: would send") == "debug"
|
|
and _level_of("main.py", "SHADOW MODE: not commanding") is None)
|
|
check("'Shadow mode is on' and 'is off', which follow a setting or a button, "
|
|
f"are INFO ({_level_of('main.py', 'Shadow mode is on')}, "
|
|
f"{_level_of('main.py', 'Shadow mode is off')})",
|
|
_level_of("main.py", "Shadow mode is on") == "info"
|
|
and _level_of("main.py", "Shadow mode is off") == "info")
|
|
_access = {}
|
|
for _lvl in (logging.INFO, logging.DEBUG):
|
|
logging.getLogger("goodwe.web").setLevel(_lvl)
|
|
_access[logging.getLevelName(_lvl)] = _web.access_log()
|
|
logging.getLogger("goodwe.web").setLevel(logging.NOTSET)
|
|
_start_src = " ".join(
|
|
ast.unparse(n) for n in ast.walk(ast.parse(
|
|
(APP / "web.py").read_text(encoding="utf-8")))
|
|
if isinstance(n, ast.AsyncFunctionDef) and n.name == "start")
|
|
check(f"the page's own polls are not logged at INFO: aiohttp's access log is "
|
|
f"off unless the log level is debug ({_access})",
|
|
_access["INFO"] is None
|
|
and isinstance(_access["DEBUG"], logging.Logger)
|
|
and "access_log=access_log()" in _start_src)
|
|
_validate.WARN_ONCE.seen.clear() # part 3 ran the validator already
|
|
_twice = _captured("goodwe.config", lambda: [
|
|
_validate.validate_startup_config(
|
|
dict(_BASE, price_provider="generic_dayahead", static_profile=""),
|
|
-5000, 5000) for _ in range(2)])
|
|
_sp = [r for r in _twice if "static_profile is empty" in r.getMessage()]
|
|
check(f"a settings warning is logged once per start, though the check runs "
|
|
f"twice ({len(_sp)} line(s))", len(_sp) == 1
|
|
and _sp[0].levelno == logging.WARNING)
|
|
|
|
|
|
def _outage():
|
|
o = _hass.Outage()
|
|
o.down("api", "error 502", "read sensor.x: 502, message='Bad Gateway'")
|
|
o.down("api", "error 502", "read sensor.y: 502")
|
|
o.down("websocket", "the meter connection closed",
|
|
"Received message 8:1000 is not str")
|
|
o.up("api")
|
|
o.up("websocket")
|
|
o.down("api", "no connection", "Cannot connect to host supervisor:80")
|
|
|
|
|
|
_out = _captured("goodwe.hass", _outage)
|
|
_warn = [r.getMessage() for r in _out if r.levelno == logging.WARNING]
|
|
_info = [r.getMessage() for r in _out if r.levelno == logging.INFO]
|
|
check(f"a Home Assistant restart is one plain WARNING line, and one INFO line "
|
|
f"when both connections are back ({_warn}, {_info})",
|
|
_warn[:1] == ["Home Assistant is not answering (error 502), perhaps it "
|
|
"is restarting. The add-on keeps trying."]
|
|
and _info == ["Home Assistant answers again."] and len(_warn) == 2)
|
|
check("...and the library's own text is kept, at DEBUG",
|
|
any("Received message 8:1000" in r.getMessage() for r in _out
|
|
if r.levelno == logging.DEBUG)
|
|
and not any("8:1000" in m or "Bad Gateway" in m for m in _warn + _info))
|
|
|
|
|
|
class _Resp400:
|
|
def __init__(self, body):
|
|
self.status, self.reason, self._body = 400, "Bad Request", body
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *a):
|
|
return False
|
|
|
|
async def text(self):
|
|
return self._body
|
|
|
|
|
|
class _Session400:
|
|
def __init__(self, body):
|
|
self.body = body
|
|
|
|
def post(self, *a, **k):
|
|
return _Resp400(self.body)
|
|
|
|
|
|
def _refused(body):
|
|
ha = _hass.HomeAssistant(_Session400(body), token="t")
|
|
ok = []
|
|
recs = _captured("goodwe.hass", lambda: ok.append(asyncio.run(
|
|
ha.call("number", "set_value", {"entity_id": "number.sp", "value": 1}))))
|
|
return (ok[0] if ok else None), [r.getMessage() for r in recs
|
|
if r.levelno == logging.WARNING]
|
|
|
|
|
|
def _one(res):
|
|
"""(write result, the one WARNING line) or a named absence that fails."""
|
|
ok, lines = res
|
|
return ok, (lines[0] if len(lines) == 1 else f"<{len(lines)} lines>")
|
|
|
|
|
|
_bare_ok, _bare = _one(_refused("400: Bad Request"))
|
|
_full_ok, _full = _one(_refused(
|
|
'{"message": "Value 6000 for number.sp is outside valid range"}'))
|
|
check(f"a write Home Assistant refuses while it loads is one plain WARNING, "
|
|
f"still a failed write, without the empty '400: Bad Request' ({_bare!r})",
|
|
_bare_ok is False
|
|
and _bare.startswith("Home Assistant refused number.set_value for "
|
|
"number.sp (error 400, no details).")
|
|
and "Bad Request" not in _bare)
|
|
check(f"...while a body that says more is still quoted ({_full!r})",
|
|
_full_ok is False and "outside valid range" in _full)
|
|
|
|
|
|
class _WS:
|
|
def __init__(self, kind):
|
|
self.kind = kind
|
|
|
|
async def receive(self):
|
|
return type("M", (), {"type": self.kind, "data": "{}"})()
|
|
|
|
|
|
def _closed_ws():
|
|
try:
|
|
asyncio.run(_p1._receive_json(_WS(_p1.aiohttp.WSMsgType.CLOSE)))
|
|
except ConnectionResetError as e:
|
|
return str(e)
|
|
return None
|
|
|
|
|
|
check(f"a socket Home Assistant closed is a plain ConnectionResetError, which "
|
|
f"the reconnect loop reports through that one line ({_closed_ws()!r})",
|
|
(_closed_ws() or "").startswith("Home Assistant closed the connection"))
|
|
|
|
# "Why: Other". Every winner and every safety reason the code can produce has
|
|
# words. The keys are found in the code: each `Claim.set(source, ..., reason)`
|
|
# through main.why_key(), plus main.py's own refusal to split ("unsplit").
|
|
_SOURCES = {}
|
|
for _f in sorted(APP.glob("*.py")):
|
|
_tree = ast.parse(_f.read_text(encoding="utf-8"))
|
|
_consts = _module_constants(_tree)
|
|
for _n in ast.walk(_tree):
|
|
if isinstance(_n, ast.Call) and ast.unparse(_n.func) == "Claim.set" \
|
|
and _n.args:
|
|
_src = _n.args[0]
|
|
_src = (_src.value if isinstance(_src, ast.Constant)
|
|
else _consts.get(getattr(_src, "id", ""), None))
|
|
_rsn = next((k.value for k in _n.keywords if k.arg == "reason"),
|
|
_n.args[3] if len(_n.args) > 3 else None)
|
|
_rsn = _rsn.value if isinstance(_rsn, ast.Constant) else ""
|
|
_SOURCES.setdefault((_src, _rsn), f"{_f.name}:{_n.lineno}")
|
|
_keys = {why_key(s, [Claim.set(s, 0, 0.0, r)]) for s, r in _SOURCES}
|
|
_keys |= {why_key("failsafe", []),
|
|
why_key("failsafe", [Claim.set("loop", 0, 1.0, "tracking")])}
|
|
check(f"the scan found the claims ({len(_SOURCES)}), including the latched "
|
|
f"per-inverter stop the 2d test saw as 'Other'",
|
|
("failsafe", "sign disagreement") in _SOURCES and len(_SOURCES) >= 8)
|
|
_missing_why = sorted(k for k in _keys if k not in messages.WHY_WINNER)
|
|
check(f"every winner and safety reason has words in messages.WHY_WINNER "
|
|
f"(missing: {_missing_why}, keys: {sorted(_keys)})", not _missing_why)
|
|
check("...so the latched stop reads as a safety stop, not 'Other' "
|
|
f"({messages.why(why_key('failsafe', [Claim.set('failsafe', 0, 0.0, 'sign disagreement')]), 0)!r})",
|
|
messages.why(why_key("failsafe", [Claim.set("failsafe", 0, 0.0,
|
|
"sign disagreement")]), 0)
|
|
== "Safety stop, power went the wrong way: 0 W")
|
|
_without = {k: v for k, v in messages.WHY_WINNER.items()
|
|
if k != "sign disagreement"}
|
|
check("...and a table that lost that entry is caught",
|
|
any(k not in _without for k in _keys))
|
|
|
|
|
|
class _H:
|
|
async def number(self, *a, **k):
|
|
return None
|
|
|
|
async def limits(self, *a):
|
|
return (-2000.0, 2000.0, 1.0)
|
|
|
|
async def call(self, *a, **k):
|
|
return True
|
|
|
|
|
|
class _S:
|
|
data = {}
|
|
|
|
def set(self, *a):
|
|
pass
|
|
|
|
def get_time(self, *a):
|
|
return None
|
|
|
|
|
|
class _P:
|
|
enabled = False
|
|
|
|
def publish(self, *a, **k):
|
|
pass
|
|
|
|
|
|
_c = Controller({"setpoint_entity": "number.sp", "meter_source": "off",
|
|
"max_w": 2000, "stale_input_s": 15}, _H(), _S(), _P(),
|
|
stop=asyncio.Event())
|
|
_c.p1_enabled = True
|
|
_c.p1 = _p1.P1Ingest(max_age_s=30.0)
|
|
|
|
|
|
def _meter_row():
|
|
"""The meter row, or a named absence that fails every check below."""
|
|
rows = [r for r in _c.checks() if r.get("id") == "meter"]
|
|
return rows[0] if len(rows) == 1 else {
|
|
"ok": None, "warn": None, "text": f"<{len(rows)} meter rows>"}
|
|
|
|
|
|
_r0 = _meter_row()
|
|
_c.p1.started_mono -= 31.0
|
|
_r1 = _meter_row()
|
|
_c.p1.submit(_p1.make_sample(_p1.SOURCE_HA, 500.0, 0.0, phases=1))
|
|
_r2 = _meter_row()
|
|
check(f"right after a start, before the first meter reading and inside its "
|
|
f"limit, the meter row is amber and waits, not red at 0 s ({_r0})",
|
|
_r0["warn"] and not _r0["ok"]
|
|
and _r0["text"] == "Meter: waiting for the first reading, limit 30 "
|
|
"seconds")
|
|
check(f"...past the limit with still no reading it is red, as before ({_r1})",
|
|
not _r1["ok"] and not _r1["warn"]
|
|
and _r1["text"].startswith("Meter: no reading for 31 seconds"))
|
|
check(f"...and a reading makes it green ({_r2['text']!r})",
|
|
_r2["ok"] and _r2["text"].startswith("Meter: 500 W"))
|
|
|
|
_c.auto = True
|
|
_c.p1_enabled = False
|
|
asyncio.run(_c.cycle())
|
|
_w0 = _c.why_words
|
|
_c._why_waiting_since -= _td(seconds=7)
|
|
_w7 = _c.why_words
|
|
check(f"'A reading is missing for N seconds' counts while control waits, "
|
|
f"instead of the age the last cycle saw ({_w0!r} -> {_w7!r})",
|
|
_w0 == "A reading is missing for 0 seconds, control waits"
|
|
and _w7 == "A reading is missing for 7 seconds, control waits")
|
|
_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")
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED")
|
|
sys.exit(1)
|
|
print("all passed")
|