adminandClaude Opus 5.5
7b65a3e69a
- not delivering: the body said control keeps running and then "start
control"; clearing it takes stop then start, and it now says so.
- peak reserve: the "reserve not set" branch is really soc_floor None,
i.e. an inverter's cut-out level missing; it now names that. The title
is true on all three branches ("Battery stopped limiting the peak").
- no semicolon run-ons in bodies (STYLE.md + test); loop-bound, circulating,
meter-frozen and multi-unit over-rating bodies rewritten to match.
- allowlist cap is now exact, so a shrink cannot leave room to regrow.
- DOCS names the settings behind "peak reserve" and "cut-out level";
DECISIONS no longer states the HA anchor behaviour as fact.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
251 lines
10 KiB
Python
251 lines
10 KiB
Python
"""Every Home Assistant notification this add-on raises, in plain words (UX-08).
|
|
|
|
The rule is `STYLE.md` at the repository root; `test_messages.py` enforces it.
|
|
Titles, bodies and severities live here and nowhere else, so `main.py` only
|
|
decides WHEN to raise a notice and this file decides WHAT it says. The detail
|
|
that used to fill the bodies is in DOCS.md, under "Notifications", one section
|
|
per title; the body ends with a link to it.
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass
|
|
|
|
# Severity (STYLE.md section 3). An ALARM may use up to three sentences so
|
|
# that no action a person must take is lost; everything else gets one.
|
|
ALARM, WARNING, INFO, RESOLVED = "alarm", "warning", "info", "resolved"
|
|
|
|
DOCS_SECTION = "Notifications"
|
|
|
|
|
|
def watts(value) -> str:
|
|
"""`7412` -> `7 412 W`: a magnitude, grouped for reading, no sign."""
|
|
return f"{abs(float(value)):,.0f} W".replace(",", " ")
|
|
|
|
|
|
def duration(seconds) -> str:
|
|
s = float(seconds)
|
|
if s < 120:
|
|
return f"{s:.0f} seconds"
|
|
return f"{s / 60:.0f} minutes"
|
|
|
|
|
|
def names(items) -> str:
|
|
items = list(items)
|
|
if len(items) <= 1:
|
|
return "".join(items)
|
|
return ", ".join(items[:-1]) + " and " + items[-1]
|
|
|
|
|
|
def addon_slug() -> str | None:
|
|
"""This add-on's Supervisor slug, from the container hostname.
|
|
|
|
The Supervisor names an add-on's host after its slug with `_` spelled `-`
|
|
(`local_goodwe_controller` -> `local-goodwe-controller`). Anything else,
|
|
e.g. a test run, gets no link rather than a wrong one.
|
|
"""
|
|
host = os.environ.get("HOSTNAME", "")
|
|
if re.fullmatch(r"[a-z0-9]+-goodwe-controller", host):
|
|
return host.replace("-", "_")
|
|
return None
|
|
|
|
|
|
def details(title: str) -> str:
|
|
where = f"section “{title}” under {DOCS_SECTION}"
|
|
slug = addon_slug()
|
|
if slug is None:
|
|
return f"Details: add-on documentation, {where}."
|
|
return (f"Details: [add-on documentation](/hassio/addon/{slug}/documentation),"
|
|
f" {where}.")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Notice:
|
|
id: str
|
|
severity: str
|
|
title: str
|
|
body: str
|
|
section: str = "" # the DOCS.md heading, when it is not the title itself
|
|
|
|
@property
|
|
def docs(self) -> str:
|
|
return self.section or self.title
|
|
|
|
def data(self) -> dict:
|
|
"""The `persistent_notification.create` payload."""
|
|
return {"notification_id": self.id, "title": self.title,
|
|
"message": f"{self.body}\n\n{details(self.docs)}"}
|
|
|
|
|
|
# --- the catalogue ----------------------------------------------------------
|
|
# Each title below is also a DOCS.md heading (test_messages.py checks it).
|
|
|
|
T_LOOP_BOUND = "Battery is not keeping up"
|
|
T_OUTPUT_CLAMP = "Power request too high"
|
|
T_PEAK_SATURATED = "Load too high for battery"
|
|
T_PEAK_RESERVE = "Battery stopped limiting the peak"
|
|
T_CIRCULATING = "Inverters working against each other"
|
|
T_NOT_DELIVERING = "{unit} is not delivering"
|
|
S_NOT_DELIVERING = "An inverter is not delivering"
|
|
T_METER_FROZEN = "Meter reading may be stuck"
|
|
|
|
ID_LOOP_BOUND = "goodwe_integrator_bound"
|
|
ID_OUTPUT_CLAMP = "goodwe_output_clamp"
|
|
ID_PEAK_SATURATED = "goodwe_peak_saturated"
|
|
ID_PEAK_RESERVE = "goodwe_peak_soc_floor"
|
|
ID_CIRCULATING = "goodwe_circulating_power"
|
|
ID_NOT_DELIVERING = "goodwe_not_delivering_{slug}"
|
|
ID_METER_FROZEN = "goodwe_meter_frozen"
|
|
|
|
# Which limit cut a command (clamp.ClampResult.bound, plus the loop's own).
|
|
LIMIT_WORDS = {
|
|
"max_w": "the configured inverter rating",
|
|
"device": "the inverter's own limit",
|
|
"max_w+device": "the configured inverter rating and the inverter's own limit",
|
|
"rating": "the inverter rating",
|
|
}
|
|
|
|
|
|
def loop_bound() -> Notice:
|
|
"""SAFETY-40: the integrator bound. Normal in short bursts; not a fault."""
|
|
return Notice(ID_LOOP_BOUND, WARNING, T_LOOP_BOUND,
|
|
"The battery cannot give what the controller asks, which is "
|
|
"normal for a moment but needs a check if it keeps happening.")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Clamp:
|
|
"""One command that was cut: what was asked, what was sent, and by what."""
|
|
unit: str | None
|
|
asked_w: float
|
|
sent_w: float
|
|
limit: str # a key of LIMIT_WORDS
|
|
|
|
|
|
def output_clamp(clamps) -> Notice:
|
|
"""SAFETY-03: a computed command above the limit. A software fault.
|
|
|
|
Asked comes before sent, always - reading them the other way round turns
|
|
the alarm backwards (test_clamp.py part 6).
|
|
"""
|
|
if not clamps:
|
|
return Notice(ID_OUTPUT_CLAMP, ALARM, T_OUTPUT_CLAMP,
|
|
"The controller asked for more power than the inverter "
|
|
"rating and was cut back. Control is still running. This "
|
|
"is a software fault, so please report it to your "
|
|
"installer.")
|
|
c = clamps[0]
|
|
who = f"For {c.unit}, the" if c.unit else "The"
|
|
others = [x.unit for x in clamps[1:] if x.unit]
|
|
running = (f"{names(others)} {'was' if len(others) == 1 else 'were'} cut "
|
|
f"too, and control is still running." if others
|
|
else "Control is still running.")
|
|
return Notice(ID_OUTPUT_CLAMP, ALARM, T_OUTPUT_CLAMP,
|
|
f"{who} controller asked for {watts(c.asked_w)}, above "
|
|
f"{LIMIT_WORDS.get(c.limit, LIMIT_WORDS['rating'])}, and sent "
|
|
f"{watts(c.sent_w)} instead. {running} "
|
|
f"This is a software fault, so please report it to your "
|
|
f"installer.")
|
|
|
|
|
|
def peak_saturated(ceiling_w) -> Notice:
|
|
"""SAFETY-07: the battery gives all it can and the peak still goes up."""
|
|
return Notice(ID_PEAK_SATURATED, ALARM, T_PEAK_SATURATED,
|
|
f"The battery is giving all it can, but this 15-minute period "
|
|
f"will still go over your {watts(ceiling_w)} peak limit and "
|
|
f"raise this month's peak. Switch off large loads in the next "
|
|
f"few minutes. This is not a fault: the load is bigger than "
|
|
f"the inverter.")
|
|
|
|
|
|
def peak_reserve(ceiling_w, *, soc_unread: bool, floor_unknown: bool,
|
|
packs=()) -> Notice:
|
|
"""SAFETY-24/25/32: the peak guard may not discharge any further.
|
|
|
|
The last sentence is SAFETY-25 and must stay: the reserve binds the peak
|
|
guard only, and normal home use goes on discharging below it.
|
|
`packs` names the batteries at their limit, only on a site with several.
|
|
`floor_unknown`: the floor the guard grades against is None, which at a
|
|
roster means some inverter's cut-out level is missing (the peak reserve
|
|
itself always has a value) - so the action names the cut-out level.
|
|
"""
|
|
if soc_unread:
|
|
why = "The battery level cannot be read"
|
|
act = "Switch off large loads, and check the battery level sensor."
|
|
elif floor_unknown:
|
|
why = "The cut-out level of a battery is not known"
|
|
act = ("Switch off large loads, and set each battery's cut-out level "
|
|
"in the settings.")
|
|
else:
|
|
why = (f"The battery of {names(packs)} is at its peak reserve" if packs
|
|
else "The battery is at its peak reserve")
|
|
act = ("Switch off large loads, or lower the peak reserve if the "
|
|
"battery allows it.")
|
|
return Notice(ID_PEAK_RESERVE, ALARM, T_PEAK_RESERVE,
|
|
f"{why}, so this 15-minute period will go over your "
|
|
f"{watts(ceiling_w)} peak limit and raise this month's peak. "
|
|
f"{act} The reserve only applies to keeping the peak down: normal home use "
|
|
f"can still discharge the battery below it, down to its "
|
|
f"cut-out level.")
|
|
|
|
|
|
def circulating(discharging=(), charging=()) -> Notice:
|
|
"""SAFETY-31: the inverters push power round each other. Control stopped."""
|
|
if discharging and charging:
|
|
what = (f"{names(discharging)} was discharging while "
|
|
f"{names(charging)} was charging, and your meter cannot see "
|
|
f"this.")
|
|
else:
|
|
what = ("One inverter was charging from another, and your meter "
|
|
"cannot see this.")
|
|
return Notice(ID_CIRCULATING, ALARM, T_CIRCULATING,
|
|
f"{what} Every inverter is set to 0 W: if the grid power does "
|
|
f"not change now, switch the inverters off at the isolator. "
|
|
f"Control stays stopped, even after a restart, until you "
|
|
f"start it on the add-on page once you know why.")
|
|
|
|
|
|
def not_delivering(unit: str, slug: str) -> Notice:
|
|
"""SAFETY-42: one inverter gives less than its share. Reports, never acts."""
|
|
return Notice(
|
|
ID_NOT_DELIVERING.format(slug=slug), ALARM,
|
|
T_NOT_DELIVERING.format(unit=unit),
|
|
f"{unit} gives less power than it is asked for, but nothing was "
|
|
f"changed and control keeps running. Check that {unit} is "
|
|
f"on and connected, that its battery is not empty or cold, and that "
|
|
f"nothing else controls it. A restart does not clear this: once you "
|
|
f"know why, stop and start control on the add-on page.",
|
|
section=S_NOT_DELIVERING)
|
|
|
|
|
|
def meter_frozen(unchanged_s) -> Notice:
|
|
"""SAFETY-18: the meter answers but its value does not move."""
|
|
return Notice(ID_METER_FROZEN, ALARM, T_METER_FROZEN,
|
|
f"The meter has shown the same value for "
|
|
f"{duration(unchanged_s)} while the controller asks the "
|
|
f"battery for full power. Control is still running and still "
|
|
f"follows this reading. Check the meter, and if it is stuck, "
|
|
f"stop control on the add-on page.")
|
|
|
|
|
|
def samples() -> list:
|
|
"""One rendering of every notice, with realistic values, for the test."""
|
|
return [
|
|
loop_bound(),
|
|
output_clamp([Clamp(None, 7412, 5000, "rating")]),
|
|
output_clamp(()),
|
|
output_clamp([Clamp("inverter-a", -3000, -2500, "device"),
|
|
Clamp("inverter-b", 6000, 4600, "max_w+device")]),
|
|
peak_saturated(5000),
|
|
peak_reserve(5000, soc_unread=False, floor_unknown=False),
|
|
peak_reserve(5000, soc_unread=False, floor_unknown=False,
|
|
packs=["inverter-b"]),
|
|
peak_reserve(5000, soc_unread=True, floor_unknown=False),
|
|
peak_reserve(5000, soc_unread=False, floor_unknown=True),
|
|
circulating(["inverter-a"], ["inverter-b"]),
|
|
circulating(),
|
|
not_delivering("inverter-b", "inverter_b"),
|
|
meter_frozen(600),
|
|
meter_frozen(90),
|
|
]
|