"""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.") # --- the status page (UX-08 phase 2b) ---------------------------------------- # The rows of the Commissioning list and the banner at the top of the add-on # page. `main.py` decides WHEN a row shows and its colour (ok / warn / bad); # these decide WHAT it says. A row is one sentence of at most 15 words; a red # row may use up to 3 sentences and 30 words. A banner is at most 12 words. # Rows never name a Home Assistant entity: they name the setting in words. # DOCS.md, "Status page", explains every row. NOT_READY = "Not ready" DOCS_STATUS = "Status page" def signed_watts(value) -> str: """`-1200` -> `-1 200 W`: like `watts()`, but the sign is kept.""" return f"{float(value):,.0f} W".replace(",", " ") def seconds(value) -> str: """Always in seconds, for a live count set against its limit.""" n = f"{float(value):.0f}" return f"{n} second" if n == "1" else f"{n} seconds" def times(n) -> str: return "1 time" if int(n) == 1 else f"{int(n)} times" def for_unit(unit) -> str: """` for inverter-b` on a site with several inverters, else nothing.""" return f" for {unit}" if unit else "" def headline(text: str) -> str: """The first sentence of a row, for the banner when the row has no own.""" first = re.split(r"(?<=[.!?])\s+(?=[A-Z0-9])", text.strip(), maxsplit=1)[0] return first.rstrip(".") # Banners (the headline at the top of the page). def banner_not_ready(row: dict) -> str: return f"{NOT_READY} — {row.get('banner') or headline(row['text'])}" def banner_circulating() -> str: return "Inverters worked against each other. Every inverter is set to 0 W." def banner_site_direction() -> str: return "Total command and meter point opposite ways. All inverters at 0 W." def banner_direction(several: bool) -> str: if several: return "Inverter and own power point opposite ways. It is at 0 W." return "Command and meter point opposite ways. Inverter set to 0 W." def banner_not_delivering(units) -> str: units = list(units) verb = "are" if len(units) > 1 else "is" return f"{names(units)} {verb} not delivering. Control keeps running." # Rows: the controller and the meter. def row_controller_starting() -> str: return "Controller: starting" def row_controller_ran(age_s) -> str: return f"Controller: last ran {seconds(age_s)} ago" def row_controller_stalled(age_s, veto_s) -> str: return (f"Controller has not run for {seconds(age_s)}. The control board " f"sets the inverter to 0 W after {seconds(veto_s)}.") def row_meter_stale(age_s, limit_s) -> str: return (f"Meter: no reading for {seconds(age_s)}, limit " f"{seconds(limit_s)}. The add-on log shows why.") def row_meter(net_w, age_s, rejected=0, unchanged_s=None) -> str: return (f"Meter: {signed_watts(net_w)}, {seconds(age_s)} old" + (f", {rejected} readings refused" if rejected else "") + ("" if unchanged_s is None else f", unchanged for {duration(unchanged_s)}")) def row_meter_frozen() -> str: return (f"{T_METER_FROZEN}. Control still follows it. Check the meter, " f"and stop control if it is stuck.") FREEZE_OFF = { "source": "it only works with the HomeWizard local meter source", "rating": "the inverter rating is not set", "range": "the inverter reports no usable power range", } def row_freeze_off(why: str) -> str: """`why` is a key of FREEZE_OFF: why the stuck-meter check cannot run.""" return f"Stuck-meter check: off, {FREEZE_OFF[why]}" def row_freeze_on(hold_s, rail_w, own_limit: bool) -> str: return (f"Stuck-meter check: alerts after {duration(hold_s)} unchanged " f"above {watts(rail_w)}" + (", the inverter's own limit" if own_limit else "")) # Rows: the peak guard. def row_peak_saturated(ceiling_w) -> str: return (f"{T_PEAK_SATURATED}. Switch off large loads to stay under the " f"{watts(ceiling_w)} peak limit.") def row_peak_reserve(ceiling_w, *, soc_unread: bool, floor_unknown: bool, packs=()) -> str: """SAFETY-24/25/32. The second sentence is SAFETY-25 and must stay.""" if soc_unread: why = "Battery level cannot be read" elif floor_unknown: why = "A battery cut-out level is not set" else: why = (f"{names(packs)} battery at its peak reserve" if packs else "Battery at its peak reserve") return (f"{why}, so the {watts(ceiling_w)} peak limit will be passed. Home " f"use still discharges below the reserve, down to the cut-out level.") def row_peak_guard_off() -> str: """No meter source: the 15-minute average cannot be measured.""" return ("Peak guard: off, it needs a meter source instead of the grid " "power sensor") def banner_peak_reserve() -> str: return T_PEAK_RESERVE # Rows: the sensors and the inverter control. LABEL_SOC = "Battery level" LABEL_BATTERY = "Battery power" LABEL_GRID = "Grid power" def row_sensor(label, value, unit=None) -> str: shown = f"{float(value):g} %" if label.startswith(LABEL_SOC) \ else signed_watts(value) return f"{label}{for_unit(unit)}: {shown}" def row_sensor_unset(label, unit=None) -> str: return f"{label}{for_unit(unit)}: no sensor set in the settings" def row_sensor_unusable(label, unit=None) -> str: """SAFETY-56: it may be missing OR held back as stale; say both.""" return (f"{label}{for_unit(unit)}: no usable reading. The sensor may be " f"missing, or its reading is held back as too old. Check it in " f"Developer Tools before replacing it.") def row_control_unset(unit=None) -> str: return f"Inverter control{for_unit(unit)}: not set in the settings" def row_control_missing(unit=None) -> str: return f"Inverter control{for_unit(unit)}: not found in Home Assistant" def row_control(lo_w, hi_w, unit=None) -> str: def w(v): return "?" if v is None else signed_watts(v) return f"Inverter control{for_unit(unit)}: ready, from {w(lo_w)} to {w(hi_w)}" # Rows: the two loop alarms. The names are the notification titles, so the # row and the card always call the same thing by the same name (UX-07 item 1). def row_power_too_high(clamps, count) -> str: c = clamps[0] if clamps else None what = (f"asked {watts(c.asked_w)}{for_unit(c.unit)}, sent " f"{watts(c.sent_w)}. Cut back" if c else "cut back") return (f"{T_OUTPUT_CLAMP}: {what} {times(count)} since start. This is a " f"software fault, so please report it.") def banner_power_too_high() -> str: return T_OUTPUT_CLAMP def row_not_keeping_up(count) -> str: return f"{T_LOOP_BOUND}: {times(count)} since start" def row_before(title, count) -> str: """After an episode clears: how often it happened, so the evidence stays.""" return f"{title}: {times(count)} since start, not now" def row_write_refused(unit=None) -> str: return (f"{unit or 'The inverter'} refused the last command. Check the " f"add-on log.") # Rows: the direction checks (SAFETY-02/30). def row_direction_trip(unit=None) -> str: """Per inverter. `unit` only at several inverters, graded on its own power.""" what = (f"{unit}: command and its own power point opposite ways, so it is" if unit else "Command and meter point opposite ways, so the " "inverter is") return (f"{what} set to 0 W. Find out why, then start control. A restart " f"does not clear this.") def row_direction_check(held_s, hold_s, unit=None) -> str: who = (f"Direction check for {unit}, against its own power" if unit else "Direction check") if held_s: return f"{who}: opposite for {held_s:.0f} of {seconds(hold_s)}" return f"{who}: on" def row_site_direction_trip() -> str: return ("Total command and meter point opposite ways, so every inverter is " "set to 0 W. The cause is the total, not one inverter. Only starting " "control clears this.") def row_site_direction_check(held_s, hold_s) -> str: if held_s: return (f"Site direction check: opposite for {held_s:.0f} of " f"{seconds(hold_s)}") return "Site direction check: on, against the meter" # Rows: circulating power (SAFETY-31) and delivery (SAFETY-42). def row_circulating_trip() -> str: return ("Inverters worked against each other, so every inverter is set to " "0 W. The meter cannot see this. Only starting control clears this.") def row_circulating_check(now_w, held_s, hold_s) -> str: return ("Circulating power check: on" + (f", now {watts(now_w)}" if now_w else "") + (f", for {held_s:.0f} of {seconds(hold_s)}" if held_s else "")) def row_one_inverter(check: str) -> str: return f"{check}: not needed with one inverter" def row_not_delivering(unit) -> str: return (f"{unit} is not delivering, and only stopping and starting control " f"clears this") def row_delivery_check(unit, short_s, hold_s) -> str: if short_s: return (f"Delivery check for {unit}: short for {short_s:.0f} of " f"{seconds(hold_s)}") return f"Delivery check for {unit}: on" def row_no_estop() -> str: return ("No emergency stop fitted: if this computer fails, the battery " "keeps its last command") # Rows: solar production (TEL-11/12/14). Sources are numbered in the order of # the solar sensor list in the settings. def source(n, of) -> str: return "the sensor" if of == 1 else f"source {n}" def pv_problem(why: str) -> str: """`pv.watts()`'s reason, in plain words.""" w = str(why) if w.startswith("no state"): return "the sensor is missing" if w.startswith("entity is"): return "the sensor is unavailable" if w.endswith("is not numeric"): return "the reading is not a number" if w.startswith("no unit"): return "the sensor has no unit" if w.startswith("unit "): return "the sensor's unit is not a power unit" return "the reading is not usable" def row_pv(total_w, count, reversed_=False) -> str: return (f"Solar production: {signed_watts(total_w)}" + (f" from {count} sources" if count > 1 else "") + (", sign reversed" if reversed_ else "")) def row_pv_none(why) -> str: return f"Solar production: no reading, {pv_problem(why)}" def row_pv_partial(missing, count) -> str: """`missing` are the 1-based positions of the sources that did not report.""" which = names([str(n) for n in missing]) return (f"Solar production: no total, source{'s' if len(missing) > 1 else ''} " f"{which} of {count} {'have' if len(missing) > 1 else 'has'} no " f"reading") def row_pv_single_ignored() -> str: return ("Solar production: clear the single sensor setting, the list " "replaces it") def row_pv_negative(n, count, value_w, flipped: bool) -> str: fix = ("remove it from" if flipped else "add it to") who = source(n, count) return (f"Solar production: {who} reads {signed_watts(value_w)}, " f"so {fix} the reversed-sign list") def row_pv_stray(n) -> str: return (f"Solar production: entry {n} of the reversed-sign list is not a " f"solar sensor in use") RED, AMBER, GREEN = "bad", "warn", "ok" def status_samples() -> list: """`(function name, colour, text)` for every row and banner, for the test. The colour is the one main.py gives the row; a red row may use three sentences. Banners are listed with the colour None. """ clamp = [Clamp(None, 7412, 5000, "rating")] return [ ("banner_not_ready", None, banner_not_ready( {"text": row_meter_frozen()})), ("banner_not_ready", None, banner_not_ready( {"text": "x", "banner": banner_peak_reserve()})), ("banner_circulating", None, banner_circulating()), ("banner_site_direction", None, banner_site_direction()), ("banner_direction", None, banner_direction(False)), ("banner_direction", None, banner_direction(True)), ("banner_not_delivering", None, banner_not_delivering(["inverter-b"])), ("banner_peak_reserve", None, banner_peak_reserve()), ("banner_power_too_high", None, banner_power_too_high()), ("row_controller_starting", GREEN, row_controller_starting()), ("row_controller_ran", GREEN, row_controller_ran(2)), ("row_controller_stalled", RED, row_controller_stalled(45, 30)), ("row_meter_stale", RED, row_meter_stale(40, 10)), ("row_meter", GREEN, row_meter(-350, 2, 3, 340)), ("row_meter_frozen", RED, row_meter_frozen()), *(("row_freeze_off", AMBER, row_freeze_off(k)) for k in FREEZE_OFF), ("row_freeze_on", GREEN, row_freeze_on(600, 4465, True)), ("row_peak_saturated", RED, row_peak_saturated(5000)), ("row_peak_guard_off", AMBER, row_peak_guard_off()), ("row_peak_reserve", RED, row_peak_reserve( 5000, soc_unread=False, floor_unknown=False, packs=["inverter-b"])), ("row_peak_reserve", RED, row_peak_reserve( 5000, soc_unread=True, floor_unknown=False)), ("row_peak_reserve", RED, row_peak_reserve( 5000, soc_unread=False, floor_unknown=True)), ("row_sensor", GREEN, row_sensor(LABEL_SOC, 54, "inverter-b")), ("row_sensor", GREEN, row_sensor(LABEL_BATTERY, -1200.0, "inverter-b")), ("row_sensor_unset", RED, row_sensor_unset(LABEL_GRID)), ("row_sensor_unusable", RED, row_sensor_unusable(LABEL_BATTERY, "inverter-b")), ("row_control_unset", RED, row_control_unset("inverter-b")), ("row_control_missing", RED, row_control_missing("inverter-b")), ("row_control", GREEN, row_control(-5000, 5000, "inverter-b")), ("row_power_too_high", RED, row_power_too_high(clamp, 3)), ("row_power_too_high", RED, row_power_too_high( [Clamp("inverter-b", 6000, 4600, "max_w+device")], 12)), ("row_power_too_high", RED, row_power_too_high((), 1)), ("row_not_keeping_up", AMBER, row_not_keeping_up(48)), ("row_before", AMBER, row_before(T_OUTPUT_CLAMP, 2)), ("row_before", AMBER, row_before(T_LOOP_BOUND, 1)), ("row_write_refused", RED, row_write_refused("inverter-b")), ("row_direction_trip", RED, row_direction_trip()), ("row_direction_trip", RED, row_direction_trip("inverter-b")), ("row_direction_check", GREEN, row_direction_check(0, 60)), ("row_direction_check", GREEN, row_direction_check(12, 60, "inverter-b")), ("row_site_direction_trip", RED, row_site_direction_trip()), ("row_site_direction_check", GREEN, row_site_direction_check(0, 60)), ("row_site_direction_check", GREEN, row_site_direction_check(5, 60)), ("row_circulating_trip", RED, row_circulating_trip()), ("row_circulating_check", GREEN, row_circulating_check(1200, 20, 60)), ("row_one_inverter", GREEN, row_one_inverter("Delivery check")), ("row_not_delivering", AMBER, row_not_delivering("inverter-b")), ("row_delivery_check", GREEN, row_delivery_check("inverter-b", 30, 120)), ("row_delivery_check", GREEN, row_delivery_check("inverter-b", 0, 120)), ("row_no_estop", AMBER, row_no_estop()), ("row_pv", GREEN, row_pv(3200, 2)), ("row_pv", GREEN, row_pv(1800, 1, True)), ("row_pv_none", AMBER, row_pv_none("no unit_of_measurement on the " "entity - it cannot be read in watts")), ("row_pv_partial", AMBER, row_pv_partial([2, 3], 3)), ("row_pv_partial", AMBER, row_pv_partial([2], 2)), ("row_pv_single_ignored", AMBER, row_pv_single_ignored()), ("row_pv_negative", AMBER, row_pv_negative(2, 3, -1500, False)), ("row_pv_negative", AMBER, row_pv_negative(1, 1, -1500, True)), ("row_pv_stray", AMBER, row_pv_stray(2)), ] 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), ]