kota-technician-tool/kota_tech/diag/kota.py

"""Knowledge about the structured "KOTA:" lines of firmware 1.0.16+.

Firmware source of truth (kota-sensorboard/main):
  boot_info.cpp  "boot fw=.. build=.. hw=.. mac=.. reset=.. mode=.. commissioned=.. last_restart=.."
                 "restart reason=.." (W, printed just before a self-restart)
  status.cpp     "status A -> B [code=N] reason=x", "led color=.. pattern=.. [n=..|period=..ms]",
                 "led previous_boot status=.. code=..", "led restart_flash color=.."
  i2c.cpp        "i2c SDA=1 SCL=1 0x39=AS7341 0x68=DS1307(missing) .. [bus=stuck]"
  ds1307.cpp     "rtc running=1 time=..", "rtc read_failed err=..", "rtc verify running=1 delta=2s ok=1"
  sd_card.cpp    "sd type=.. size=.. fs=.. free=.. env_keys=A,B buffer_lines=N", "sd mounted=0 [err=..]"
  main.cpp       "matter fabrics=N window=open|closed qr=MT:.. manual=.."
  microphone.cpp "mic enabled=0 (GPIO3=1, expected 0)"
  health.cpp     "health up=60s status=OK wifi=connected rssi=-61 time_valid=1 sd=ok buffer=0
                  heap=123456 psram=7.8M last_upload=12s|never ota=registered|unregistered|selfcheck"

Error IDs ("[E-SD-03] ...") come from error_ids.csv, a copy of the
firmware's docs/error_ids.csv (the firmware build generates its IDs from it).
"""

from __future__ import annotations

import csv
from dataclasses import dataclass
from pathlib import Path

ERROR_IDS_PATH = Path(__file__).with_name("error_ids.csv")

# Firmware colour names (status.cpp colors[]) -> engine LED keys.
LED_COLOURS = {
    "green": "GREEN", "cyan": "CYAN", "white": "WHITE", "red": "RED",
    "blue": "BLUE", "purple": "PURPLE", "yellow": "YELLOW", "orange": "ORANGE",
}

RED_CODES = {1: "BME688", 2: "SEN66", 3: "AS7341", 4: "Microphone", 5: "Flicker sensor", 6: "Self-check"}
I2C_DEVICES = {"0x39": "AS7341", "0x68": "DS1307", "0x6B": "SEN66", "0x76": "BME688"}

# last_restart values written by kota_restart() that are part of normal
# operation (button presses, a finished setup, an installed update).
INTENTIONAL_RESTARTS = {
    "ota_update", "button_provisioning_abort", "button_wifi_cleared",
    "button_provisioning_toggle", "provisioning_saved",
    "usb_wifi_set", "usb_wifi_clear",        # Wi-Fi changed from the technician tool (usb_console)
}

MODULES = {
    "SD": "SD card", "ENV": "ENV.TXT", "INF": "InfluxDB upload", "OTA": "Firmware update",
    "WIFI": "Wi-Fi", "PROV": "Wi-Fi setup", "MAIN": "Start-up", "STAT": "Status LED", "SPI": "SPI bus",
    "RTC": "Clock (RTC)", "BME": "BME688", "SEN": "SEN66", "AS": "AS7341", "MIC": "Microphone",
    "FLK": "Flicker sensor", "DEV": "Device", "GATH": "Sensor readings",
}


@dataclass(frozen=True)
class ErrorId:
    id: str
    module: str
    severity: str        # error | warning | retired
    message: str
    action: str

    @property
    def module_name(self) -> str:
        return MODULES.get(self.module, self.module)


def load_error_ids(path: Path = ERROR_IDS_PATH) -> dict[str, ErrorId]:
    if not path.exists():
        return {}
    with path.open(encoding="utf-8", newline="") as fh:
        return {r["id"]: ErrorId(r["id"], r["module"], r["severity"], r["message"], r["action"])
                for r in csv.DictReader(fh)}


def led_for_status(status: str, code: int) -> tuple[str, str]:
    """(LED key, pattern text) for a status - mirrors Status::pattern_for().
    Used when the box's own "led" line was not captured (connected mid-run)."""
    if status == "OK":
        return "GREEN", "solid"
    if status == "ERROR":
        return ("RED", f"{code} pulses") if 1 <= code <= 6 else ("RED", "solid")
    if status == "OUT_OF_RESOURCES":
        return "RED", "solid"
    if status == "UPDATING":
        return "YELLOW", "blinking"
    if status == "SELF_CHECK":
        return "ORANGE", "solid"
    if status == "CONNECTION_ERROR":
        return ("PURPLE", "blinking") if code == 1 else ("PURPLE", "solid")
    if status == "RTC_ERROR":
        return "BLUE", "2 pulses" if code == 2 else "1 pulse"
    if status == "SD_ERROR":
        return "WHITE", "fast blinking"
    if status == "RESETING":
        return "WHITE", "solid"
    if status == "HTTP_SERVER":
        return "WHITE", "breathing"
    return "UNKNOWN", ""


def pattern_text(fields: dict[str, str]) -> str:
    """'led color=red pattern=pulses n=2' -> '2 pulses'."""
    p = fields.get("pattern", "")
    if p == "pulses":
        n = fields.get("n", "?")
        return f"{n} pulse" if n == "1" else f"{n} pulses"
    if p == "blink":
        period = fields.get("period", "").removesuffix("ms")
        fast = period.isdigit() and int(period) <= 500
        return ("fast blinking" if fast else "blinking") + (f" ({period} ms)" if period else "")
    if p == "breathe":
        return "breathing"
    return p   # solid / off


def led_meaning(colour: str, pattern: str) -> str:
    """Technician meaning of a colour + pattern (firmware diagnostics doc)."""
    if colour == "GREEN":
        return "OK"
    if colour == "CYAN":
        return "phone connected for pairing" if "fast" in pattern else "waiting to be paired in Matter"
    if colour == "WHITE":
        if "fast" in pattern:
            return "SD card or ENV.TXT error"
        if "breath" in pattern:
            return "Wi-Fi setup portal"
        return "factory reset"
    if colour == "RED":
        if "pulse" in pattern:
            n = int(pattern.split()[0]) if pattern.split()[0].isdigit() else 0
            return f"{RED_CODES.get(n, 'sensor')} failed"
        return "internal error / out of memory"
    if colour == "BLUE":
        return "I2C bus stuck (sensor cable)" if pattern.startswith("2") else "clock (RTC) not responding"
    if colour == "PURPLE":
        return "server rejected the box (check ENV.TXT credentials)" if "blink" in pattern else "no network"
    if colour == "YELLOW":
        return "downloading a firmware update"
    if colour == "ORANGE":
        return "self-check after an update"
    return ""


# Finding for the current firmware status (Status::set_status). key suffix ->
# (severity, title, meaning, actions, explained_by). The status finding is
# grouped under a more specific finding when one explains it.
_SENSOR_ROOTS = ("sensor.init_failed", "sensor.selfcheck_failed", "kota.i2c_stuck", "kota.sensor_missing",
                 "sen66.not_ready", "bme688.read_fail", "as7341.read_fail", "mic.read_fail",
                 "ota.selfcheck_failed")
_NET_ROOTS = ("matter.not_commissioned", "wifi.disconnected", "wifi.setup_portal", "kota.wifi_down",
              "net.dns_fail", "net.tls_fail")


def status_finding(status: str, code: int) -> tuple[str, str, str, str, tuple[str, ...], tuple[str, ...]] | None:
    """(key, severity, title, meaning, actions, explained_by) or None for healthy states."""
    if status == "ERROR":
        if 1 <= code <= 5:
            s = RED_CODES[code]
            return (f"status.error.{code}", "critical", f"{s} fault (LED red, {code} pulses)",
                    f"The firmware reports the {s} as failed.",
                    (f"Open the case and check the {s} connector/cable.",
                     "If several sensors fail: check the shared I2C cable.",
                     "The box restarts itself after 150 s in this state."), _SENSOR_ROOTS)
        if code == 6:
            return ("status.error.6", "error", "Post-update self-check failed (LED red, 6 pulses)",
                    "The new firmware failed its self-check and rolls back.",
                    ("Check the sensor named in the log (inside the case), then power-cycle.",), _SENSOR_ROOTS)
        return ("status.error", "error", "Internal error (LED red)",
                "The firmware is in its general error state.",
                ("Save the session log and send it to the firmware team.",
                 "The box restarts itself after 150 s in this state."), _SENSOR_ROOTS)
    if status == "OUT_OF_RESOURCES":
        return ("status.oom", "error", "Firmware ran out of memory (LED red)",
                "The firmware could not allocate memory; it restarts itself after 30 s.",
                ("Power-cycle. If it repeats, save the session log for the firmware team.",),
                ("sys.out_of_resources",))
    if status == "CONNECTION_ERROR":
        if code == 1:
            return ("status.conn.1", "critical", "Server rejected the box (LED purple, blinking)",
                    "The network works but the server refused the box (HTTP 401/403/404): "
                    "INFLUX_URL / INFLUX_TOKEN in ENV.TXT are wrong or revoked.",
                    ("Re-provision the SD card with the current ENV.TXT (SD card is inside the case).",),
                    ("influx.auth", "influx.not_found"))
        return ("status.conn", "error", "No connection to the servers (LED purple)",
                "Uploads or the update check fail: no Wi-Fi, no Internet or the server is unreachable.",
                ("Check the box is paired/connected to Wi-Fi and the network has Internet.",),
                _NET_ROOTS + ("influx.buffer_fail", "ota.check_failed", "influx.server_error"))
    if status == "RTC_ERROR":
        if code == 2:
            return ("status.rtc.2", "critical", "I2C bus stuck (LED blue, 2 pulses)",
                    "SDA or SCL is held low: a shorted or broken sensor cable, or a faulty sensor.",
                    ("Open the case and check the I2C cable between the boards.",
                     "Unplug sensors one by one to find the one holding the bus."), ("kota.i2c_stuck",))
        return ("status.rtc.1", "critical", "Clock chip (RTC) not responding (LED blue, 1 pulse)",
                "The DS1307 clock does not answer on the I2C bus.",
                ("Open the case and check the I2C cable/connector between the boards.",
                 "If the cable is fine: the RTC module is faulty."),
                ("rtc.init_failed", "kota.rtc_read_failed", "kota.sensor_missing"))
    if status == "SD_ERROR":
        return ("status.sd", "critical", "SD card or ENV.TXT error (LED white, fast blinking)",
                "The SD card cannot be read or ENV.TXT is missing settings.",
                ("Open the case, re-seat the SD card or re-provision it with the tool.",),
                ("kota.sd_unmounted", "influx.env_missing", "sd.card_fail", "sd.mount_fail",
                 "sd.env_open_fail", "influx.init_fail"))
    if status == "RESETING":
        return ("status.reset", "warning", "Factory reset in progress (LED white, solid)",
                "The Matter button was held: the box forgets its Matter network.",
                ("Pair the box again in the Matter app afterwards.",), ("sys.factory_reset",))
    if status == "HTTP_SERVER":
        return ("status.portal", "warning", "Box is in Wi-Fi setup mode (LED white, breathing)",
                "The box hosts its own Wi-Fi network for configuration.",
                ("Connect a phone to the box's network and open http://sensorbox.local/,",
                 "or short-press the Matter button to leave setup mode."), ("wifi.setup_portal",))
    return None   # OK, UPDATING, SELF_CHECK: covered by milestones / OTA rules


# last_restart -> (severity, title, meaning, actions); None = not a fault.
def restart_finding(reason: str) -> tuple[str, str, str, tuple[str, ...]] | None:
    if reason in INTENTIONAL_RESTARTS or reason in ("none", "", "-"):
        return None
    if reason == "status_watchdog":
        return ("warning", "Box restarted itself after 150 s in the red error state",
                "By design the firmware restarts when it stays in the error state (red) for 150 s.",
                ("Fix the grouped cause; the restarts stop by themselves.",))
    if reason == "status_oom_watchdog":
        return ("error", "Box restarted itself because it ran out of memory",
                "The firmware was out of memory for 30 s and restarted.",
                ("Save the session log for the firmware team.",))
    if reason == "provisioning_failed":
        return ("warning", "Wi-Fi setup portal failed to start",
                "The setup portal could not start; the box cleared setup mode and restarted.",
                ("Retry Wi-Fi setup (short button press); restart the device if it persists.",))
    if reason == "panic":
        return ("critical", "Firmware crashed (panic) and restarted",
                "The firmware hit a fatal error.",
                ("Press Reset & capture to record the crash; save the session log for the firmware team.",))
    if reason in ("task_wdt", "int_wdt", "wdt"):
        return ("error", "Firmware froze and was restarted by the watchdog",
                f"The firmware stopped responding ({reason}); the watchdog restarted it.",
                ("Save the session log for the firmware team.",))
    if reason == "brownout":
        return ("error", "Power supply voltage dropped (brown-out)",
                "The supply voltage fell too low and the chip reset itself.",
                ("Check the power supply, cable and connectors.",))
    if reason == "unknown_sw":
        return ("warning", "Unexpected software restart",
                "The firmware restarted without recording a reason.",
                ("Watch whether it repeats; save the session log for the firmware team.",))
    return ("warning", f"Box restarted itself ({reason})",
            f"The firmware recorded the restart reason '{reason}'.",
            ("Save the session log for the firmware team if it repeats.",))


def parse_size_gb(value: str) -> float | None:
    """'7.3GB' -> 7.3 (sd_card.cpp format_gb)."""
    v = value.upper().removesuffix("GB")
    try:
        return float(v)
    except ValueError:
        return None


def parse_seconds(value: str) -> int | None:
    v = value.removesuffix("s")
    return int(v) if v.lstrip("-").isdigit() else None