goodwe-addon/goodwe_controller/app/messages.py

"""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 at its peak reserve"
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 is not following the controller; short moments "
                  "are normal, repeated ones need a check.")


@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]
    also = (f"; {names(others)} {'was' if len(others) == 1 else 'were'} cut too"
            if others else "")
    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{also}. Control is still 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, reserve_unset: 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.
    """
    if soc_unread:
        why = "The battery level cannot be read"
        act = "Switch off large loads, and check the battery level sensor."
    elif reserve_unset:
        why = "No peak reserve is set"
        act = "Switch off large loads, and set a peak reserve 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 limits peak shaving: 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} Control is stopped and every inverter is set to "
                  f"0 W; if the grid power does not change now, switch the "
                  f"inverters off at the isolator. A restart does not clear "
                  f"this: start control 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, so the site is short; "
        f"nothing was 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: start "
        f"control on the add-on page once you know why.",
        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 follows this "
                  f"reading. Check the meter; if it is stuck, stop control on "
                  f"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, reserve_unset=False),
        peak_reserve(5000, soc_unread=False, reserve_unset=False,
                     packs=["inverter-b"]),
        peak_reserve(5000, soc_unread=True, reserve_unset=False),
        peak_reserve(5000, soc_unread=False, reserve_unset=True),
        circulating(["inverter-a"], ["inverter-b"]),
        circulating(),
        not_delivering("inverter-b", "inverter_b"),
        meter_frozen(600),
        meter_frozen(90),
    ]