kkallas/kota-technician-tool
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Technician tool v0.2

Browse Source 
Imported from the Google Drive folder 004-technician_tool/v0.2.
This commit is contained in:
kkallas committed 2026-09-26 16:53:59 +02:00
1 parent e5a33a1c34
commit e8b72b743d
17 files changed
+2410 -283
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

kota_tech/__init__.py
+1 -1
View File
@@ -1,3 +1,3 @@
"""KOTA sensorbox technician tool."""
__version__ = "0.1.0"
__version__ = "0.2.0"
kota_tech/core/logfile.py
+32
View File
@@ -0,0 +1,32 @@
"""Load saved logs for offline analysis.
Accepts both raw serial captures and this tool's session logs, whose lines
look like "[14:02:11.532] I (1234) TAG: msg" and whose host events look
like "[14:02:11.532] --- reset issued by tool ---".
"""
from __future__ import annotations
import re
from pathlib import Path
from .logline import LineKind, LogLine, parse_line, split_glued
_SESSION_PREFIX = re.compile(r"^\[(\d\d):(\d\d):(\d\d)\.(\d{3})\] ")
_HOST_EVENT = re.compile(r"^--- .* ---$")
def read_log_file(path: Path) -> list[LogLine]:
text = path.read_bytes().decode("utf-8", errors="replace")
out: list[LogLine] = []
for i, raw in enumerate(text.splitlines()):
host_time = i * 0.001 # raw capture: keep order, no wall clock
if m := _SESSION_PREFIX.match(raw):
h, mi, s, ms = (int(g) for g in m.groups())
host_time = h * 3600 + mi * 60 + s + ms / 1000
raw = raw[m.end():]
if _HOST_EVENT.match(raw):
out.append(LogLine(raw, LineKind.HOST, host_time))
continue
out.extend(parse_line(part, host_time) for part in split_glued(raw))
return out
kota_tech/core/logline.py
+36 -30
View File
@@ -76,65 +76,71 @@ class LineFramer:
A partial line is released after `idle_flush_s` without new bytes so
prompts / unterminated output are not held back forever.
Lines are also split where a new log line starts in the middle
("I (2302) chip[DL]: I (2302) chip[DL]: Posting..."): the two S3 cores
interleave output, and the USB-Serial-JTAG FIFO keeps the tail of an old
line while no host reads, releasing it glued in front of the next live
output. After a (re)open, the fragment in front of the first such split
is stale and dropped. Timing cannot identify it, because the FIFO only
flushes when the board next prints, possibly seconds later.
"""
# Stale FIFO bytes arrive immediately after open; anything later is live.
STALE_WINDOW_S = 0.25
def __init__(self, idle_flush_s: float = 0.3, max_line: int = 4096,
drop_first_partial: bool = False, now: float | None = None):
drop_first_partial: bool = False):
self._buf = bytearray()
self._last_rx = 0.0
self.idle_flush_s = idle_flush_s
self.max_line = max_line
# The S3 USB-Serial-JTAG FIFO keeps the tail of an old line while no
# host is reading and flushes it on open; drop bytes up to the first
# newline after a (re)connect so it is not glued to a fresh line.
# Only bytes arriving within STALE_WINDOW_S of open are considered
# stale, otherwise an idle board's next boot banner would be lost.
self._dropping = drop_first_partial
self._drop_deadline = (time.monotonic() if now is None else now) + self.STALE_WINDOW_S
self._first_line_pending = drop_first_partial
def cancel_drop(self) -> None:
self._dropping = False
self._first_line_pending = False
def feed(self, data: bytes, now: float | None = None) -> list[str]:
now = time.monotonic() if now is None else now
if self._dropping and now > self._drop_deadline:
self._dropping = False
if data:
self._buf.extend(data)
self._last_rx = now
if self._dropping:
i = self._buf.find(b"\n")
if i < 0:
return []
del self._buf[: i + 1]
self._dropping = False
out: list[str] = []
while (i := self._buf.find(b"\n")) >= 0:
out.append(self._decode(self._buf[:i]))
out += self._emit(self._buf[:i])
del self._buf[: i + 1]
if len(self._buf) >= self.max_line:
out.append(self._decode(self._buf))
out += self._emit(self._buf)
self._buf.clear()
return out
def poll_idle(self, now: float | None = None) -> list[str]:
now = time.monotonic() if now is None else now
if self._buf and now - self._last_rx >= self.idle_flush_s:
line = self._decode(self._buf)
self._buf.clear()
return [line]
return self.flush()
return []
def flush(self) -> list[str]:
if not self._buf:
return []
line = self._decode(self._buf)
out = self._emit(self._buf)
self._buf.clear()
return [line]
return out
@staticmethod
def _decode(b: bytes | bytearray) -> str:
return bytes(b).decode("utf-8", errors="replace")
def _emit(self, raw: bytes | bytearray) -> list[str]:
parts = split_glued(bytes(raw).decode("utf-8", errors="replace"))
if self._first_line_pending:
self._first_line_pending = False
if len(parts) > 1:
parts = parts[1:] # stale FIFO fragment from before the open
return parts
# Start of a log line (optionally ANSI-coloured) or ROM banner.
_LINE_START = re.compile(r"(?:\x1b\[[0-9;]*m)?(?:[EWIDV] \(\d+\) \S|ESP-ROM:)")
def split_glued(text: str) -> list[str]:
# Left-to-right scan so an ANSI colour prefix stays with its line.
cuts = [m.start() for m in _LINE_START.finditer(text) if m.start() > 0]
if not cuts:
return [text]
bounds = [0, *cuts, len(text)]
return [text[a:b] for a, b in zip(bounds, bounds[1:]) if a < b]
kota_tech/core/session.py
+7
View File
@@ -33,6 +33,13 @@ class SessionLog:
self._fh.write(f"[{ts}] {line.text}\n")
self._fh.flush()
def write_report(self, text: str) -> None:
"""Append the diagnosis at the end of the session (for the firmware team)."""
if self._fh is None:
return
self._fh.write("\n===== Diagnosis at end of session =====\n" + text + "\n")
self._fh.flush()
def stop(self) -> None:
if self._fh is not None:
self._fh.close()
kota_tech/diag/__init__.py
View File
Whitespace-only changes.
kota_tech/diag/engine.py
+554
View File
@@ -0,0 +1,554 @@
"""Diagnostics engine: parsed log lines -> findings, milestones, LED, verdict.
Line-level knowledge lives in rules.toml. This module adds what a single
line cannot express:
* boot tracking (ROM banner, reset reason, uptime drops when the banner
was missed) and classification of why the previous boot ended;
* the status-LED model (Status::set_status / set_status_conditional);
* crash capture (panic / ESP_ERROR_CHECK output);
* root-cause grouping via `explained_by` chains.
Findings are session-wide (a reboot loop must keep its cause visible);
milestones and the LED are per boot.
"""
from __future__ import annotations
import re
import tomllib
from collections import Counter, deque
from dataclasses import dataclass, field
from enum import IntEnum
from pathlib import Path
from ..core.logline import LineKind, LogLine
from .firmware import FirmwareInfo
RULES_PATH = Path(__file__).with_name("rules.toml")
class Severity(IntEnum):
IGNORE = -1
INFO = 1
WARNING = 2
ERROR = 3
CRITICAL = 4
@property
def label(self) -> str:
return self.name.capitalize()
# name -> (label, swatch colour)
LED = {
"UNKNOWN": ("Unknown", "#888888"),
"OFF": ("Off", "#444444"),
"GREEN": ("Green", "#2e9e44"),
"PURPLE": ("Purple", "#8e44ad"),
"RED": ("Red", "#d64541"),
"BLUE": ("Blue", "#2f6fd6"),
"YELLOW": ("Yellow", "#e6c229"),
"ORANGE": ("Orange", "#f08c1a"),
"WHITE": ("White", "#f4f4f4"),
}
# (id, label, optional) - optional milestones are listed only once reached.
MILESTONES = [
("rom_boot", "Board booted", False),
("app_started", "Firmware started", False),
("sd_mounted", "SD card mounted", False),
("env_loaded", "ENV.TXT read (InfluxDB settings)", False),
("matter_started", "Matter started", True),
("local_wifi_mode", "Local Wi-Fi mode (SSID from SD card)", True),
("sensors_ready", "All sensors started", False),
("wifi_connected", "Connected to Wi-Fi", False),
("time_synced", "Clock set from the Internet", False),
("data_uploaded", "Data uploaded to InfluxDB", False),
("ota_registered", "Registered for firmware updates", False),
("ota_checked", "Checked for firmware updates", False),
("ota_update_started", "Firmware update started", True),
("ota_update_done", "Firmware update installed", True),
("ota_update_accepted", "Updated firmware accepted", True),
]
# Findings that stop start-up; a crash/reboot right after them is their effect.
_INIT_FAILURES = [
"rtc.init_failed", "sensor.init_failed", "sd.card_fail", "sd.mount_fail",
"sd.env_open_fail", "influx.env_missing", "influx.init_fail",
]
_NETWORK_ROOTS = [
"influx.buffer_fail", "ota.check_failed", "net.dns_fail",
"matter.not_commissioned", "wifi.disconnected",
]
_RESET_BY_TOOL_WINDOW_S = 5.0
_SHORT_BOOT_MS = 60_000
class _Fmt(dict):
def __missing__(self, key: str) -> str:
return "?"
def _fmt(s: str, groups: dict[str, str]) -> str:
return s.format_map(_Fmt(groups)) if s else s
@dataclass(frozen=True)
class Rule:
id: str
pattern: re.Pattern[str]
severity: Severity
title: str = ""
meaning: str = ""
actions: tuple[str, ...] = ()
explained_by: tuple[str, ...] = ()
milestone: str | None = None
clears: tuple[str, ...] = ()
led: str | None = None
led_from: str | None = None
split_by: str | None = None
min_count: int = 1
min_uptime_ms: int = 0
def load_rules(path: Path = RULES_PATH) -> list[Rule]:
data = tomllib.loads(path.read_text(encoding="utf-8"))
rules = []
for r in data["rule"]:
rules.append(Rule(
id=r["id"],
pattern=re.compile(r["pattern"]),
severity=Severity[r["severity"].upper()],
title=r.get("title", ""),
meaning=r.get("meaning", ""),
actions=tuple(r.get("actions", ())),
explained_by=tuple(r.get("explained_by", ())),
milestone=r.get("milestone"),
clears=tuple(r.get("clears", ())),
led=r.get("led"),
led_from=r.get("led_from"),
split_by=r.get("split_by"),
min_count=int(r.get("min_count", 1)),
min_uptime_ms=int(r.get("min_uptime_ms", 0)),
))
_validate(rules)
return rules
def _validate(rules: list[Rule]) -> None:
ids = [r.id for r in rules]
dupes = [i for i, n in Counter(ids).items() if n > 1]
if dupes:
raise ValueError(f"duplicate rule ids: {dupes}")
known = set(ids)
for r in rules:
for ref in (*r.explained_by, *r.clears):
if ref not in known:
raise ValueError(f"{r.id}: unknown rule id {ref!r}")
for colour in (r.led, r.led_from):
if colour and colour not in LED:
raise ValueError(f"{r.id}: unknown LED colour {colour!r}")
if r.severity != Severity.IGNORE and not r.title:
raise ValueError(f"{r.id}: visible rule needs a title")
@dataclass
class Finding:
key: str # rule id, or "rule id:split value"
rule_id: str
severity: Severity
title: str
meaning: str
actions: tuple[str, ...]
explained_by: tuple[str, ...] = ()
min_count: int = 1
count: int = 0
first_uptime_ms: int | None = None
last_uptime_ms: int | None = None
last_seen: float = 0.0
boot: int = 0
resolved: bool = False
resolved_by: str | None = None
evidence: deque[str] = field(default_factory=lambda: deque(maxlen=3))
@property
def active(self) -> bool:
return self.count >= self.min_count and not self.resolved
@dataclass
class BootRecord:
index: int
host_time: float
explicit: bool # ROM banner seen (vs. inferred)
reset_reason: str | None = None
tool_reset: bool = False
last_uptime_ms: int = 0
saw_log: bool = False
led_at_end: str | None = None
panic_lines: list[str] = field(default_factory=list)
recent_errors: deque[str] = field(default_factory=lambda: deque(maxlen=5))
intentional_restart: bool = False # factory reset / setup-mode toggle
@dataclass
class RootCause:
finding: Finding
consequences: list[Finding]
@property
def severity(self) -> Severity:
return max([self.finding.severity, *(c.severity for c in self.consequences)])
@dataclass
class Report:
verdict: str
verdict_severity: Severity | None # None = healthy / nothing found
led: str
led_reason: str
firmware: FirmwareInfo
milestones: list[tuple[str, str, int | None]] # (id, label, uptime_ms)
roots: list[RootCause]
earlier: list[Finding] # seen only in previous boots
resolved: list[Finding]
boots: list[BootRecord]
lines: int
mid_run: bool # no boot seen yet
def to_text(self) -> str:
out = [f"VERDICT: {self.verdict}",
f"Expected status LED: {LED[self.led][0]}"
+ (f" (last change: {self.led_reason})" if self.led_reason else ""),
f"Firmware: {self.firmware.describe()}",
f"Boots seen: {max(len(self.boots) - (0 if self.boots and self.boots[0].explicit else 1), 0)}",
"", "Milestones (current boot):"]
for _, label, t in self.milestones:
out.append(f" [{'x' if t is not None else ' '}] {label}"
+ (f" @ {t / 1000:.1f} s" if t is not None else ""))
out.append("")
out.append("Findings:" if self.roots else "Findings: none")
for rc in self.roots:
out += _finding_text(rc.finding, " ")
for c in rc.consequences:
out.append(" consequence:")
out += _finding_text(c, " ")
if self.earlier:
out.append("")
out.append("Seen in earlier boots only:")
for f in self.earlier:
out.append(f" - [{f.severity.label}] {f.title} (x{f.count})")
if self.resolved:
out.append("")
out.append("Resolved during this session:")
for f in self.resolved:
out.append(f" - {f.title}" + (f" (by: {f.resolved_by})" if f.resolved_by else ""))
return "\n".join(out)
def _finding_text(f: Finding, ind: str) -> list[str]:
lines = [f"{ind}- [{f.severity.label}] {f.title} (x{f.count})", f"{ind} {f.meaning}"]
lines += [f"{ind} {i}. {a}" for i, a in enumerate(f.actions, 1)]
if f.evidence:
lines.append(f"{ind} log: {f.evidence[-1]}")
return lines
class DiagnosticsEngine:
def __init__(self, rules: list[Rule] | None = None):
self.rules = rules if rules is not None else load_rules()
self._rule_by_id = {r.id: r for r in self.rules}
self.reset()
# ---------------------------------------------------------------- API
def reset(self) -> None:
self.findings: dict[str, Finding] = {}
self.boots: list[BootRecord] = [BootRecord(0, 0.0, explicit=False)]
self.milestones: dict[str, int] = {}
self.led = "UNKNOWN"
self.led_reason = ""
self.firmware = FirmwareInfo()
self.lines = 0
self._tool_reset_at: float | None = None
self._panic_tail = 0 # RAW lines after a panic line to keep
@property
def boot(self) -> BootRecord:
return self.boots[-1]
def feed(self, line: LogLine) -> None:
self.lines += 1
text = line.text
if line.kind == LineKind.HOST:
if "reset issued by tool" in text:
self._tool_reset_at = line.host_time
return
if line.kind == LineKind.BOOT:
self._new_boot(line.host_time, explicit=True)
self.milestones["rom_boot"] = 0
return
if line.kind == LineKind.RESET:
self.boot.reset_reason = line.reset_reason
if line.reset_reason and "POWERON" in line.reset_reason:
self.led, self.led_reason = "OFF", "power-on - firmware not started yet"
return
if line.kind == LineKind.LOG and line.uptime_ms is not None:
b = self.boot
# Uptime jumped back to a fresh-boot value: the ROM banner was
# missed (port reconnect, capture started late).
if b.saw_log and line.uptime_ms + 3000 < b.last_uptime_ms and line.uptime_ms < 10_000:
self._new_boot(line.host_time, explicit=False)
b = self.boot
b.saw_log = True
b.last_uptime_ms = max(b.last_uptime_ms, line.uptime_ms)
if line.kind == LineKind.PANIC:
self._on_panic(line)
elif self._panic_tail and line.kind == LineKind.RAW:
self.boot.panic_lines.append(text)
self._panic_tail -= 1
self._crash_finding().evidence.append(text)
self.firmware.observe(text)
matched = False
for rule in self.rules:
m = rule.pattern.search(text)
if not m:
continue
matched = True
if rule.min_uptime_ms and (line.uptime_ms or 0) < rule.min_uptime_ms:
continue
groups = {k: v for k, v in m.groupdict().items() if v is not None}
if rule.milestone and rule.milestone not in self.milestones:
self.milestones[rule.milestone] = line.uptime_ms or 0
if rule.led:
self._set_led(rule.led, rule.led_from, text)
for cid in rule.clears:
for f in self.findings.values():
if f.rule_id == cid and not f.resolved:
f.resolved = True
f.resolved_by = rule.milestone or rule.id
if rule.id in ("sys.factory_reset",) or "Short press:" in text:
self.boot.intentional_restart = True
if rule.severity != Severity.IGNORE:
self._hit(rule, groups, line)
self._stateful(line)
if line.level == "E":
self.boot.recent_errors.append(text)
if not matched and line.tag:
self._hit_other(line)
def report(self) -> Report:
all_f = list(self.findings.values()) + self._reboot_findings()
cur = self.boot.index
active = [f for f in all_f if f.active]
current = [f for f in active if f.boot == cur]
earlier = [f for f in active if f not in current]
resolved = [f for f in all_f if f.resolved and f.count >= f.min_count]
roots = _group(current)
# Worst consequence first, then the root's own severity, then order seen.
roots.sort(key=lambda rc: (-rc.severity, -rc.finding.severity, rc.finding.first_uptime_ms or 0))
mid_run = not any(b.explicit for b in self.boots) and self.boot.index == 0
if self.lines == 0:
verdict, vsev = "Waiting for data from the sensorbox...", None
elif roots and roots[0].severity >= Severity.WARNING:
vsev = roots[0].severity
prefix = "Problem" if vsev >= Severity.ERROR else "Attention"
verdict = f"{prefix}: {roots[0].finding.title}"
elif "data_uploaded" in self.milestones:
verdict, vsev = "Healthy: data is reaching InfluxDB", None
else:
verdict, vsev = "No problems found so far", None
if mid_run and self.lines:
verdict += " (connected mid-run - press Reset & capture for a full check)"
ms = [(mid, label, self.milestones.get(mid)) for mid, label, optional in MILESTONES
if not optional or mid in self.milestones]
# Reached milestones in the order they happened, then the pending ones.
ms.sort(key=lambda m: (m[2] is None, m[2] or 0))
return Report(verdict, vsev, self.led, self.led_reason, self.firmware, ms, roots,
earlier, resolved, list(self.boots), self.lines, mid_run)
# ------------------------------------------------------------ internals
def _new_boot(self, host_time: float, explicit: bool) -> None:
prev = self.boot
# An inferred boot right after a banner-started boot is the same boot.
if not explicit and prev.explicit and not prev.saw_log:
return
prev.led_at_end = self.led
b = BootRecord(prev.index + 1, host_time, explicit)
if self._tool_reset_at is not None and host_time - self._tool_reset_at <= _RESET_BY_TOOL_WINDOW_S:
b.tool_reset = True
self._tool_reset_at = None
self.boots.append(b)
self.milestones = {}
# The WS2812 latches its colour through a software reset until
# Status::init() rewrites it; only a power-on turns it dark.
self.led_reason = "restarting - LED keeps its last colour until the firmware starts"
self.firmware = FirmwareInfo()
self._panic_tail = 0
def _set_led(self, colour: str, led_from: str | None, text: str) -> None:
if led_from and self.led != led_from:
return
self.led = colour
self.led_reason = text
def _hit(self, rule: Rule, groups: dict[str, str], line: LogLine) -> None:
key = rule.id
if rule.split_by and rule.split_by in groups:
key = f"{rule.id}:{groups[rule.split_by]}"
f = self.findings.get(key)
if f is None:
f = self.findings[key] = Finding(
key, rule.id, rule.severity, _fmt(rule.title, groups), _fmt(rule.meaning, groups),
tuple(_fmt(a, groups) for a in rule.actions), rule.explained_by, rule.min_count)
self._touch(f, line)
def _touch(self, f: Finding, line: LogLine) -> None:
f.count += 1
# Uptime restarts every boot, so "first seen" is per current boot.
if f.first_uptime_ms is None or f.boot != self.boot.index:
f.first_uptime_ms = line.uptime_ms
f.last_uptime_ms = line.uptime_ms
f.last_seen = line.host_time
f.boot = self.boot.index
f.evidence.append(line.text)
if f.resolved:
f.resolved, f.resolved_by = False, None # problem came back
def _custom(self, key: str, severity: Severity, title: str, meaning: str,
actions: tuple[str, ...], explained_by: tuple[str, ...] = ()) -> Finding:
f = self.findings.get(key)
if f is None:
f = self.findings[key] = Finding(key, key, severity, title, meaning, actions, explained_by)
return f
def _hit_other(self, line: LogLine) -> None:
f = self._custom(
f"other.{line.tag}", Severity.INFO, f"Unrecognised error messages from {line.tag}",
"These errors are not in the rule set yet; they may be harmless.",
("If the box misbehaves, save the session log and send it to the firmware team.",))
self._touch(f, line)
def _crash_finding(self) -> Finding:
return self._custom(
"sys.crash", Severity.CRITICAL, "Firmware stopped with an error and restarts",
"The firmware hit a fatal error. The error printed just before it is the real cause.",
("See the grouped cause; otherwise save the session log for the firmware team.",),
tuple(_INIT_FAILURES))
def _on_panic(self, line: LogLine) -> None:
b = self.boot
if len(b.panic_lines) < 8:
b.panic_lines.append(line.text)
self._panic_tail = 4 # file/func/expression lines follow ESP_ERROR_CHECK
f = self._crash_finding()
if f.count == 0 and b.recent_errors:
f.meaning += f" Last error before it: {b.recent_errors[-1]}"
self._touch(f, line)
def _stateful(self, line: LogLine) -> None:
# RTC set by SNTP but reads "uninitialized" again later: the chip
# does not keep time (crystal/module fault), not just a flat cell.
synced = self.milestones.get("time_synced")
if (synced is not None and line.uptime_ms is not None
and line.uptime_ms > synced + 10_000
and "RTC is uninitialized" in line.text):
f = self._custom(
"rtc.lost_after_sync", Severity.ERROR,
"Clock stops again after being set from the Internet",
"The clock was set from the Internet but stopped running shortly after. The RTC module or its crystal is faulty.",
("Replace the RTC module (inside the case).",))
self._touch(f, line)
def _reboot_findings(self) -> list[Finding]:
"""Classify why each earlier boot ended (needs the next boot's reset reason)."""
out: list[Finding] = []
short_unexpected: list[BootRecord] = []
for prev, nxt in zip(self.boots, self.boots[1:]):
if not prev.saw_log and prev.index == 0:
continue # nothing captured before the first boot
reason = nxt.reset_reason or "unknown"
if nxt.tool_reset or prev.intentional_restart:
continue
key, sev, title, meaning, actions, expl = None, Severity.INFO, "", "", (), ()
if "USB_UART_CHIP_RESET" in reason:
key, title = "reboot.usb", "Box was reset over USB"
meaning = "The board was reset through its USB port (a flashing tool, a serial program, or this tool in an earlier session)."
elif "POWERON" in reason:
key, title = "reboot.power", "Box was switched off and on"
meaning = "Power was interrupted (power switch, cable or supply)."
actions = ("If nobody switched it off: check the power supply and cable.",)
elif "BROWN" in reason:
key, sev, title = "reboot.brownout", Severity.ERROR, "Power supply voltage dropped (brown-out)"
meaning = "The supply voltage fell too low and the chip reset itself."
actions = ("Check the power supply, cable and connectors.",)
elif "WDT" in reason:
key, sev, title = "reboot.watchdog", Severity.ERROR, "Firmware froze and was restarted by the watchdog"
meaning = "The firmware stopped responding; the hardware watchdog restarted it."
actions = ("Save the session log for the firmware team.",)
elif prev.panic_lines:
pass # covered by sys.crash
elif prev.led_at_end in ("PURPLE", "RED") and prev.last_uptime_ms >= 140_000:
key, title = "reboot.status_watchdog", "Box restarted itself after 2.5 minutes of errors"
meaning = "By design the firmware restarts when it stays in an error state (purple/red) for 150 s."
actions = ("Fix the grouped cause; the restarts stop by themselves.",)
expl = tuple(_NETWORK_ROOTS) + ("sys.out_of_resources",)
else:
key, sev, title = "reboot.unexpected", Severity.WARNING, "Unexpected restart"
meaning = f"The firmware restarted (reset reason {reason})."
actions = ("Watch whether it repeats; save the session log for the firmware team.",)
if key:
out.append(self._reboot_finding(key, sev, title, meaning, actions, expl, nxt))
if prev.last_uptime_ms < _SHORT_BOOT_MS and "POWERON" not in reason:
short_unexpected.append(prev)
if len(short_unexpected) >= 2:
avg = sum(b.last_uptime_ms for b in short_unexpected) / len(short_unexpected) / 1000
f = self._reboot_finding(
"reboot.loop", Severity.CRITICAL, "Reboot loop",
f"The box restarted {len(short_unexpected)} times, about every {avg:.0f} s. "
"The error printed just before each restart is the cause.",
("Fix the grouped cause below.",), tuple(_INIT_FAILURES), self.boot)
f.count = len(short_unexpected)
out.append(f)
return out
def _reboot_finding(self, key, sev, title, meaning, actions, expl, boot: BootRecord) -> Finding:
f = Finding(key, key, sev, title, meaning, actions, expl)
f.count, f.boot, f.last_seen = 1, self.boot.index, boot.host_time
if boot.reset_reason:
f.evidence.append(f"reset reason: {boot.reset_reason}")
return f
def _group(findings: list[Finding]) -> list[RootCause]:
"""Attach each finding to the root of its explained_by chain."""
by_rule: dict[str, list[Finding]] = {}
for f in findings:
by_rule.setdefault(f.rule_id, []).append(f)
def parent(f: Finding) -> Finding | None:
for rid in f.explained_by:
if by_rule.get(rid):
return by_rule[rid][0]
return None
def root(f: Finding) -> Finding:
seen = {f.key}
while (p := parent(f)) is not None and p.key not in seen:
seen.add(p.key)
f = p
return f
roots: dict[str, RootCause] = {}
for f in findings:
r = root(f)
rc = roots.setdefault(r.key, RootCause(r, []))
if r is not f:
rc.consequences.append(f)
for rc in roots.values():
rc.consequences.sort(key=lambda c: (-c.severity, c.first_uptime_ms or 0))
return list(roots.values())
kota_tech/diag/firmware.py
+91
View File
@@ -0,0 +1,91 @@
"""Firmware identification from the boot log.
The app version string is "1.0" in every build seen so far, so builds are
told apart by the ELF SHA256 prefix printed by app_init. Boxes receive new
firmware over OTA; an unknown SHA usually means a box has already updated.
"""
from __future__ import annotations
import re
from dataclasses import dataclass, field
@dataclass(frozen=True)
class KnownFirmware:
name: str
source: str
sd_wifi: bool # reads SSID/PASSWORD from ENV.TXT
setup_portal: bool # short press -> Wi-Fi setup portal
KNOWN: dict[str, KnownFirmware] = {
"45fa9d5b9": KnownFirmware(
name="v1.0 (built Jun 29 2026)",
source="factory image new_system.bin",
sd_wifi=False,
setup_portal=False,
),
}
_ELF = re.compile(r"app_init: ELF file SHA256:\s+(?P<sha>[0-9a-f]{6,})")
_VER = re.compile(r"app_init: App version:\s+(?P<v>\S+)")
_BUILT = re.compile(r"app_init: Compile time:\s+(?P<t>.+)$")
_IDF = re.compile(r"app_init: ESP-IDF:\s+(?P<v>\S+)")
# Evidence of features, for builds not in KNOWN.
_FEATURE_SD_WIFI = re.compile(r"WIFI_CREDENTIALS: ")
_FEATURE_PORTAL = re.compile(r"WIFI_PROVISIONING: |Short press: ")
# Deployed v1.0 logs the SD mount under the INFLUX_BUFFER tag.
_LEGACY_SD_TAG = re.compile(r"INFLUX_BUFFER: SD card mounted")
@dataclass
class FirmwareInfo:
elf_sha: str | None = None
version: str | None = None
built: str | None = None
idf: str | None = None
evidence: set[str] = field(default_factory=set)
def observe(self, text: str) -> None:
if m := _ELF.search(text):
self.elf_sha = m.group("sha")
elif m := _VER.search(text):
self.version = m.group("v")
elif m := _BUILT.search(text):
self.built = m.group("t").strip()
elif m := _IDF.search(text):
self.idf = m.group("v")
if _FEATURE_SD_WIFI.search(text):
self.evidence.add("sd_wifi")
if _FEATURE_PORTAL.search(text):
self.evidence.add("setup_portal")
if _LEGACY_SD_TAG.search(text):
self.evidence.add("legacy_v1")
@property
def known(self) -> KnownFirmware | None:
if not self.elf_sha:
return None
return next((k for sha, k in KNOWN.items() if self.elf_sha.startswith(sha)), None)
@property
def supports_sd_wifi(self) -> bool | None:
"""True/False when determinable, None when unknown (no boot seen)."""
if k := self.known:
return k.sd_wifi
if "sd_wifi" in self.evidence:
return True
if "legacy_v1" in self.evidence:
return False
return None
def describe(self) -> str:
if k := self.known:
return f"{k.name} - {k.source}"
if self.elf_sha:
built = f", built {self.built}" if self.built else ""
return f"Unknown build {self.elf_sha[:9]}{built} (probably updated over OTA)"
if "legacy_v1" in self.evidence:
return "Probably v1.0 factory image (press Reset & capture to confirm)"
return "Not identified yet - press Reset & capture"
kota_tech/diag/rules.toml
+565
View File
@@ -0,0 +1,565 @@
# Log signatures for the KOTA sensorbox firmware.
#
# Derived from the firmware source (log strings, Status::set_status calls) and
# checked against real captures (tests/fixtures). Patterns match the cleaned
# line text, not the tag: the deployed v1.0 build logs some messages under
# different tags than the source (e.g. SD mount under INFLUX_BUFFER).
#
# Fields:
# id unique, "<category>.<name>"
# pattern Python regex (search). Named groups can be used in `title`.
# severity critical | error | warning | info | ignore
# ignore = never shown (benign ESP_LOGE lines, LED-only rules)
# title short technician-facing headline
# meaning what it means, in plain words
# actions what to do, in order (prefer no-unscrewing actions first)
# explained_by list of rule ids; while one of them is active this finding
# is shown as a consequence of it instead of a root cause
# milestone boot milestone reached when the line is seen
# clears finding ids resolved when this line is seen
# led status LED colour this line sets (firmware Status::set_status)
# led_from only change the LED if it currently has this colour
# (Status::set_status_conditional)
# split_by named group; one finding per distinct value (e.g. per sensor)
# min_count finding becomes active only after this many occurrences
# min_uptime_ms ignore occurrences earlier than this uptime
# ---------------------------------------------------------------- milestones
[[rule]]
id = "boot.app_started"
pattern = 'app_init: Project name:\s+kota-sensorboard'
severity = "ignore"
milestone = "app_started"
led = "GREEN" # Status::init() -> STATUS_OK
[[rule]]
id = "boot.sd_mounted"
pattern = 'SD card mounted successfully'
severity = "ignore"
milestone = "sd_mounted"
[[rule]]
id = "boot.env_loaded"
pattern = 'Read INFLUX_URL from \.env file'
severity = "ignore"
milestone = "env_loaded"
[[rule]]
id = "boot.matter_started"
pattern = 'ESP-Matter started successfully'
severity = "ignore"
milestone = "matter_started"
[[rule]]
id = "boot.local_wifi_mode"
pattern = 'SSID/PASSWORD found in \.env'
severity = "ignore"
milestone = "local_wifi_mode"
[[rule]]
id = "boot.sensors_ready"
# DeviceStatus::init installs the temperature sensor; it is the last init
# step, and any earlier init failure aborts before reaching it.
pattern = 'temperature_sensor: Range \['
severity = "ignore"
milestone = "sensors_ready"
[[rule]]
id = "net.wifi_connected"
pattern = 'sta ip: \d|got ip:|WIFI_SERVICE: Got IP address|IPv4 address changed|Interface IP Address changed'
severity = "ignore"
milestone = "wifi_connected"
clears = ["matter.not_commissioned", "wifi.disconnected"]
[[rule]]
id = "net.time_synced"
pattern = 'SNTP: Time synchronized'
severity = "ignore"
milestone = "time_synced"
clears = ["rtc.clock_not_running", "net.dns_fail"]
[[rule]]
id = "influx.upload_ok"
pattern = 'INFLUX: HTTP request performed, status code: 2\d\d'
severity = "ignore"
milestone = "data_uploaded"
clears = ["influx.buffer_fail", "net.dns_fail", "net.tls_fail"]
[[rule]]
id = "influx.write_ok"
pattern = 'Successfully wrote to Influx'
severity = "ignore"
milestone = "data_uploaded"
led = "GREEN"
led_from = "PURPLE"
[[rule]]
id = "ota.registered"
pattern = 'Device registered successfully|OTA token loaded from NVS'
severity = "ignore"
milestone = "ota_registered"
clears = ["ota.registration_failed"]
[[rule]]
id = "ota.checked"
pattern = 'OTA_CLIENT: No update available'
severity = "ignore"
milestone = "ota_checked"
clears = ["ota.check_failed"]
led = "GREEN"
led_from = "PURPLE"
# ----------------------------------------------------------- benign E lines
[[rule]]
id = "benign.mac"
pattern = 'ID: Device MAC address'
severity = "ignore"
[[rule]]
id = "benign.attr_cb"
pattern = 'MAIN: Attribute update callback'
severity = "ignore"
[[rule]]
id = "benign.zcl"
pattern = 'chip\[ZCL\]: WRITE ERR|data_model: Cluster cannot be NULL'
severity = "ignore"
[[rule]]
id = "benign.connect_chain"
# Transport-level echo of a failure that a more specific rule reports.
pattern = 'esp-tls: Failed to open new connection|transport_base: Failed to open a new connection|HTTP_CLIENT: Connection failed, sock < 0|INFLUX: Failed to perform HTTP request: 0,|INFLUX_BUFFER: Failed to write chunk to Influx|OTA_CLIENT: Error ESP_ERR_HTTP_CONNECT at'
severity = "ignore"
# ------------------------------------------------------------------ network
[[rule]]
id = "matter.not_commissioned"
pattern = 'CHIPoBLE advertising started|MAIN: Commissioning window opened'
severity = "warning"
title = "Box is not connected to Wi-Fi (not commissioned in Matter)"
meaning = "The box is waiting to be added to a Matter network over Bluetooth. Until then it has no Wi-Fi, so it cannot upload data or receive firmware updates."
actions = [
"Add the box in the customer's Matter app (Apple Home, Google Home, ...) by scanning the QR sticker, or enter the manual code printed under it.",
"Use a 2.4 GHz Wi-Fi network; keep the phone close to the box during pairing.",
"If pairing keeps failing: hold the Matter button 5 s (factory reset) and try again.",
]
milestone = "ble_advertising"
[[rule]]
id = "matter.commissioning_failed"
pattern = 'Commissioning failed, fail safe timer expired'
severity = "error"
title = "Matter pairing attempt failed"
meaning = "A phone started adding the box to Matter but did not finish in time."
actions = [
"Retry pairing from the Matter app with the phone next to the box.",
"Check the Wi-Fi is 2.4 GHz and the password entered in the app is correct.",
]
[[rule]]
id = "net.dns_fail"
pattern = '''couldn't get hostname for :(?P<host>[\w.-]+)'''
severity = "error"
title = "No Internet: server names cannot be resolved"
meaning = "The box could not look up a server name. Either it has no Wi-Fi connection, or the network has no working Internet/DNS."
actions = [
"Check the site Wi-Fi has Internet (open a website on a phone connected to the same network).",
"If the box was never paired: commission it in Matter first.",
"Check the router does not block new devices (guest isolation, MAC filtering).",
]
explained_by = ["matter.not_commissioned", "wifi.disconnected", "wifi.setup_portal"]
[[rule]]
id = "net.tls_fail"
pattern = 'mbedtls_ssl_handshake returned|Failed to verify peer certificate|esp-tls-mbedtls: .*[Hh]andshake'
severity = "error"
title = "Secure connection to the server failed"
meaning = "The box reached the server but the encrypted (HTTPS) connection failed. Common causes: wrong clock on the box, or a network that intercepts HTTPS (corporate proxy/firewall)."
actions = [
"Try another network (e.g. phone hotspot) to rule out a firewall.",
"Check the clock finding below: HTTPS needs the correct date.",
]
explained_by = ["rtc.clock_not_running"]
[[rule]]
id = "wifi.disconnected"
pattern = 'WiFi disconnected, retrying in'
severity = "error"
min_count = 3
title = "Box cannot connect to the configured Wi-Fi"
meaning = "The Wi-Fi name/password from the SD card does not work: wrong password, network out of range, or not 2.4 GHz."
actions = [
"Check SSID and PASSWORD in ENV.TXT (re-provision the SD card with the tool).",
"Make sure the network is 2.4 GHz and in range.",
]
[[rule]]
id = "wifi.setup_portal"
pattern = 'WiFi provisioning portal ready: connect to "(?P<ssid>[^"]+)"'
severity = "warning"
title = "Box is in Wi-Fi setup mode (network {ssid})"
meaning = "The box is hosting its own Wi-Fi network for configuration instead of connecting to the site Wi-Fi."
actions = [
"Connect a phone to {ssid} and open http://sensorbox.local/ to enter the Wi-Fi details,",
"or short-press the Matter button (< 2 s) to leave setup mode.",
]
led = "WHITE"
# ------------------------------------------------------------------- InfluxDB
[[rule]]
id = "influx.buffer_fail"
pattern = 'Failed to write chunk with \d+ retries to Influx|MAIN: Failed to flush Influx buffer'
severity = "error"
title = "Measurements cannot be uploaded to InfluxDB"
meaning = "Uploads keep failing, so readings are stored on the SD card and sent later. The LED turns purple and the box restarts itself after about 2.5 minutes in this state."
actions = [
"Fix the network problem first (see the finding this is grouped under).",
"Buffered data is uploaded automatically once the connection works.",
]
explained_by = ["net.dns_fail", "net.tls_fail", "matter.not_commissioned", "wifi.disconnected", "wifi.setup_portal"]
[[rule]]
id = "influx.purple"
pattern = 'Failed to write chunk with \d+ retries to Influx|Failed to flush buffer before writing|Failed to append new entry to buffer'
severity = "ignore"
led = "PURPLE"
[[rule]]
id = "influx.auth"
pattern = 'INFLUX: Failed to perform HTTP request: (?P<code>401|403)'
severity = "critical"
title = "InfluxDB rejected the token (HTTP {code})"
meaning = "The box reaches the database but the INFLUX_TOKEN on the SD card is wrong or revoked."
actions = [
"Re-provision the SD card with the current ENV.TXT (SD card is inside the case).",
"If all boxes show this: the shared token was rotated - update ENV.TXT for everyone.",
]
[[rule]]
id = "influx.not_found"
pattern = 'INFLUX: Failed to perform HTTP request: (?P<code>404)'
severity = "critical"
title = "InfluxDB bucket or organisation not found (HTTP 404)"
meaning = "INFLUX_URL on the SD card points to a bucket/org that does not exist."
actions = ["Re-provision the SD card with the correct ENV.TXT."]
[[rule]]
id = "influx.server_error"
pattern = 'INFLUX: Failed to perform HTTP request: (?P<code>5\d\d)'
severity = "warning"
title = "InfluxDB server error (HTTP {code})"
meaning = "The database server has a problem. The box is fine; data is buffered on the SD card."
actions = ["Nothing to fix on the box. Report to the backend team if it persists."]
[[rule]]
id = "influx.bad_request"
pattern = 'INFLUX: Bad request, check line protocol format'
severity = "warning"
title = "InfluxDB refused some measurements (HTTP 400)"
meaning = "A measurement was malformed and dropped (firmware issue, not a hardware fault)."
actions = ["Save the session log and send it to the firmware team."]
[[rule]]
id = "influx.env_missing"
split_by = "key"
pattern = 'Failed to read (?P<key>INFLUX_URL|INFLUX_TOKEN) from \.env file'
severity = "critical"
title = "ENV.TXT on the SD card is missing {key}"
meaning = "The box cannot start without the InfluxDB settings; it shows white and restarts."
actions = ["Re-provision the SD card with the tool (SD card is inside the case)."]
[[rule]]
id = "influx.init_fail"
pattern = 'MAIN: Failed to initialize Influx buffer'
severity = "critical"
title = "Data upload could not be initialised"
meaning = "Start-up stopped because the InfluxDB settings could not be read from the SD card. The box restarts."
actions = ["Re-provision the SD card with the tool (SD card is inside the case)."]
explained_by = ["influx.env_missing", "sd.mount_fail", "sd.card_fail", "sd.env_open_fail"]
led = "WHITE"
# ----------------------------------------------------------------- SD card
[[rule]]
id = "sd.card_fail"
pattern = 'Failed to initialize the card \((?P<err>[^)]+)\)'
severity = "critical"
title = "SD card not detected ({err})"
meaning = "The SD card does not respond: missing, not seated, or damaged. The box shows white and restarts."
actions = [
"Open the case and re-seat the SD card.",
"Try a freshly provisioned card.",
]
led = "WHITE"
[[rule]]
id = "sd.mount_fail"
pattern = 'Failed to mount filesystem'
severity = "critical"
title = "SD card is not formatted correctly"
meaning = "The card responds but has no readable FAT32 file system."
actions = ["Re-provision the card with the tool (FAT32, MBR) - SD card is inside the case."]
led = "WHITE"
[[rule]]
id = "sd.env_open_fail"
pattern = 'Failed to open \.env file for reading'
severity = "critical"
title = "ENV.TXT not found on the SD card"
meaning = "The card is readable but the configuration file is missing."
actions = ["Re-provision the card with the tool (SD card is inside the case)."]
[[rule]]
id = "sd.write_fail"
pattern = 'Failed to (create buffer file|open buffer file for appending)'
severity = "error"
title = "Cannot write to the SD card"
meaning = "Offline data cannot be stored: the card is full, write-protected or failing."
actions = ["Replace / re-provision the SD card (inside the case)."]
# --------------------------------------------------------------- RTC / I2C
[[rule]]
id = "rtc.init_failed"
pattern = 'MAIN: Failed to initialize RTC: (?P<err>\S+)'
severity = "critical"
title = "Clock chip (RTC) or I2C bus not responding ({err})"
meaning = "The RTC is the first chip the firmware talks to over I2C, so this also appears when the I2C cable between the boards is loose or broken. LED blue, the box restarts."
actions = [
"Open the case and check the I2C cable/connector between the boards.",
"If the cable is fine: the RTC module is faulty.",
]
led = "BLUE"
[[rule]]
id = "rtc.clock_not_running"
pattern = 'RTC is uninitialized, using current time as timestamp'
severity = "warning"
min_count = 5
title = "Clock is not set - sensor readings are on hold"
meaning = "The clock chip answers but its clock is not running (it reads 1 Jan 2000; typical after its coin cell was removed or is empty), and there is no Internet time yet. Readings are held until the time is set, so no data is recorded."
actions = [
"Give the box Internet access: the clock is set automatically on the first connection.",
"If this comes back after every power-off: replace the RTC coin cell (inside the case).",
]
[[rule]]
id = "i2c.errors"
pattern = 'I2C (read|write) failed: (?P<err>\S+)|Failed to add I2C device'
severity = "error"
min_count = 2
title = "I2C bus communication errors"
meaning = "Sensors are not answering reliably on the I2C bus: loose connector, damaged cable, or a failing sensor."
actions = ["Open the case and check the I2C cable/connectors between the boards."]
# ------------------------------------------------------------------ sensors
[[rule]]
id = "sensor.init_failed"
split_by = "sensor"
pattern = 'MAIN: Failed to initialize (?P<sensor>BME688|SEN66|AS7341|Microphone|Flicker sensor|DeviceStatus): (?P<err>.+)'
severity = "critical"
title = "{sensor} failed to start"
meaning = "Start-up stopped at the {sensor} sensor ({err}). LED red, the box restarts."
actions = [
"Open the case and check the {sensor} connector/cable.",
"If several sensors fail in turn: check the shared I2C cable.",
]
led = "RED"
[[rule]]
id = "sensor.selfcheck_failed"
split_by = "sensor"
pattern = '(?P<sensor>BME688|SEN66|AS7341|Microphone|Flicker sensor) self check failed'
severity = "error"
title = "{sensor} failed the post-update self-check"
meaning = "After a firmware update the {sensor} did not work; the box rolls back to the previous firmware."
actions = ["Check the {sensor} connector (inside the case), then power-cycle."]
[[rule]]
id = "sen66.not_ready"
pattern = 'SEN66: (Failed while awaiting sensor ready|Self-check failed: Sensor not ready|Failed to read measurement)'
severity = "error"
title = "Air-quality sensor (SEN66) not delivering data"
meaning = "The SEN66 answers but never reports a measurement."
actions = ["Check the SEN66 cable (inside the case); the sensor may be faulty."]
[[rule]]
id = "bme688.read_fail"
pattern = 'Failed to get (BME688 )?data from BME688|No new data available from BME688'
severity = "warning"
min_count = 3
title = "Temperature/humidity sensor (BME688) read errors"
meaning = "The BME688 repeatedly fails to return a reading."
actions = ["Check the BME688 connection (inside the case)."]
[[rule]]
id = "as7341.read_fail"
pattern = 'Failed to read (spectral|basic counts) data'
severity = "warning"
min_count = 2
title = "Light sensor (AS7341) read errors"
meaning = "The AS7341 repeatedly fails to return a reading."
actions = ["Check the AS7341 connection (inside the case)."]
[[rule]]
id = "mic.silent"
pattern = 'MICROPHONE: .*dB peak: -inf'
severity = "warning"
min_count = 2
title = "Microphone records pure silence"
meaning = "Every audio sample is zero. A real room is never perfectly silent, so the microphone or its wiring is likely faulty."
actions = [
"Make some noise near the box and watch whether this finding keeps growing.",
"If it does: check the microphone board/cable (inside the case).",
]
[[rule]]
id = "mic.read_fail"
pattern = 'Failed to read samples'
severity = "error"
title = "Microphone read error"
meaning = "The audio interface returned an error."
actions = ["Check the microphone board/cable (inside the case)."]
[[rule]]
id = "gather.stalled"
pattern = 'tasks still running, waiting'
severity = "warning"
min_uptime_ms = 180000
min_count = 5
title = "Sensor readings are stuck"
meaning = "A sensor has not finished its reading for minutes (the first 2 minutes of warm-up are normal)."
actions = ["See the grouped cause, or note which sensor stopped logging and check its cable."]
explained_by = ["rtc.clock_not_running", "sen66.not_ready"]
# ---------------------------------------------------------- OTA / firmware
[[rule]]
id = "ota.registration_failed"
pattern = 'Device registration failed'
severity = "warning"
title = "Box is not registered for firmware updates"
meaning = "The box could not reach the update server (bin.kota-app.com) to register, so it will not receive new firmware."
actions = ["Fix the network problem; registration is retried automatically every few seconds."]
explained_by = ["net.dns_fail", "net.tls_fail", "matter.not_commissioned", "wifi.disconnected", "wifi.setup_portal"]
[[rule]]
id = "ota.check_failed"
pattern = 'Failed to check for updates|OTA Update check failed with status code: (?P<code>\d+)'
severity = "error"
title = "Firmware update check failed"
meaning = "The update server could not be reached or refused the request. LED purple."
actions = ["Fix the network problem; the check is retried every 15 minutes."]
explained_by = ["net.dns_fail", "net.tls_fail", "matter.not_commissioned", "wifi.disconnected"]
led = "PURPLE"
[[rule]]
id = "ota.unauthorized"
pattern = 'Unauthorized, clearing saved token'
severity = "warning"
title = "Update server did not accept the box - it will re-register"
meaning = "The saved update token was rejected; the box registers again automatically."
actions = ["Nothing to do unless it repeats. Then report the box MAC to the backend team."]
[[rule]]
id = "ota.updating"
pattern = 'New update available, starting OTA update'
severity = "info"
title = "Firmware update is downloading - do not switch off"
meaning = "The box found new firmware and is installing it (LED yellow). It restarts when done."
actions = ["Wait. Do not switch the power off."]
milestone = "ota_update_started"
led = "YELLOW"
[[rule]]
id = "ota.update_failed"
pattern = 'OTA update failed: (?P<err>\S+)'
severity = "error"
title = "Firmware update failed ({err})"
meaning = "The download or installation was interrupted; the box keeps its current firmware and retries later."
actions = ["Check the network is stable; the update is retried automatically."]
[[rule]]
id = "ota.update_ok"
pattern = 'OTA update successful, rebooting'
severity = "info"
title = "Firmware update installed - restarting"
meaning = "New firmware was installed. After the restart it runs a self-check (LED orange, ~2.5 min)."
actions = ["Wait for green."]
milestone = "ota_update_done"
[[rule]]
id = "ota.selfcheck"
pattern = 'Device is in selfcheck mode'
severity = "info"
title = "Post-update self-check running (~2.5 min)"
meaning = "The new firmware is testing all sensors before it is accepted (LED orange)."
actions = ["Wait. Do not switch off."]
led = "ORANGE"
[[rule]]
id = "ota.selfcheck_failed"
pattern = 'Failing selfcheck: (?P<reason>.+)'
severity = "error"
title = "Update self-check failed: {reason}"
meaning = "The new firmware rejected itself; the box rolls back to the previous version."
actions = ["Check the named sensor (inside the case), then power-cycle."]
led = "RED"
[[rule]]
id = "ota.selfcheck_passed"
pattern = 'Passing selfcheck'
severity = "ignore"
milestone = "ota_update_accepted"
led = "GREEN"
[[rule]]
id = "ota.reset_during_update"
pattern = 'Device was reset during update'
severity = "warning"
title = "Box was switched off during a firmware update"
meaning = "An update was interrupted; the box reports it and retries."
actions = ["Leave the box powered while the LED is yellow or orange."]
# --------------------------------------------------------------- system
[[rule]]
id = "sys.out_of_resources"
pattern = 'Failed to create (task for sensor|gather task|main gather task)'
severity = "error"
title = "Firmware ran out of memory"
meaning = "The firmware could not start an internal task (LED red); it restarts itself after 30 s."
actions = ["Power-cycle. If it repeats, save the session log for the firmware team."]
led = "RED"
[[rule]]
id = "sys.factory_reset"
pattern = 'Factory reset button pressed|Factory resetting device'
severity = "warning"
title = "Factory reset triggered with the Matter button"
meaning = "The box forgets its Matter network and must be paired again with the QR sticker."
actions = ["Re-commission the box in the Matter app."]
led = "WHITE"
[[rule]]
id = "mic.disabled"
pattern = 'MICROPHONE: Microphone is in shutdown state'
severity = "info"
title = "Microphone is switched off - no noise measurements"
meaning = "The firmware reads the microphone enable line (GPIO3) as 'off', which it treats as disabled on purpose. Noise levels are not recorded."
actions = ["If this box should measure noise: check the microphone enable switch/jumper and its wiring (inside the case)."]
[[rule]]
id = "gather.empty_reading"
pattern = 'SENSOR_GATHERER: Failed to get line protocol from sensor'
severity = "warning"
min_count = 2
title = "A sensor returned no data"
meaning = "One of the sensors skipped its reading; the preceding log lines name the sensor."
actions = ["Check the log lines just before this message for the sensor name."]
explained_by = ["mic.disabled", "mic.read_fail", "bme688.read_fail", "as7341.read_fail", "rtc.clock_not_running"]
kota_tech/ui/diag_panel.py
+230
View File
@@ -0,0 +1,230 @@
"""Diagnostics side panel: renders a diag.engine.Report.
Layout: verdict banner, expected LED, firmware, start-up milestones, a
findings tree (root causes; their consequences as children; titles only)
and a wrapped detail pane for the selected finding.
"""
from __future__ import annotations
from PySide6.QtCore import Qt
from PySide6.QtGui import QBrush, QColor
from PySide6.QtWidgets import (
QFrame,
QHBoxLayout,
QLabel,
QListWidget,
QListWidgetItem,
QSizePolicy,
QSplitter,
QTextBrowser,
QTreeWidget,
QTreeWidgetItem,
QVBoxLayout,
QWidget,
)
from ..diag.engine import LED, Finding, Report, Severity
SEV_COLOR = {
Severity.CRITICAL: "#b03a2e",
Severity.ERROR: "#d9534f",
Severity.WARNING: "#c77c02",
Severity.INFO: "#2e6da4",
}
# verdict banner (background, text) by verdict severity; None = healthy
BANNER = {
Severity.CRITICAL: ("#f6d5d1", "#7b1e14"),
Severity.ERROR: ("#f9dedc", "#8a2019"),
Severity.WARNING: ("#fdebc8", "#6b4400"),
None: ("#d8f0dd", "#1b5e2a"),
}
_KEY = Qt.ItemDataRole.UserRole
def _heading(text: str) -> QLabel:
lbl = QLabel(text)
f = lbl.font()
f.setBold(True)
lbl.setFont(f)
return lbl
def _esc(s: str) -> str:
return s.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
class DiagnosticsPanel(QWidget):
def __init__(self, parent: QWidget | None = None):
super().__init__(parent)
self.setMinimumWidth(380)
self.verdict = QLabel(wordWrap=True)
self.verdict.setMargin(10)
vf = self.verdict.font()
vf.setPointSize(vf.pointSize() + 2)
vf.setBold(True)
self.verdict.setFont(vf)
self.led_swatch = QLabel()
self.led_swatch.setFixedSize(22, 22)
self.led_text = QLabel(wordWrap=True)
led_row = QHBoxLayout()
led_row.addWidget(self.led_swatch, 0, Qt.AlignmentFlag.AlignTop)
led_row.addWidget(self.led_text, 1)
self.firmware = QLabel(wordWrap=True)
self.firmware.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
self.milestones = QListWidget()
self.milestones.setFrameShape(QFrame.Shape.NoFrame)
self.milestones.setSelectionMode(QListWidget.SelectionMode.NoSelection)
self.milestones.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Maximum)
self.tree = QTreeWidget()
self.tree.setHeaderHidden(True)
self.tree.setIndentation(16)
self.tree.setTextElideMode(Qt.TextElideMode.ElideRight)
self.detail = QTextBrowser()
self.detail.setOpenExternalLinks(False)
findings = QSplitter(Qt.Orientation.Vertical)
findings.addWidget(self.tree)
findings.addWidget(self.detail)
findings.setSizes([220, 260])
lay = QVBoxLayout(self)
lay.setContentsMargins(4, 0, 0, 0)
lay.addWidget(self.verdict)
lay.addWidget(_heading("Status LED should be"))
lay.addLayout(led_row)
lay.addWidget(_heading("Firmware"))
lay.addWidget(self.firmware)
lay.addWidget(_heading("Start-up progress (current boot)"))
lay.addWidget(self.milestones)
lay.addWidget(_heading("Findings - root causes first; click one for details"))
lay.addWidget(findings, 1)
self._findings: dict[str, Finding] = {}
self._selected: str | None = None
# key -> expanded, chosen by the user; survives the rebuild on refresh.
self._expanded: dict[str, bool] = {}
self.tree.currentItemChanged.connect(self._on_select)
self.tree.itemCollapsed.connect(lambda it: self._remember(it, False))
self.tree.itemExpanded.connect(lambda it: self._remember(it, True))
# ------------------------------------------------------------------
def show_report(self, r: Report) -> None:
bg, fg = BANNER.get(r.verdict_severity, BANNER[Severity.WARNING])
if r.lines == 0:
bg, fg = "#e6e6e6", "#333333"
self.verdict.setText(r.verdict)
self.verdict.setStyleSheet(f"QLabel {{ background:{bg}; color:{fg}; border-radius:6px; }}")
name, colour = LED[r.led]
self.led_swatch.setStyleSheet(
f"QLabel {{ background:{colour}; border:1px solid #777; border-radius:11px; }}")
ask = "" if r.led == "UNKNOWN" else " - <i>check it matches the box</i>"
reason = f"<br><small style='color:#666'>from: {_esc(r.led_reason)}</small>" if r.led_reason else ""
self.led_text.setText(f"<b>{name}</b>{ask}{reason}")
fw = r.firmware
sd = {True: "reads Wi-Fi settings from the SD card",
False: "does not read Wi-Fi settings from the SD card",
None: "Wi-Fi-from-SD support unknown"}
self.firmware.setText(f"{_esc(fw.describe())}<br><small>{sd[fw.supports_sd_wifi]}</small>")
self.milestones.clear()
for _, label, t in r.milestones:
done = t is not None
it = QListWidgetItem(("✔ " if done else "○ ") + label
+ (f" ({t / 1000:.1f} s)" if done else ""))
it.setForeground(QBrush(QColor("#1b7a33" if done else "#8a8a8a")))
self.milestones.addItem(it)
rows = min(max(self.milestones.count(), 1), 15)
self.milestones.setFixedHeight(rows * max(self.milestones.sizeHintForRow(0), 16) + 6)
self._fill_tree(r)
def _fill_tree(self, r: Report) -> None:
self.tree.blockSignals(True)
self.tree.clear()
self._findings = {}
if not r.roots:
it = QTreeWidgetItem(["No problems detected in the log so far."])
it.setForeground(0, QBrush(QColor("#1b7a33")))
self.tree.addTopLevelItem(it)
for rc in r.roots:
top = self._item(rc.finding, bold=True)
self.tree.addTopLevelItem(top)
for c in rc.consequences:
top.addChild(self._item(c, prefix="↳ "))
top.setExpanded(self._expanded.get(rc.finding.key, True))
for title, key, items in (("Seen in earlier boots only", "grp:earlier", r.earlier),
("Resolved during this session", "grp:resolved", r.resolved)):
if not items:
continue
grp = QTreeWidgetItem([f"{title} ({len(items)})"])
grp.setData(0, _KEY, key)
grp.setForeground(0, QBrush(QColor("#666666")))
self.tree.addTopLevelItem(grp)
for f in items:
grp.addChild(self._item(f, prefix="✔ " if key == "grp:resolved" else ""))
grp.setExpanded(self._expanded.get(key, False))
# Keep the selection across refreshes; default to the top root cause.
target = self._selected if self._selected in self._findings else (
r.roots[0].finding.key if r.roots else None)
self.tree.blockSignals(False)
self._select(target)
def _remember(self, item: QTreeWidgetItem, expanded: bool) -> None:
if key := item.data(0, _KEY):
self._expanded[key] = expanded
def _item(self, f: Finding, prefix: str = "", bold: bool = False) -> QTreeWidgetItem:
it = QTreeWidgetItem([f"{prefix}[{f.severity.label}] {f.title}"])
it.setData(0, _KEY, f.key)
it.setForeground(0, QBrush(QColor(SEV_COLOR.get(f.severity, "#333333"))))
it.setToolTip(0, f.title)
if bold:
font = it.font(0)
font.setBold(True)
it.setFont(0, font)
self._findings[f.key] = f
return it
def _select(self, key: str | None) -> None:
if key is None:
self.detail.setHtml("<p style='color:#666'>Nothing to show yet.</p>")
return
match = self.tree.findItems("*", Qt.MatchFlag.MatchWildcard | Qt.MatchFlag.MatchRecursive)
for it in match:
if it.data(0, _KEY) == key:
self.tree.setCurrentItem(it)
return
def _on_select(self, cur: QTreeWidgetItem | None, _prev) -> None:
key = cur.data(0, _KEY) if cur else None
f = self._findings.get(key or "")
if f is None:
return
self._selected = key
self.detail.setHtml(_detail_html(f))
def _detail_html(f: Finding) -> str:
colour = SEV_COLOR.get(f.severity, "#333")
actions = "".join(f"<li>{_esc(a)}</li>" for a in f.actions)
seen = f"Seen {f.count}x"
if f.first_uptime_ms is not None:
seen += f", first at {f.first_uptime_ms / 1000:.1f} s after start"
if f.resolved and f.resolved_by:
seen += f" - resolved ({_esc(f.resolved_by)})"
evidence = "<br>".join(_esc(e) for e in f.evidence)
return (
f"<h3 style='color:{colour}; margin:0'>{_esc(f.title)}</h3>"
f"<p><b>What it means:</b> {_esc(f.meaning)}</p>"
+ (f"<p><b>What to do:</b></p><ol>{actions}</ol>" if actions else "")
+ f"<p style='color:#666'>{seen}</p>"
+ (f"<p style='color:#555; font-family:Consolas,Menlo,monospace; font-size:9pt'>{evidence}</p>"
if evidence else "")
)
kota_tech/ui/main_window.py
+299 -234
View File
@@ -1,234 +1,299 @@
from __future__ import annotations
from PySide6.QtCore import QObject, Qt, QTimer, QUrl, Signal
from PySide6.QtGui import QColor, QDesktopServices, QFont, QTextCharFormat, QTextCursor
from PySide6.QtWidgets import (
QCheckBox,
QComboBox,
QFileDialog,
QHBoxLayout,
QLabel,
QMainWindow,
QPlainTextEdit,
QPushButton,
QTabWidget,
QTextBrowser,
QVBoxLayout,
QWidget,
)
from .. import __version__
from ..core.logline import LineKind, LogLine
from ..core.session import SessionLog
from ..serialio.manager import SerialManager, State
from ..serialio.ports import PortInfo, list_serial_ports
from .cheatsheet import cheatsheet_html
MAX_LINES = 20000
LEVEL_FILTERS = {"All lines": None, "Warnings + errors": {"W", "E"}, "Errors only": {"E"}}
COLORS = {
"E": "#ff6b6b",
"W": "#f5c451",
"I": "#d4d4d4",
LineKind.BOOT: "#6fc3ff",
LineKind.RESET: "#6fc3ff",
LineKind.PANIC: "#ff4dd2",
LineKind.HOST: "#7ee07e",
LineKind.RAW: "#a8a8a8",
}
class _Bridge(QObject):
"""Marshals SerialManager callbacks (reader thread) onto the UI thread."""
line = Signal(object)
state = Signal(object, str)
class SerialTab(QWidget):
def __init__(self, parent: QWidget | None = None):
super().__init__(parent)
self._bridge = _Bridge()
self._bridge.line.connect(self._on_line)
self._bridge.state.connect(self._on_state)
self._manager = SerialManager(self._bridge.line.emit, self._bridge.state.emit)
self._session = SessionLog()
self._ports: list[PortInfo] = []
self._counts = {"E": 0, "W": 0, "boots": 0}
# --- controls
self.port_combo = QComboBox(minimumWidth=320)
self.all_ports = QCheckBox("Show all ports")
self.refresh_btn = QPushButton("Refresh")
self.connect_btn = QPushButton("Connect")
self.reset_btn = QPushButton("Reset && capture")
self.reset_btn.setToolTip("Reboot the sensorbox and record the log from the first boot line")
self.filter_combo = QComboBox()
self.filter_combo.addItems(LEVEL_FILTERS)
self.clear_btn = QPushButton("Clear view")
self.save_btn = QPushButton("Save view as...")
self.folder_btn = QPushButton("Open logs folder")
row1 = QHBoxLayout()
for w in (QLabel("Port:"), self.port_combo, self.all_ports, self.refresh_btn,
self.connect_btn, self.reset_btn):
row1.addWidget(w)
row1.addStretch(1)
row2 = QHBoxLayout()
for w in (QLabel("Show:"), self.filter_combo, self.clear_btn, self.save_btn, self.folder_btn):
row2.addWidget(w)
row2.addStretch(1)
self.view = QPlainTextEdit(readOnly=True)
self.view.setMaximumBlockCount(MAX_LINES)
self.view.setLineWrapMode(QPlainTextEdit.LineWrapMode.NoWrap)
mono = QFont("Menlo" if QFont("Menlo").exactMatch() else "Consolas")
mono.setStyleHint(QFont.StyleHint.Monospace)
mono.setPointSize(10)
self.view.setFont(mono)
self.view.setStyleSheet("QPlainTextEdit { background:#1e1e1e; color:#d4d4d4; }")
self.status = QLabel()
layout = QVBoxLayout(self)
layout.addLayout(row1)
layout.addLayout(row2)
layout.addWidget(self.view, 1)
layout.addWidget(self.status)
# --- wiring
self.refresh_btn.clicked.connect(self.refresh_ports)
self.all_ports.toggled.connect(self.refresh_ports)
self.connect_btn.clicked.connect(self._toggle_connect)
self.reset_btn.clicked.connect(self._manager.reset_board)
self.clear_btn.clicked.connect(self.view.clear)
self.save_btn.clicked.connect(self._save_view)
self.folder_btn.clicked.connect(self._open_folder)
# Auto-detect a box being plugged in while idle.
self._poll = QTimer(self, interval=2000)
self._poll.timeout.connect(self._auto_refresh)
self._poll.start()
self.refresh_ports()
self._on_state(State.DISCONNECTED, "")
# ---- ports -------------------------------------------------------
def refresh_ports(self) -> None:
current = self.port_combo.currentData()
self._ports = list_serial_ports(include_all=self.all_ports.isChecked())
self.port_combo.clear()
for p in self._ports:
self.port_combo.addItem(p.label, p.device)
if not self._ports:
self.port_combo.addItem("No sensorbox found - check USB cable and power switch", None)
idx = self.port_combo.findData(current)
self.port_combo.setCurrentIndex(max(idx, 0))
def _auto_refresh(self) -> None:
if self._manager.is_running:
return
devices = [p.device for p in list_serial_ports(self.all_ports.isChecked())]
if devices != [p.device for p in self._ports]:
self.refresh_ports()
def _selected_port(self) -> PortInfo | None:
dev = self.port_combo.currentData()
return next((p for p in self._ports if p.device == dev), None)
# ---- connection --------------------------------------------------
def _toggle_connect(self) -> None:
if self._manager.is_running:
# Session file is closed in _on_state(DISCONNECTED), after the
# reader thread's final queued lines have been written.
self._manager.disconnect()
return
port = self._selected_port()
if port is None:
return
self._counts = {"E": 0, "W": 0, "boots": 0}
self._session.start(port.mac or port.device)
self._manager.connect(port)
def _on_state(self, state: State, detail: str) -> None:
running = state != State.DISCONNECTED
if not running:
self._session.stop()
self.connect_btn.setText("Disconnect" if running else "Connect")
self.reset_btn.setEnabled(state == State.CONNECTED)
self.port_combo.setEnabled(not running)
self.all_ports.setEnabled(not running)
self.refresh_btn.setEnabled(not running)
self._state_text = f"{state.value}" + (f" ({detail})" if detail else "")
self._update_status()
def _update_status(self) -> None:
sess = f" | Session log: {self._session.path}" if self._session.path else ""
c = self._counts
self.status.setText(
f"{getattr(self, '_state_text', '')} | boots seen: {c['boots']} "
f"errors: {c['E']} warnings: {c['W']}{sess}"
)
# ---- lines -------------------------------------------------------
def _on_line(self, line: LogLine) -> None:
self._session.write(line)
if line.kind == LineKind.BOOT:
self._counts["boots"] += 1
if line.level in ("E", "W"):
self._counts[line.level] += 1
self._update_status()
allowed = LEVEL_FILTERS[self.filter_combo.currentText()]
if allowed is not None and line.kind != LineKind.HOST and not (
line.level in allowed or line.kind in (LineKind.PANIC, LineKind.BOOT, LineKind.RESET)
):
return
self._append(line)
def _append(self, line: LogLine) -> None:
sb = self.view.verticalScrollBar()
at_bottom = sb.value() >= sb.maximum() - 4
fmt = QTextCharFormat()
fmt.setForeground(QColor(COLORS.get(line.level) or COLORS.get(line.kind, "#d4d4d4")))
if line.kind in (LineKind.PANIC, LineKind.BOOT, LineKind.HOST):
fmt.setFontWeight(QFont.Weight.Bold)
cur = QTextCursor(self.view.document())
cur.movePosition(QTextCursor.MoveOperation.End)
if not self.view.document().isEmpty():
cur.insertBlock()
cur.insertText(line.text, fmt)
if at_bottom:
sb.setValue(sb.maximum())
# ---- misc --------------------------------------------------------
def _save_view(self) -> None:
path, _ = QFileDialog.getSaveFileName(self, "Save log view", "sensorbox_log.txt", "Text (*.txt)")
if path:
with open(path, "w", encoding="utf-8") as fh:
fh.write(self.view.toPlainText())
def _open_folder(self) -> None:
self._session.root.mkdir(parents=True, exist_ok=True)
QDesktopServices.openUrl(QUrl.fromLocalFile(str(self._session.root)))
def shutdown(self) -> None:
self._manager.disconnect()
self._session.stop()
class MainWindow(QMainWindow):
def __init__(self):
super().__init__()
self.setWindowTitle(f"KOTA Sensorbox Technician Tool v{__version__}")
self.resize(1200, 800)
tabs = QTabWidget()
guide = QTextBrowser(openExternalLinks=True)
guide.setHtml(cheatsheet_html())
self.serial_tab = SerialTab()
tabs.addTab(guide, "Quick start")
tabs.addTab(self.serial_tab, "Serial monitor")
self.setCentralWidget(tabs)
def closeEvent(self, event): # noqa: N802 (Qt naming)
self.serial_tab.shutdown()
super().closeEvent(event)
from __future__ import annotations
from pathlib import Path
from PySide6.QtCore import QObject, Qt, QTimer, QUrl, Signal
from PySide6.QtGui import QColor, QDesktopServices, QFont, QGuiApplication, QTextCharFormat, QTextCursor
from PySide6.QtWidgets import (
QCheckBox,
QComboBox,
QFileDialog,
QHBoxLayout,
QLabel,
QMainWindow,
QPlainTextEdit,
QPushButton,
QSplitter,
QTabWidget,
QTextBrowser,
QVBoxLayout,
QWidget,
)
from .. import __version__
from ..core.logfile import read_log_file
from ..core.logline import LineKind, LogLine
from ..core.session import SessionLog
from ..diag.engine import DiagnosticsEngine
from ..serialio.manager import SerialManager, State
from ..serialio.ports import PortInfo, list_serial_ports
from .cheatsheet import cheatsheet_html
from .diag_panel import DiagnosticsPanel
MAX_LINES = 20000
LEVEL_FILTERS = {"All lines": None, "Warnings + errors": {"W", "E"}, "Errors only": {"E"}}
COLORS = {
"E": "#ff6b6b",
"W": "#f5c451",
"I": "#d4d4d4",
LineKind.BOOT: "#6fc3ff",
LineKind.RESET: "#6fc3ff",
LineKind.PANIC: "#ff4dd2",
LineKind.HOST: "#7ee07e",
LineKind.RAW: "#a8a8a8",
}
class _Bridge(QObject):
"""Marshals SerialManager callbacks (reader thread) onto the UI thread."""
line = Signal(object)
state = Signal(object, str)
class SerialTab(QWidget):
def __init__(self, parent: QWidget | None = None):
super().__init__(parent)
self._bridge = _Bridge()
self._bridge.line.connect(self._on_line)
self._bridge.state.connect(self._on_state)
self._manager = SerialManager(self._bridge.line.emit, self._bridge.state.emit)
self._session = SessionLog()
self._ports: list[PortInfo] = []
self._counts = {"E": 0, "W": 0, "boots": 0}
self._engine = DiagnosticsEngine()
self._diag_dirty = True
# --- controls
self.port_combo = QComboBox(minimumWidth=320)
self.all_ports = QCheckBox("Show all ports")
self.refresh_btn = QPushButton("Refresh")
self.connect_btn = QPushButton("Connect")
self.reset_btn = QPushButton("Reset && capture")
self.reset_btn.setToolTip("Reboot the sensorbox and record the log from the first boot line")
self.filter_combo = QComboBox()
self.filter_combo.addItems(LEVEL_FILTERS)
self.clear_btn = QPushButton("Clear view")
self.save_btn = QPushButton("Save view as...")
self.folder_btn = QPushButton("Open logs folder")
self.analyze_btn = QPushButton("Analyze log file...")
self.analyze_btn.setToolTip("Load a saved session log or serial capture and diagnose it")
self.copy_btn = QPushButton("Copy report")
self.copy_btn.setToolTip("Copy the diagnosis as text (e.g. to send to the firmware team)")
row1 = QHBoxLayout()
for w in (QLabel("Port:"), self.port_combo, self.all_ports, self.refresh_btn,
self.connect_btn, self.reset_btn):
row1.addWidget(w)
row1.addStretch(1)
row2 = QHBoxLayout()
for w in (QLabel("Show:"), self.filter_combo, self.clear_btn, self.save_btn, self.folder_btn,
self.analyze_btn, self.copy_btn):
row2.addWidget(w)
row2.addStretch(1)
self.view = QPlainTextEdit(readOnly=True)
self.view.setMaximumBlockCount(MAX_LINES)
self.view.setLineWrapMode(QPlainTextEdit.LineWrapMode.NoWrap)
mono = QFont("Menlo" if QFont("Menlo").exactMatch() else "Consolas")
mono.setStyleHint(QFont.StyleHint.Monospace)
mono.setPointSize(10)
self.view.setFont(mono)
self.view.setStyleSheet("QPlainTextEdit { background:#1e1e1e; color:#d4d4d4; }")
self.diag = DiagnosticsPanel()
split = QSplitter(Qt.Orientation.Horizontal)
split.addWidget(self.view)
split.addWidget(self.diag)
split.setStretchFactor(0, 3)
split.setStretchFactor(1, 2)
split.setSizes([700, 480])
self.status = QLabel()
layout = QVBoxLayout(self)
layout.addLayout(row1)
layout.addLayout(row2)
layout.addWidget(split, 1)
layout.addWidget(self.status)
# --- wiring
self.refresh_btn.clicked.connect(self.refresh_ports)
self.all_ports.toggled.connect(self.refresh_ports)
self.connect_btn.clicked.connect(self._toggle_connect)
self.reset_btn.clicked.connect(self._manager.reset_board)
self.clear_btn.clicked.connect(self.view.clear)
self.save_btn.clicked.connect(self._save_view)
self.folder_btn.clicked.connect(self._open_folder)
self.analyze_btn.clicked.connect(self._analyze_file)
self.copy_btn.clicked.connect(self._copy_report)
# Report refresh is throttled; lines can arrive at >100/s.
self._diag_timer = QTimer(self, interval=400)
self._diag_timer.timeout.connect(self._refresh_diag)
self._diag_timer.start()
# Auto-detect a box being plugged in while idle.
self._poll = QTimer(self, interval=2000)
self._poll.timeout.connect(self._auto_refresh)
self._poll.start()
self.refresh_ports()
self._on_state(State.DISCONNECTED, "")
# ---- ports -------------------------------------------------------
def refresh_ports(self) -> None:
current = self.port_combo.currentData()
self._ports = list_serial_ports(include_all=self.all_ports.isChecked())
self.port_combo.clear()
for p in self._ports:
self.port_combo.addItem(p.label, p.device)
if not self._ports:
self.port_combo.addItem("No sensorbox found - check USB cable and power switch", None)
idx = self.port_combo.findData(current)
self.port_combo.setCurrentIndex(max(idx, 0))
def _auto_refresh(self) -> None:
if self._manager.is_running:
return
devices = [p.device for p in list_serial_ports(self.all_ports.isChecked())]
if devices != [p.device for p in self._ports]:
self.refresh_ports()
def _selected_port(self) -> PortInfo | None:
dev = self.port_combo.currentData()
return next((p for p in self._ports if p.device == dev), None)
# ---- connection --------------------------------------------------
def _toggle_connect(self) -> None:
if self._manager.is_running:
# Session file is closed in _on_state(DISCONNECTED), after the
# reader thread's final queued lines have been written.
self._manager.disconnect()
return
port = self._selected_port()
if port is None:
return
self._counts = {"E": 0, "W": 0, "boots": 0}
self._engine.reset()
self._diag_dirty = True
self._session.start(port.mac or port.device)
self._manager.connect(port)
def _on_state(self, state: State, detail: str) -> None:
running = state != State.DISCONNECTED
if not running and self._session.path is not None:
self._session.write_report(self._engine.report().to_text())
self._session.stop()
self.connect_btn.setText("Disconnect" if running else "Connect")
self.reset_btn.setEnabled(state == State.CONNECTED)
self.port_combo.setEnabled(not running)
self.all_ports.setEnabled(not running)
self.refresh_btn.setEnabled(not running)
# Offline analysis would mix with the live session; disconnect first.
self.analyze_btn.setEnabled(not running)
self._state_text = f"{state.value}" + (f" ({detail})" if detail else "")
self._update_status()
def _update_status(self) -> None:
sess = f" | Session log: {self._session.path}" if self._session.path else ""
c = self._counts
self.status.setText(
f"{getattr(self, '_state_text', '')} | boots seen: {c['boots']} "
f"errors: {c['E']} warnings: {c['W']}{sess}"
)
# ---- lines -------------------------------------------------------
def _on_line(self, line: LogLine) -> None:
self._session.write(line)
self._engine.feed(line)
self._diag_dirty = True
if line.kind == LineKind.BOOT:
self._counts["boots"] += 1
if line.level in ("E", "W"):
self._counts[line.level] += 1
self._update_status()
allowed = LEVEL_FILTERS[self.filter_combo.currentText()]
if allowed is not None and line.kind != LineKind.HOST and not (
line.level in allowed or line.kind in (LineKind.PANIC, LineKind.BOOT, LineKind.RESET)
):
return
self._append(line)
def _append(self, line: LogLine) -> None:
sb = self.view.verticalScrollBar()
at_bottom = sb.value() >= sb.maximum() - 4
fmt = QTextCharFormat()
fmt.setForeground(QColor(COLORS.get(line.level) or COLORS.get(line.kind, "#d4d4d4")))
if line.kind in (LineKind.PANIC, LineKind.BOOT, LineKind.HOST):
fmt.setFontWeight(QFont.Weight.Bold)
cur = QTextCursor(self.view.document())
cur.movePosition(QTextCursor.MoveOperation.End)
if not self.view.document().isEmpty():
cur.insertBlock()
cur.insertText(line.text, fmt)
if at_bottom:
sb.setValue(sb.maximum())
# ---- diagnostics -------------------------------------------------
def _refresh_diag(self) -> None:
if self._diag_dirty:
self._diag_dirty = False
self.diag.show_report(self._engine.report())
def _copy_report(self) -> None:
QGuiApplication.clipboard().setText(self._engine.report().to_text())
def _analyze_file(self) -> None:
path, _ = QFileDialog.getOpenFileName(
self, "Analyze log file", str(self._session.root), "Logs (*.log *.txt);;All files (*)")
if not path:
return
lines = read_log_file(Path(path))
self.view.clear()
self._engine.reset()
self._counts = {"E": 0, "W": 0, "boots": 0}
for line in lines:
self._engine.feed(line)
if line.kind == LineKind.BOOT:
self._counts["boots"] += 1
if line.level in ("E", "W"):
self._counts[line.level] += 1
self._append(line)
self._state_text = f"Analyzed file: {Path(path).name} ({len(lines)} lines)"
self._update_status()
self._diag_dirty = True
self._refresh_diag()
# ---- misc --------------------------------------------------------
def _save_view(self) -> None:
path, _ = QFileDialog.getSaveFileName(self, "Save log view", "sensorbox_log.txt", "Text (*.txt)")
if path:
with open(path, "w", encoding="utf-8") as fh:
fh.write(self.view.toPlainText())
def _open_folder(self) -> None:
self._session.root.mkdir(parents=True, exist_ok=True)
QDesktopServices.openUrl(QUrl.fromLocalFile(str(self._session.root)))
def shutdown(self) -> None:
self._manager.disconnect()
self._session.stop()
class MainWindow(QMainWindow):
def __init__(self):
super().__init__()
self.setWindowTitle(f"KOTA Sensorbox Technician Tool v{__version__}")
self.resize(1200, 800)
tabs = QTabWidget()
guide = QTextBrowser(openExternalLinks=True)
guide.setHtml(cheatsheet_html())
self.serial_tab = SerialTab()
tabs.addTab(guide, "Quick start")
tabs.addTab(self.serial_tab, "Serial monitor")
self.setCentralWidget(tabs)
def closeEvent(self, event): # noqa: N802 (Qt naming)
self.serial_tab.shutdown()
super().closeEvent(event)