from __future__ import annotations import os import threading from pathlib import Path from PySide6.QtCore import QCoreApplication, QObject, QSettings, Qt, QTimer, QUrl, Signal from PySide6.QtGui import QColor, QDesktopServices, QFont, QGuiApplication, QTextCharFormat, QTextCursor from PySide6.QtWidgets import ( QCheckBox, QComboBox, QDialog, QDialogButtonBox, QFileDialog, QFormLayout, QHBoxLayout, QInputDialog, QLabel, QLineEdit, QMainWindow, QMessageBox, QPlainTextEdit, QPushButton, QSplitter, QTabWidget, QTextBrowser, QVBoxLayout, QWidget, ) from .. import __version__ from ..core.logfile import read_log_file from ..core.logline import LineKind, LogLine, host_event from ..core.session import SessionLog from ..diag.engine import DiagnosticsEngine from ..diag.guides import GuideRunner from ..serialio.commands import loggable from ..serialio.manager import SerialManager, State from ..serialio.ports import PortInfo, list_serial_ports from ..matter.payload import SetupPayload, payload_from_log from ..service import influx from ..service.report import DEFAULT_BASE, normalize_mac, report_url from .cheatsheet import cheatsheet_html from .diag_panel import DiagnosticsPanel from .flash_tab import FlashTab from .matter_qr import MatterQrDialog from .guide_panel import GuidePanel from .sd_tab import SdTab from .wifi_tab import WifiTab MAX_LINES = 20000 # "KOTA" = the firmware's structured diagnostic lines (1.0.16+) plus errors. LEVEL_FILTERS = {"All lines": None, "KOTA diagnostics + errors": "KOTA", "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.KOTA: "#5fd7e8", LineKind.RAW: "#a8a8a8", } class _Bridge(QObject): """Marshals SerialManager callbacks (reader thread) onto the UI thread.""" line = Signal(object) state = Signal(object, str) influx = Signal(object) # InfluxCheck from the query thread 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._bridge.influx.connect(self._on_influx) 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 self._runner = GuideRunner(self._engine, on_action=self._guide_action, on_log=self._guide_log) self._line_listeners: list = [] # other tabs following the live log (Wi-Fi over USB) # --- 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)") self.service_btn = QPushButton("Service report...") self.service_btn.setToolTip("Open this box's page in the KOTA service system with the diagnosis " "filled in; check it and press Save there") self._settings = QSettings("KOTA", "TechnicianTool") self.auto_service = QCheckBox("Open service report on disconnect") self.auto_service.setChecked(self._settings.value("service/auto", False, type=bool)) self.auto_service.toggled.connect(lambda on: self._settings.setValue("service/auto", on)) self._active_port: PortInfo | None = None 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, self.service_btn, self.auto_service): 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() self.guide = GuidePanel(self._runner, self._engine, matter_payload=self.matter_payload) self.side_tabs = QTabWidget() self.side_tabs.addTab(self.diag, "Diagnosis") self.side_tabs.addTab(self.guide, "Guided troubleshooting") split = QSplitter(Qt.Orientation.Horizontal) split.addWidget(self.view) split.addWidget(self.side_tabs) 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) self.service_btn.clicked.connect(lambda: self._open_service_report(auto=False)) self.diag.qr_btn.clicked.connect(lambda: MatterQrDialog(self.matter_payload(), self).exec()) # 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._connect(port) def _connect(self, port: PortInfo) -> None: self._counts = {"E": 0, "W": 0, "boots": 0} self._engine.reset() self._runner.cancel() self._diag_dirty = True self._session.start(port.mac or port.device) self._active_port = port self._manager.connect(port) def matter_payload(self) -> SetupPayload: """Pairing payload reported by this box (KOTA: matter), from its log, or the defaults.""" return payload_from_log([self._engine.matter_advert] if self._engine.matter_advert else [], self._engine.matter_qr, self._engine.matter_manual) # ---- service system ------------------------------------------------- def _service_base(self) -> str: return str(self._settings.value("service/base", DEFAULT_BASE)) def _open_service_report(self, auto: bool) -> None: mac = normalize_mac(self._engine.mac or (self._active_port.mac if self._active_port else "") or "") if len(mac) != 12: if not auto: QMessageBox.information(self, "Service report", "The box's MAC address is not known yet. " "Connect to the box and press Reset & capture first.") return url = report_url(mac, self._engine.report(), self._service_base()) QDesktopServices.openUrl(QUrl(url)) # Log that it happened, not the report itself (it is already in the session log). self._guide_log(f"service report for {mac} opened in the browser - check it and press Save there") # ---- hand-off with the flash tab ------------------------------------ @property def engine(self) -> DiagnosticsEngine: return self._engine # ---- commands to the box (firmware usb_console) ----------------------- @property def is_connected(self) -> bool: return self._manager.state == State.CONNECTED @property def active_port(self) -> PortInfo | None: return self._active_port if self._manager.is_running else None def add_line_listener(self, callback) -> None: self._line_listeners.append(callback) def send_command(self, command: str) -> bool: """Write one "kota ..." line to the box; False when not connected. The session log gets the command without the Wi-Fi password.""" if not self.is_connected: return False self._manager.write((command + "\n").encode("ascii")) if not command.startswith(("kota health", "kota ping")): # polled; would flood the log self._guide_log(f"sent to the box: {loggable(command)}") return True def connect_first_box(self) -> bool: """Connect to the first sensorbox port found (for tabs that need the box).""" if self._manager.is_running: return True self.refresh_ports() port = next((p for p in self._ports if p.is_sensorbox), None) if port is None: return False self.connect_port(port, capture=False) return True def release_port(self) -> None: """Stop monitoring so esptool can open the port exclusively.""" if self._manager.is_running: self._no_auto_service = True # a flash hand-off is not the end of a session self._manager.disconnect() # joins the reader thread QCoreApplication.processEvents() # deliver its queued final lines/state now self._no_auto_service = False def connect_port(self, port: PortInfo, capture: bool = True) -> None: """Monitor `port`; with capture, restart the box once connected to record its boot.""" self.release_port() idx = self.port_combo.findData(port.device) if idx < 0: self.port_combo.addItem(port.label, port.device) self._ports.append(port) idx = self.port_combo.count() - 1 self.port_combo.setCurrentIndex(idx) self._reset_on_connect = capture self._connect(port) def _on_state(self, state: State, detail: str) -> None: running = state != State.DISCONNECTED if not running and self._manager.is_running: return # stale event from a session that was already replaced if state == State.CONNECTED and getattr(self, "_reset_on_connect", False): self._reset_on_connect = False QTimer.singleShot(300, self._manager.reset_board) if not running and self._session.path is not None: self._session.write_report(self._engine.report().to_text()) if (self.auto_service.isChecked() and self._engine.lines and not getattr(self, "_no_auto_service", False)): self._open_service_report(auto=True) 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._runner.feed(line) for listener in self._line_listeners: listener(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() if self._visible(line): self._append(line) def _visible(self, line: LogLine) -> bool: allowed = LEVEL_FILTERS[self.filter_combo.currentText()] if allowed is None or line.kind in (LineKind.HOST, LineKind.PANIC, LineKind.BOOT, LineKind.RESET): return True if allowed == "KOTA": return line.kind == LineKind.KOTA or line.level == "E" return line.level in allowed def _append(self, line: LogLine) -> None: sb = self.view.verticalScrollBar() at_bottom = sb.value() >= sb.maximum() - 4 fmt = QTextCharFormat() colour = COLORS.get(line.kind) if line.kind == LineKind.KOTA and line.level == "I" else None fmt.setForeground(QColor(colour or COLORS.get(line.level) or COLORS.get(line.kind, "#d4d4d4"))) if line.kind in (LineKind.PANIC, LineKind.BOOT, LineKind.HOST) or line.record in ("boot", "status"): 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()) self.guide.refresh() # countdowns and timeouts need a steady tick # ---- guided troubleshooting hooks ------------------------------------ def _guide_action(self, action: str) -> bool: if action == "reset" and self._manager.state == State.CONNECTED: self._manager.reset_board() return True if action == "influx": cfg = self.influx_config() mac = self._engine.mac or (self._active_port.mac if self._active_port else "") or "" if cfg is None: self._runner.note = "InfluxDB access is not set up: Settings > InfluxDB access." return False threading.Thread(target=lambda: self._bridge.influx.emit(influx.check_device(cfg, mac)), name="influx-check", daemon=True).start() return True return False def influx_config(self) -> influx.InfluxConfig | None: """Read token: KOTA_INFLUX_TOKEN, else the tool settings. None when not set up.""" token = os.environ.get("KOTA_INFLUX_TOKEN") or str(self._settings.value("influx/token", "")) if not token: return None return influx.InfluxConfig( token=token, url=str(self._settings.value("influx/url", influx.DEFAULT_URL)), org=str(self._settings.value("influx/org", influx.DEFAULT_ORG)), bucket=str(self._settings.value("influx/bucket", influx.DEFAULT_BUCKET))) def _on_influx(self, check: influx.InfluxCheck) -> None: self._runner.influx_result(check.outcome, check.summary, check.details) self._diag_dirty = True self.guide.refresh() def _guide_log(self, text: str) -> None: # Technician answers go into the session log and the log view. line = host_event(text) self._session.write(line) self._append(line) self._diag_dirty = True 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._runner.cancel() 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() self.flash_tab = FlashTab(self.serial_tab) tabs.addTab(guide, "Quick start") tabs.addTab(self.serial_tab, "Serial monitor") tabs.addTab(self.flash_tab, "Flash firmware") self.wifi_tab = WifiTab(self.serial_tab) tabs.addTab(self.wifi_tab, "Wi-Fi (USB)") self.sd_tab = SdTab(self.serial_tab) tabs.addTab(self.sd_tab, "SD card") self.setCentralWidget(tabs) settings_menu = self.menuBar().addMenu("Settings") settings_menu.addAction("Service system address...", self._edit_service_base) settings_menu.addAction("InfluxDB access...", self._edit_influx) def _edit_service_base(self) -> None: st = self.serial_tab text, ok = QInputDialog.getText( self, "Service system address", "Address of the KOTA service system (the box MAC is added at the end):", text=st._service_base()) if ok and text.strip().startswith(("https://", "http://")): st._settings.setValue("service/base", text.strip()) elif ok: QMessageBox.warning(self, "Service system address", "The address must start with https://") def _edit_influx(self) -> None: st = self.serial_tab._settings dlg = QDialog(self) dlg.setWindowTitle("InfluxDB access") form = QFormLayout(dlg) form.addRow(QLabel("Used by the check 'Sensor data in the database'. Ask the firmware team for a " "read-only token for the bucket.", wordWrap=True)) url = QLineEdit(str(st.value("influx/url", influx.DEFAULT_URL))) org = QLineEdit(str(st.value("influx/org", influx.DEFAULT_ORG))) bucket = QLineEdit(str(st.value("influx/bucket", influx.DEFAULT_BUCKET))) token = QLineEdit(echoMode=QLineEdit.EchoMode.Password) token.setPlaceholderText("stored - leave empty to keep" if st.value("influx/token", "") else "paste the read token") clear = QCheckBox("Remove the stored token") for label, w in (("Address", url), ("Organisation", org), ("Bucket", bucket), ("Token", token), ("", clear)): form.addRow(label, w) buttons = QDialogButtonBox(QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel) buttons.accepted.connect(dlg.accept) buttons.rejected.connect(dlg.reject) form.addRow(buttons) if dlg.exec() != QDialog.DialogCode.Accepted: return if not url.text().strip().startswith("https://"): QMessageBox.warning(self, "InfluxDB access", "The address must start with https://") return st.setValue("influx/url", url.text().strip()) st.setValue("influx/org", org.text().strip() or influx.DEFAULT_ORG) st.setValue("influx/bucket", bucket.text().strip() or influx.DEFAULT_BUCKET) if clear.isChecked(): st.remove("influx/token") elif token.text().strip(): st.setValue("influx/token", token.text().strip()) def closeEvent(self, event): # noqa: N802 (Qt naming) self.serial_tab.shutdown() super().closeEvent(event)