kkallas/kota-technician-tool
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
feature/influx-check
kota-technician-tool/kota_tech/ui/main_window.py
T
KrzyszurandClaude Opus 5.5 bb933ff748 Serial monitor redesign: Test box button, InfluxDB row in the checklist 
- Test box: restart + start-up capture + online InfluxDB check (data newer
  than the restart), with a status strip; checklist and findings are the
  main view, live log / guided troubleshooting / box details are tabs,
  rarely used actions in a More menu
- light palette on all systems (macOS dark mode made the colours unreadable)
- new rule influx.invalid_value: -inf from a silent microphone makes
  InfluxDB reject every upload (HTTP 400); guide end names it
- exclusive serial port on macOS/Linux: two tool windows no longer split
  the log between them
- InfluxDB check: `since` filter, neutral wording when nothing new arrives

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-01 00:06:48 +02:00

690 lines
31 KiB
Python
Raw Permalink Blame History

from __future__ import annotations
import datetime as dt
import os
import sys
import threading
import time
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,
QMenu,
QMessageBox,
QPlainTextEdit,
QPushButton,
QSplitter,
QTabWidget,
QTextBrowser,
QToolButton,
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
TEST_INFLUX_AFTER_S = 30 # Test box: start the online check this long after the restart
INFLUX_ROW = "Sensor data in InfluxDB (online check)"
# test status strip: (background, text colour)
STRIP = {"idle": ("#eef1f4", "#333333"), "running": ("#dde9f7", "#123a63"), "ok": ("#d8f0dd", "#1b5e2a"),
"info": ("#dde9f7", "#123a63"), "warning": ("#fdebc8", "#6b4400"), "error": ("#f9dedc", "#8a2019"),
"critical": ("#f6d5d1", "#7b1e14")}
CLOUD_STATE = {"ok": "ok", "sensors_missing": "bad", "silent": "warn", "never": "warn", "unavailable": "off"}
# "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",
}
def _esc(s: str) -> str:
return s.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
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._settings = QSettings("KOTA", "TechnicianTool")
self.port_combo = QComboBox(minimumWidth=360)
self.refresh_btn = QPushButton("↻")
self.refresh_btn.setToolTip("Look for boxes again")
self.refresh_btn.setFixedWidth(40)
self.test_btn = QPushButton("▶ Test box")
self.test_btn.setToolTip("Restart the box, check its start-up and whether its sensor data reaches "
"InfluxDB (about 4 minutes)")
self.test_btn.setMinimumHeight(38)
self.test_btn.setStyleSheet(
"QPushButton { background:#1f7a3a; color:white; font-weight:bold; font-size:15px; "
"padding:4px 22px; border-radius:6px; border:none; }"
"QPushButton:hover { background:#25904a; } QPushButton:disabled { background:#9dbfa6; }")
self.connect_btn = QPushButton("Connect")
self.connect_btn.setMinimumHeight(32)
# Less frequent actions live in the "More" menu.
self.more_btn = QToolButton(text="More ▾")
self.more_btn.setPopupMode(QToolButton.ToolButtonPopupMode.InstantPopup)
self.more_btn.setMinimumHeight(32)
menu = QMenu(self.more_btn)
self.reset_btn = menu.addAction("Reset && capture (restart the box, no online check)")
menu.addSeparator()
self.service_btn = menu.addAction("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.copy_btn = menu.addAction("Copy report")
self.save_btn = menu.addAction("Save log view as...")
self.folder_btn = menu.addAction("Open logs folder")
self.analyze_btn = menu.addAction("Analyze log file...")
menu.addSeparator()
self.auto_service = menu.addAction("Open service report on disconnect")
self.auto_service.setCheckable(True)
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.all_ports = menu.addAction("Show all serial ports")
self.all_ports.setCheckable(True)
self.more_btn.setMenu(menu)
self._active_port: PortInfo | None = None
top = QHBoxLayout()
box_lbl = QLabel("<b>Box:</b>")
for w in (box_lbl, self.port_combo, self.refresh_btn):
top.addWidget(w)
top.addSpacing(12)
top.addWidget(self.test_btn)
top.addWidget(self.connect_btn)
top.addStretch(1)
top.addWidget(self.more_btn)
# Test status strip (what the test is doing / its result).
self.test_status = QLabel(wordWrap=True)
self.test_status.setMargin(8)
tf = self.test_status.font()
tf.setPointSize(tf.pointSize() + 1)
self.test_status.setFont(tf)
# --- live log (a tab on the right)
self.filter_combo = QComboBox()
self.filter_combo.addItems(LEVEL_FILTERS)
self.clear_btn = QPushButton("Clear view")
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(12 if sys.platform == "darwin" else 10)
self.view.setFont(mono)
self.view.setStyleSheet("QPlainTextEdit { background:#1e1e1e; color:#d4d4d4; }")
log_page = QWidget()
ll = QVBoxLayout(log_page)
ll.setContentsMargins(0, 4, 0, 0)
lrow = QHBoxLayout()
for w in (QLabel("Show:"), self.filter_combo, self.clear_btn):
lrow.addWidget(w)
lrow.addStretch(1)
ll.addLayout(lrow)
ll.addWidget(self.view, 1)
self.diag = DiagnosticsPanel()
self.guide = GuidePanel(self._runner, self._engine, matter_payload=self.matter_payload)
self.side_tabs = QTabWidget()
self.side_tabs.addTab(log_page, "Live log")
self.side_tabs.addTab(self.guide, "Guided troubleshooting")
self.side_tabs.addTab(self.diag.details, "Box details")
split = QSplitter(Qt.Orientation.Horizontal)
split.addWidget(self.diag)
split.addWidget(self.side_tabs)
split.setStretchFactor(0, 1)
split.setStretchFactor(1, 1)
split.setSizes([560, 620])
self.status = QLabel()
self.status.setStyleSheet("color:#555")
layout = QVBoxLayout(self)
layout.addLayout(top)
layout.addWidget(self.test_status)
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.test_btn.clicked.connect(self._test_box)
self.connect_btn.clicked.connect(self._toggle_connect)
self.reset_btn.triggered.connect(self._manager.reset_board)
self.clear_btn.clicked.connect(self.view.clear)
self.save_btn.triggered.connect(self._save_view)
self.folder_btn.triggered.connect(self._open_folder)
self.analyze_btn.triggered.connect(self._analyze_file)
self.copy_btn.triggered.connect(self._copy_report)
self.service_btn.triggered.connect(lambda: self._open_service_report(auto=False))
self.diag.qr_btn.clicked.connect(lambda: MatterQrDialog(self.matter_payload(), self).exec())
# Test box state: None = no test; else monotonic start time.
self._test_t0: float | None = None
self._test_since: dt.datetime | None = None # UTC; InfluxDB data must be newer
self._influx_started = False
self._cloud: influx.InfluxCheck | None = None # last InfluxDB answer
self._cloud_checking = False
# 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._cloud, self._cloud_checking = None, False
self._manager.connect(port)
# ---- Test box ----------------------------------------------------------
def _test_box(self) -> None:
"""Restart the box, record its start-up, then check InfluxDB online."""
if self._manager.is_running:
self._counts = {"E": 0, "W": 0, "boots": 0}
self._engine.reset()
self._runner.cancel()
self._cloud, self._cloud_checking = None, False
self._manager.reset_board()
else:
port = self._selected_port() or next((p for p in self._ports if p.is_sensorbox), None)
if port is None:
QMessageBox.information(self, "Test box", "No sensorbox found. Check the USB cable (data cable) "
"and that the box's power switch is ON, then press ↻.")
return
self.connect_port(port, capture=True)
self._test_t0 = time.monotonic()
self._test_since = dt.datetime.now(dt.timezone.utc)
self._influx_started = False
self._test_end = None
self._diag_dirty = True
self._guide_log("Test box started: restarting the box and recording its start-up")
def _tick_test(self) -> None:
if self._test_t0 is None:
return
self._diag_dirty = True # countdowns in the strip and the checklist
if not self._manager.is_running:
self._finish_test(None, "Box disconnected - test stopped.")
return
elapsed = time.monotonic() - self._test_t0
if not self._influx_started and elapsed >= TEST_INFLUX_AFTER_S and self.is_connected:
self._influx_started = True
self._runner.start("influx") # polls, routes and records findings like a guided check
self.guide.refresh()
elif self._influx_started and (self._runner.end is not None or self._runner.guide is None):
self._finish_test(self._runner.end)
def _finish_test(self, end, text: str = "") -> None:
self._test_t0 = None
self._test_since = None
self._test_end = end
self._test_text = text
self._guide_log("Test box finished: " + (end.title if end else text or "stopped"))
def _cloud_rows(self) -> list[tuple[str, str]]:
in_check = self._runner.step is not None and self._runner.step.type == "influx"
if self.influx_config() is None:
return [("off", f"{INFLUX_ROW}: not set up - Settings > InfluxDB access")]
if in_check:
left = int(self._runner.view().remaining_s)
last = f" - last answer: {self._cloud.summary}" if self._cloud else ""
rows = [("running", f"{INFLUX_ROW}: checking... {left // 60}:{left % 60:02d} left{last}")]
elif self._cloud is not None:
rows = [(CLOUD_STATE.get(self._cloud.outcome, "warn"), f"{INFLUX_ROW}: {self._cloud.summary}")]
else:
when = (f"starts {TEST_INFLUX_AFTER_S} s after the restart" if self._test_t0 is not None
else "press Test box")
return [("pending", f"{INFLUX_ROW} - {when}")]
if self._cloud is not None:
rows += [("note", f" {d}") for d in self._cloud.details]
return rows
def _update_test_strip(self) -> None:
if self._test_t0 is not None:
el = int(time.monotonic() - self._test_t0)
if self._influx_started:
doing = "checking that the data reaches InfluxDB - " + (
self._runner.view().note or "waiting for the first answer")
else:
ms = self._engine.report().milestones
done = sum(t is not None for _, _, t in ms)
doing = f"recording the start-up ({done} of {len(ms)} checklist steps done)"
key, text = "running", f"<b>Test running {el // 60}:{el % 60:02d}</b> - {_esc(doing)}"
elif getattr(self, "_test_end", None) is not None:
e = self._test_end
key = e.severity
text = (f"<b>Test finished: {_esc(e.title)}</b><br>{_esc(e.text)} "
"Details: Findings below and the <i>Guided troubleshooting</i> tab.")
elif getattr(self, "_test_text", ""):
key, text = "warning", f"<b>{_esc(self._test_text)}</b>"
else:
key = "idle"
text = ("Press <b>▶ Test box</b>: the tool restarts the box, checks its start-up (sensors, SD card, "
"Wi-Fi, clock) and then whether its sensor data reaches InfluxDB. Takes about 4 minutes.")
bg, fg = STRIP[key]
self.test_status.setStyleSheet(f"QLabel {{ background:{bg}; color:{fg}; border-radius:6px; }}")
self.test_status.setText(text)
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.test_btn.setEnabled(state != State.WAITING)
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:
self._tick_test()
if self._diag_dirty:
self._diag_dirty = False
self.diag.show_report(self._engine.report(), self._cloud_rows())
self._update_test_strip()
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
since = self._test_since
self._cloud_checking = True
threading.Thread(target=lambda: self._bridge.influx.emit(influx.check_device(cfg, mac, since=since)),
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._cloud, self._cloud_checking = check, False
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 "
"<b>read-only</b> 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)
Reference in new issue View Git Blame Copy Permalink