"""Hand-off of a diagnosis to the KOTA service system (Google Apps Script). The tool opens the box's service page in the technician's browser: #kota-report= The service page (v1.1.0+) reads the fragment and pre-fills status and note; the technician checks it, adds the location and saves - so the report is authored by their own Google login and the tool holds no credentials. The fragment is never sent to any server (redirect host or Apps Script). """ from __future__ import annotations import base64 import json import re from .. import __version__ from ..diag.engine import Report, Severity DEFAULT_BASE = "https://service.kota-app.com/" NOTE_LIMIT = 4800 # the service form and Code.gs accept 5000; leave room for the technician # Status names exactly as in the service system (Code.gs STATUSES). OK = "Działa poprawnie" CHECK = "Do sprawdzenia" IN_SERVICE = "W trakcie serwisu" HARDWARE = "Błąd hardware" SOFTWARE = "Błąd software" POWER = "Problem z zasilaniem" CONNECTIVITY = "Problem z łącznością" # Root-finding rule id prefix -> service status (first match wins). _STATUS_BY_RULE = [ ("reboot.brownout", POWER), ("reboot.power", POWER), ("tech.power.", POWER), ("sys.crash", SOFTWARE), ("reboot.watchdog", SOFTWARE), ("sys.out_of_resources", SOFTWARE), ("influx.bad_request", SOFTWARE), ("influx.invalid_value", SOFTWARE), ("ota.selfcheck_failed", SOFTWARE), ("tech.quick.hung", SOFTWARE), ("rtc.init_failed", HARDWARE), ("rtc.lost_after_sync", HARDWARE), ("i2c.", HARDWARE), ("sensor.", HARDWARE), ("sd.", HARDWARE), ("sen66.", HARDWARE), ("bme688.", HARDWARE), ("as7341.", HARDWARE), ("mic.read_fail", HARDWARE), ("mic.silent", HARDWARE), ("tech.rtc_i2c.", HARDWARE), ("tech.sensors.", HARDWARE), ("tech.mic.mic_faulty", HARDWARE), ("tech.mic.mic_switch_faulty", HARDWARE), ("tech.white.sd_", HARDWARE), ("matter.", CONNECTIVITY), ("net.", CONNECTIVITY), ("wifi.", CONNECTIVITY), ("influx.", CONNECTIVITY), ("ota.", CONNECTIVITY), ("tech.network.", CONNECTIVITY), ("tech.influx.invalid_value", SOFTWARE), ("tech.influx.sensors_clock", HARDWARE), ("tech.influx.silent", CONNECTIVITY), ("tech.influx.never", CONNECTIVITY), ("tech.influx.", CHECK), # firmware 1.0.16+ (KOTA: lines, status, error IDs) ("reboot.crash", SOFTWARE), ("reboot.status_oom_watchdog", SOFTWARE), ("status.oom", SOFTWARE), ("kota.low_heap", SOFTWARE), ("kota.i2c_stuck", HARDWARE), ("kota.sensor_missing", HARDWARE), ("kota.rtc_", HARDWARE), ("kota.sd_", HARDWARE), ("status.error", HARDWARE), ("status.rtc", HARDWARE), ("status.sd", HARDWARE), ("kota.wifi_down", CONNECTIVITY), ("kota.weak_wifi", CONNECTIVITY), ("kota.no_upload", CONNECTIVITY), ("kota.ota_unregistered", CONNECTIVITY), ("kota.buffer_backlog", CONNECTIVITY), ("status.conn", CONNECTIVITY), ("kota.time_invalid", CHECK), *((f"eid.{m}", HARDWARE) for m in ("SD", "ENV", "RTC", "BME", "SEN", "AS", "MIC", "FLK", "SPI")), *((f"eid.{m}", CONNECTIVITY) for m in ("INF", "WIFI", "OTA", "PROV")), ("eid.", SOFTWARE), ("rtc.clock_not_running", CHECK), ("reboot.", CHECK), ] _MAC = re.compile(r"^[0-9A-F]{12}$") def normalize_mac(mac: str) -> str: return re.sub(r"[^0-9A-Fa-f]", "", mac or "").upper() def service_status(report: Report) -> str: """Suggested status; the technician can change it in the form.""" problems = [rc for rc in report.roots if rc.severity >= Severity.WARNING] if not problems: return OK if any(m == "data_uploaded" and t is not None for m, _, t in report.milestones) else CHECK rid = problems[0].finding.rule_id for prefix, status in _STATUS_BY_RULE: if rid.startswith(prefix): return status return CHECK BOX_MARK = {"ok": "✔", "warn": "!", "bad": "✖", "info": "i"} # same marks as the Diagnosis panel SD_WIFI = {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"} def service_note(report: Report, limit: int = NOTE_LIMIT) -> str: """The Diagnosis panel as text: verdict, LED, firmware, box self-report, start-up progress with times, findings with details and when they were seen. Falls back to shorter finding details when the full text does not fit the service form (consequences' explanations first, then all), then truncates. """ for detail in ("all", "roots", "none"): text = _note(report, detail) if len(text) <= limit: return text return text[:limit - 20].rstrip() + "\n...(shortened)" def _note(report: Report, detail: str) -> str: fw = report.firmware source = " (reported by the box)" if report.led_exact else "" lines = [f"[KOTA Technician Tool {__version__}] {report.verdict}", "", f"Status LED should be: {report.led_text()}{source}"] if report.led_reason: lines.append(f" from: {report.led_reason}") lines += ["", f"Firmware: {fw.describe()}", f" {SD_WIFI[fw.supports_sd_wifi]}"] if report.box: lines += ["", "Box self-report (KOTA lines):"] lines += [f" {BOX_MARK.get(st, '-')} {topic}: {text}" for topic, text, st in report.box] lines += ["", "Start-up progress (current boot):"] lines += [f" ✔ {label} ({t / 1000:.1f} s)" if t is not None else f" ○ {label}" for _, label, t in report.milestones] lines += ["", "Findings - root causes first:" if report.roots else "Findings: no problems detected in the log"] for rc in report.roots: lines += _finding(rc.finding, "", detail in ("all", "roots")) for c in rc.consequences: lines += _finding(c, " ↳ ", detail == "all") for title, items in (("Seen in earlier boots only", report.earlier), ("Resolved during this session", report.resolved)): if items: lines += ["", f"{title} ({len(items)}):"] lines += [f" - [{f.severity.label}] {f.title} - {f.seen_text()}" for f in items] if report.technician: lines += ["", "Technician checks:"] lines += [f" * {t}" for t in report.technician[-12:]] return "\n".join(lines) def _finding(f, prefix: str, detailed: bool) -> list[str]: ind = " " * len(prefix) + " " out = [f"{prefix}[{f.severity.label}] {f.title}"] if detailed: out.append(f"{ind}What it means: {f.meaning}") if f.actions: out.append(f"{ind}What to do:") out += [f"{ind} {i}. {a}" for i, a in enumerate(f.actions, 1)] out += [f"{ind}Error {eid}" + (f": {a}" if a else "") for eid, a in f.error_ids.items()] out.append(f"{ind}{f.seen_text()}") if f.evidence: out.append(f"{ind}Log: {f.evidence[-1]}") return out def report_url(mac: str, report: Report, base: str = DEFAULT_BASE) -> str: mac = normalize_mac(mac) if not _MAC.match(mac): raise ValueError(f"not a box MAC address: {mac!r}") payload = {"v": 1, "mac": mac, "status": service_status(report), "note": service_note(report), "tool": f"v{__version__}"} raw = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8") frag = base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=") base = base if base.endswith("/") else base + "/" return f"{base}{mac}#kota-report={frag}" def decode_fragment(url: str) -> dict: """Inverse of report_url (used by tests and for troubleshooting).""" frag = url.split("#kota-report=", 1)[1] return json.loads(base64.urlsafe_b64decode(frag + "=" * (-len(frag) % 4)).decode("utf-8"))