402 lines
17 KiB
Python
402 lines
17 KiB
Python
"""Guided troubleshooting: executes the decision trees in guides.toml.
|
|
|
|
GUI-independent. The runner is fed the same LogLines as the diagnostics
|
|
engine, reads the engine's report to route and to skip questions the log
|
|
already answers, asks the host to perform tool actions (board reset), and
|
|
records every answer and conclusion in the engine so they appear in the
|
|
report and the session log.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import time
|
|
import tomllib
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
from ..core.logline import LogLine
|
|
from .engine import LED, DiagnosticsEngine, Severity
|
|
|
|
GUIDES_PATH = Path(__file__).with_name("guides.toml")
|
|
# LED answers a technician can give (engine colours + a reboot-loop pattern).
|
|
LED_ANSWERS = ["GREEN", "PURPLE", "RED", "BLUE", "WHITE", "YELLOW", "ORANGE", "CYAN", "OFF", "CYCLING"]
|
|
LED_ANSWER_LABELS = {**{k: v[0] for k, v in LED.items()},
|
|
"CYCLING": "Keeps changing / goes dark every few seconds"}
|
|
STEP_TYPES = {"led", "question", "info", "wait", "route"}
|
|
END_SEVERITY = {"ok": None, "info": Severity.INFO, "warning": Severity.WARNING,
|
|
"error": Severity.ERROR, "critical": Severity.CRITICAL}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Option:
|
|
label: str
|
|
next: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Step:
|
|
id: str
|
|
type: str
|
|
text: str = ""
|
|
options: tuple[Option, ...] = ()
|
|
route: dict[str, str] = field(default_factory=dict) # led colour -> target
|
|
routes: dict[str, str] = field(default_factory=dict) # rule id -> target
|
|
default: str | None = None
|
|
next: str | None = None
|
|
tool_action: str | None = None
|
|
until: re.Pattern[str] | None = None
|
|
until_milestone: str | None = None
|
|
fail_on: re.Pattern[str] | None = None
|
|
timeout_s: float = 60.0
|
|
pass_on_timeout: bool = False
|
|
on_ok: str | None = None
|
|
on_fail: str | None = None
|
|
on_timeout: str | None = None
|
|
skip_if_finding: str | None = None
|
|
skip_unless_feature: str | None = None
|
|
skip_to: str | None = None
|
|
show_qr: bool = False # show the Matter pairing QR next to the step text
|
|
|
|
def targets(self) -> list[str]:
|
|
t = [o.next for o in self.options] + list(self.route.values()) + list(self.routes.values())
|
|
t += [x for x in (self.default, self.next, self.on_ok, self.on_fail, self.on_timeout, self.skip_to) if x]
|
|
return t
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class End:
|
|
id: str
|
|
severity: str
|
|
title: str
|
|
text: str
|
|
actions: tuple[str, ...] = ()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Guide:
|
|
id: str
|
|
title: str
|
|
summary: str
|
|
recommend: tuple[str, ...]
|
|
start: str
|
|
steps: dict[str, Step]
|
|
ends: dict[str, End]
|
|
|
|
|
|
def load_guides(path: Path = GUIDES_PATH) -> dict[str, Guide]:
|
|
data = tomllib.loads(path.read_text(encoding="utf-8"))
|
|
guides: dict[str, Guide] = {}
|
|
for g in data["guide"]:
|
|
steps = {}
|
|
for s in g.get("step", []):
|
|
steps[s["id"]] = Step(
|
|
id=s["id"], type=s["type"], text=s.get("text", ""),
|
|
options=tuple(Option(o["label"], o["next"]) for o in s.get("options", ())),
|
|
route=dict(s.get("route", {})), routes=dict(s.get("routes", {})),
|
|
default=s.get("default"), next=s.get("next"), tool_action=s.get("tool_action"),
|
|
until=re.compile(s["until"]) if "until" in s else None,
|
|
until_milestone=s.get("until_milestone"),
|
|
fail_on=re.compile(s["fail_on"]) if "fail_on" in s else None,
|
|
timeout_s=float(s.get("timeout_s", 60)), pass_on_timeout=bool(s.get("pass_on_timeout", False)),
|
|
on_ok=s.get("on_ok"), on_fail=s.get("on_fail"), on_timeout=s.get("on_timeout"),
|
|
skip_if_finding=s.get("skip_if_finding"), skip_unless_feature=s.get("skip_unless_feature"),
|
|
skip_to=s.get("skip_to"), show_qr=bool(s.get("show_qr", False)),
|
|
)
|
|
ends = {e["id"]: End(e["id"], e["severity"], e["title"], e["text"], tuple(e.get("actions", ())))
|
|
for e in g.get("end", [])}
|
|
guides[g["id"]] = Guide(g["id"], g["title"], g["summary"], tuple(g.get("recommend", ())),
|
|
g["start"], steps, ends)
|
|
_validate(guides)
|
|
return guides
|
|
|
|
|
|
def _validate(guides: dict[str, Guide]) -> None:
|
|
for g in guides.values():
|
|
if g.start not in g.steps:
|
|
raise ValueError(f"guide {g.id}: start step {g.start!r} missing")
|
|
for s in g.steps.values():
|
|
if s.type not in STEP_TYPES:
|
|
raise ValueError(f"{g.id}.{s.id}: unknown step type {s.type!r}")
|
|
if s.type == "led" and set(s.route) - set(LED_ANSWERS):
|
|
raise ValueError(f"{g.id}.{s.id}: unknown LED colours {set(s.route) - set(LED_ANSWERS)}")
|
|
if s.type == "wait" and not (s.until or s.until_milestone or s.pass_on_timeout):
|
|
raise ValueError(f"{g.id}.{s.id}: wait step needs until/until_milestone/pass_on_timeout")
|
|
if s.tool_action not in (None, "reset"):
|
|
raise ValueError(f"{g.id}.{s.id}: unknown tool_action {s.tool_action!r}")
|
|
for t in s.targets():
|
|
kind, _, ref = t.partition(":")
|
|
if kind == "end" and ref not in g.ends:
|
|
raise ValueError(f"{g.id}.{s.id}: unknown end {ref!r}")
|
|
elif kind == "guide" and ref not in guides:
|
|
raise ValueError(f"{g.id}.{s.id}: unknown guide {ref!r}")
|
|
elif not ref and t not in g.steps:
|
|
raise ValueError(f"{g.id}.{s.id}: unknown step {t!r}")
|
|
for e in g.ends.values():
|
|
if e.severity not in END_SEVERITY:
|
|
raise ValueError(f"{g.id}.end.{e.id}: unknown severity {e.severity!r}")
|
|
|
|
|
|
def recommended(guides: dict[str, Guide], engine: DiagnosticsEngine) -> list[Guide]:
|
|
"""Guides matching the current root findings (in root order), then the LED."""
|
|
report = engine.report()
|
|
keys: list[str] = []
|
|
for rc in report.roots:
|
|
keys.append(rc.finding.rule_id)
|
|
keys += [c.rule_id for c in rc.consequences]
|
|
if report.led not in ("UNKNOWN",):
|
|
keys.append(f"led:{report.led}")
|
|
out: list[Guide] = [g for g in guides.values() if "always" in g.recommend]
|
|
for k in keys:
|
|
for g in guides.values():
|
|
if k in g.recommend and g not in out:
|
|
out.append(g)
|
|
return out
|
|
|
|
|
|
@dataclass
|
|
class View:
|
|
"""What the UI should show right now."""
|
|
guide: Guide | None = None
|
|
step: Step | None = None
|
|
text: str = ""
|
|
options: list[tuple[str, str]] = field(default_factory=list) # (key, label)
|
|
waiting: bool = False
|
|
remaining_s: float = 0.0
|
|
note: str = ""
|
|
end: End | None = None
|
|
can_back: bool = False
|
|
|
|
|
|
class GuideRunner:
|
|
def __init__(self, engine: DiagnosticsEngine, guides: dict[str, Guide] | None = None,
|
|
on_action: Callable[[str], bool] | None = None,
|
|
on_log: Callable[[str], None] | None = None,
|
|
clock: Callable[[], float] = time.monotonic):
|
|
self.engine = engine
|
|
self.guides = guides if guides is not None else load_guides()
|
|
self._on_action = on_action or (lambda _a: False)
|
|
self._on_log = on_log or (lambda _t: None)
|
|
self._clock = clock
|
|
self.cancel()
|
|
|
|
# ------------------------------------------------------------ control
|
|
def cancel(self) -> None:
|
|
self.guide: Guide | None = None
|
|
self.step: Step | None = None
|
|
self.end: End | None = None
|
|
self.note = ""
|
|
self._stack: list[tuple[str, str]] = [] # (guide id, step id) for Back
|
|
self._wait_started = 0.0
|
|
self._wait_boot = 0
|
|
self._wait_milestones: set[str] = set()
|
|
|
|
def start(self, guide_id: str) -> None:
|
|
self.cancel()
|
|
self._enter_guide(guide_id)
|
|
|
|
def answer(self, key: str) -> None:
|
|
"""Pick an option: LED colour name for led steps, option index otherwise."""
|
|
s = self.step
|
|
if s is None:
|
|
return
|
|
if s.type == "led":
|
|
label = LED_ANSWER_LABELS.get(key, key)
|
|
self._record(s, label)
|
|
self._check_led(key)
|
|
target = s.route.get(key)
|
|
elif s.type == "question":
|
|
opt = s.options[int(key)]
|
|
self._record(s, opt.label)
|
|
target = opt.next
|
|
else:
|
|
return
|
|
if target:
|
|
self._push()
|
|
self._go(target)
|
|
|
|
def proceed(self) -> None:
|
|
"""Continue on info steps; 'skip waiting' on wait steps."""
|
|
s = self.step
|
|
if s is None:
|
|
return
|
|
if s.type == "info" and s.next:
|
|
self._push()
|
|
self._go(s.next)
|
|
elif s.type == "wait":
|
|
self._record(s, "technician skipped the wait")
|
|
self._finish_wait("timeout")
|
|
|
|
def back(self) -> None:
|
|
# Never step back into a wait: re-entering it could restart the box.
|
|
while self._stack:
|
|
gid, sid = self._stack.pop()
|
|
if self.guides[gid].steps[sid].type != "wait":
|
|
self.guide, self.end, self.note = self.guides[gid], None, ""
|
|
self._enter_step(sid)
|
|
return
|
|
|
|
# --------------------------------------------------------------- feed
|
|
def feed(self, line: LogLine) -> None:
|
|
s = self.step
|
|
if s is None or s.type != "wait":
|
|
return
|
|
if s.fail_on and s.fail_on.search(line.text):
|
|
self._record(s, f"log: {line.text}")
|
|
self._finish_wait("fail")
|
|
elif s.until and s.until.search(line.text):
|
|
self._record(s, f"log: {line.text}")
|
|
self._finish_wait("ok")
|
|
elif s.until_milestone and self._milestone_new(s.until_milestone):
|
|
self._record(s, f"milestone reached: {s.until_milestone}")
|
|
self._finish_wait("ok")
|
|
|
|
def tick(self) -> None:
|
|
s = self.step
|
|
if s is not None and s.type == "wait" and self._clock() - self._wait_started >= s.timeout_s:
|
|
self._finish_wait("ok" if s.pass_on_timeout else "timeout")
|
|
|
|
# --------------------------------------------------------------- view
|
|
def view(self) -> View:
|
|
v = View(guide=self.guide, step=self.step, end=self.end, note=self.note, can_back=bool(self._stack))
|
|
s = self.step
|
|
if s is None:
|
|
return v
|
|
v.text = self._render(s.text)
|
|
if s.type == "led":
|
|
v.options = [(k, LED_ANSWER_LABELS[k]) for k in LED_ANSWERS if k in s.route]
|
|
elif s.type == "question":
|
|
v.options = [(str(i), o.label) for i, o in enumerate(s.options)]
|
|
elif s.type == "wait":
|
|
v.waiting = True
|
|
v.remaining_s = max(0.0, s.timeout_s - (self._clock() - self._wait_started))
|
|
return v
|
|
|
|
# ---------------------------------------------------------- internals
|
|
def _push(self) -> None:
|
|
if self.guide and self.step:
|
|
self._stack.append((self.guide.id, self.step.id))
|
|
|
|
def _enter_guide(self, guide_id: str) -> None:
|
|
self.guide = self.guides[guide_id]
|
|
self._on_log(f"technician: started guide '{self.guide.title}'")
|
|
self._enter_step(self.guide.start)
|
|
|
|
def _go(self, target: str) -> None:
|
|
kind, _, ref = target.partition(":")
|
|
if kind == "end" and ref:
|
|
self._conclude(self.guide.ends[ref])
|
|
elif kind == "guide" and ref:
|
|
self._enter_guide(ref)
|
|
else:
|
|
self._enter_step(target)
|
|
|
|
def _enter_step(self, step_id: str) -> None:
|
|
s = self.guide.steps[step_id]
|
|
self.step, self.note = s, ""
|
|
if s.skip_if_finding and self._finding_active(s.skip_if_finding):
|
|
self._auto(s, f"skipped - log already shows {s.skip_if_finding}", s.skip_to)
|
|
return
|
|
if s.skip_unless_feature and not self._feature(s.skip_unless_feature):
|
|
self._auto(s, f"skipped - firmware has no {s.skip_unless_feature}", s.skip_to)
|
|
return
|
|
if s.type == "route":
|
|
target = self._route(s)
|
|
self._auto(s, f"routed by log findings -> {target}", target)
|
|
return
|
|
if s.type == "wait":
|
|
self._wait_started = self._clock()
|
|
self._wait_boot = self.engine.boot.index
|
|
self._wait_milestones = set(self.engine.milestones)
|
|
if s.tool_action and not self._on_action(s.tool_action):
|
|
self.note = "Could not restart the box from the tool - connect to it first, or switch it off and on."
|
|
# A milestone that is already reached satisfies the wait at once
|
|
# unless the step first restarts the box.
|
|
if s.until_milestone and not s.tool_action and s.until_milestone in self.engine.milestones:
|
|
self._record(s, f"already reached: {s.until_milestone}")
|
|
self._finish_wait("ok")
|
|
|
|
def _auto(self, s: Step, why: str, target: str | None) -> None:
|
|
self._on_log(f"technician guide: {s.id}: {why}")
|
|
if target:
|
|
self._go(target)
|
|
|
|
def _finish_wait(self, outcome: str) -> None:
|
|
s = self.step
|
|
target = {"ok": s.on_ok, "fail": s.on_fail or s.on_timeout, "timeout": s.on_timeout or s.on_fail}[outcome]
|
|
if target is None:
|
|
return
|
|
self._push()
|
|
self._go(target)
|
|
|
|
def _milestone_new(self, mid: str) -> bool:
|
|
if mid not in self.engine.milestones:
|
|
return False
|
|
return self.engine.boot.index != self._wait_boot or mid not in self._wait_milestones
|
|
|
|
def _route(self, s: Step) -> str:
|
|
report = self.engine.report()
|
|
for rc in report.roots:
|
|
for f in (rc.finding, *rc.consequences):
|
|
if f.rule_id in s.routes:
|
|
return s.routes[f.rule_id]
|
|
return s.default or "end:" + next(iter(self.guide.ends))
|
|
|
|
def _check_led(self, seen: str) -> None:
|
|
expected = self.engine.led
|
|
if expected in ("UNKNOWN", "OFF") or seen == "CYCLING" or seen == expected:
|
|
return
|
|
exp_label, see_label = LED[expected][0], LED_ANSWER_LABELS[seen]
|
|
self.engine.add_confirmed(
|
|
"tech.led_mismatch", Severity.WARNING,
|
|
f"LED shows {see_label}, but the log implies {exp_label}",
|
|
f"The firmware state in the log corresponds to {exp_label}, while the technician sees {see_label}. "
|
|
"Either the LED just changed (check again in a few seconds), the LED/its wiring is faulty, "
|
|
"or the box runs different firmware. Trust the log findings.",
|
|
("Look at the LED again for 10 seconds.", "If it stays different: note it for repair (LED board)."))
|
|
|
|
def _conclude(self, end: End) -> None:
|
|
self.step, self.end = None, end
|
|
self._on_log(f"technician guide result: {end.title}")
|
|
sev = END_SEVERITY[end.severity]
|
|
self.engine.note_technician(f"{self.guide.title}: {end.title}")
|
|
if sev is not None:
|
|
self.engine.add_confirmed(f"tech.{self.guide.id}.{end.id}", sev, end.title, end.text, end.actions)
|
|
|
|
def _record(self, s: Step, answer: str) -> None:
|
|
q = self._render(s.text)
|
|
self.engine.note_technician(f"{q} -> {answer}")
|
|
self._on_log(f"technician: {q[:80]} -> {answer}")
|
|
|
|
def _finding_active(self, rule_id: str) -> bool:
|
|
return any(f.rule_id == rule_id and f.active for f in self.engine.findings.values())
|
|
|
|
def _feature(self, name: str) -> bool:
|
|
fw = self.engine.firmware
|
|
if name == "sd_wifi":
|
|
return fw.supports_sd_wifi is not False
|
|
if name == "setup_portal":
|
|
if fw.known is not None:
|
|
return fw.known.setup_portal
|
|
if "setup_portal" in fw.evidence:
|
|
return True
|
|
return "legacy_v1" not in fw.evidence # unknown firmware: ask
|
|
return False
|
|
|
|
def _render(self, text: str) -> str:
|
|
report = self.engine.report()
|
|
sensors = sorted({f.title.removesuffix(" failed to start") for f in self.engine.findings.values()
|
|
if f.rule_id == "sensor.init_failed" and f.active})
|
|
ctx = {
|
|
"expected_led": LED[report.led][0],
|
|
"failing_sensors": ", ".join(sensors) or "see the Diagnosis tab",
|
|
"top_finding": report.roots[0].finding.title if report.roots else "none",
|
|
}
|
|
return text.format_map(_Fmt(ctx))
|
|
|
|
|
|
class _Fmt(dict):
|
|
def __missing__(self, key: str) -> str:
|
|
return "{" + key + "}"
|