diff --git a/README.md b/README.md index 64b29fd..d0e843d 100644 --- a/README.md +++ b/README.md @@ -55,6 +55,15 @@ Sesje, kopie zapasowe flash i ustawienia są w `~/KotaTechnicianTool` (Windows: Kabel USB musi przenosić dane (nie tylko ładowanie). Przy restartach z komunikatem o brownout użyj innego portu lub zasilacza. +## Wi-Fi przez USB (bez otwierania obudowy) + +Zakładka **Wi-Fi (USB)**: podłącz box kablem USB, kliknij **Connect to the box**, wpisz nazwę sieci i hasło, +kliknij **Write to the box and connect**. Box zapisuje je na karcie SD, restartuje się i łączy z siecią +(zamiast Matter); zakładka pokazuje każdy krok aż do „Connected ... (signal -58 dBm)”. +**Read current Wi-Fi settings** pokazuje zapisaną sieć (bez hasła), **Remove Wi-Fi** wraca do Matter. +Wymaga firmware z komendami USB (plik `firmware/merged-binary-v13.bin` lub nowszy); starszy firmware +wgraj w zakładce **Flash firmware** (tryb App only zachowuje parowanie i ustawienia). + ## Firmware Wgrać można tylko pliki `.bin` z folderu `firmware/`. Lista pokazuje datę kompilacji zapisaną w pliku, diff --git a/docs/CHANGES-v0.10.md b/docs/CHANGES-v0.10.md new file mode 100644 index 0000000..9b8ed13 --- /dev/null +++ b/docs/CHANGES-v0.10.md @@ -0,0 +1,14 @@ +# Technician tool v0.10 - changes (draft) + +- New tab **Wi-Fi (USB)**: write the Wi-Fi name and password to the box's SD card over the USB cable, without + opening the case (firmware usb_console, `kota wifi set`). Shows each step: command support check, stored, + restart in Wi-Fi mode, connected with signal strength (polls `kota health`). Also reads the stored network + (`kota wifi get`, password never shown) and removes it (`kota wifi clear`, back to Matter). +- The password is sent hex-encoded, never written to the session log or the service report. +- Diagnosis: restarts `usb_wifi_set` / `usb_wifi_clear` are intentional (no "unexpected restart"); box + self-report shows the stored Wi-Fi network; `cmd health` replies update the health line. +- firmware/: `merged-binary-v13.bin` (usb_console, build 3ae273ec) replaces `merged-binary.bin` (build 4f10989e, + still in git tag v0.9). Note: v13 reports the same version 1.0.16 and compile date as the previous build; + the firmware team should bump the version for the next build. +- error_ids.csv updated from the firmware (E-CON-01). +- Test that builds the main window (catches start-up errors). diff --git a/firmware/merged-binary.bin b/firmware/merged-binary-v13.bin similarity index 61% rename from firmware/merged-binary.bin rename to firmware/merged-binary-v13.bin index 106443d..bbb3ab6 100644 Binary files a/firmware/merged-binary.bin and b/firmware/merged-binary-v13.bin differ diff --git a/kota_tech/diag/engine.py b/kota_tech/diag/engine.py index 3805f34..d06c2af 100644 --- a/kota_tech/diag/engine.py +++ b/kota_tech/diag/engine.py @@ -335,6 +335,8 @@ class DiagnosticsEngine: self.matter_qr: str | None = None # "KOTA: matter ... qr=MT:..." self.matter_manual: str | None = None self.kota_seen = False # firmware prints KOTA: lines + self.console_api: int | None = None # "cmd ping ... api=N": box accepts USB commands + self.wifi_config: dict[str, str] | None = None # last "cmd wifi_get" reply (never a password) self._reset_led_state() def _reset_led_state(self) -> None: @@ -714,6 +716,20 @@ class DiagnosticsEngine: fd = self._custom(key, Severity[sev.upper()], title, meaning, actions, expl) self._touch(fd, line) + def _kota_cmd(self, line: LogLine, f: dict[str, str]) -> None: + """Replies to commands sent by this tool (firmware usb_console).""" + name = line.words[0] if line.words else "" + if f.get("ok") != "1": + return # the Wi-Fi tab reports failures itself + if name == "ping": + api = f.get("api", "") + self.console_api = int(api) if api.isdigit() else None + elif name == "health": + self.boot.kota["health"] = line # same fields as the periodic health line + self._kota_health(line, f) + elif name == "wifi_get": + self.wifi_config = {k: f.get(k, "?") for k in ("ssid", "password_set", "mode")} + def _kota_led(self, line: LogLine, f: dict[str, str]) -> None: if "previous_boot" in line.words: code = f.get("code", "0") @@ -845,6 +861,12 @@ class DiagnosticsEngine: if ln := k.get("mic"): on = ln.fields.get("enabled") == "1" out.append(("Microphone", "on" if on else "switched OFF (switch on the back)", "ok" if on else "info")) + if self.wifi_config: + from ..serialio.commands import unhex_utf8 + w = self.wifi_config + ssid = unhex_utf8(w["ssid"]) + out.append(("Wi-Fi config", (f"network '{ssid}', password {'set' if w['password_set'] == '1' else 'NOT set'}" + if ssid else "no network on the SD card") + f", mode {w['mode']}", "info")) if ln := k.get("health"): f = ln.fields wifi = f.get("wifi", "?") diff --git a/kota_tech/diag/error_ids.csv b/kota_tech/diag/error_ids.csv index b38ceaf..a7de0ef 100644 --- a/kota_tech/diag/error_ids.csv +++ b/kota_tech/diag/error_ids.csv @@ -191,3 +191,4 @@ E-GATH-01,GATH,error,Failed to take line protocol mutex,Internal error - send th E-GATH-02,GATH,error,Failed to get line protocol from sensor,Device ran out of memory; restart it and send the log to the firmware team E-GATH-03,GATH,error,Failed to create task for sensor,Internal error - send the log to the firmware team E-GATH-04,GATH,error,Failed to create main gather task,Internal error - send the log to the firmware team +E-CON-01,CON,error,Failed to install USB-Serial-JTAG driver: %s,Internal error - send the log to the firmware team diff --git a/kota_tech/diag/kota.py b/kota_tech/diag/kota.py index 95c730a..7eb84fd 100644 --- a/kota_tech/diag/kota.py +++ b/kota_tech/diag/kota.py @@ -39,6 +39,7 @@ I2C_DEVICES = {"0x39": "AS7341", "0x68": "DS1307", "0x6B": "SEN66", "0x76": "BME INTENTIONAL_RESTARTS = { "ota_update", "button_provisioning_abort", "button_wifi_cleared", "button_provisioning_toggle", "provisioning_saved", + "usb_wifi_set", "usb_wifi_clear", # Wi-Fi changed from the technician tool (usb_console) } MODULES = { diff --git a/kota_tech/serialio/commands.py b/kota_tech/serialio/commands.py new file mode 100644 index 0000000..56b4f8a --- /dev/null +++ b/kota_tech/serialio/commands.py @@ -0,0 +1,283 @@ +"""Commands to the box over the USB serial port (firmware with usb_console, api=1). + +Protocol (kota-sensorboard main/usb_console.hpp): ASCII lines "kota ...", +one reply line each, logged as "KOTA: cmd ok=1|0 [key=value ...]": + + kota ping -> cmd ping ok=1 fw= api=1 + kota health -> cmd health ok=1 | ok=0 err=not_ready + kota wifi get -> cmd wifi_get ok=1 ssid= password_set=0|1 mode=... + kota wifi set -> cmd wifi_set ok=1 restart=1, then restart (usb_wifi_set) + kota wifi clear -> cmd wifi_clear ok=1 restart=1, then restart (usb_wifi_clear) + +Firmware without the console never answers. The password is only ever sent +hex-encoded and is never written to the log by the box or by this tool. + +WifiFlow runs a whole "write / clear Wi-Fi" job as a small state machine +driven by `feed(line)` and `tick(now)`, so it is testable without Qt or a box. +""" + +from __future__ import annotations + +import enum +import time +from collections.abc import Callable +from dataclasses import dataclass, field + +from ..core.logline import LineKind, LogLine + +API_VERSION = 1 + +# Technician-facing text for the firmware's err= codes. +ERRORS = { + "bad_args": "The box did not understand the command (tool/firmware mismatch).", + "bad_hex": "The box could not decode the settings (tool/firmware mismatch).", + "bad_char": "The name or password contains a line break.", + "ssid_length": "The network name must be 1-32 bytes.", + "password_length": "The password must be 8-63 characters.", + "sd_not_mounted": "The box cannot read its SD card, so it cannot store the settings.", + "write_failed": "Writing ENV.TXT on the SD card failed.", + "verify_failed": "ENV.TXT did not read back correctly after writing.", + "unknown_command": "This firmware does not know the command.", + "not_ready": "The box is still starting; try again in a few seconds.", +} + + +def hex_utf8(text: str) -> str: + return text.encode("utf-8").hex() + + +def unhex_utf8(value: str) -> str | None: + if not value or value == "-": + return None + try: + return bytes.fromhex(value).decode("utf-8", errors="replace") + except ValueError: + return None + + +def validate_wifi(ssid: str, password: str) -> list[str]: + """Same limits as the firmware (usb_console.cpp); empty list = OK.""" + problems = [] + if not ssid: + problems.append("Enter the Wi-Fi network name (SSID).") + elif len(ssid.encode("utf-8")) > 32: + problems.append("The network name is longer than 32 bytes.") + if not 8 <= len(password.encode("utf-8")) <= 63: + problems.append("The password must be 8-63 characters (WPA2). Open networks are not supported.") + if any(c in ssid + password for c in "\r\n\0"): + problems.append("The name or password contains a line break.") + return problems + + +def wifi_set_command(ssid: str, password: str) -> str: + problems = validate_wifi(ssid, password) + if problems: + raise ValueError(" ".join(problems)) + return f"kota wifi set {hex_utf8(ssid)} {hex_utf8(password)}" + + +def loggable(command: str) -> str: + """The command as it may appear in the session log: never the password.""" + parts = command.split() + if parts[:3] == ["kota", "wifi", "set"] and len(parts) == 5: + return f"kota wifi set (network '{unhex_utf8(parts[3])}', password hidden)" + return command + + +@dataclass(frozen=True) +class Reply: + name: str # ping | health | wifi_get | wifi_set | wifi_clear | unknown + ok: bool + fields: dict[str, str] + + @property + def error(self) -> str: + code = self.fields.get("err", "") + return ERRORS.get(code, f"The box reported an error ({code or 'unknown'}).") + + +def parse_reply(line: LogLine) -> Reply | None: + """'I (5) KOTA: cmd wifi_set ok=1 restart=1' -> Reply('wifi_set', True, {...}).""" + if line.kind != LineKind.KOTA or line.record != "cmd" or not line.words: + return None + return Reply(line.words[0], line.fields.get("ok") == "1", dict(line.fields)) + + +@dataclass(frozen=True) +class WifiInfo: + """What `kota wifi get` reported.""" + ssid: str | None + password_set: bool + mode: str + + @classmethod + def from_reply(cls, r: Reply) -> WifiInfo: + return cls(unhex_utf8(r.fields.get("ssid", "-")), r.fields.get("password_set") == "1", + r.fields.get("mode", "?")) + + def describe(self) -> str: + mode = {"wifi": "local Wi-Fi from the SD card", "matter": "Matter (Wi-Fi given by the Matter app)", + "provisioning": "Wi-Fi setup portal"}.get(self.mode, self.mode) + net = f"network '{self.ssid}'" + (", password set" if self.password_set else ", NO password") + return f"Mode: {mode}. " + (net if self.ssid else "No network stored on the SD card.") + + +# ------------------------------------------------------------------ flows + +class Step(str, enum.Enum): + PING = "Checking the box accepts USB commands" + READ = "Reading the Wi-Fi settings" + SEND = "Storing the settings on the box's SD card" + RESTART = "Waiting for the box to restart" + CONNECT = "Waiting for the box to join the Wi-Fi network" + DONE = "Done" + FAILED = "Failed" + + +@dataclass +class FlowEvent: + step: Step + text: str + ok: bool | None = None # None = in progress, True = finished OK, False = failed + + +PING_TIMEOUT_S = 3.0 +REPLY_TIMEOUT_S = 8.0 +RESTART_TIMEOUT_S = 40.0 +CONNECT_TIMEOUT_S = 90.0 +HEALTH_POLL_S = 5.0 + + +@dataclass +class WifiFlow: + """One technician job: "read", "set" (ssid+password) or "clear". + + `send(text)` writes one command line to the box (returns False when the + port is not connected); `on_event` reports progress for the UI. + """ + action: str + send: Callable[[str], bool] + on_event: Callable[[FlowEvent], None] + ssid: str = "" + password: str = "" + now: Callable[[], float] = time.monotonic + step: Step = Step.PING + info: WifiInfo | None = None + _deadline: float = 0.0 + _next_poll: float = 0.0 + _events: list[FlowEvent] = field(default_factory=list) + + @property + def finished(self) -> bool: + return self.step in (Step.DONE, Step.FAILED) + + def start(self) -> None: + if self.action == "set": + problems = validate_wifi(self.ssid, self.password) + if problems: + return self._fail(" ".join(problems)) + self._go(Step.PING, "kota ping", PING_TIMEOUT_S) + + def cancel(self) -> None: + if not self.finished: + self._fail("Cancelled.") + + # -------------------------------------------------------- driving + def feed(self, line: LogLine) -> None: + if self.finished: + return + if self.step == Step.RESTART and line.kind == LineKind.KOTA and line.record == "boot": + return self._on_boot(line.fields) + r = parse_reply(line) + if r is None: + return + if self.step == Step.PING and r.name == "ping": + if not r.ok or int(r.fields.get("api", "0") or 0) < API_VERSION: + return self._fail("The box answered, but its command version is not supported.") + self._emit(f"Firmware {r.fields.get('fw', '?')} accepts USB commands.", True) + if self.action == "read": + return self._go(Step.READ, "kota wifi get", REPLY_TIMEOUT_S) + if self.action == "set": + return self._go(Step.SEND, wifi_set_command(self.ssid, self.password), REPLY_TIMEOUT_S) + return self._go(Step.SEND, "kota wifi clear", REPLY_TIMEOUT_S) + if self.step == Step.READ and r.name == "wifi_get": + if not r.ok: + return self._fail(r.error) + self.info = WifiInfo.from_reply(r) + return self._finish(self.info.describe()) + if self.step == Step.SEND and r.name in ("wifi_set", "wifi_clear"): + if not r.ok: + return self._fail(r.error) + self._emit("The box stored the settings and is restarting.", True) + self.step, self._deadline = Step.RESTART, self.now() + RESTART_TIMEOUT_S + return self._emit(Step.RESTART.value + "...") + if self.step == Step.CONNECT and r.name == "health" and r.ok: + return self._on_health(r.fields) + + def tick(self) -> None: + """Call periodically (e.g. every 0.5 s): timeouts and health polling.""" + if self.finished: + return + t = self.now() + if self.step == Step.CONNECT and t >= self._next_poll: + self._next_poll = t + HEALTH_POLL_S + self.send("kota health") + if t < self._deadline: + return + if self.step == Step.PING: + self._fail("The box does not answer USB commands. It needs firmware with the USB console " + "(Flash firmware tab), or it is still starting - try again in 10 s.") + elif self.step == Step.RESTART: + self._fail("The box stored the settings but no restart was seen. Press Reset & capture in the " + "Serial monitor and check the Diagnosis panel.") + elif self.step == Step.CONNECT: + self._fail(f"The box did not join '{self.ssid}' within {int(CONNECT_TIMEOUT_S)} s. Check the network " + "name and password, that it is a 2.4 GHz network and that it is in range. The settings stay " + "on the box; it keeps retrying.") + else: + self._fail("The box did not answer. Check the USB connection and try again.") + + # -------------------------------------------------------- internals + def _on_boot(self, f: dict[str, str]) -> None: + mode, last = f.get("mode", "?"), f.get("last_restart", "?") + want = "usb_wifi_set" if self.action == "set" else "usb_wifi_clear" + if last != want: + return self._fail(f"The box restarted for another reason ({last}); the settings may not be applied.") + self._emit(f"Restarted in {mode} mode.", True) + if self.action == "clear": + ok = mode == "matter" + return self._finish("Wi-Fi removed; the box uses Matter again (pair it in the Matter app if needed).") \ + if ok else self._fail(f"The box restarted in {mode} mode instead of Matter.") + if mode != "wifi": + return self._fail(f"The box restarted in {mode} mode instead of local Wi-Fi.") + self.step = Step.CONNECT + self._deadline = self.now() + CONNECT_TIMEOUT_S + self._next_poll = self.now() + HEALTH_POLL_S + self._emit(Step.CONNECT.value + f" '{self.ssid}'...") + + def _on_health(self, f: dict[str, str]) -> None: + if f.get("wifi") != "connected": + return + rssi = f.get("rssi", "?") + note = " - weak signal, consider moving the box or the router" if rssi.lstrip("-").isdigit() \ + and int(rssi) < -75 else "" + self._finish(f"Connected to '{self.ssid}' (signal {rssi} dBm{note}). Data upload starts within a minute.") + + def _go(self, step: Step, command: str, timeout: float) -> None: + self.step, self._deadline = step, self.now() + timeout + self._emit(step.value + "...") + if not self.send(command): + self._fail("The box is not connected. Connect it first.") + + def _emit(self, text: str, ok: bool | None = None) -> None: + ev = FlowEvent(self.step, text, ok) + self._events.append(ev) + self.on_event(ev) + + def _finish(self, text: str) -> None: + self.step = Step.DONE + self._emit(text, True) + + def _fail(self, text: str) -> None: + self.step = Step.FAILED + self._emit(text, False) diff --git a/kota_tech/ui/main_window.py b/kota_tech/ui/main_window.py index 2dd9359..7ae146c 100644 --- a/kota_tech/ui/main_window.py +++ b/kota_tech/ui/main_window.py @@ -28,6 +28,7 @@ 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 @@ -38,6 +39,7 @@ 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. @@ -76,6 +78,7 @@ class SerialTab(QWidget): 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) @@ -237,6 +240,39 @@ class SerialTab(QWidget): 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: @@ -293,6 +329,8 @@ class SerialTab(QWidget): 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 @@ -402,6 +440,8 @@ class MainWindow(QMainWindow): 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) diff --git a/kota_tech/ui/wifi_tab.py b/kota_tech/ui/wifi_tab.py new file mode 100644 index 0000000..259ebd9 --- /dev/null +++ b/kota_tech/ui/wifi_tab.py @@ -0,0 +1,207 @@ +"""Wi-Fi over USB: read / write / remove the box's Wi-Fi without opening the case. + +Needs firmware with the USB console (usb_console, api=1). The job itself is +serialio.commands.WifiFlow; this tab only collects the input and shows the +steps. Lines arrive from the Serial monitor's connection (one port owner). +""" + +from __future__ import annotations + +from PySide6.QtCore import Qt, QTimer +from PySide6.QtWidgets import ( + QCheckBox, + QFormLayout, + QGroupBox, + QHBoxLayout, + QLabel, + QLineEdit, + QListWidget, + QListWidgetItem, + QMessageBox, + QPushButton, + QVBoxLayout, + QWidget, +) +from PySide6.QtGui import QBrush, QColor + +from ..core.logline import LogLine +from ..serialio.commands import FlowEvent, WifiFlow, validate_wifi + +INTRO = ( + "Stores the Wi-Fi name and password on the box's SD card over the USB cable - no need to open the " + "case. The box then restarts and uses this network instead of Matter. Needs firmware with USB commands " + "(Flash firmware tab, newest file). Only 2.4 GHz networks with a password (8-63 characters)." +) +MARK = {True: ("✔", "#1b7a33"), False: ("✖", "#b03a2e"), None: ("…", "#2e6da4")} + + +class WifiTab(QWidget): + def __init__(self, serial_tab, parent: QWidget | None = None): + super().__init__(parent) + self.serial_tab = serial_tab + self._flow: WifiFlow | None = None + + intro = QLabel(INTRO, wordWrap=True) + + # 1. box + box = QGroupBox("1. Sensorbox") + self.box_state = QLabel(wordWrap=True) + self.connect_btn = QPushButton("Connect to the box") + self.connect_btn.clicked.connect(self._connect) + self.read_btn = QPushButton("Read current Wi-Fi settings") + self.read_btn.clicked.connect(lambda: self._run("read")) + bl = QVBoxLayout(box) + bl.addWidget(self.box_state) + brow = QHBoxLayout() + brow.addWidget(self.connect_btn) + brow.addWidget(self.read_btn) + brow.addStretch(1) + bl.addLayout(brow) + + # 2. network + net = QGroupBox("2. Wi-Fi network for this box") + self.ssid = QLineEdit(placeholderText="network name (SSID), exactly as on the router") + self.password = QLineEdit(placeholderText="8-63 characters") + self.password.setEchoMode(QLineEdit.EchoMode.Password) + show = QCheckBox("Show password") + show.toggled.connect(lambda on: self.password.setEchoMode( + QLineEdit.EchoMode.Normal if on else QLineEdit.EchoMode.Password)) + form = QFormLayout() + form.addRow("Network (SSID)", self.ssid) + prow = QHBoxLayout() + prow.addWidget(self.password, 1) + prow.addWidget(show) + form.addRow("Password", prow) + self.input_problem = QLabel(wordWrap=True) + self.input_problem.setStyleSheet("color:#b03a2e") + self.write_btn = QPushButton("Write to the box and connect") + self.write_btn.setMinimumHeight(34) + self.write_btn.clicked.connect(lambda: self._run("set")) + self.clear_btn = QPushButton("Remove Wi-Fi (back to Matter)") + self.clear_btn.clicked.connect(self._clear) + nl = QVBoxLayout(net) + nl.addLayout(form) + nl.addWidget(self.input_problem) + arow = QHBoxLayout() + arow.addWidget(self.write_btn) + arow.addWidget(self.clear_btn) + arow.addStretch(1) + nl.addLayout(arow) + + # 3. progress + prog = QGroupBox("3. Progress") + self.steps = QListWidget() + self.steps.setSelectionMode(QListWidget.SelectionMode.NoSelection) + self.result = QLabel(wordWrap=True) + self.result.setMargin(8) + self.result.setVisible(False) + self.cancel_btn = QPushButton("Cancel") + self.cancel_btn.clicked.connect(self._cancel) + pl = QVBoxLayout(prog) + pl.addWidget(self.steps) + pl.addWidget(self.result) + pl.addWidget(self.cancel_btn, 0, Qt.AlignmentFlag.AlignLeft) + + lay = QVBoxLayout(self) + lay.addWidget(intro) + lay.addWidget(box) + lay.addWidget(net) + lay.addWidget(prog, 1) + + serial_tab.add_line_listener(self._on_line) + for w in (self.ssid, self.password): + w.textChanged.connect(self._update) + self._timer = QTimer(self, interval=500) + self._timer.timeout.connect(self._tick) + self._timer.start() + self._update() + + # ------------------------------------------------------------ state + def showEvent(self, event): # noqa: N802 - refresh when the tab is opened + self._update() + super().showEvent(event) + + def _update(self) -> None: + connected = self.serial_tab.is_connected + busy = self._flow is not None and not self._flow.finished + port = self.serial_tab.active_port + eng = self.serial_tab.engine + if connected and port: + fw = eng.firmware.describe() if eng.firmware.elf_sha else "firmware not identified yet" + api = {None: "", 1: " - USB commands supported"}.get(eng.console_api, f" - USB commands api {eng.console_api}") + self.box_state.setText(f"Connected: {port.label}
{fw}{api}") + elif self.serial_tab.active_port: + self.box_state.setText("Waiting for the box... (it may be restarting)") + else: + self.box_state.setText("Not connected. Plug in the sensorbox (native USB) and press " + "Connect to the box.") + problems = validate_wifi(self.ssid.text(), self.password.text()) + typed = bool(self.ssid.text() or self.password.text()) + self.input_problem.setText(" ".join(problems) if typed else "") + self.connect_btn.setVisible(not self.serial_tab.active_port) + for w in (self.read_btn, self.clear_btn): + w.setEnabled(connected and not busy) + self.write_btn.setEnabled(connected and not busy and not problems) + self.ssid.setEnabled(not busy) + self.password.setEnabled(not busy) + self.cancel_btn.setVisible(busy) + + def _tick(self) -> None: + if self._flow is not None: + self._flow.tick() + self._update() + + # ------------------------------------------------------------ actions + def _connect(self) -> None: + if not self.serial_tab.connect_first_box(): + QMessageBox.information(self, "Wi-Fi over USB", "No sensorbox found. Check the USB cable (data-capable) " + "and that the box is powered, then try again.") + self._update() + + def _clear(self) -> None: + if QMessageBox.question( + self, "Remove Wi-Fi", + "Remove the Wi-Fi network from the box?\n\nThe box restarts and uses Matter again. If it was never " + "paired in Matter, it has no network until you pair it in the Matter app.", + QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No) == QMessageBox.StandardButton.Yes: + self._run("clear") + + def _run(self, action: str) -> None: + if action == "set" and QMessageBox.question( + self, "Write Wi-Fi", + f"Store network '{self.ssid.text()}' on the box?\n\nThe box restarts and connects to this Wi-Fi " + "instead of Matter (about 30 s).", + QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No) != QMessageBox.StandardButton.Yes: + return + self.steps.clear() + self.result.setVisible(False) + self._flow = WifiFlow(action, self.serial_tab.send_command, self._on_event, + ssid=self.ssid.text(), password=self.password.text()) + self._flow.start() + self._update() + + def _cancel(self) -> None: + if self._flow is not None: + self._flow.cancel() + self._update() + + # ------------------------------------------------------------ events + def _on_line(self, line: LogLine) -> None: + if self._flow is not None: + self._flow.feed(line) + + def _on_event(self, ev: FlowEvent) -> None: + mark, colour = MARK[ev.ok] + item = QListWidgetItem(f"{mark} {ev.text}") + item.setForeground(QBrush(QColor(colour))) + self.steps.addItem(item) + self.steps.scrollToBottom() + if self._flow is not None and self._flow.finished or ev.ok is False: + ok = ev.ok is True + border, bg = ("#2e9e44", "#e3f4e7") if ok else ("#d9534f", "#f9e0df") + self.result.setStyleSheet(f"QLabel {{ border:2px solid {border}; background:{bg}; border-radius:6px; }}") + self.result.setText(("Done. " if ok else "Not done. ") + ev.text) + self.result.setVisible(True) + if ok and self._flow and self._flow.info and self._flow.info.ssid and not self.ssid.text(): + self.ssid.setText(self._flow.info.ssid) # read: prefill the stored network name + self._update() diff --git a/tests/test_commands.py b/tests/test_commands.py new file mode 100644 index 0000000..4b97c96 --- /dev/null +++ b/tests/test_commands.py @@ -0,0 +1,162 @@ +"""Wi-Fi over USB (firmware usb_console api=1): encoding, replies and the whole +write/clear/read job, driven with reply lines in the firmware's exact format.""" + +import pytest + +from kota_tech.core.logline import parse_line +from kota_tech.diag.engine import DiagnosticsEngine +from kota_tech.serialio.commands import ( + CONNECT_TIMEOUT_S, PING_TIMEOUT_S, Step, WifiFlow, loggable, parse_reply, validate_wifi, wifi_set_command) + +PING = "I (5000) KOTA: cmd ping ok=1 fw=1.0.16 api=1" +BOOT = ("I (2843) KOTA: boot fw=1.0.16 build=3ae273ec hw=2 mac=A4CB8FC946F8 reset=RTC_SW_SYS_RST " + "mode={mode} commissioned=- last_restart={last}") +HEALTH = ("I (9000) KOTA: cmd health ok=1 up=9s status=OK wifi={wifi} rssi={rssi} time_valid=1 sd=ok buffer=0 " + "heap=90000 psram=7.6M last_upload=never ota=registered") + + +class Clock: + def __init__(self): + self.t = 0.0 + + def __call__(self): + return self.t + + +def make(action, **kw): + sent, events, clock = [], [], Clock() + flow = WifiFlow(action, lambda c: sent.append(c) or True, events.append, now=clock, **kw) + return flow, sent, events, clock + + +def feed(flow, text): + flow.feed(parse_line(text)) + + +# ------------------------------------------------------------- encoding + +def test_wifi_set_command_is_hex_utf8(): + assert wifi_set_command("Biuro 2.4", "tajne=haslo1") == \ + "kota wifi set 426975726f20322e34 74616a6e653d6861736c6f31" + assert wifi_set_command("Łąka", "zażółć1234").split()[3] == "c581c4856b61" + + +@pytest.mark.parametrize("ssid,password,problem", [ + ("", "12345678", "network name"), + ("x" * 33, "12345678", "32 bytes"), + ("net", "1234567", "8-63"), + ("net", "x" * 64, "8-63"), + ("net\n", "12345678", "line break"), +]) +def test_validation_matches_firmware_limits(ssid, password, problem): + assert any(problem in p for p in validate_wifi(ssid, password)) + with pytest.raises(ValueError): + wifi_set_command(ssid, password) + + +def test_password_never_logged(): + cmd = wifi_set_command("Biuro", "sekretne-haslo") + assert "sekretne" not in loggable(cmd) and "7365" not in loggable(cmd) + assert "Biuro" in loggable(cmd) + + +def test_reply_parsing(): + r = parse_reply(parse_line("I (1) KOTA: cmd wifi_set ok=0 err=password_length")) + assert r.name == "wifi_set" and not r.ok and "8-63" in r.error + assert parse_reply(parse_line("I (1) KOTA: health up=1s")) is None + + +# ----------------------------------------------------------------- flows + +def test_write_wifi_happy_path(): + flow, sent, events, clock = make("set", ssid="Biuro", password="haslo1234") + flow.start() + assert sent == ["kota ping"] + feed(flow, PING) + assert sent[-1] == wifi_set_command("Biuro", "haslo1234") and flow.step == Step.SEND + feed(flow, "I (5100) KOTA: cmd wifi_set ok=1 restart=1") + assert flow.step == Step.RESTART + feed(flow, "W (5101) KOTA: restart reason=usb_wifi_set") + feed(flow, BOOT.format(mode="wifi", last="usb_wifi_set")) + assert flow.step == Step.CONNECT + clock.t += 6 + flow.tick() + assert sent[-1] == "kota health" + feed(flow, HEALTH.format(wifi="disconnected", rssi=0)) + assert flow.step == Step.CONNECT + feed(flow, HEALTH.format(wifi="connected", rssi=-58)) + assert flow.step == Step.DONE and "Connected to 'Biuro'" in events[-1].text and "-58" in events[-1].text + + +def test_old_firmware_without_console_times_out_on_ping(): + flow, sent, events, clock = make("read") + flow.start() + clock.t += PING_TIMEOUT_S + 0.1 + flow.tick() + assert flow.step == Step.FAILED and "USB console" in events[-1].text + + +def test_box_rejects_settings(): + flow, sent, events, _ = make("set", ssid="Biuro", password="haslo1234") + flow.start() + feed(flow, PING) + feed(flow, "I (5100) KOTA: cmd wifi_set ok=0 err=sd_not_mounted") + assert flow.step == Step.FAILED and "SD card" in events[-1].text + + +def test_wrong_password_does_not_connect(): + flow, _, events, clock = make("set", ssid="Biuro", password="zlehaslo1") + flow.start() + feed(flow, PING) + feed(flow, "I (5100) KOTA: cmd wifi_set ok=1 restart=1") + feed(flow, BOOT.format(mode="wifi", last="usb_wifi_set")) + clock.t += CONNECT_TIMEOUT_S + 1 + flow.tick() + assert flow.step == Step.FAILED and "2.4 GHz" in events[-1].text + + +def test_clear_returns_to_matter(): + flow, sent, events, _ = make("clear") + flow.start() + feed(flow, PING) + assert sent[-1] == "kota wifi clear" + feed(flow, "I (5100) KOTA: cmd wifi_clear ok=1 restart=1") + feed(flow, BOOT.format(mode="matter", last="usb_wifi_clear")) + assert flow.step == Step.DONE and "Matter" in events[-1].text + + +def test_read_current_settings(): + flow, sent, events, _ = make("read") + flow.start() + feed(flow, PING) + assert sent[-1] == "kota wifi get" + feed(flow, "I (5100) KOTA: cmd wifi_get ok=1 ssid=426975726f password_set=1 mode=wifi") + assert flow.step == Step.DONE and flow.info.ssid == "Biuro" and flow.info.password_set + assert "local Wi-Fi" in events[-1].text + + +def test_not_connected(): + events = [] + flow = WifiFlow("read", lambda c: False, events.append) + flow.start() + assert flow.step == Step.FAILED and "not connected" in events[-1].text + + +# ---------------------------------------------------------------- engine + +def test_engine_treats_usb_wifi_restart_as_intentional_and_reads_replies(): + e = DiagnosticsEngine() + for t in ["ESP-ROM:esp32s3-20210327", "I (1790) KOTA: i2c SDA=1 SCL=1 0x39=AS7341", + BOOT.format(mode="matter", last="none"), PING, + "I (5100) KOTA: cmd wifi_set ok=1 restart=1", "W (5101) KOTA: restart reason=usb_wifi_set", + "ESP-ROM:esp32s3-20210327", "rst:0x3 (RTC_SW_SYS_RST),boot:0xb (SPI_FAST_FLASH_BOOT)", + BOOT.format(mode="wifi", last="usb_wifi_set"), + "I (6000) KOTA: cmd wifi_get ok=1 ssid=426975726f password_set=1 mode=wifi", + HEALTH.format(wifi="connected", rssi=-60)]: + e.feed(parse_line(t)) + r = e.report() + assert not any(f.rule_id.startswith("reboot.") for rc in r.roots for f in [rc.finding, *rc.consequences]) + assert e.console_api == 1 + box = dict((t, x) for t, x, _ in r.box) + assert "network 'Biuro'" in box["Wi-Fi config"] and "Wi-Fi connected (-60 dBm)" in box["Health"] + assert any(m == "wifi_connected" and t is not None for m, _, t in r.milestones) diff --git a/tests/test_ui.py b/tests/test_ui.py new file mode 100644 index 0000000..0779e9d --- /dev/null +++ b/tests/test_ui.py @@ -0,0 +1,19 @@ +"""The whole window builds (catches errors that only show when the app starts).""" + +import os + +os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") + + +def test_main_window_builds_with_all_tabs(): + from PySide6.QtWidgets import QApplication + + from kota_tech.ui.main_window import MainWindow + app = QApplication.instance() or QApplication([]) + w = MainWindow() + tabs = w.centralWidget() + assert [tabs.tabText(i) for i in range(tabs.count())] == \ + ["Quick start", "Serial monitor", "Flash firmware", "Wi-Fi (USB)", "SD card"] + assert not w.wifi_tab.write_btn.isEnabled() # nothing connected, nothing typed + w.serial_tab.shutdown() + app.processEvents()