Wi-Fi over USB tab for firmware with the USB console (v13)

Writes / reads / removes the box's Wi-Fi over the serial port with the
usb_console commands (kota ping, wifi set|get|clear, health) and follows
the box through the restart until it joins the network. The password is
sent hex-encoded and never logged.

firmware/ now holds merged-binary-v13.bin (build 3ae273ec) instead of the
previous 1.0.16 build; error_ids.csv synced with the firmware.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
kkallasandClaude Opus 5.5 committed 2026-09-29 17:30:35 +02:00
1 parent 0d21c5c205
commit ab420220e1
11 files changed
+758

No files matched your search

+9
View File
@@ -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,
+14
View File
@@ -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).
+22
View File
@@ -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", "?")
+1
View File
@@ -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
1 id module severity message action
191 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
192 E-GATH-03 GATH error Failed to create task for sensor Internal error - send the log to the firmware team
193 E-GATH-04 GATH error Failed to create main gather task Internal error - send the log to the firmware team
194 E-CON-01 CON error Failed to install USB-Serial-JTAG driver: %s Internal error - send the log to the firmware team
+1
View File
@@ -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 = {
+283
View File
@@ -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 <name> ok=1|0 [key=value ...]":
kota ping -> cmd ping ok=1 fw=<version> api=1
kota health -> cmd health ok=1 <health fields> | ok=0 err=not_ready
kota wifi get -> cmd wifi_get ok=1 ssid=<hex|-> password_set=0|1 mode=...
kota wifi set <ssid_hex> <pass_hex> -> 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)
+40
View File
@@ -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)
+207
View File
@@ -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 - <b>no need to open the "
"case</b>. The box then restarts and uses this network instead of Matter. Needs firmware with USB commands "
"(<i>Flash firmware</i> 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: <b>{port.label}</b><br><small>{fw}{api}</small>")
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 "
"<b>Connect to the box</b>.")
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(("<b>Done.</b> " if ok else "<b>Not done.</b> ") + 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()
+162
View File
@@ -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)
+19
View File
@@ -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()