Store all text files with LF line endings
Some files had been saved with CRLF on Windows, so a one-line change showed as a whole-file diff. .gitattributes now stores text as LF (.bat stays CRLF for cmd, .command/.sh LF for macOS). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
kkallasandClaude Opus 5.5 committed
1 parent
3fe45fa4d5
commit
0d21c5c205
19 files changed
+8290
-8288
No files matched your search
+999
-999
File diff suppressed because it is too large.
Load diff
+147
-147
@@ -1,147 +1,147 @@
|
||||
"""Firmware identification from the boot log.
|
||||
|
||||
Up to firmware 1.0 the app version string was "1.0" in every build, 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.
|
||||
|
||||
Firmware 1.0.16+ prints "KOTA: boot fw=1.0.16 build=4f10989e ..." (version
|
||||
unique per build, first 8 hex digits of the ELF SHA256). Builds found in the
|
||||
tool's firmware/ folder are registered with `register_local_build` so the
|
||||
box's build can be shown with its compile date.
|
||||
"""
|
||||
|
||||
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,
|
||||
),
|
||||
"c94ba0e9b": KnownFirmware(
|
||||
name="v1.0 source build (Sep 26 2026)",
|
||||
source="developer build, built by Krzysztof Kallas",
|
||||
sd_wifi=True,
|
||||
setup_portal=True,
|
||||
),
|
||||
}
|
||||
|
||||
# elf sha (full or prefix) -> (version, compile time, file name) of builds in firmware/
|
||||
LOCAL_BUILDS: dict[str, tuple[str, str, str]] = {}
|
||||
|
||||
|
||||
def register_local_build(elf_sha: str, version: str, built: str, file_name: str) -> None:
|
||||
LOCAL_BUILDS[elf_sha.lower()] = (version, built, file_name)
|
||||
|
||||
|
||||
def same_build(a: str | None, b: str | None) -> bool:
|
||||
"""ELF SHA prefixes of different lengths (8 from KOTA: boot, 9+ from
|
||||
app_init, 64 from an image) name the same build if they agree."""
|
||||
if not a or not b:
|
||||
return False
|
||||
n = min(len(a), len(b))
|
||||
return n >= 6 and a[:n].lower() == b[:n].lower()
|
||||
|
||||
|
||||
def local_build(elf_sha: str | None) -> tuple[str, str, str] | None:
|
||||
return next((v for sha, v in LOCAL_BUILDS.items() if same_build(sha, elf_sha)), None)
|
||||
|
||||
|
||||
_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
|
||||
hw: str | None = None
|
||||
evidence: set[str] = field(default_factory=set)
|
||||
|
||||
def observe_kota(self, fields: dict[str, str]) -> None:
|
||||
"""'KOTA: boot fw=1.0.16 build=4f10989e hw=2 ...'"""
|
||||
if fields.get("fw"):
|
||||
self.version = fields["fw"]
|
||||
build = fields.get("build", "")
|
||||
# app_init's longer prefix is kept if it agrees.
|
||||
if build and not same_build(self.elf_sha, build):
|
||||
self.elf_sha = build.lower()
|
||||
if fields.get("hw"):
|
||||
self.hw = fields["hw"]
|
||||
self.evidence.add("kota") # 1.0.16+: SD Wi-Fi and setup portal exist
|
||||
|
||||
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 same_build(self.elf_sha, 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 or "kota" 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 and (self.version not in (None, "1.0") or local_build(self.elf_sha)):
|
||||
local = local_build(self.elf_sha)
|
||||
built = self.built or (local[1] if local else None)
|
||||
text = f"v{self.version or (local[0] if local else '?')} (build {self.elf_sha[:8]}"
|
||||
text += f", compiled {built})" if built else ")"
|
||||
if self.hw:
|
||||
text += f", hardware {self.hw}"
|
||||
if local:
|
||||
text += f" - same as firmware/{local[2]}"
|
||||
return text
|
||||
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"
|
||||
"""Firmware identification from the boot log.
|
||||
|
||||
Up to firmware 1.0 the app version string was "1.0" in every build, 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.
|
||||
|
||||
Firmware 1.0.16+ prints "KOTA: boot fw=1.0.16 build=4f10989e ..." (version
|
||||
unique per build, first 8 hex digits of the ELF SHA256). Builds found in the
|
||||
tool's firmware/ folder are registered with `register_local_build` so the
|
||||
box's build can be shown with its compile date.
|
||||
"""
|
||||
|
||||
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,
|
||||
),
|
||||
"c94ba0e9b": KnownFirmware(
|
||||
name="v1.0 source build (Sep 26 2026)",
|
||||
source="developer build, built by Krzysztof Kallas",
|
||||
sd_wifi=True,
|
||||
setup_portal=True,
|
||||
),
|
||||
}
|
||||
|
||||
# elf sha (full or prefix) -> (version, compile time, file name) of builds in firmware/
|
||||
LOCAL_BUILDS: dict[str, tuple[str, str, str]] = {}
|
||||
|
||||
|
||||
def register_local_build(elf_sha: str, version: str, built: str, file_name: str) -> None:
|
||||
LOCAL_BUILDS[elf_sha.lower()] = (version, built, file_name)
|
||||
|
||||
|
||||
def same_build(a: str | None, b: str | None) -> bool:
|
||||
"""ELF SHA prefixes of different lengths (8 from KOTA: boot, 9+ from
|
||||
app_init, 64 from an image) name the same build if they agree."""
|
||||
if not a or not b:
|
||||
return False
|
||||
n = min(len(a), len(b))
|
||||
return n >= 6 and a[:n].lower() == b[:n].lower()
|
||||
|
||||
|
||||
def local_build(elf_sha: str | None) -> tuple[str, str, str] | None:
|
||||
return next((v for sha, v in LOCAL_BUILDS.items() if same_build(sha, elf_sha)), None)
|
||||
|
||||
|
||||
_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
|
||||
hw: str | None = None
|
||||
evidence: set[str] = field(default_factory=set)
|
||||
|
||||
def observe_kota(self, fields: dict[str, str]) -> None:
|
||||
"""'KOTA: boot fw=1.0.16 build=4f10989e hw=2 ...'"""
|
||||
if fields.get("fw"):
|
||||
self.version = fields["fw"]
|
||||
build = fields.get("build", "")
|
||||
# app_init's longer prefix is kept if it agrees.
|
||||
if build and not same_build(self.elf_sha, build):
|
||||
self.elf_sha = build.lower()
|
||||
if fields.get("hw"):
|
||||
self.hw = fields["hw"]
|
||||
self.evidence.add("kota") # 1.0.16+: SD Wi-Fi and setup portal exist
|
||||
|
||||
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 same_build(self.elf_sha, 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 or "kota" 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 and (self.version not in (None, "1.0") or local_build(self.elf_sha)):
|
||||
local = local_build(self.elf_sha)
|
||||
built = self.built or (local[1] if local else None)
|
||||
text = f"v{self.version or (local[0] if local else '?')} (build {self.elf_sha[:8]}"
|
||||
text += f", compiled {built})" if built else ")"
|
||||
if self.hw:
|
||||
text += f", hardware {self.hw}"
|
||||
if local:
|
||||
text += f" - same as firmware/{local[2]}"
|
||||
return text
|
||||
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"
|
||||
+401
-401
@@ -1,401 +1,401 @@
|
||||
"""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 + "}"
|
||||
"""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 + "}"
|
||||
+740
-740
File diff suppressed because it is too large.
Load diff
+732
-732
File diff suppressed because it is too large.
Load diff
Reference in new issue
Block a user