diff --git a/environment.yml b/environment.yml index 4961af5..d7de9d9 100644 --- a/environment.yml +++ b/environment.yml @@ -8,4 +8,5 @@ dependencies: - pyserial>=3.5 - pyside6>=6.7 - esptool>=4.8,<5 + - segno>=1.6 - pytest>=8 diff --git a/firmware/firmware.toml b/firmware/firmware.toml deleted file mode 100644 index fe67c1a..0000000 --- a/firmware/firmware.toml +++ /dev/null @@ -1,11 +0,0 @@ -# Official firmware images the technician tool may flash. -# Boxes normally receive new firmware over OTA; this is the factory image a -# box updates itself from once it is online. Add a new image only when the -# firmware team releases one: copy the .bin here and add an entry with its -# SHA-256 (the tool refuses files that do not match). - -[[image]] -file = "new_system.bin" -sha256 = "1d58a8f4bb60f66009bd6df32e6e860c545c0c668de955af60d049ee3b9a212d" -name = "v1.0 factory image (built Jun 29 2026)" -notes = "Official factory image. Once online, the box registers for OTA and updates itself." diff --git a/firmware/new_system.bin b/firmware/merged-binary.bin similarity index 51% rename from firmware/new_system.bin rename to firmware/merged-binary.bin index 730ce4b..106443d 100644 Binary files a/firmware/new_system.bin and b/firmware/merged-binary.bin differ diff --git a/kota_tech/__init__.py b/kota_tech/__init__.py index 765cd5d..fd2bb79 100644 --- a/kota_tech/__init__.py +++ b/kota_tech/__init__.py @@ -1,3 +1,3 @@ """KOTA sensorbox technician tool.""" -__version__ = "0.5.0" +__version__ = "0.6.0" diff --git a/kota_tech/diag/engine.py b/kota_tech/diag/engine.py index de61b70..bfe1d16 100644 --- a/kota_tech/diag/engine.py +++ b/kota_tech/diag/engine.py @@ -280,6 +280,7 @@ class DiagnosticsEngine: self._panic_tail = 0 # RAW lines after a panic line to keep self.technician: list[str] = [] self.mac: str | None = None # from "ID: Device MAC address: ..." at boot + self.matter_advert: str | None = None # "Advertise commission parameter vendorID=..." line @property def boot(self) -> BootRecord: @@ -321,6 +322,8 @@ class DiagnosticsEngine: self.firmware.observe(text) if m := _MAC_LINE.search(text): self.mac = m.group(1).upper() + if "Advertise commission parameter" in text: + self.matter_advert = text matched = False for rule in self.rules: m = rule.pattern.search(text) diff --git a/kota_tech/diag/firmware.py b/kota_tech/diag/firmware.py index 590b0de..3d1440f 100644 --- a/kota_tech/diag/firmware.py +++ b/kota_tech/diag/firmware.py @@ -26,6 +26,12 @@ KNOWN: dict[str, KnownFirmware] = { sd_wifi=False, setup_portal=False, ), + "c94ba0e9b": KnownFirmware( + name="v1.0 source build (Sep 26 2026)", + source="developer build, built by Krzysztof Kallas", + sd_wifi=True, + setup_portal=True, + ), } _ELF = re.compile(r"app_init: ELF file SHA256:\s+(?P[0-9a-f]{6,})") diff --git a/kota_tech/diag/guides.py b/kota_tech/diag/guides.py index cb4d5e0..9400ec5 100644 --- a/kota_tech/diag/guides.py +++ b/kota_tech/diag/guides.py @@ -57,6 +57,7 @@ class Step: skip_if_finding: str | None = None skip_unless_feature: str | None = None skip_to: str | None = None + show_qr: bool = False # show the Matter pairing QR next to the step text def targets(self) -> list[str]: t = [o.next for o in self.options] + list(self.route.values()) + list(self.routes.values()) @@ -101,7 +102,7 @@ def load_guides(path: Path = GUIDES_PATH) -> dict[str, Guide]: timeout_s=float(s.get("timeout_s", 60)), pass_on_timeout=bool(s.get("pass_on_timeout", False)), on_ok=s.get("on_ok"), on_fail=s.get("on_fail"), on_timeout=s.get("on_timeout"), skip_if_finding=s.get("skip_if_finding"), skip_unless_feature=s.get("skip_unless_feature"), - skip_to=s.get("skip_to"), + skip_to=s.get("skip_to"), show_qr=bool(s.get("show_qr", False)), ) ends = {e["id"]: End(e["id"], e["severity"], e["title"], e["text"], tuple(e.get("actions", ()))) for e in g.get("end", [])} diff --git a/kota_tech/diag/guides.toml b/kota_tech/diag/guides.toml index 6330403..632f4f4 100644 --- a/kota_tech/diag/guides.toml +++ b/kota_tech/diag/guides.toml @@ -113,7 +113,8 @@ next = "pair_start" [[guide.step]] id = "pair_start" type = "wait" -text = "In the Matter app choose 'Add device' and scan the QR code on the sticker (or type the 11-digit code printed under it). Keep the phone within 1 m of the box." +text = "In the Matter app choose 'Add device' and scan the QR code shown here (the same as the sticker), or type the 11-digit manual code. Keep the phone within 1 m of the box." +show_qr = true until = 'MAIN: Commissioning session started' timeout_s = 300 on_ok = "pair_wifi" diff --git a/kota_tech/flash/image.py b/kota_tech/flash/image.py index 0e3633e..4e1d67b 100644 --- a/kota_tech/flash/image.py +++ b/kota_tech/flash/image.py @@ -86,6 +86,7 @@ class FirmwareImage: app_offset: int # where the app starts inside the file app_length: int # exact app image length (from its structure) app: AppInfo + built_by: str = "" # who compiled a developer build (shown in the UI) @property def data(self) -> bytes: @@ -109,7 +110,10 @@ class FirmwareImage: return b"\xff" * self.partition("otadata").size def label(self) -> str: - tag = "factory" if self.origin == "factory" else "developer build" + if self.origin == "factory": + tag = "factory" + else: + tag = "developer build" + (f" - built by {self.built_by}" if self.built_by else "") return f"{self.app.version} built {self.app.built} [{self.app.elf_sha[:9]}] - {tag}" @@ -181,7 +185,7 @@ def _app_info(data: bytes, offset: int) -> AppInfo: def inspect_image(path: Path, name: str = "", notes: str = "", origin: str = "factory", - reference: FirmwareImage | None = None) -> FirmwareImage: + reference: FirmwareImage | None = None, built_by: str = "") -> FirmwareImage: """Validate a merged or app-only image. App-only images need a `reference` (the factory image) that provides the flash layout.""" data = path.read_bytes() @@ -208,13 +212,14 @@ def inspect_image(path: Path, name: str = "", notes: str = "", origin: str = "fa raise ImageError(f"firmware ({length} bytes) does not fit the ota_0 partition ({ota0.size} bytes)") app = _app_info(data, app_off) return FirmwareImage(path, name or path.name, notes, origin, kind, hashlib.sha256(data).hexdigest(), - flash_size, parts, table, app_off, length, app) + flash_size, parts, table, app_off, length, app, built_by) # ------------------------------------------------------------- catalogue -def factory_images(manifest: Path = MANIFEST) -> list[FirmwareImage]: - """Bundled images whose file exists and matches the SHA-256 in the manifest.""" +def bundled_images(manifest: Path = MANIFEST) -> list[FirmwareImage]: + """Images shipped with the tool whose file exists and matches its SHA-256. + Factory images first; bundled developer builds keep their origin/built_by.""" if not manifest.exists(): return [] out = [] @@ -222,11 +227,19 @@ def factory_images(manifest: Path = MANIFEST) -> list[FirmwareImage]: path = manifest.parent / e["file"] if not path.exists(): continue - img = inspect_image(path, e.get("name", ""), e.get("notes", ""), "factory") + origin = e.get("origin", "factory") + if origin not in ("factory", "developer"): + raise ImageError(f"{path.name}: unknown origin {origin!r} in the manifest") + img = inspect_image(path, e.get("name", ""), e.get("notes", ""), origin, built_by=e.get("built_by", "")) if img.sha256 != e["sha256"].lower(): raise ImageError(f"{path.name}: SHA-256 does not match the manifest - file changed or corrupt") out.append(img) - return out + return sorted(out, key=lambda i: i.origin != "factory") + + +def factory_images(manifest: Path = MANIFEST) -> list[FirmwareImage]: + """Only the official factory image(s).""" + return [i for i in bundled_images(manifest) if i.origin == "factory"] class FirmwareLibrary: @@ -249,7 +262,8 @@ class FirmwareLibrary: for e in self._index(): path = self.root / e["file"] try: - img = inspect_image(path, e.get("name", ""), e.get("notes", ""), "developer", reference) + img = inspect_image(path, e.get("name", ""), e.get("notes", ""), "developer", reference, + e.get("built_by", "")) if img.sha256 != e["sha256"]: raise ImageError("file changed since it was imported") imgs.append(img) @@ -257,21 +271,24 @@ class FirmwareLibrary: problems.append(f"{e.get('original', e['file'])}: {exc}") return imgs, problems - def import_file(self, src: Path, reference: FirmwareImage | None, notes: str = "") -> FirmwareImage: + def import_file(self, src: Path, reference: FirmwareImage | None, notes: str = "", + built_by: str = "") -> FirmwareImage: img = inspect_image(src, origin="developer", reference=reference) # validate before copying entries = self._index() existing = next((e for e in entries if e["sha256"] == img.sha256), None) if existing: return inspect_image(self.root / existing["file"], existing["name"], existing["notes"], - "developer", reference) + "developer", reference, existing.get("built_by", "")) self.root.mkdir(parents=True, exist_ok=True) fname = f"{img.app.elf_sha[:9]}_{img.sha256[:8]}.bin" shutil.copyfile(src, self.root / fname) - name = f"Developer build {img.app.version} ({img.app.built})" + built_by = built_by.strip() + name = f"Developer build {img.app.version} ({img.app.built})" + (f", built by {built_by}" if built_by else "") entries.append({"file": fname, "sha256": img.sha256, "name": name, "original": src.name, - "imported": dt.datetime.now().isoformat(timespec="seconds"), "notes": notes}) + "imported": dt.datetime.now().isoformat(timespec="seconds"), "notes": notes, + "built_by": built_by}) self._save(entries) - return inspect_image(self.root / fname, name, notes, "developer", reference) + return inspect_image(self.root / fname, name, notes, "developer", reference, built_by) def remove(self, sha256: str) -> None: entries = self._index() diff --git a/kota_tech/matter/__init__.py b/kota_tech/matter/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/kota_tech/matter/payload.py b/kota_tech/matter/payload.py new file mode 100644 index 0000000..ce11d91 --- /dev/null +++ b/kota_tech/matter/payload.py @@ -0,0 +1,96 @@ +"""Matter onboarding payloads: the "MT:..." QR string and the 11-digit manual code. + +Implements the Matter spec (5.1.3 QR payload, 5.1.4 manual pairing code): +bit fields packed LSB-first, Base38 in 3-byte chunks, Verhoeff check digit. + +The sensorbox firmware uses the Matter SDK example commissionable data +(CONFIG_EXAMPLE_COMMISSIONABLE_DATA_PROVIDER): every v1.0 box advertises +VID 0xFFF1 / PID 0x8000 / discriminator 3840 with passcode 20202021, which is +exactly the sticker (QR "MT:Y.K9042C00KA0648G00", manual 3497-011-2332). +VID/PID/discriminator are taken from the box's own log when available; the +passcode is never logged, so the known default is used. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass + +BASE38 = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ-." +RENDEZVOUS_BLE = 1 << 1 +DEFAULT_PASSCODE = 20202021 # Matter SDK test passcode used by the v1.0 firmware +DEFAULT = dict(vendor_id=0xFFF1, product_id=0x8000, discriminator=3840) + +# chip[DIS]: Advertise commission parameter vendorID=65521 productID=32768 discriminator=3840/15 cm=1 cp=0 +_ADVERTISE = re.compile(r"vendorID=(\d+) productID=(\d+) discriminator=(\d+)") + + +@dataclass(frozen=True) +class SetupPayload: + vendor_id: int = DEFAULT["vendor_id"] + product_id: int = DEFAULT["product_id"] + discriminator: int = DEFAULT["discriminator"] + passcode: int = DEFAULT_PASSCODE + rendezvous: int = RENDEZVOUS_BLE + flow: int = 0 # standard commissioning flow + version: int = 0 + from_log: bool = False # VID/PID/discriminator read from this box's log + + def qr_text(self) -> str: + fields = [(self.version, 3), (self.vendor_id, 16), (self.product_id, 16), (self.flow, 2), + (self.rendezvous, 8), (self.discriminator, 12), (self.passcode, 27), (0, 4)] + value, shift = 0, 0 + for v, bits in fields: + if not 0 <= v < (1 << bits): + raise ValueError(f"field value {v} does not fit in {bits} bits") + value |= v << shift + shift += bits + return "MT:" + _base38(value.to_bytes(shift // 8, "little")) + + def manual_code(self) -> str: + short = self.discriminator >> 8 # upper 4 bits + chunk1 = short >> 2 # vid_pid_present = 0 + chunk2 = ((short & 0x3) << 14) | (self.passcode & 0x3FFF) + chunk3 = self.passcode >> 14 + digits = f"{chunk1}{chunk2:05d}{chunk3:04d}" + return digits + str(verhoeff_check(digits)) + + def manual_code_pretty(self) -> str: + c = self.manual_code() + return f"{c[:4]}-{c[4:7]}-{c[7:]}" + + +def _base38(data: bytes) -> str: + out = [] + for i in range(0, len(data), 3): + chunk = data[i:i + 3] + value = int.from_bytes(chunk, "little") + for _ in range({3: 5, 2: 4, 1: 2}[len(chunk)]): + value, rem = divmod(value, 38) + out.append(BASE38[rem]) + return "".join(out) + + +_D = [[0, 1, 2, 3, 4, 5, 6, 7, 8, 9], [1, 2, 3, 4, 0, 6, 7, 8, 9, 5], [2, 3, 4, 0, 1, 7, 8, 9, 5, 6], + [3, 4, 0, 1, 2, 8, 9, 5, 6, 7], [4, 0, 1, 2, 3, 9, 5, 6, 7, 8], [5, 9, 8, 7, 6, 0, 4, 3, 2, 1], + [6, 5, 9, 8, 7, 1, 0, 4, 3, 2], [7, 6, 5, 9, 8, 2, 1, 0, 4, 3], [8, 7, 6, 5, 9, 3, 2, 1, 0, 4], + [9, 8, 7, 6, 5, 4, 3, 2, 1, 0]] +_P = [[0, 1, 2, 3, 4, 5, 6, 7, 8, 9], [1, 5, 7, 6, 2, 8, 3, 0, 9, 4], [5, 8, 0, 3, 7, 9, 6, 1, 4, 2], + [8, 9, 1, 6, 0, 4, 3, 5, 2, 7], [9, 4, 5, 3, 1, 2, 6, 8, 7, 0], [4, 2, 8, 6, 5, 7, 3, 9, 0, 1], + [2, 7, 9, 3, 8, 0, 6, 4, 1, 5], [7, 0, 4, 6, 9, 1, 3, 2, 5, 8]] +_INV = [0, 4, 3, 2, 1, 5, 6, 7, 8, 9] + + +def verhoeff_check(digits: str) -> int: + c = 0 + for i, d in enumerate(reversed(digits)): + c = _D[c][_P[(i + 1) % 8][int(d)]] + return _INV[c] + + +def payload_from_log(lines: list[str]) -> SetupPayload: + """Use the box's advertised VID/PID/discriminator if seen; defaults otherwise.""" + for text in reversed(lines): + if m := _ADVERTISE.search(text): + return SetupPayload(int(m.group(1)), int(m.group(2)), int(m.group(3)), from_log=True) + return SetupPayload() diff --git a/kota_tech/sd/disks.py b/kota_tech/sd/disks.py index a3ff356..7450d7e 100644 --- a/kota_tech/sd/disks.py +++ b/kota_tech/sd/disks.py @@ -111,10 +111,10 @@ def parse_macos(list_plist: bytes, infos: dict[str, bytes]) -> list[Disk]: def list_cards() -> tuple[list[Disk], list[Disk]]: """(cards that may be formatted, readers/disks shown but not offered).""" if sys.platform == "win32": - out = subprocess.run(["powershell", "-NoProfile", "-NonInteractive", "-Command", WIN_LIST_PS], - capture_output=True, text=True, timeout=90, - creationflags=subprocess.CREATE_NO_WINDOW).stdout - disks = parse_windows(out) + # Driver-level listing (milliseconds, 1.5 s worst case) instead of + # Get-Disk/WMI, which stalled 5-21 s on an empty card-reader slot. + from .win_disks import list_disks + disks = list_disks(CARD_BUSES) elif sys.platform == "darwin": lst = subprocess.run(["diskutil", "list", "-plist", "physical"], capture_output=True, timeout=30).stdout infos = {} diff --git a/kota_tech/sd/fat32.py b/kota_tech/sd/fat32.py new file mode 100644 index 0000000..4b64c3f --- /dev/null +++ b/kota_tech/sd/fat32.py @@ -0,0 +1,135 @@ +"""FAT32 + MBR layout for an SD card, as raw byte writes (no OS formatter). + +Follows the Microsoft FAT specification (fatgen103): BPB at sector 0, +FSInfo at 1, backup boot sector at 6, two FATs, root directory in cluster 2 +holding the volume label. Cluster sizes follow Windows' defaults. The +partition starts at 1 MiB (SD erase-block friendly), MBR type 0x0C +(FAT32 LBA), which is what the firmware's FatFs expects. + +Pure functions: the result is a list of (offset, bytes) writes plus zero +ranges relative to the start of the DISK, so it can be tested on a buffer. +""" + +from __future__ import annotations + +import math +import struct +from dataclasses import dataclass + +PART_START = 1024 * 1024 +FAT32_MAX_BYTES = 32 * 1024**3 # Windows convention, and all the box needs +MIN_CLUSTERS, MAX_CLUSTERS = 65525, 0x0FFFFFF5 +RESERVED = 32 +NUM_FATS = 2 +MEDIA = 0xF8 + + +@dataclass(frozen=True) +class Fat32Layout: + bytes_per_sector: int + part_start: int # bytes from disk start + part_sectors: int + sectors_per_cluster: int + fat_sectors: int + clusters: int + + @property + def data_start(self) -> int: # sectors from partition start + return RESERVED + NUM_FATS * self.fat_sectors + + +def _spc(sectors: int, bps: int) -> int: + size = sectors * bps + for limit, cluster in ((260 * 1024**2, 512), (8 * 1024**3, 4096), (16 * 1024**3, 8192), + (32 * 1024**3, 16384)): + if size <= limit: + return max(1, cluster // bps) + return max(1, 32768 // bps) + + +def plan(disk_bytes: int, bps: int = 512) -> Fat32Layout: + part_bytes = min(disk_bytes - PART_START, FAT32_MAX_BYTES) + part_bytes -= part_bytes % (1024 * 1024) # whole MiB + sectors = part_bytes // bps + if sectors <= 0: + raise ValueError("card is too small") + spc = _spc(sectors, bps) + fat = 1 + for _ in range(20): # converge FAT size <-> cluster count + clusters = (sectors - RESERVED - NUM_FATS * fat) // spc + need = math.ceil((clusters + 2) * 4 / bps) + if need == fat: + break + fat = need + clusters = (sectors - RESERVED - NUM_FATS * fat) // spc + if not MIN_CLUSTERS <= clusters < MAX_CLUSTERS: + raise ValueError(f"card size gives {clusters} clusters - not valid for FAT32") + return Fat32Layout(bps, PART_START, sectors, spc, fat, clusters) + + +def _boot_sector(lay: Fat32Layout, label: bytes, vol_id: int) -> bytes: + bps = lay.bytes_per_sector + b = bytearray(bps) + b[0:3] = b"\xEB\x58\x90" + b[3:11] = b"MSWIN4.1" + struct.pack_into(" bytes: + b = bytearray(lay.bytes_per_sector) + struct.pack_into(" bytes: + b = bytearray(512) + struct.pack_into(" bytes: + label = text.upper().encode("ascii", "replace")[:11] + return label.ljust(11, b" ") + + +def build(lay: Fat32Layout, label: str, vol_id: int, disk_signature: int, + disk_bytes: int) -> tuple[list[tuple[int, int]], list[tuple[int, bytes]]]: + """(zero ranges, writes) in disk byte offsets; apply zero ranges first.""" + bps, p = lay.bytes_per_sector, lay.part_start + lab = volume_label(label) + fat_bytes = lay.fat_sectors * bps + fat1 = p + RESERVED * bps + fat2 = fat1 + fat_bytes + root = p + lay.data_start * bps + cluster = lay.sectors_per_cluster * bps + zeros = [ + (0, PART_START), # old MBR / GPT header and gap + (disk_bytes - PART_START, PART_START), # backup GPT at the end of the disk + (p, RESERVED * bps), # reserved region + (fat1, NUM_FATS * fat_bytes), # both FATs + (root, cluster), # root directory cluster + ] + boot, info = _boot_sector(lay, lab, vol_id), _fsinfo(lay) + first_fat = struct.pack(" str: - buses = ",".join(f"'{b}'" for b in sorted(CARD_BUSES)) - return f""" -$ErrorActionPreference = 'Stop' -$result = @{{ ok = $false; message = ''; drive = '' }} -try {{ - $n = {int(disk.id)} - $d = Get-Disk -Number $n - if ($d.IsSystem -or $d.IsBoot) {{ throw 'Refusing: this is a system/boot disk.' }} - if (@({buses}) -notcontains [string]$d.BusType) {{ throw "Refusing: disk bus is $($d.BusType), not an SD/USB card." }} - if ([int64]$d.Size -ne {int(disk.size)}) {{ throw 'Refusing: the disk changed since it was selected (card swapped?).' }} - if ($d.PartitionStyle -ne 'RAW') {{ Clear-Disk -Number $n -RemoveData -RemoveOEM -Confirm:$false }} - Initialize-Disk -Number $n -PartitionStyle MBR -ErrorAction SilentlyContinue - $size = [Math]::Min([int64]$d.Size - 2MB, [int64]{FAT32_MAX}) - $p = New-Partition -DiskNumber $n -Size $size -MbrType FAT32 -AssignDriveLetter - Format-Volume -Partition $p -FileSystem FAT32 -NewFileSystemLabel '{VOLUME_LABEL}' -Confirm:$false -Force | Out-Null - Start-Sleep -Seconds 2 - $letter = (Get-Partition -DiskNumber $n -PartitionNumber $p.PartitionNumber).DriveLetter - if (-not $letter) {{ throw 'Formatted, but Windows gave the card no drive letter.' }} - $result.ok = $true; $result.drive = "$($letter):\\"; $result.message = 'Formatted FAT32 (MBR).' -}} catch {{ $result.message = $_.Exception.Message }} -$result | ConvertTo-Json | Out-File -FilePath '{result_file}' -Encoding utf8 -""" - - WORK_ROOT = Path.home() / "KotaTechnicianTool" / "tmp" @@ -86,42 +61,54 @@ def _workdir() -> Path: return work -def _query_windows_card(number: int) -> tuple[str, str, str]: - """(drive, filesystem, label) of the card's first partition - no admin rights needed.""" - ps = (f"$p = Get-Partition -DiskNumber {number} -ErrorAction SilentlyContinue | Select-Object -First 1; " - "if ($p) { $v = $p | Get-Volume -ErrorAction SilentlyContinue; " - "\"$($p.DriveLetter)|$($v.FileSystem)|$($v.FileSystemLabel)\" }") - out = subprocess.run(["powershell", "-NoProfile", "-Command", ps], capture_output=True, text=True, - timeout=90, creationflags=subprocess.CREATE_NO_WINDOW).stdout.strip() - drive, fs, label = (out.split("|") + ["", "", ""])[:3] - return (f"{drive}:\\" if drive.strip() else ""), fs, label +def win_format_command(disk: Disk, result: Path) -> tuple[str, str, str]: + """(program, parameters, working dir) for the elevated helper.""" + root = Path(__file__).resolve().parents[2] # the version folder holding kota_tech/ + params = f'-m kota_tech.sd.win_format --disk {int(disk.id)} --size {int(disk.size)} ' \ + f'--label {VOLUME_LABEL} --result "{result}"' + return sys.executable, params, str(root) + + +def _run_elevated(program: str, params: str, cwd: str, timeout_s: int = 300) -> int | None: + """ShellExecuteEx 'runas' + wait. None if the technician refused the UAC prompt.""" + import ctypes + from ctypes import wintypes + + class SHELLEXECUTEINFOW(ctypes.Structure): + _fields_ = [("cbSize", wintypes.DWORD), ("fMask", ctypes.c_ulong), ("hwnd", wintypes.HWND), + ("lpVerb", wintypes.LPCWSTR), ("lpFile", wintypes.LPCWSTR), ("lpParameters", wintypes.LPCWSTR), + ("lpDirectory", wintypes.LPCWSTR), ("nShow", ctypes.c_int), ("hInstApp", wintypes.HINSTANCE), + ("lpIDList", ctypes.c_void_p), ("lpClass", wintypes.LPCWSTR), ("hkeyClass", wintypes.HKEY), + ("dwHotKey", wintypes.DWORD), ("hIconOrMonitor", wintypes.HANDLE), ("hProcess", wintypes.HANDLE)] + + SEE_MASK_NOCLOSEPROCESS, SW_HIDE = 0x40, 0 + info = SHELLEXECUTEINFOW(cbSize=ctypes.sizeof(SHELLEXECUTEINFOW), fMask=SEE_MASK_NOCLOSEPROCESS, + lpVerb="runas", lpFile=program, lpParameters=params, lpDirectory=cwd, nShow=SW_HIDE) + shell32 = ctypes.WinDLL("shell32", use_last_error=True) + k32 = ctypes.WinDLL("kernel32", use_last_error=True) + if not shell32.ShellExecuteExW(ctypes.byref(info)): + return None # ERROR_CANCELLED (1223) or failure + k32.WaitForSingleObject(info.hProcess, timeout_s * 1000) + code = wintypes.DWORD() + k32.GetExitCodeProcess(info.hProcess, ctypes.byref(code)) + k32.CloseHandle(info.hProcess) + return code.value def _format_windows(disk: Disk, log: list[str]) -> Outcome: - work = _workdir() - script, result = work / "format_card.ps1", work / "result.json" - script.write_text(windows_format_script(disk, result), encoding="utf-8-sig") + result = _workdir() / "result.json" log.append("Asking Windows for administrator rights to format the card (UAC prompt)...") - launcher = ("$p = Start-Process powershell -Verb RunAs -Wait -PassThru -WindowStyle Hidden " - f"-ArgumentList '-NoProfile -ExecutionPolicy Bypass -File \"{script}\"'; exit $p.ExitCode") - proc = subprocess.run(["powershell", "-NoProfile", "-Command", launcher], capture_output=True, text=True, - timeout=600, creationflags=subprocess.CREATE_NO_WINDOW) - if not result.exists(): - err = (proc.stderr or "").strip().splitlines() - cancelled = any("canceled" in e.lower() or "cancelled" in e.lower() for e in err) - return Outcome(False, "Administrator permission was refused - the card was not formatted." if cancelled - else "Formatting did not run: " + (err[-1] if err else "unknown error"), log=log) + t = time.perf_counter() + code = _run_elevated(*win_format_command(disk, result)) + if code is None: + return Outcome(False, "Administrator permission was refused - the card was not formatted.", log=log) try: - res = json.loads(result.read_text(encoding="utf-8-sig")) + res = json.loads(result.read_text(encoding="utf-8")) except (OSError, ValueError) as exc: - # The script ran but its report is unreadable: judge by the card itself. - log.append(f"Could not read the format result ({exc.__class__.__name__}); checking the card directly.") - drive, fs, label = _query_windows_card(int(disk.id)) - ok = bool(drive) and fs.upper() == "FAT32" and label == VOLUME_LABEL - res = {"ok": ok, "drive": drive, - "message": "Formatted FAT32 (MBR)." if ok else f"Card state after formatting: {fs or 'no file system'} " - f"'{label}' - formatting probably failed."} - log.append(res.get("message", "")) + return Outcome(False, f"The formatter did not report back ({exc.__class__.__name__}, exit {code}). " + "Check the card with 'Check card'.", log=log) + steps = ", ".join(f"{k} {v:.1f} s" for k, v in res.get("timings", {}).items()) + log.append(f"{res.get('message', '')} ({time.perf_counter() - t:.1f} s incl. permission prompt; {steps})") return Outcome(bool(res.get("ok")), res.get("message", ""), res.get("drive", ""), log) diff --git a/kota_tech/sd/win_disks.py b/kota_tech/sd/win_disks.py new file mode 100644 index 0000000..db27bf7 --- /dev/null +++ b/kota_tech/sd/win_disks.py @@ -0,0 +1,161 @@ +"""Fast Windows card discovery straight from the storage drivers (ctypes). + +Why not Get-Disk / WMI: every fresh PowerShell/WMI query enumerates ALL +disks, which wakes sleeping hard drives - measured 1.6 to 21 s per listing +on a PC with two SATA HDDs, freezing the UI each time. Here each physical +drive is opened with access 0 (no admin needed, no I/O to the medium) and +only its cached device descriptor is read; size and volumes are queried only +for USB/SD/MMC disks. A full listing takes milliseconds. +""" + +from __future__ import annotations + +import concurrent.futures +import ctypes +import string +from ctypes import wintypes + +from .disks import Disk, Volume + +IOCTL_STORAGE_QUERY_PROPERTY = 0x002D1400 +IOCTL_STORAGE_GET_DEVICE_NUMBER = 0x002D1080 +IOCTL_DISK_GET_DRIVE_GEOMETRY_EX = 0x000700A0 +BUS_TYPES = {1: "SCSI", 2: "ATAPI", 3: "ATA", 4: "1394", 7: "USB", 8: "RAID", 10: "SAS", 11: "SATA", + 12: "SD", 13: "MMC", 14: "Virtual", 15: "FileBackedVirtual", 17: "NVMe", 18: "SCM", 19: "UFS"} +SHARE_RW, OPEN_EXISTING = 0x3, 3 +DRIVE_REMOVABLE, DRIVE_FIXED = 2, 3 +MAX_DRIVES = 32 + +_k32 = ctypes.WinDLL("kernel32", use_last_error=True) +_k32.CreateFileW.restype = wintypes.HANDLE +_k32.CreateFileW.argtypes = [wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD, wintypes.LPVOID, + wintypes.DWORD, wintypes.DWORD, wintypes.HANDLE] +_k32.DeviceIoControl.argtypes = [wintypes.HANDLE, wintypes.DWORD, wintypes.LPVOID, wintypes.DWORD, + wintypes.LPVOID, wintypes.DWORD, ctypes.POINTER(wintypes.DWORD), wintypes.LPVOID] +_k32.GetDriveTypeW.argtypes = [wintypes.LPCWSTR] +_k32.GetVolumeInformationW.argtypes = [wintypes.LPCWSTR, wintypes.LPWSTR, wintypes.DWORD, + ctypes.POINTER(wintypes.DWORD), ctypes.POINTER(wintypes.DWORD), + ctypes.POINTER(wintypes.DWORD), wintypes.LPWSTR, wintypes.DWORD] +_INVALID = wintypes.HANDLE(-1).value + + +class _Query(ctypes.Structure): # STORAGE_PROPERTY_QUERY (StorageDeviceProperty, standard) + _fields_ = [("PropertyId", wintypes.DWORD), ("QueryType", wintypes.DWORD), ("Extra", ctypes.c_ubyte * 4)] + + +class _DevNumber(ctypes.Structure): # STORAGE_DEVICE_NUMBER + _fields_ = [("DeviceType", wintypes.DWORD), ("DeviceNumber", wintypes.DWORD), + ("PartitionNumber", wintypes.DWORD)] + + +def _open(path: str): + h = _k32.CreateFileW(path, 0, SHARE_RW, None, OPEN_EXISTING, 0, None) + return None if h in (None, _INVALID) else h + + +def _ioctl(h, code: int, inbuf=None, insize: int = 0, outsize: int = 1024) -> bytes | None: + out = ctypes.create_string_buffer(outsize) + got = wintypes.DWORD() + ok = _k32.DeviceIoControl(h, code, inbuf, insize, out, outsize, ctypes.byref(got), None) + return out.raw[:got.value] if ok else None + + +def _descriptor(h) -> tuple[str, str] | None: + """(bus, "vendor product") from STORAGE_DEVICE_DESCRIPTOR - served from the driver's cache.""" + q = _Query(0, 0) + raw = _ioctl(h, IOCTL_STORAGE_QUERY_PROPERTY, ctypes.byref(q), ctypes.sizeof(q)) + if not raw or len(raw) < 32: + return None + vendor_off, product_off = int.from_bytes(raw[12:16], "little"), int.from_bytes(raw[16:20], "little") + bus = int.from_bytes(raw[28:32], "little") + + def text(off: int) -> str: + if not off or off >= len(raw): + return "" + return raw[off:].split(b"\0", 1)[0].decode("ascii", "replace").strip() + + return BUS_TYPES.get(bus, f"bus{bus}"), " ".join(t for t in (text(vendor_off), text(product_off)) if t) or "Disk" + + +def _size(h) -> int: + raw = _ioctl(h, IOCTL_DISK_GET_DRIVE_GEOMETRY_EX, outsize=256) # fails when no card is inserted + return int.from_bytes(raw[24:32], "little") if raw and len(raw) >= 32 else 0 + + +def _volumes_by_disk() -> dict[int, list[Volume]]: + found: dict[int, list[Volume]] = {} + mask = _k32.GetLogicalDrives() + for i, letter in enumerate(string.ascii_uppercase): + if not mask & (1 << i) or _k32.GetDriveTypeW(f"{letter}:\\") not in (DRIVE_REMOVABLE, DRIVE_FIXED): + continue + h = _open(f"\\\\.\\{letter}:") + if h is None: + continue + try: + raw = _ioctl(h, IOCTL_STORAGE_GET_DEVICE_NUMBER, outsize=ctypes.sizeof(_DevNumber)) + finally: + _k32.CloseHandle(h) + if raw and len(raw) >= 8: + num = int.from_bytes(raw[4:8], "little") + found.setdefault(num, []).append(Volume(f"{letter}:\\", "", "")) + return found + + +def _fill_volume_info(v: Volume) -> None: + name, fs = ctypes.create_unicode_buffer(64), ctypes.create_unicode_buffer(32) + serial, maxlen, flags = wintypes.DWORD(), wintypes.DWORD(), wintypes.DWORD() + if _k32.GetVolumeInformationW(v.mount, name, 64, ctypes.byref(serial), ctypes.byref(maxlen), + ctypes.byref(flags), fs, 32): + v.filesystem, v.label = fs.value, name.value + + +MEDIA_TIMEOUT_S = 1.5 +# Slot queries still running from an earlier listing (an empty card-reader +# slot can block ~13 s on its first media query). Never start a second one. +_pending: dict[int, concurrent.futures.Future] = {} +_pool = concurrent.futures.ThreadPoolExecutor(max_workers=8, thread_name_prefix="sd-media") + + +def _media_size(n: int) -> int: + h = _open(f"\\\\.\\PhysicalDrive{n}") + if h is None: + return 0 + try: + return _size(h) + finally: + _k32.CloseHandle(h) + + +def list_disks(buses: set[str]) -> list[Disk]: + """Disks on the given buses (e.g. {"USB", "SD", "MMC"}), with size and volumes. + + Media queries run in parallel with a short timeout: a slot that does not + answer in time is reported as "no card" instead of freezing the caller. + """ + candidates: list[tuple[int, tuple[str, str]]] = [] + for n in range(MAX_DRIVES): + h = _open(f"\\\\.\\PhysicalDrive{n}") + if h is None: + continue + try: + desc = _descriptor(h) + finally: + _k32.CloseHandle(h) + if desc is not None and desc[0] in buses: # internal disks: only the cached descriptor is read + candidates.append((n, desc)) + for n, _ in candidates: + if n not in _pending or _pending[n].done(): + _pending[n] = _pool.submit(_media_size, n) + concurrent.futures.wait([_pending[n] for n, _ in candidates], timeout=MEDIA_TIMEOUT_S) + volumes: dict[int, list[Volume]] | None = None + disks = [] + for n, (bus, name) in candidates: + fut = _pending[n] + size = fut.result() if fut.done() and fut.exception() is None else 0 + if size and volumes is None: + volumes = _volumes_by_disk() + vols = (volumes or {}).get(n, []) if size else [] + for v in vols: + _fill_volume_info(v) + disks.append(Disk(str(n), name, size, bus, vols, has_media=size > 0)) + return disks diff --git a/kota_tech/sd/win_format.py b/kota_tech/sd/win_format.py new file mode 100644 index 0000000..7aac4c6 --- /dev/null +++ b/kota_tech/sd/win_format.py @@ -0,0 +1,192 @@ +"""Elevated Windows SD-card formatter (runs as: python -m kota_tech.sd.win_format ...). + +Replaces Clear-Disk / New-Partition / Format-Volume: every Storage cmdlet +goes through the storage provider, which stalls ~13 s on an empty card +reader slot, several times per format. This helper touches only the chosen +disk through its own driver: + + 1. guard: bus is USB/SD/MMC, size is exactly the one the technician + confirmed, and the disk does not hold the Windows drive; + 2. lock + dismount the card's volumes; + 3. raw writes: zero old MBR/GPT, FAT32 structures (kota_tech.sd.fat32), + then the new MBR last; + 4. unlock, let Windows re-read the disk, wait for the FAT32 'KOTA' volume; + 5. read-only `chkdsk` of the new volume. + +The result (with per-step timings) goes to a JSON file for the non-elevated +tool, which then writes ENV.TXT through the normal file API. +""" + +from __future__ import annotations + +import argparse +import ctypes +import json +import os +import random +import subprocess +import sys +import time +from ctypes import wintypes + +from . import fat32, win_disks + +GENERIC_RW, SHARE_RW, OPEN_EXISTING = 0xC0000000, 0x3, 3 +FSCTL_LOCK_VOLUME, FSCTL_UNLOCK_VOLUME, FSCTL_DISMOUNT_VOLUME = 0x00090018, 0x0009001C, 0x00090020 +IOCTL_DISK_UPDATE_PROPERTIES = 0x00070140 +CARD_BUSES = {"USB", "SD", "MMC"} +CHUNK = 1024 * 1024 + +k32 = win_disks._k32 +k32.WriteFile.argtypes = [wintypes.HANDLE, wintypes.LPCVOID, wintypes.DWORD, + ctypes.POINTER(wintypes.DWORD), wintypes.LPVOID] +k32.SetFilePointerEx.argtypes = [wintypes.HANDLE, ctypes.c_longlong, ctypes.POINTER(ctypes.c_longlong), wintypes.DWORD] +k32.VirtualAlloc.restype = wintypes.LPVOID +k32.VirtualAlloc.argtypes = [wintypes.LPVOID, ctypes.c_size_t, wintypes.DWORD, wintypes.DWORD] + + +class FormatError(RuntimeError): + pass + + +def _open_rw(path: str): + h = k32.CreateFileW(path, GENERIC_RW, SHARE_RW, None, OPEN_EXISTING, 0, None) + if h in (None, win_disks._INVALID): + raise FormatError(f"cannot open {path} (error {ctypes.get_last_error()})") + return h + + +def _ioctl(h, code: int) -> bool: + got = wintypes.DWORD() + return bool(k32.DeviceIoControl(h, code, None, 0, None, 0, ctypes.byref(got), None)) + + +class RawDisk: + """Sector-aligned writes through a page-aligned buffer (device alignment rules).""" + + def __init__(self, number: int, bps: int): + self.h = _open_rw(rf"\\.\PhysicalDrive{number}") + self.bps = bps + self.buf = k32.VirtualAlloc(None, CHUNK, 0x3000, 0x04) # MEM_COMMIT|RESERVE, RW + + def write(self, offset: int, data: bytes) -> None: + if offset % self.bps or len(data) % self.bps: + raise FormatError("unaligned write") + pos = 0 + while pos < len(data): + n = min(CHUNK, len(data) - pos) + ctypes.memmove(self.buf, data[pos:pos + n], n) + self._raw(offset + pos, n) + pos += n + + def zero(self, offset: int, length: int) -> None: + ctypes.memset(self.buf, 0, CHUNK) + pos = 0 + while pos < length: + n = min(CHUNK, length - pos) + self._raw(offset + pos, n) + pos += n + + def _raw(self, offset: int, n: int) -> None: + if not k32.SetFilePointerEx(self.h, offset, None, 0): + raise FormatError(f"seek failed (error {ctypes.get_last_error()})") + done = wintypes.DWORD() + if not k32.WriteFile(self.h, self.buf, n, ctypes.byref(done), None) or done.value != n: + raise FormatError(f"write at {offset} failed (error {ctypes.get_last_error()}) - write-protected card?") + + def close(self) -> None: + k32.FlushFileBuffers(self.h) + _ioctl(self.h, IOCTL_DISK_UPDATE_PROPERTIES) # re-read the new partition table + k32.CloseHandle(self.h) + + +def format_card(number: int, expected_size: int, label: str, timings: dict) -> str: + t0 = time.perf_counter() + disk = next((d for d in win_disks.list_disks(CARD_BUSES) if d.id == str(number)), None) + if disk is None: + raise FormatError("Refusing: the disk is not a USB/SD/MMC card reader.") + if disk.size != expected_size: + raise FormatError("Refusing: the disk changed since it was selected (card swapped?).") + system = os.environ.get("SystemDrive", "C:").upper().rstrip("\\") + "\\" + if any(v.mount.upper() == system for v in disk.volumes): + raise FormatError("Refusing: this disk holds the Windows drive.") + timings["guard"] = time.perf_counter() - t0 + + t = time.perf_counter() + locked = [] + for v in disk.volumes: + h = _open_rw(rf"\\.\{v.mount[0]}:") + for _ in range(20): + if _ioctl(h, FSCTL_LOCK_VOLUME): + break + time.sleep(0.25) + else: + k32.CloseHandle(h) + raise FormatError(f"{v.mount} is in use by another program - close it and try again.") + _ioctl(h, FSCTL_DISMOUNT_VOLUME) + locked.append(h) + timings["lock_volumes"] = time.perf_counter() - t + + t = time.perf_counter() + lay = fat32.plan(disk.size) + zeros, writes = fat32.build(lay, label, random.getrandbits(32), random.getrandbits(32), disk.size) + raw = RawDisk(number, lay.bytes_per_sector) + try: + for off, n in zeros: + raw.zero(off, n) + for off, data in writes: # MBR is last: the partition appears complete + raw.write(off, data) + finally: + for h in locked: + _ioctl(h, FSCTL_UNLOCK_VOLUME) + k32.CloseHandle(h) + raw.close() + timings["write"] = time.perf_counter() - t + + t = time.perf_counter() + drive = "" + for _ in range(80): # Windows mounts the new volume by itself + d = next((x for x in win_disks.list_disks(CARD_BUSES) if x.id == str(number)), None) + vol = next((v for v in (d.volumes if d else []) if v.filesystem.upper() == "FAT32"), None) + if vol and vol.label.upper() == label.upper(): + drive = vol.mount + break + time.sleep(0.25) + timings["mount"] = time.perf_counter() - t + if not drive: + raise FormatError("Formatted, but Windows did not mount the card - remove and re-insert it.") + + t = time.perf_counter() + chk = subprocess.run(["chkdsk", drive.rstrip("\\")], capture_output=True, text=True, timeout=120, + creationflags=subprocess.CREATE_NO_WINDOW) + timings["chkdsk"] = time.perf_counter() - t + if chk.returncode != 0: + tail = " ".join(chk.stdout.split()[-30:]) + raise FormatError(f"chkdsk reported problems on the new file system: {tail}") + return drive + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--disk", type=int, required=True) + ap.add_argument("--size", type=int, required=True) + ap.add_argument("--label", default="KOTA") + ap.add_argument("--result", required=True) + a = ap.parse_args(argv) + timings: dict[str, float] = {} + result = {"ok": False, "message": "", "drive": "", "timings": timings} + try: + result["drive"] = format_card(a.disk, a.size, a.label, timings) + result["ok"] = True + result["message"] = "Formatted FAT32 (MBR), checked with chkdsk." + except FormatError as exc: + result["message"] = str(exc) + except Exception as exc: # report everything to the tool + result["message"] = f"Unexpected error: {exc!r}" + with open(a.result, "w", encoding="utf-8") as fh: + json.dump(result, fh) + return 0 if result["ok"] else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/kota_tech/ui/diag_panel.py b/kota_tech/ui/diag_panel.py index 7172f99..11c4e4f 100644 --- a/kota_tech/ui/diag_panel.py +++ b/kota_tech/ui/diag_panel.py @@ -1,230 +1,234 @@ -"""Diagnostics side panel: renders a diag.engine.Report. - -Layout: verdict banner, expected LED, firmware, start-up milestones, a -findings tree (root causes; their consequences as children; titles only) -and a wrapped detail pane for the selected finding. -""" - -from __future__ import annotations - -from PySide6.QtCore import Qt -from PySide6.QtGui import QBrush, QColor -from PySide6.QtWidgets import ( - QFrame, - QHBoxLayout, - QLabel, - QListWidget, - QListWidgetItem, - QSizePolicy, - QSplitter, - QTextBrowser, - QTreeWidget, - QTreeWidgetItem, - QVBoxLayout, - QWidget, -) - -from ..diag.engine import LED, Finding, Report, Severity - -SEV_COLOR = { - Severity.CRITICAL: "#b03a2e", - Severity.ERROR: "#d9534f", - Severity.WARNING: "#c77c02", - Severity.INFO: "#2e6da4", -} -# verdict banner (background, text) by verdict severity; None = healthy -BANNER = { - Severity.CRITICAL: ("#f6d5d1", "#7b1e14"), - Severity.ERROR: ("#f9dedc", "#8a2019"), - Severity.WARNING: ("#fdebc8", "#6b4400"), - None: ("#d8f0dd", "#1b5e2a"), -} -_KEY = Qt.ItemDataRole.UserRole - - -def _heading(text: str) -> QLabel: - lbl = QLabel(text) - f = lbl.font() - f.setBold(True) - lbl.setFont(f) - return lbl - - -def _esc(s: str) -> str: - return s.replace("&", "&").replace("<", "<").replace(">", ">") - - -class DiagnosticsPanel(QWidget): - def __init__(self, parent: QWidget | None = None): - super().__init__(parent) - self.setMinimumWidth(380) - - self.verdict = QLabel(wordWrap=True) - self.verdict.setMargin(10) - vf = self.verdict.font() - vf.setPointSize(vf.pointSize() + 2) - vf.setBold(True) - self.verdict.setFont(vf) - - self.led_swatch = QLabel() - self.led_swatch.setFixedSize(22, 22) - self.led_text = QLabel(wordWrap=True) - led_row = QHBoxLayout() - led_row.addWidget(self.led_swatch, 0, Qt.AlignmentFlag.AlignTop) - led_row.addWidget(self.led_text, 1) - - self.firmware = QLabel(wordWrap=True) - self.firmware.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse) - - self.milestones = QListWidget() - self.milestones.setFrameShape(QFrame.Shape.NoFrame) - self.milestones.setSelectionMode(QListWidget.SelectionMode.NoSelection) - self.milestones.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Maximum) - - self.tree = QTreeWidget() - self.tree.setHeaderHidden(True) - self.tree.setIndentation(16) - self.tree.setTextElideMode(Qt.TextElideMode.ElideRight) - self.detail = QTextBrowser() - self.detail.setOpenExternalLinks(False) - findings = QSplitter(Qt.Orientation.Vertical) - findings.addWidget(self.tree) - findings.addWidget(self.detail) - findings.setSizes([220, 260]) - - lay = QVBoxLayout(self) - lay.setContentsMargins(4, 0, 0, 0) - lay.addWidget(self.verdict) - lay.addWidget(_heading("Status LED should be")) - lay.addLayout(led_row) - lay.addWidget(_heading("Firmware")) - lay.addWidget(self.firmware) - lay.addWidget(_heading("Start-up progress (current boot)")) - lay.addWidget(self.milestones) - lay.addWidget(_heading("Findings - root causes first; click one for details")) - lay.addWidget(findings, 1) - - self._findings: dict[str, Finding] = {} - self._selected: str | None = None - # key -> expanded, chosen by the user; survives the rebuild on refresh. - self._expanded: dict[str, bool] = {} - self.tree.currentItemChanged.connect(self._on_select) - self.tree.itemCollapsed.connect(lambda it: self._remember(it, False)) - self.tree.itemExpanded.connect(lambda it: self._remember(it, True)) - - # ------------------------------------------------------------------ - def show_report(self, r: Report) -> None: - bg, fg = BANNER.get(r.verdict_severity, BANNER[Severity.WARNING]) - if r.lines == 0: - bg, fg = "#e6e6e6", "#333333" - self.verdict.setText(r.verdict) - self.verdict.setStyleSheet(f"QLabel {{ background:{bg}; color:{fg}; border-radius:6px; }}") - - name, colour = LED[r.led] - self.led_swatch.setStyleSheet( - f"QLabel {{ background:{colour}; border:1px solid #777; border-radius:11px; }}") - ask = "" if r.led == "UNKNOWN" else " - check it matches the box" - reason = f"
from: {_esc(r.led_reason)}" if r.led_reason else "" - self.led_text.setText(f"{name}{ask}{reason}") - - fw = r.firmware - sd = {True: "reads Wi-Fi settings from the SD card", - False: "does not read Wi-Fi settings from the SD card", - None: "Wi-Fi-from-SD support unknown"} - self.firmware.setText(f"{_esc(fw.describe())}
{sd[fw.supports_sd_wifi]}") - - self.milestones.clear() - for _, label, t in r.milestones: - done = t is not None - it = QListWidgetItem(("✔ " if done else "○ ") + label - + (f" ({t / 1000:.1f} s)" if done else "")) - it.setForeground(QBrush(QColor("#1b7a33" if done else "#8a8a8a"))) - self.milestones.addItem(it) - rows = min(max(self.milestones.count(), 1), 15) - self.milestones.setFixedHeight(rows * max(self.milestones.sizeHintForRow(0), 16) + 6) - - self._fill_tree(r) - - def _fill_tree(self, r: Report) -> None: - self.tree.blockSignals(True) - self.tree.clear() - self._findings = {} - if not r.roots: - it = QTreeWidgetItem(["No problems detected in the log so far."]) - it.setForeground(0, QBrush(QColor("#1b7a33"))) - self.tree.addTopLevelItem(it) - for rc in r.roots: - top = self._item(rc.finding, bold=True) - self.tree.addTopLevelItem(top) - for c in rc.consequences: - top.addChild(self._item(c, prefix="↳ ")) - top.setExpanded(self._expanded.get(rc.finding.key, True)) - for title, key, items in (("Seen in earlier boots only", "grp:earlier", r.earlier), - ("Resolved during this session", "grp:resolved", r.resolved)): - if not items: - continue - grp = QTreeWidgetItem([f"{title} ({len(items)})"]) - grp.setData(0, _KEY, key) - grp.setForeground(0, QBrush(QColor("#666666"))) - self.tree.addTopLevelItem(grp) - for f in items: - grp.addChild(self._item(f, prefix="✔ " if key == "grp:resolved" else "")) - grp.setExpanded(self._expanded.get(key, False)) - # Keep the selection across refreshes; default to the top root cause. - target = self._selected if self._selected in self._findings else ( - r.roots[0].finding.key if r.roots else None) - self.tree.blockSignals(False) - self._select(target) - - def _remember(self, item: QTreeWidgetItem, expanded: bool) -> None: - if key := item.data(0, _KEY): - self._expanded[key] = expanded - - def _item(self, f: Finding, prefix: str = "", bold: bool = False) -> QTreeWidgetItem: - it = QTreeWidgetItem([f"{prefix}[{f.severity.label}] {f.title}"]) - it.setData(0, _KEY, f.key) - it.setForeground(0, QBrush(QColor(SEV_COLOR.get(f.severity, "#333333")))) - it.setToolTip(0, f.title) - if bold: - font = it.font(0) - font.setBold(True) - it.setFont(0, font) - self._findings[f.key] = f - return it - - def _select(self, key: str | None) -> None: - if key is None: - self.detail.setHtml("

Nothing to show yet.

") - return - match = self.tree.findItems("*", Qt.MatchFlag.MatchWildcard | Qt.MatchFlag.MatchRecursive) - for it in match: - if it.data(0, _KEY) == key: - self.tree.setCurrentItem(it) - return - - def _on_select(self, cur: QTreeWidgetItem | None, _prev) -> None: - key = cur.data(0, _KEY) if cur else None - f = self._findings.get(key or "") - if f is None: - return - self._selected = key - self.detail.setHtml(_detail_html(f)) - - -def _detail_html(f: Finding) -> str: - colour = SEV_COLOR.get(f.severity, "#333") - actions = "".join(f"
  • {_esc(a)}
    • " for a in f.actions) - seen = f"Seen {f.count}x" - if f.first_uptime_ms is not None: - seen += f", first at {f.first_uptime_ms / 1000:.1f} s after start" - if f.resolved and f.resolved_by: - seen += f" - resolved ({_esc(f.resolved_by)})" - evidence = "
    ".join(_esc(e) for e in f.evidence) - return ( - f"

    {_esc(f.title)}

    " - f"

    What it means: {_esc(f.meaning)}

    " - + (f"

    What to do:

      {actions}
    " if actions else "") - + f"

    {seen}

    " - + (f"

    {evidence}

    " - if evidence else "") - ) +"""Diagnostics side panel: renders a diag.engine.Report. + +Layout: verdict banner, expected LED, firmware, start-up milestones, a +findings tree (root causes; their consequences as children; titles only) +and a wrapped detail pane for the selected finding. +""" + +from __future__ import annotations + +from PySide6.QtCore import Qt +from PySide6.QtGui import QBrush, QColor +from PySide6.QtWidgets import ( + QFrame, + QHBoxLayout, + QLabel, + QListWidget, + QListWidgetItem, + QPushButton, + QSizePolicy, + QSplitter, + QTextBrowser, + QTreeWidget, + QTreeWidgetItem, + QVBoxLayout, + QWidget, +) + +from ..diag.engine import LED, Finding, Report, Severity + +SEV_COLOR = { + Severity.CRITICAL: "#b03a2e", + Severity.ERROR: "#d9534f", + Severity.WARNING: "#c77c02", + Severity.INFO: "#2e6da4", +} +# verdict banner (background, text) by verdict severity; None = healthy +BANNER = { + Severity.CRITICAL: ("#f6d5d1", "#7b1e14"), + Severity.ERROR: ("#f9dedc", "#8a2019"), + Severity.WARNING: ("#fdebc8", "#6b4400"), + None: ("#d8f0dd", "#1b5e2a"), +} +_KEY = Qt.ItemDataRole.UserRole + + +def _heading(text: str) -> QLabel: + lbl = QLabel(text) + f = lbl.font() + f.setBold(True) + lbl.setFont(f) + return lbl + + +def _esc(s: str) -> str: + return s.replace("&", "&").replace("<", "<").replace(">", ">") + + +class DiagnosticsPanel(QWidget): + def __init__(self, parent: QWidget | None = None): + super().__init__(parent) + self.setMinimumWidth(380) + + self.verdict = QLabel(wordWrap=True) + self.verdict.setMargin(10) + vf = self.verdict.font() + vf.setPointSize(vf.pointSize() + 2) + vf.setBold(True) + self.verdict.setFont(vf) + + self.led_swatch = QLabel() + self.led_swatch.setFixedSize(22, 22) + self.led_text = QLabel(wordWrap=True) + led_row = QHBoxLayout() + led_row.addWidget(self.led_swatch, 0, Qt.AlignmentFlag.AlignTop) + led_row.addWidget(self.led_text, 1) + + self.firmware = QLabel(wordWrap=True) + self.firmware.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse) + + self.milestones = QListWidget() + self.milestones.setFrameShape(QFrame.Shape.NoFrame) + self.milestones.setSelectionMode(QListWidget.SelectionMode.NoSelection) + self.milestones.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Maximum) + + self.tree = QTreeWidget() + self.tree.setHeaderHidden(True) + self.tree.setIndentation(16) + self.tree.setTextElideMode(Qt.TextElideMode.ElideRight) + self.detail = QTextBrowser() + self.detail.setOpenExternalLinks(False) + findings = QSplitter(Qt.Orientation.Vertical) + findings.addWidget(self.tree) + findings.addWidget(self.detail) + findings.setSizes([220, 260]) + + lay = QVBoxLayout(self) + lay.setContentsMargins(4, 0, 0, 0) + lay.addWidget(self.verdict) + lay.addWidget(_heading("Status LED should be")) + lay.addLayout(led_row) + lay.addWidget(_heading("Firmware")) + lay.addWidget(self.firmware) + self.qr_btn = QPushButton("Matter pairing QR code...") + self.qr_btn.setToolTip("Show the box's Matter QR code on screen to scan with the phone") + lay.addWidget(self.qr_btn, 0, Qt.AlignmentFlag.AlignLeft) + lay.addWidget(_heading("Start-up progress (current boot)")) + lay.addWidget(self.milestones) + lay.addWidget(_heading("Findings - root causes first; click one for details")) + lay.addWidget(findings, 1) + + self._findings: dict[str, Finding] = {} + self._selected: str | None = None + # key -> expanded, chosen by the user; survives the rebuild on refresh. + self._expanded: dict[str, bool] = {} + self.tree.currentItemChanged.connect(self._on_select) + self.tree.itemCollapsed.connect(lambda it: self._remember(it, False)) + self.tree.itemExpanded.connect(lambda it: self._remember(it, True)) + + # ------------------------------------------------------------------ + def show_report(self, r: Report) -> None: + bg, fg = BANNER.get(r.verdict_severity, BANNER[Severity.WARNING]) + if r.lines == 0: + bg, fg = "#e6e6e6", "#333333" + self.verdict.setText(r.verdict) + self.verdict.setStyleSheet(f"QLabel {{ background:{bg}; color:{fg}; border-radius:6px; }}") + + name, colour = LED[r.led] + self.led_swatch.setStyleSheet( + f"QLabel {{ background:{colour}; border:1px solid #777; border-radius:11px; }}") + ask = "" if r.led == "UNKNOWN" else " - check it matches the box" + reason = f"
    from: {_esc(r.led_reason)}" if r.led_reason else "" + self.led_text.setText(f"{name}{ask}{reason}") + + fw = r.firmware + sd = {True: "reads Wi-Fi settings from the SD card", + False: "does not read Wi-Fi settings from the SD card", + None: "Wi-Fi-from-SD support unknown"} + self.firmware.setText(f"{_esc(fw.describe())}
    {sd[fw.supports_sd_wifi]}") + + self.milestones.clear() + for _, label, t in r.milestones: + done = t is not None + it = QListWidgetItem(("✔ " if done else "○ ") + label + + (f" ({t / 1000:.1f} s)" if done else "")) + it.setForeground(QBrush(QColor("#1b7a33" if done else "#8a8a8a"))) + self.milestones.addItem(it) + rows = min(max(self.milestones.count(), 1), 15) + self.milestones.setFixedHeight(rows * max(self.milestones.sizeHintForRow(0), 16) + 6) + + self._fill_tree(r) + + def _fill_tree(self, r: Report) -> None: + self.tree.blockSignals(True) + self.tree.clear() + self._findings = {} + if not r.roots: + it = QTreeWidgetItem(["No problems detected in the log so far."]) + it.setForeground(0, QBrush(QColor("#1b7a33"))) + self.tree.addTopLevelItem(it) + for rc in r.roots: + top = self._item(rc.finding, bold=True) + self.tree.addTopLevelItem(top) + for c in rc.consequences: + top.addChild(self._item(c, prefix="↳ ")) + top.setExpanded(self._expanded.get(rc.finding.key, True)) + for title, key, items in (("Seen in earlier boots only", "grp:earlier", r.earlier), + ("Resolved during this session", "grp:resolved", r.resolved)): + if not items: + continue + grp = QTreeWidgetItem([f"{title} ({len(items)})"]) + grp.setData(0, _KEY, key) + grp.setForeground(0, QBrush(QColor("#666666"))) + self.tree.addTopLevelItem(grp) + for f in items: + grp.addChild(self._item(f, prefix="✔ " if key == "grp:resolved" else "")) + grp.setExpanded(self._expanded.get(key, False)) + # Keep the selection across refreshes; default to the top root cause. + target = self._selected if self._selected in self._findings else ( + r.roots[0].finding.key if r.roots else None) + self.tree.blockSignals(False) + self._select(target) + + def _remember(self, item: QTreeWidgetItem, expanded: bool) -> None: + if key := item.data(0, _KEY): + self._expanded[key] = expanded + + def _item(self, f: Finding, prefix: str = "", bold: bool = False) -> QTreeWidgetItem: + it = QTreeWidgetItem([f"{prefix}[{f.severity.label}] {f.title}"]) + it.setData(0, _KEY, f.key) + it.setForeground(0, QBrush(QColor(SEV_COLOR.get(f.severity, "#333333")))) + it.setToolTip(0, f.title) + if bold: + font = it.font(0) + font.setBold(True) + it.setFont(0, font) + self._findings[f.key] = f + return it + + def _select(self, key: str | None) -> None: + if key is None: + self.detail.setHtml("

    Nothing to show yet.

    ") + return + match = self.tree.findItems("*", Qt.MatchFlag.MatchWildcard | Qt.MatchFlag.MatchRecursive) + for it in match: + if it.data(0, _KEY) == key: + self.tree.setCurrentItem(it) + return + + def _on_select(self, cur: QTreeWidgetItem | None, _prev) -> None: + key = cur.data(0, _KEY) if cur else None + f = self._findings.get(key or "") + if f is None: + return + self._selected = key + self.detail.setHtml(_detail_html(f)) + + +def _detail_html(f: Finding) -> str: + colour = SEV_COLOR.get(f.severity, "#333") + actions = "".join(f"
  • {_esc(a)}
    • " for a in f.actions) + seen = f"Seen {f.count}x" + if f.first_uptime_ms is not None: + seen += f", first at {f.first_uptime_ms / 1000:.1f} s after start" + if f.resolved and f.resolved_by: + seen += f" - resolved ({_esc(f.resolved_by)})" + evidence = "
    ".join(_esc(e) for e in f.evidence) + return ( + f"

    {_esc(f.title)}

    " + f"

    What it means: {_esc(f.meaning)}

    " + + (f"

    What to do:

      {actions}
    " if actions else "") + + f"

    {seen}

    " + + (f"

    {evidence}

    " + if evidence else "") + ) diff --git a/kota_tech/ui/flash_tab.py b/kota_tech/ui/flash_tab.py index 9636e20..4700d4f 100644 --- a/kota_tech/ui/flash_tab.py +++ b/kota_tech/ui/flash_tab.py @@ -11,6 +11,7 @@ from PySide6.QtWidgets import ( QButtonGroup, QCheckBox, QFileDialog, + QInputDialog, QComboBox, QGroupBox, QHBoxLayout, @@ -25,7 +26,7 @@ from PySide6.QtWidgets import ( ) from ..core.session import SESSIONS_DIR -from ..flash.image import FirmwareImage, FirmwareLibrary, ImageError, factory_images +from ..flash.image import FirmwareImage, FirmwareLibrary, ImageError, bundled_images from ..flash.plan import Mode, build_plan from ..flash.runner import FlashRunner, Result from ..serialio.ports import PortInfo, list_serial_ports @@ -168,12 +169,14 @@ class FlashTab(QWidget): def _load_images(self, select_sha: str | None = None) -> None: self._problems = [] try: - factory = factory_images() + bundled = bundled_images() except ImageError as exc: - factory = [] - self._problems.append(f"factory image: {exc}") + bundled = [] + self._problems.append(f"bundled image: {exc}") + factory = [i for i in bundled if not i.is_developer] self._reference = factory[0] if factory else None dev, problems = self._library.images(self._reference) + dev += [i for i in bundled if i.is_developer] self._problems += problems # Newest developer builds first, then the factory image. dev.sort(key=lambda i: i.app.built_at or dt.datetime.min, reverse=True) @@ -196,8 +199,12 @@ class FlashTab(QWidget): "Firmware (*.bin);;All files (*)") if not path: return + built_by, ok = QInputDialog.getText(self, "Who built this firmware?", + "Name of the developer who built this image (shown next to it):") + if not ok: + return try: - img = self._library.import_file(Path(path), self._reference) + img = self._library.import_file(Path(path), self._reference, built_by=built_by) except (ImageError, OSError) as exc: QMessageBox.warning(self, "Cannot use this file", f"This file cannot be flashed:\n\n{exc}") return diff --git a/kota_tech/ui/guide_panel.py b/kota_tech/ui/guide_panel.py index 193f30a..289d90e 100644 --- a/kota_tech/ui/guide_panel.py +++ b/kota_tech/ui/guide_panel.py @@ -1,216 +1,225 @@ -"""Guided troubleshooting panel: renders a diag.guides.GuideRunner.""" - -from __future__ import annotations - -from PySide6.QtCore import QSize, Qt -from PySide6.QtGui import QBrush, QColor, QIcon, QLinearGradient, QPainter, QPixmap -from PySide6.QtWidgets import ( - QFrame, - QHBoxLayout, - QLabel, - QListWidget, - QListWidgetItem, - QProgressBar, - QPushButton, - QVBoxLayout, - QWidget, -) - -from ..diag.engine import LED, DiagnosticsEngine -from ..diag.guides import GuideRunner, recommended - -END_STYLE = { # end severity -> (border, background) - "ok": ("#2e9e44", "#e3f4e7"), - "info": ("#2e6da4", "#e3edf7"), - "warning": ("#c77c02", "#fdf0d6"), - "error": ("#d9534f", "#f9e0df"), - "critical": ("#b03a2e", "#f6d5d1"), -} - - -def _swatch(key: str, size: int = 22) -> QIcon: - """Round LED-colour icon; CYCLING is drawn as a blue-dark-red gradient.""" - pm = QPixmap(size, size) - pm.fill(QColor(0, 0, 0, 0)) - p = QPainter(pm) - p.setRenderHint(QPainter.RenderHint.Antialiasing) - if key == "CYCLING": - g = QLinearGradient(0, 0, size, 0) - for stop, c in ((0.0, "#2f6fd6"), (0.5, "#444444"), (1.0, "#d64541")): - g.setColorAt(stop, QColor(c)) - p.setBrush(QBrush(g)) - else: - p.setBrush(QColor(LED.get(key, ("", "#888888"))[1])) - p.setPen(QColor("#777777")) - p.drawEllipse(1, 1, size - 2, size - 2) - p.end() - return QIcon(pm) - - -def _esc(s: str) -> str: - return s.replace("&", "&").replace("<", "<").replace(">", ">") - - -class GuidePanel(QWidget): - def __init__(self, runner: GuideRunner, engine: DiagnosticsEngine, parent: QWidget | None = None): - super().__init__(parent) - self.runner, self.engine = runner, engine - self._shown_key: tuple | None = None - - # --- picker (no guide running) - self.picker = QWidget() - pl = QVBoxLayout(self.picker) - pl.setContentsMargins(0, 0, 0, 0) - pl.addWidget(QLabel("Recommended checks for this box (based on the log):")) - self.guide_list = QListWidget() - self.guide_list.itemDoubleClicked.connect(lambda it: self._start(it.data(Qt.ItemDataRole.UserRole))) - pl.addWidget(self.guide_list, 1) - self.start_btn = QPushButton("Start selected check") - self.start_btn.clicked.connect(self._start_selected) - pl.addWidget(self.start_btn) - - # --- running guide - self.card = QFrame() - self.card.setFrameShape(QFrame.Shape.StyledPanel) - cl = QVBoxLayout(self.card) - self.guide_title = QLabel() - self.guide_title.setStyleSheet("color:#666") - self.step_text = QLabel(wordWrap=True) - f = self.step_text.font() - f.setPointSize(f.pointSize() + 2) - self.step_text.setFont(f) - self.step_text.setTextFormat(Qt.TextFormat.RichText) - self.options = QVBoxLayout() - self.wait_bar = QProgressBar() - self.wait_bar.setTextVisible(False) - self.wait_label = QLabel() - self.note = QLabel(wordWrap=True) - self.note.setStyleSheet("color:#b03a2e") - self.continue_btn = QPushButton("Continue") - self.continue_btn.clicked.connect(self._proceed) - self.skip_btn = QPushButton("Skip waiting / it did not work") - self.skip_btn.clicked.connect(self._proceed) - self.end_box = QLabel(wordWrap=True) - self.end_box.setTextFormat(Qt.TextFormat.RichText) - self.end_box.setMargin(10) - for w in (self.guide_title, self.step_text): - cl.addWidget(w) - cl.addLayout(self.options) - for w in (self.wait_label, self.wait_bar, self.note, self.continue_btn, self.skip_btn, self.end_box): - cl.addWidget(w) - cl.addStretch(1) - nav = QHBoxLayout() - self.back_btn = QPushButton("← Back") - self.back_btn.clicked.connect(self._back) - self.done_btn = QPushButton("Back to checks list") - self.done_btn.clicked.connect(self._cancel) - nav.addWidget(self.back_btn) - nav.addStretch(1) - nav.addWidget(self.done_btn) - cl.addLayout(nav) - - lay = QVBoxLayout(self) - lay.setContentsMargins(4, 0, 0, 0) - lay.addWidget(self.picker, 1) - lay.addWidget(self.card, 1) - self.refresh() - - # ------------------------------------------------------------ actions - def _start_selected(self) -> None: - it = self.guide_list.currentItem() or self.guide_list.item(0) - if it: - self._start(it.data(Qt.ItemDataRole.UserRole)) - - def _start(self, guide_id: str) -> None: - self.runner.start(guide_id) - self.refresh() - - def _answer(self, key: str) -> None: - self.runner.answer(key) - self.refresh() - - def _proceed(self) -> None: - self.runner.proceed() - self.refresh() - - def _back(self) -> None: - self.runner.back() - self.refresh() - - def _cancel(self) -> None: - self.runner.cancel() - self.refresh() - - # ------------------------------------------------------------- render - def refresh(self) -> None: - """Called after every interaction and periodically (timeouts, countdown).""" - self.runner.tick() - v = self.runner.view() - running = v.guide is not None - self.picker.setVisible(not running) - self.card.setVisible(running) - if not running: - self._fill_picker() - self._shown_key = None - return - key = (v.guide.id, v.step.id if v.step else None, v.end.id if v.end else None) - if key != self._shown_key: - self._shown_key = key - self._render_step(v) - if v.waiting: - total = v.step.timeout_s - self.wait_bar.setMaximum(int(total)) - self.wait_bar.setValue(int(total - v.remaining_s)) - m, s = divmod(int(v.remaining_s), 60) - self.wait_label.setText(f"Watching the live log... {m}:{s:02d} left") - self.note.setText(v.note) - self.note.setVisible(bool(v.note)) - self.back_btn.setEnabled(v.can_back) - - def _fill_picker(self) -> None: - wanted = [(g.id, g.title, g.summary) for g in recommended(self.runner.guides, self.engine)] - current = [(self.guide_list.item(i).data(Qt.ItemDataRole.UserRole)) for i in range(self.guide_list.count())] - if [w[0] for w in wanted] == current: - return - self.guide_list.clear() - for gid, title, summary in wanted: - it = QListWidgetItem(f"{title}\n {summary}") - it.setData(Qt.ItemDataRole.UserRole, gid) - self.guide_list.addItem(it) - self.guide_list.setCurrentRow(0) - - def _render_step(self, v) -> None: - self.guide_title.setText(v.guide.title) - while self.options.count(): - w = self.options.takeAt(0).widget() - if w: - # Detach now; deleteLater alone leaves it visible until the - # event loop processes the deferred delete. - w.hide() - w.setParent(None) - w.deleteLater() - is_end = v.end is not None - self.step_text.setVisible(not is_end) - self.step_text.setText(_esc(v.text)) - for opt_key, label in v.options: - b = QPushButton(label) - b.setMinimumHeight(34) - if v.step and v.step.type == "led": - b.setIcon(_swatch(opt_key)) - b.setIconSize(QSize(22, 22)) - b.setStyleSheet("QPushButton { text-align:left; padding-left:10px; }") - b.clicked.connect(lambda _=False, k=opt_key: self._answer(k)) - self.options.addWidget(b) - step_type = v.step.type if v.step else None - self.continue_btn.setVisible(step_type == "info") - self.skip_btn.setVisible(step_type == "wait") - self.wait_bar.setVisible(step_type == "wait") - self.wait_label.setVisible(step_type == "wait") - self.end_box.setVisible(is_end) - if is_end: - e = v.end - border, bg = END_STYLE.get(e.severity, END_STYLE["info"]) - acts = "".join(f"
  • {_esc(a)}
    • " for a in e.actions) - self.end_box.setStyleSheet(f"QLabel {{ border:2px solid {border}; background:{bg}; border-radius:6px; }}") - self.end_box.setText(f"

    {_esc(e.title)}

    {_esc(e.text)}

    " - + (f"
      {acts}
    " if acts else "")) +"""Guided troubleshooting panel: renders a diag.guides.GuideRunner.""" + +from __future__ import annotations + +from PySide6.QtCore import QSize, Qt +from PySide6.QtGui import QBrush, QColor, QIcon, QLinearGradient, QPainter, QPixmap +from PySide6.QtWidgets import ( + QFrame, + QHBoxLayout, + QLabel, + QListWidget, + QListWidgetItem, + QProgressBar, + QPushButton, + QVBoxLayout, + QWidget, +) + +from ..diag.engine import LED, DiagnosticsEngine +from ..diag.guides import GuideRunner, recommended +from ..matter.payload import SetupPayload +from .matter_qr import MatterQrWidget + +END_STYLE = { # end severity -> (border, background) + "ok": ("#2e9e44", "#e3f4e7"), + "info": ("#2e6da4", "#e3edf7"), + "warning": ("#c77c02", "#fdf0d6"), + "error": ("#d9534f", "#f9e0df"), + "critical": ("#b03a2e", "#f6d5d1"), +} + + +def _swatch(key: str, size: int = 22) -> QIcon: + """Round LED-colour icon; CYCLING is drawn as a blue-dark-red gradient.""" + pm = QPixmap(size, size) + pm.fill(QColor(0, 0, 0, 0)) + p = QPainter(pm) + p.setRenderHint(QPainter.RenderHint.Antialiasing) + if key == "CYCLING": + g = QLinearGradient(0, 0, size, 0) + for stop, c in ((0.0, "#2f6fd6"), (0.5, "#444444"), (1.0, "#d64541")): + g.setColorAt(stop, QColor(c)) + p.setBrush(QBrush(g)) + else: + p.setBrush(QColor(LED.get(key, ("", "#888888"))[1])) + p.setPen(QColor("#777777")) + p.drawEllipse(1, 1, size - 2, size - 2) + p.end() + return QIcon(pm) + + +def _esc(s: str) -> str: + return s.replace("&", "&").replace("<", "<").replace(">", ">") + + +class GuidePanel(QWidget): + def __init__(self, runner: GuideRunner, engine: DiagnosticsEngine, parent: QWidget | None = None, + matter_payload=None): + super().__init__(parent) + self.runner, self.engine = runner, engine + self._matter_payload = matter_payload or SetupPayload # callable -> SetupPayload + self._shown_key: tuple | None = None + + # --- picker (no guide running) + self.picker = QWidget() + pl = QVBoxLayout(self.picker) + pl.setContentsMargins(0, 0, 0, 0) + pl.addWidget(QLabel("Recommended checks for this box (based on the log):")) + self.guide_list = QListWidget() + self.guide_list.itemDoubleClicked.connect(lambda it: self._start(it.data(Qt.ItemDataRole.UserRole))) + pl.addWidget(self.guide_list, 1) + self.start_btn = QPushButton("Start selected check") + self.start_btn.clicked.connect(self._start_selected) + pl.addWidget(self.start_btn) + + # --- running guide + self.card = QFrame() + self.card.setFrameShape(QFrame.Shape.StyledPanel) + cl = QVBoxLayout(self.card) + self.guide_title = QLabel() + self.guide_title.setStyleSheet("color:#666") + self.step_text = QLabel(wordWrap=True) + f = self.step_text.font() + f.setPointSize(f.pointSize() + 2) + self.step_text.setFont(f) + self.step_text.setTextFormat(Qt.TextFormat.RichText) + self.options = QVBoxLayout() + self.wait_bar = QProgressBar() + self.wait_bar.setTextVisible(False) + self.wait_label = QLabel() + self.note = QLabel(wordWrap=True) + self.note.setStyleSheet("color:#b03a2e") + self.continue_btn = QPushButton("Continue") + self.continue_btn.clicked.connect(self._proceed) + self.skip_btn = QPushButton("Skip waiting / it did not work") + self.skip_btn.clicked.connect(self._proceed) + self.end_box = QLabel(wordWrap=True) + self.end_box.setTextFormat(Qt.TextFormat.RichText) + self.end_box.setMargin(10) + self.qr = MatterQrWidget(SetupPayload(), 220) + for w in (self.guide_title, self.step_text, self.qr): + cl.addWidget(w) + cl.addLayout(self.options) + for w in (self.wait_label, self.wait_bar, self.note, self.continue_btn, self.skip_btn, self.end_box): + cl.addWidget(w) + cl.addStretch(1) + nav = QHBoxLayout() + self.back_btn = QPushButton("← Back") + self.back_btn.clicked.connect(self._back) + self.done_btn = QPushButton("Back to checks list") + self.done_btn.clicked.connect(self._cancel) + nav.addWidget(self.back_btn) + nav.addStretch(1) + nav.addWidget(self.done_btn) + cl.addLayout(nav) + + lay = QVBoxLayout(self) + lay.setContentsMargins(4, 0, 0, 0) + lay.addWidget(self.picker, 1) + lay.addWidget(self.card, 1) + self.refresh() + + # ------------------------------------------------------------ actions + def _start_selected(self) -> None: + it = self.guide_list.currentItem() or self.guide_list.item(0) + if it: + self._start(it.data(Qt.ItemDataRole.UserRole)) + + def _start(self, guide_id: str) -> None: + self.runner.start(guide_id) + self.refresh() + + def _answer(self, key: str) -> None: + self.runner.answer(key) + self.refresh() + + def _proceed(self) -> None: + self.runner.proceed() + self.refresh() + + def _back(self) -> None: + self.runner.back() + self.refresh() + + def _cancel(self) -> None: + self.runner.cancel() + self.refresh() + + # ------------------------------------------------------------- render + def refresh(self) -> None: + """Called after every interaction and periodically (timeouts, countdown).""" + self.runner.tick() + v = self.runner.view() + running = v.guide is not None + self.picker.setVisible(not running) + self.card.setVisible(running) + if not running: + self._fill_picker() + self._shown_key = None + return + key = (v.guide.id, v.step.id if v.step else None, v.end.id if v.end else None) + if key != self._shown_key: + self._shown_key = key + self._render_step(v) + if v.waiting: + total = v.step.timeout_s + self.wait_bar.setMaximum(int(total)) + self.wait_bar.setValue(int(total - v.remaining_s)) + m, s = divmod(int(v.remaining_s), 60) + self.wait_label.setText(f"Watching the live log... {m}:{s:02d} left") + self.note.setText(v.note) + self.note.setVisible(bool(v.note)) + self.back_btn.setEnabled(v.can_back) + + def _fill_picker(self) -> None: + wanted = [(g.id, g.title, g.summary) for g in recommended(self.runner.guides, self.engine)] + current = [(self.guide_list.item(i).data(Qt.ItemDataRole.UserRole)) for i in range(self.guide_list.count())] + if [w[0] for w in wanted] == current: + return + self.guide_list.clear() + for gid, title, summary in wanted: + it = QListWidgetItem(f"{title}\n {summary}") + it.setData(Qt.ItemDataRole.UserRole, gid) + self.guide_list.addItem(it) + self.guide_list.setCurrentRow(0) + + def _render_step(self, v) -> None: + self.guide_title.setText(v.guide.title) + while self.options.count(): + w = self.options.takeAt(0).widget() + if w: + # Detach now; deleteLater alone leaves it visible until the + # event loop processes the deferred delete. + w.hide() + w.setParent(None) + w.deleteLater() + is_end = v.end is not None + self.step_text.setVisible(not is_end) + self.step_text.setText(_esc(v.text)) + show_qr = bool(v.step and v.step.show_qr) + if show_qr: + self.qr.set_payload(self._matter_payload()) + self.qr.setVisible(show_qr) + for opt_key, label in v.options: + b = QPushButton(label) + b.setMinimumHeight(34) + if v.step and v.step.type == "led": + b.setIcon(_swatch(opt_key)) + b.setIconSize(QSize(22, 22)) + b.setStyleSheet("QPushButton { text-align:left; padding-left:10px; }") + b.clicked.connect(lambda _=False, k=opt_key: self._answer(k)) + self.options.addWidget(b) + step_type = v.step.type if v.step else None + self.continue_btn.setVisible(step_type == "info") + self.skip_btn.setVisible(step_type == "wait") + self.wait_bar.setVisible(step_type == "wait") + self.wait_label.setVisible(step_type == "wait") + self.end_box.setVisible(is_end) + if is_end: + e = v.end + border, bg = END_STYLE.get(e.severity, END_STYLE["info"]) + acts = "".join(f"
  • {_esc(a)}
    • " for a in e.actions) + self.end_box.setStyleSheet(f"QLabel {{ border:2px solid {border}; background:{bg}; border-radius:6px; }}") + self.end_box.setText(f"

    {_esc(e.title)}

    {_esc(e.text)}

    " + + (f"
      {acts}
    " if acts else "")) diff --git a/kota_tech/ui/main_window.py b/kota_tech/ui/main_window.py index ec57664..4edecf6 100644 --- a/kota_tech/ui/main_window.py +++ b/kota_tech/ui/main_window.py @@ -30,10 +30,12 @@ from ..diag.engine import DiagnosticsEngine from ..diag.guides import GuideRunner from ..serialio.manager import SerialManager, State from ..serialio.ports import PortInfo, list_serial_ports +from ..matter.payload import SetupPayload, payload_from_log 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 @@ -118,7 +120,7 @@ class SerialTab(QWidget): self.view.setStyleSheet("QPlainTextEdit { background:#1e1e1e; color:#d4d4d4; }") self.diag = DiagnosticsPanel() - self.guide = GuidePanel(self._runner, self._engine) + self.guide = GuidePanel(self._runner, self._engine, matter_payload=self.matter_payload) self.side_tabs = QTabWidget() self.side_tabs.addTab(self.diag, "Diagnosis") self.side_tabs.addTab(self.guide, "Guided troubleshooting") @@ -147,6 +149,7 @@ class SerialTab(QWidget): self.analyze_btn.clicked.connect(self._analyze_file) self.copy_btn.clicked.connect(self._copy_report) self.service_btn.clicked.connect(lambda: self._open_service_report(auto=False)) + self.diag.qr_btn.clicked.connect(lambda: MatterQrDialog(self.matter_payload(), self).exec()) # Report refresh is throttled; lines can arrive at >100/s. self._diag_timer = QTimer(self, interval=400) @@ -205,6 +208,10 @@ class SerialTab(QWidget): self._active_port = port self._manager.connect(port) + def matter_payload(self) -> SetupPayload: + """Pairing payload from this box's log (VID/PID/discriminator), defaults otherwise.""" + return payload_from_log([self._engine.matter_advert] if self._engine.matter_advert else []) + # ---- service system ------------------------------------------------- def _service_base(self) -> str: return str(self._settings.value("service/base", DEFAULT_BASE)) diff --git a/kota_tech/ui/matter_qr.py b/kota_tech/ui/matter_qr.py new file mode 100644 index 0000000..c687f1d --- /dev/null +++ b/kota_tech/ui/matter_qr.py @@ -0,0 +1,72 @@ +"""Matter pairing QR code on screen, so the technician can scan it from the laptop.""" + +from __future__ import annotations + +import segno +from PySide6.QtCore import Qt +from PySide6.QtGui import QColor, QImage, QPainter, QPixmap +from PySide6.QtWidgets import QDialog, QDialogButtonBox, QLabel, QVBoxLayout, QWidget + +from ..matter.payload import SetupPayload + + +def qr_pixmap(text: str, size: int = 260) -> QPixmap: + """Black-on-white QR with the quiet zone phones need (4 modules).""" + qr = segno.make(text, error="m") + matrix = [list(row) for row in qr.matrix_iter(scale=1, border=4)] + n = len(matrix) + module = max(1, size // n) + img = QImage(n * module, n * module, QImage.Format.Format_RGB32) + img.fill(QColor("white")) + p = QPainter(img) + p.setPen(Qt.PenStyle.NoPen) + p.setBrush(QColor("black")) + for y, row in enumerate(matrix): + for x, dark in enumerate(row): + if dark: + p.drawRect(x * module, y * module, module, module) + p.end() + return QPixmap.fromImage(img) + + +class MatterQrWidget(QWidget): + def __init__(self, payload: SetupPayload, size: int = 260, parent: QWidget | None = None): + super().__init__(parent) + self.image = QLabel() + self.image.setAlignment(Qt.AlignmentFlag.AlignCenter) + self.code = QLabel() + self.code.setAlignment(Qt.AlignmentFlag.AlignCenter) + self.code.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse) + f = self.code.font() + f.setPointSize(f.pointSize() + 4) + f.setBold(True) + self.code.setFont(f) + self.note = QLabel(wordWrap=True) + self.note.setAlignment(Qt.AlignmentFlag.AlignCenter) + self.note.setStyleSheet("color:#555") + lay = QVBoxLayout(self) + lay.addWidget(self.image) + lay.addWidget(self.code) + lay.addWidget(self.note) + self._size = size + self.set_payload(payload) + + def set_payload(self, payload: SetupPayload) -> None: + self.payload = payload + self.image.setPixmap(qr_pixmap(payload.qr_text(), self._size)) + self.code.setText(f"Manual code: {payload.manual_code_pretty()}") + src = ("from this box's log" if payload.from_log else + "standard code of the v1.0 firmware - connect the box to confirm") + self.note.setText(f"Scan with the Matter app (Apple Home, Google Home, ...), or type the manual code. " + f"Discriminator {payload.discriminator} ({src}). Same as the sticker on the box.") + + +class MatterQrDialog(QDialog): + def __init__(self, payload: SetupPayload, parent: QWidget | None = None): + super().__init__(parent) + self.setWindowTitle("Matter pairing code") + lay = QVBoxLayout(self) + lay.addWidget(MatterQrWidget(payload, 320)) + buttons = QDialogButtonBox(QDialogButtonBox.StandardButton.Close) + buttons.rejected.connect(self.reject) + lay.addWidget(buttons) diff --git a/kota_tech/ui/sd_tab.py b/kota_tech/ui/sd_tab.py index 43ae8aa..9dce3ff 100644 --- a/kota_tech/ui/sd_tab.py +++ b/kota_tech/ui/sd_tab.py @@ -7,7 +7,6 @@ from pathlib import Path from PySide6.QtCore import QObject, QSettings, Qt, Signal from PySide6.QtWidgets import ( - QApplication, QCheckBox, QFileDialog, QFormLayout, @@ -34,6 +33,7 @@ DEFAULT_TEMPLATE = Path(__file__).resolve().parents[4] / "002-SD_card_format" / class _Bridge(QObject): done = Signal(object) + listed = Signal(object) def _secret_field() -> tuple[QLineEdit, QCheckBox]: @@ -52,6 +52,7 @@ class SdTab(QWidget): self._settings = QSettings("KOTA", "TechnicianTool") self._cards: list[Disk] = [] self._busy = False + self._listing = False # --- 1. settings sbox = QGroupBox("1. Settings written to the card (ENV.TXT)") @@ -135,6 +136,7 @@ class SdTab(QWidget): self._bridge = _Bridge() self._bridge.done.connect(self._on_done) + self._bridge.listed.connect(self._on_listed) for w in (self.url, self.token, self.ssid, self.password): w.textChanged.connect(self._update) self.wifi_box.toggled.connect(self._update) @@ -199,13 +201,28 @@ class SdTab(QWidget): super().showEvent(event) def refresh_cards(self) -> None: - if self._busy: + """List cards in the background: the UI never waits for a slow card reader.""" + if self._busy or self._listing: + return + self._listing = True + self.others.setText("Looking for SD cards...") + + def worker(): + try: + result = list_cards() + except Exception as exc: # show it, never leave the tab stuck + result = ([], [], str(exc)) + self._bridge.listed.emit(result) + + threading.Thread(target=worker, daemon=True, name="sd-list").start() + + def _on_listed(self, result) -> None: + self._listing = False + self._cards, others = result[0], result[1] + if len(result) > 2: + self.others.setText(f"Could not list cards: {result[2]}") + self._update() return - QApplication.setOverrideCursor(Qt.CursorShape.WaitCursor) - try: - self._cards, others = list_cards() - finally: - QApplication.restoreOverrideCursor() self.card_list.clear() for d in self._cards: self.card_list.addItem(QListWidgetItem(d.label)) diff --git a/pyproject.toml b/pyproject.toml index 7327dc0..8f0f00e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,16 +1,17 @@ [project] name = "kota-technician-tool" -version = "0.5.0" +version = "0.6.0" description = "KOTA sensorbox technician diagnostic tool (demo)" requires-python = ">=3.11,<3.14" dependencies = [ "pyserial>=3.5", "PySide6-Essentials>=6.7", "esptool~=4.8", # 4.x CLI (write_flash/read_flash); v5 renamed commands + "segno>=1.6", # QR code for Matter pairing (pure Python) ] [dependency-groups] -dev = ["pytest>=8"] +dev = ["pytest>=8", "pyfatfs>=1.0", "setuptools<81"] # pyfatfs (independent FAT32 check) needs pkg_resources [tool.uv] package = false diff --git a/run_macos.command b/run_macos.command index acd3418..05d1c31 100755 --- a/run_macos.command +++ b/run_macos.command @@ -3,7 +3,7 @@ cd "$(dirname "$0")" || exit 1 # Keep the virtual environment out of the (Google Drive) tool folder. -export UV_PROJECT_ENVIRONMENT="$HOME/Library/Caches/KotaTechnicianTool/venv-0.5" +export UV_PROJECT_ENVIRONMENT="$HOME/Library/Caches/KotaTechnicianTool/venv-0.6" export PATH="$HOME/.local/bin:$PATH" if ! command -v uv >/dev/null 2>&1; then diff --git a/run_windows.bat b/run_windows.bat index a979e72..3ec2a56 100644 --- a/run_windows.bat +++ b/run_windows.bat @@ -3,7 +3,7 @@ setlocal cd /d "%~dp0" rem Keep the virtual environment out of the (Google Drive) tool folder. -set "UV_PROJECT_ENVIRONMENT=%LOCALAPPDATA%\KotaTechnicianTool\venv-0.5" +set "UV_PROJECT_ENVIRONMENT=%LOCALAPPDATA%\KotaTechnicianTool\venv-0.6" set "PATH=%USERPROFILE%\.local\bin;%PATH%" where uv >nul 2>nul diff --git a/tests/test_fat32.py b/tests/test_fat32.py new file mode 100644 index 0000000..59323f5 --- /dev/null +++ b/tests/test_fat32.py @@ -0,0 +1,103 @@ +"""FAT32/MBR layout: field-level checks against the FAT spec on an in-memory +disk, plus an independent FAT implementation (pyfatfs) mounting the result.""" + +import struct + +import pytest + +from kota_tech.sd import fat32 + +MiB, GiB = 1024**2, 1024**3 + + +def make_disk(size: int, label="KOTA") -> tuple[bytearray, fat32.Fat32Layout]: + lay = fat32.plan(size) + disk = bytearray(b"\xAA" * size) # old garbage everywhere + zeros, writes = fat32.build(lay, label, 0x12345678, 0xCAFEBABE, size) + for off, n in zeros: + disk[off:off + n] = bytes(n) + for off, data in writes: + disk[off:off + len(data)] = data + return disk, lay + + +@pytest.fixture(scope="module") +def small(): + return make_disk(96 * MiB) + + +def test_mbr_has_one_fat32_lba_partition(small): + disk, lay = small + assert disk[510:512] == b"\x55\xAA" + status, ptype, lba, count = struct.unpack_from("