diff --git a/docs/mac_agent_command.txt b/docs/mac_agent_command.txt new file mode 100644 index 0000000..cf5ae64 --- /dev/null +++ b/docs/mac_agent_command.txt @@ -0,0 +1,30 @@ +You are testing the KOTA sensorbox technician tool v0.5 on macOS. The current folder is +004-technician_tool/v0.5 (a Python/PySide6 desktop app; it has only been tested on Windows so far). + +Rules: +- Python only via uv or conda/mamba. Never use bare/system python or pip. +- Do not build, modify or flash firmware other than the bundled firmware/new_system.bin. +- Do not push anything to Git or GitLab. This is a local demo. +- Ask me before any destructive action: formatting an SD card, Full-mode flashing, or deleting files. + Before formatting, show me the exact disk (identifier, name, size, current files). +- Do not change code without telling me. Report bugs with evidence (commands and output). + +Tests, in order. Report pass/fail with evidence for each: +1. Launch: `bash run_macos.command` (it offers to install uv). The app window opens; note any warnings. +2. Tests: `UV_PROJECT_ENVIRONMENT="$HOME/Library/Caches/KotaTechnicianTool/venv-0.5" uv run --frozen pytest -q` +3. Serial: plug in the sensorbox (native USB). The port list shows a /dev/cu.usbmodem* "Sensorbox (native USB)" + entry with a MAC. Connect, press "Reset & capture": the boot log starts at "ESP-ROM:esp32s3" and the + Diagnosis panel identifies firmware v1.0 (build 45fa9d5b9). Check whether the port disappears after the + reset and whether the tool reconnects. +4. Guided troubleshooting: run "Quick check" and answer truthfully with what I tell you the LED shows. +5. Flash (App only, factory image; the box already runs it): the tool must enter flashing mode by itself, + write, restart the box and report "Success ... build 45fa9d5b9". Capture the esptool output. +6. SD card tab: the settings load from ../../002-SD_card_format/ENV.txt. The card reader and card are listed + (internal disks must NOT be listed). Run "Check card", then, after my OK, "Format card and write ENV.TXT". + It must use `diskutil eraseDisk FAT32 KOTA MBRFormat` on the correct disk, write ENV.TXT byte-identical + to the template, and eject the card. +7. "Service report..." opens the browser at https://service.kota-app.com/#kota-report=... (it is fine + if the page itself cannot prefill yet). + +At the end, write a short report: macOS version, Mac model/architecture, each test with result, errors, and +suggested fixes (file:line). \ No newline at end of file diff --git a/kota_tech/__init__.py b/kota_tech/__init__.py index e3f79e4..765cd5d 100644 --- a/kota_tech/__init__.py +++ b/kota_tech/__init__.py @@ -1,3 +1,3 @@ """KOTA sensorbox technician tool.""" -__version__ = "0.4.0" +__version__ = "0.5.0" diff --git a/kota_tech/diag/engine.py b/kota_tech/diag/engine.py index 3d8b28c..de61b70 100644 --- a/kota_tech/diag/engine.py +++ b/kota_tech/diag/engine.py @@ -25,6 +25,7 @@ from ..core.logline import LineKind, LogLine from .firmware import FirmwareInfo RULES_PATH = Path(__file__).with_name("rules.toml") +_MAC_LINE = re.compile(r"Device MAC address: ([0-9A-Fa-f]{2}(?::[0-9A-Fa-f]{2}){5})") class Severity(IntEnum): @@ -278,6 +279,7 @@ class DiagnosticsEngine: self._tool_reset_at: float | None = None 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 @property def boot(self) -> BootRecord: @@ -317,6 +319,8 @@ class DiagnosticsEngine: self._crash_finding().evidence.append(text) self.firmware.observe(text) + if m := _MAC_LINE.search(text): + self.mac = m.group(1).upper() matched = False for rule in self.rules: m = rule.pattern.search(text) diff --git a/kota_tech/diag/guides.toml b/kota_tech/diag/guides.toml index 6999aca..6330403 100644 --- a/kota_tech/diag/guides.toml +++ b/kota_tech/diag/guides.toml @@ -379,7 +379,7 @@ severity = "critical" title = "SD card is unreadable or has wrong settings" text = "Even after re-seating, the card or its ENV.TXT cannot be used." actions = [ - "Re-provision the card with the tool (SD tab, coming in v0.5) or use a new card.", + "Re-provision the card with the tool (SD card tab) or use a new card.", "The Diagnosis tab shows which step failed (card, file system or ENV.TXT keys).", ] diff --git a/kota_tech/flash/image.py b/kota_tech/flash/image.py index a12cbb6..0e3633e 100644 --- a/kota_tech/flash/image.py +++ b/kota_tech/flash/image.py @@ -1,18 +1,26 @@ -"""Official firmware images: validation and layout. +"""Firmware images: validation, integrity and layout. -Only images listed in firmware/firmware.toml with a matching SHA-256 are -offered for flashing: boxes normally get new firmware over OTA, so the tool -only ever writes the official factory image. +Two origins: + * factory - bundled with the tool (firmware/firmware.toml), SHA-256 locked; + * developer - a newer build the firmware team sends to technicians, + imported into a per-user library (~/KotaTechnicianTool/firmware). +Two kinds: + * merged - bootloader at 0x0 + partition table + app (like new_system.bin); + App-only or Full flashing; + * app - the application binary only (build/kota-sensorboard.bin); + App-only flashing, laid out like the factory image. -The factory image is a merged ESP32-S3 image (bootloader at 0x0, partition -table at 0x8000, app at ota_0). App-only flashing takes the app out of it -and writes it to ota_0; the layout comes from the image's own partition -table, not from constants. +Every image is checked structurally (ESP32-S3, kota-sensorboard app +descriptor, segment checksum and the appended SHA-256) so a truncated or +corrupted file is refused before anything is written. """ from __future__ import annotations +import datetime as dt import hashlib +import json +import shutil import struct import tomllib from dataclasses import dataclass @@ -20,6 +28,8 @@ from pathlib import Path FIRMWARE_DIR = Path(__file__).resolve().parents[2] / "firmware" MANIFEST = FIRMWARE_DIR / "firmware.toml" +LIBRARY_DIR = Path.home() / "KotaTechnicianTool" / "firmware" +LIBRARY_INDEX = "library.json" PT_OFFSET = 0x8000 PT_SIZE = 0xC00 @@ -27,6 +37,8 @@ ESP32S3_CHIP_ID = 9 IMAGE_MAGIC = 0xE9 APP_DESC_MAGIC = 0xABCD5432 FLASH_SIZES = {0: "1MB", 1: "2MB", 2: "4MB", 3: "8MB", 4: "16MB", 5: "32MB"} +_MONTHS = {m: i for i, m in enumerate( + ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"], 1)} class ImageError(ValueError): @@ -46,9 +58,18 @@ class Partition: class AppInfo: project: str version: str - built: str + built: str # "Jun 29 2026 11:03:32" as in the app descriptor idf: str - elf_sha: str # hex + elf_sha: str # hex + + @property + def built_at(self) -> dt.datetime | None: + try: + mon, day, year, hms = self.built.split() + h, m, s = (int(x) for x in hms.split(":")) + return dt.datetime(int(year), _MONTHS[mon], int(day), h, m, s) + except (ValueError, KeyError): + return None @dataclass(frozen=True) @@ -56,36 +77,43 @@ class FirmwareImage: path: Path name: str notes: str + origin: str # "factory" | "developer" + kind: str # "merged" | "app" sha256: str flash_size: str partitions: tuple[Partition, ...] + partition_table: bytes # expected table on the box + app_offset: int # where the app starts inside the file + app_length: int # exact app image length (from its structure) app: AppInfo @property def data(self) -> bytes: return self.path.read_bytes() + @property + def is_developer(self) -> bool: + return self.origin == "developer" + def partition(self, label: str) -> Partition: for p in self.partitions: if p.label == label: return p raise ImageError(f"partition {label!r} not in image") - @property - def partition_table(self) -> bytes: - return self.data[PT_OFFSET:PT_OFFSET + PT_SIZE] - def app_bytes(self) -> bytes: - """The application image stored in the ota_0 region of the merged image.""" - # No stripping of trailing 0xFF: the image ends in a SHA-256 digest - # that may legitimately end in 0xFF. The merged file ends with the app. - ota0 = self.partition("ota_0") - return self.data[ota0.offset:ota0.offset + ota0.size] + return self.data[self.app_offset:self.app_offset + self.app_length] def blank_otadata(self) -> bytes: """Erased otadata: the bootloader then boots ota_0.""" return b"\xff" * self.partition("otadata").size + def label(self) -> str: + tag = "factory" if self.origin == "factory" else "developer build" + return f"{self.app.version} built {self.app.built} [{self.app.elf_sha[:9]}] - {tag}" + + +# ---------------------------------------------------------------- parsing def parse_partition_table(raw: bytes) -> tuple[Partition, ...]: parts = [] @@ -103,50 +131,156 @@ def parse_partition_table(raw: bytes) -> tuple[Partition, ...]: return tuple(parts) -def _app_info(data: bytes, offset: int) -> AppInfo: - if len(data) < offset + 0x20 + 256 or data[offset] != IMAGE_MAGIC: - raise ImageError(f"no application image at 0x{offset:x}") - d = data[offset + 0x20:offset + 0x20 + 256] - if struct.unpack_from(" int: + """Validate an ESP app image at `offset`; return its exact length. - -def inspect_image(path: Path, name: str = "", notes: str = "") -> FirmwareImage: - data = path.read_bytes() - if len(data) < PT_OFFSET + PT_SIZE or data[0] != IMAGE_MAGIC: - raise ImageError("not a merged ESP32 image (no bootloader at offset 0)") - chip_id = struct.unpack_from("> 4, "?") - parts = parse_partition_table(data[PT_OFFSET:PT_OFFSET + PT_SIZE]) - labels = {p.label for p in parts} - for need in ("nvs", "otadata", "ota_0"): - if need not in labels: - raise ImageError(f"partition table has no {need!r} partition") - ota0 = next(p for p in parts if p.label == "ota_0") - app = _app_info(data, ota0.offset) - if app.project != "kota-sensorboard": - raise ImageError(f"image is {app.project!r}, not kota-sensorboard") - return FirmwareImage(path, name or path.name, notes, hashlib.sha256(data).hexdigest(), - flash_size, parts, app) + hash_appended = data[offset + 23] == 1 + pos, checksum = offset + 24, 0xEF + for _ in range(segments): + if pos + 8 > len(data): + raise ImageError("image is truncated (segment header missing)") + _, seg_len = struct.unpack_from(" len(data): + raise ImageError("image is truncated (segment data missing)") + for b in data[pos:pos + seg_len]: + checksum ^= b + pos += seg_len + pos += 15 - ((pos - offset) % 16) # pad so the checksum ends a 16-byte block + if pos >= len(data) or data[pos] != checksum: + raise ImageError("image checksum is wrong - the file is corrupted") + pos += 1 + if hash_appended: + if pos + 32 > len(data): + raise ImageError("image is truncated (SHA-256 missing)") + if hashlib.sha256(data[offset:pos]).digest() != data[pos:pos + 32]: + raise ImageError("image SHA-256 does not match its content - the file is corrupted") + pos += 32 + return pos - offset -def official_images(manifest: Path = MANIFEST) -> list[FirmwareImage]: - """Images from the manifest whose file exists and matches its SHA-256.""" +def _app_info(data: bytes, offset: int) -> AppInfo: + d = data[offset + 0x20:offset + 0x20 + 256] + if len(d) < 256 or struct.unpack_from(" 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() + if len(data) >= PT_OFFSET + PT_SIZE and data[0] == IMAGE_MAGIC and data[PT_OFFSET:PT_OFFSET + 2] == b"\xaa\x50": + app_image_length(data, 0) # bootloader must be intact + table = data[PT_OFFSET:PT_OFFSET + PT_SIZE] + parts = parse_partition_table(table) + labels = {p.label for p in parts} + for need in ("nvs", "otadata", "ota_0"): + if need not in labels: + raise ImageError(f"partition table has no {need!r} partition") + ota0 = next(p for p in parts if p.label == "ota_0") + kind, app_off, flash_size = "merged", ota0.offset, FLASH_SIZES.get(data[3] >> 4, "?") + elif data[:1] == bytes([IMAGE_MAGIC]): + if reference is None: + raise ImageError("app-only image needs the factory image as layout reference") + kind, app_off = "app", 0 + parts, table, flash_size = reference.partitions, reference.partition_table, reference.flash_size + ota0 = next(p for p in parts if p.label == "ota_0") + else: + raise ImageError("not an ESP32 firmware file") + length = app_image_length(data, app_off) + if length > ota0.size: + 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) + + +# ------------------------------------------------------------- catalogue + +def factory_images(manifest: Path = MANIFEST) -> list[FirmwareImage]: + """Bundled images whose file exists and matches the SHA-256 in the manifest.""" if not manifest.exists(): return [] - entries = tomllib.loads(manifest.read_text(encoding="utf-8")).get("image", []) out = [] - for e in entries: + for e in tomllib.loads(manifest.read_text(encoding="utf-8")).get("image", []): path = manifest.parent / e["file"] if not path.exists(): continue - img = inspect_image(path, e.get("name", ""), e.get("notes", "")) + img = inspect_image(path, e.get("name", ""), e.get("notes", ""), "factory") 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 + + +class FirmwareLibrary: + """Developer builds imported by the technician (persist across tool versions).""" + + def __init__(self, root: Path = LIBRARY_DIR): + self.root = root + + def _index(self) -> list[dict]: + f = self.root / LIBRARY_INDEX + return json.loads(f.read_text(encoding="utf-8")) if f.exists() else [] + + def _save(self, entries: list[dict]) -> None: + self.root.mkdir(parents=True, exist_ok=True) + (self.root / LIBRARY_INDEX).write_text(json.dumps(entries, indent=2), encoding="utf-8") + + def images(self, reference: FirmwareImage | None) -> tuple[list[FirmwareImage], list[str]]: + """(valid images, problems) - a changed or broken file is reported, not offered.""" + imgs, problems = [], [] + for e in self._index(): + path = self.root / e["file"] + try: + img = inspect_image(path, e.get("name", ""), e.get("notes", ""), "developer", reference) + if img.sha256 != e["sha256"]: + raise ImageError("file changed since it was imported") + imgs.append(img) + except (OSError, ImageError) as exc: + problems.append(f"{e.get('original', e['file'])}: {exc}") + return imgs, problems + + def import_file(self, src: Path, reference: FirmwareImage | None, notes: 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) + 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})" + entries.append({"file": fname, "sha256": img.sha256, "name": name, "original": src.name, + "imported": dt.datetime.now().isoformat(timespec="seconds"), "notes": notes}) + self._save(entries) + return inspect_image(self.root / fname, name, notes, "developer", reference) + + def remove(self, sha256: str) -> None: + entries = self._index() + keep = [e for e in entries if e["sha256"] != sha256] + for e in entries: + if e["sha256"] == sha256: + (self.root / e["file"]).unlink(missing_ok=True) + self._save(keep) + + +# Backwards-compatible name used by earlier callers/tests. +official_images = factory_images diff --git a/kota_tech/flash/plan.py b/kota_tech/flash/plan.py index ba6e99d..a9da22f 100644 --- a/kota_tech/flash/plan.py +++ b/kota_tech/flash/plan.py @@ -84,8 +84,11 @@ def build_plan(image: FirmwareImage, mode: Mode, port: str, workdir: Path, hex(otadata.offset), str(ota_file), hex(ota0.offset), str(app_file)], est_s=20)) else: + if image.kind != "merged": + raise ValueError("Full mode needs a complete image (bootloader + partition table + app); " + "this file contains the app only - use App-only mode.") plan.steps.append(Step( - "write", "Write the complete factory image (erases pairing and OTA registration)", + "write", "Write the complete image (erases pairing and OTA registration)", _common(port, "hard_reset") + write_opts + ["0x0", str(image.path)], est_s=25)) return plan diff --git a/kota_tech/sd/__init__.py b/kota_tech/sd/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/kota_tech/sd/disks.py b/kota_tech/sd/disks.py new file mode 100644 index 0000000..a3ff356 --- /dev/null +++ b/kota_tech/sd/disks.py @@ -0,0 +1,128 @@ +"""Find SD cards (removable disks) on Windows and macOS. + +Safety first: only USB/SD/MMC-attached disks that are not the system or +boot disk and not larger than MAX_CARD_BYTES are ever offered. Parsers are +pure functions over the OS tool output so they can be tested without +hardware. +""" + +from __future__ import annotations + +import json +import plistlib +import subprocess +import sys +from dataclasses import dataclass, field + +MAX_CARD_BYTES = 256 * 1024**3 # larger "removable" disks are probably not SD cards +CARD_BUSES = {"USB", "SD", "MMC"} + + +@dataclass +class Volume: + mount: str # "E:\\" on Windows, "/Volumes/KOTA" on macOS ("" if not mounted) + filesystem: str # "FAT32", "exFAT", "MS-DOS FAT32", ... + label: str + + +@dataclass +class Disk: + id: str # Windows disk number / macOS "/dev/disk4" + name: str + size: int + bus: str + volumes: list[Volume] = field(default_factory=list) + has_media: bool = True + + @property + def size_text(self) -> str: + return f"{self.size / 1e9:.1f} GB" if self.size else "no card" + + @property + def label(self) -> str: + vols = ", ".join(f"{v.mount or '-'} {v.filesystem} '{v.label}'" for v in self.volumes) or "no volume" + return f"Disk {self.id}: {self.name} - {self.size_text} ({self.bus}) - {vols}" + + +# ------------------------------------------------------------------ Windows + +WIN_LIST_PS = r""" +$ErrorActionPreference = 'SilentlyContinue' +# Filter by bus BEFORE touching partitions: querying volumes of every +# internal disk made this take >30 s on a machine with several HDDs. +$out = @(Get-Disk | Where-Object { -not $_.IsSystem -and -not $_.IsBoot -and + @('USB','SD','MMC') -contains [string]$_.BusType } | ForEach-Object { + $d = $_ + $vols = @(Get-Partition -DiskNumber $d.Number | ForEach-Object { + $v = $_ | Get-Volume + [pscustomobject]@{ DriveLetter = [string]$_.DriveLetter; FileSystem = [string]$v.FileSystem; Label = [string]$v.FileSystemLabel } + }) + [pscustomobject]@{ Number = $d.Number; FriendlyName = $d.FriendlyName; Size = [int64]$d.Size; + BusType = [string]$d.BusType; Volumes = $vols } +}) +ConvertTo-Json -InputObject $out -Depth 4 -Compress +""" + + +def parse_windows(json_text: str) -> list[Disk]: + data = json.loads(json_text or "[]") + if isinstance(data, dict): # ConvertTo-Json quirk for one element + data = [data] + disks = [] + for d in data: + if str(d.get("BusType")) not in CARD_BUSES: + continue + vols = d.get("Volumes") or [] + if isinstance(vols, dict): + vols = [vols] + volumes = [Volume(f"{v['DriveLetter']}:\\" if v.get("DriveLetter") and v["DriveLetter"] not in ("\x00", "") else "", + v.get("FileSystem") or "", v.get("Label") or "") for v in vols] + size = int(d.get("Size") or 0) + disks.append(Disk(str(d["Number"]), d.get("FriendlyName") or "Disk", size, str(d["BusType"]), + [v for v in volumes if v.mount or v.filesystem], has_media=size > 0)) + return disks + + +# -------------------------------------------------------------------- macOS + +def parse_macos(list_plist: bytes, infos: dict[str, bytes]) -> list[Disk]: + """`diskutil list -plist physical` + `diskutil info -plist ` for each.""" + listing = plistlib.loads(list_plist) + disks = [] + for entry in listing.get("AllDisksAndPartitions", []): + ident = entry.get("DeviceIdentifier", "") + info = plistlib.loads(infos[ident]) if ident in infos else {} + protocol = str(info.get("BusProtocol", "")) + removable = bool(info.get("RemovableMedia") or info.get("Removable") or info.get("Ejectable")) + internal = bool(info.get("Internal", True)) + is_root = any(p.get("MountPoint") == "/" for p in entry.get("Partitions", []) + entry.get("APFSVolumes", [])) + if ident == "disk0" or is_root or not (removable or not internal or protocol in ("USB", "Secure Digital", "SD")): + continue + vols = [Volume(p.get("MountPoint", ""), p.get("Content", ""), p.get("VolumeName", "")) + for p in entry.get("Partitions", [])] + size = int(entry.get("Size") or info.get("TotalSize") or 0) + bus = "SD" if "Secure Digital" in protocol or protocol == "SD" else (protocol or "USB") + disks.append(Disk(f"/dev/{ident}", info.get("MediaName", "Disk"), size, bus, vols, has_media=size > 0)) + return disks + + +# ------------------------------------------------------------------ public + +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) + elif sys.platform == "darwin": + lst = subprocess.run(["diskutil", "list", "-plist", "physical"], capture_output=True, timeout=30).stdout + infos = {} + for e in plistlib.loads(lst).get("AllDisksAndPartitions", []): + ident = e.get("DeviceIdentifier", "") + infos[ident] = subprocess.run(["diskutil", "info", "-plist", ident], capture_output=True, timeout=30).stdout + disks = parse_macos(lst, infos) + else: + return [], [] + ok = [d for d in disks if d.has_media and 0 < d.size <= MAX_CARD_BYTES] + return ok, [d for d in disks if d not in ok] diff --git a/kota_tech/sd/envfile.py b/kota_tech/sd/envfile.py new file mode 100644 index 0000000..1a1633c --- /dev/null +++ b/kota_tech/sd/envfile.py @@ -0,0 +1,151 @@ +"""ENV.TXT for the sensorbox SD card. + +Parsing mirrors EnvironmentService::read_env_var in the firmware exactly: +lines read with a 512-byte buffer, split at the FIRST '=', key compared +verbatim (no trimming), CR/LF removed from the value only, first match wins. +So the checks below report what would actually break a box - e.g. a UTF-8 +BOM (Windows Notepad) hides the first key, and "KEY = value" is not "KEY". + +The firmware FAT driver has no long-file-name support: the file must be +the 8.3 name ENV.TXT (case-insensitive) in the card root. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from pathlib import Path + +ENV_NAME = "ENV.TXT" +BUFFER_NAME = "BUFFER.TXT" # offline measurements waiting for upload +LINE_BUF = 512 # char line[512] in the firmware +REQUIRED = ("INFLUX_URL", "INFLUX_TOKEN") +OPTIONAL = ("SSID", "PASSWORD") +_BOM = b"\xef\xbb\xbf" + + +@dataclass +class Issue: + level: str # "error" | "warning" | "info" + text: str + + +@dataclass +class EnvConfig: + influx_url: str = "" + influx_token: str = "" + ssid: str = "" + password: str = "" + + @property + def local_wifi(self) -> bool: + return bool(self.ssid or self.password) + + def lines(self) -> list[tuple[str, str]]: + out = [("INFLUX_TOKEN", self.influx_token), ("INFLUX_URL", self.influx_url)] + if self.local_wifi: + out += [("SSID", self.ssid), ("PASSWORD", self.password)] + return out + + def render(self) -> bytes: + # LF only, no BOM, no quoting: exactly what the firmware parser expects. + return "".join(f"{k}={v}\n" for k, v in self.lines()).encode("utf-8") + + def problems(self) -> list[Issue]: + issues: list[Issue] = [] + for key, value in self.lines(): + if any(c in value for c in "\r\n"): + issues.append(Issue("error", f"{key} contains a line break.")) + if value != value.strip(): + issues.append(Issue("error", f"{key} starts or ends with a space; the box would use the space too.")) + if len(f"{key}={value}".encode()) >= LINE_BUF - 1: + issues.append(Issue("error", f"{key} is too long for the box (max {LINE_BUF - 2} bytes per line).")) + url = self.influx_url.strip() + if not url: + issues.append(Issue("error", "INFLUX_URL is empty.")) + elif not re.match(r"^https?://", url): + issues.append(Issue("error", "INFLUX_URL must start with https://")) + elif not ("/api/v2/write" in url and "bucket=" in url and "org=" in url): + issues.append(Issue("error", "INFLUX_URL must be a write URL: .../api/v2/write?bucket=...&org=...")) + elif url.startswith("http://"): + issues.append(Issue("warning", "INFLUX_URL uses http:// (unencrypted).")) + if not self.influx_token: + issues.append(Issue("error", "INFLUX_TOKEN is empty.")) + elif re.search(r"\s", self.influx_token): + issues.append(Issue("error", "INFLUX_TOKEN contains spaces.")) + elif len(self.influx_token) < 20: + issues.append(Issue("warning", "INFLUX_TOKEN looks too short.")) + if self.local_wifi: + if not self.ssid or not self.password: + issues.append(Issue("error", "Wi-Fi needs both SSID and PASSWORD (the box ignores one without the other).")) + if len(self.ssid.encode()) > 32: + issues.append(Issue("error", "SSID is longer than 32 bytes.")) + pw = self.password + if pw and not (8 <= len(pw) <= 63 or re.fullmatch(r"[0-9A-Fa-f]{64}", pw)): + issues.append(Issue("error", "Wi-Fi PASSWORD must be 8-63 characters (WPA2).")) + return issues + + +@dataclass +class EnvCheck: + values: dict[str, str] + issues: list[Issue] + + @property + def ok(self) -> bool: + return not any(i.level == "error" for i in self.issues) + + def masked(self) -> dict[str, str]: + out = {} + for k, v in self.values.items(): + out[k] = (v[:4] + "..." + v[-4:] if len(v) > 12 else "***") if k in ("INFLUX_TOKEN", "PASSWORD") else v + return out + + +def parse_env(raw: bytes) -> EnvCheck: + """Read ENV.TXT the way the firmware does and report what it would see.""" + issues: list[Issue] = [] + if raw.startswith(_BOM): + issues.append(Issue("error", "File starts with a UTF-8 BOM (saved by Notepad?): the box cannot read the first key. " + "Re-write the card with the tool.")) + text = raw.decode("utf-8", errors="replace") + values: dict[str, str] = {} + for n, line in enumerate(text.split("\n"), 1): + if len(line.encode()) >= LINE_BUF - 1: + issues.append(Issue("error", f"Line {n} is longer than the box can read.")) + if "=" not in line: + continue + key, value = line.split("=", 1) + value = value.replace("\r", "") + if key != key.strip(): + issues.append(Issue("error", f"Line {n}: spaces around the key '{key.strip()}' - the box will not find it.")) + if key in values: + issues.append(Issue("warning", f"{key} appears twice; the box uses the first one.")) + continue + values[key] = value + for key in REQUIRED: + if key not in values: + issues.append(Issue("error", f"{key} is missing - the box stops at start-up (white LED).")) + elif not values[key]: + issues.append(Issue("error", f"{key} is empty.")) + if values.get("HTTP_SERVER") == "true": + issues.append(Issue("warning", "HTTP_SERVER=true: a box with newer firmware starts in Wi-Fi setup mode.")) + if bool(values.get("SSID")) != bool(values.get("PASSWORD")): + issues.append(Issue("warning", "Only one of SSID / PASSWORD is set; the box ignores Wi-Fi settings.")) + elif values.get("SSID"): + issues.append(Issue("info", "Wi-Fi settings present: newer firmware uses this Wi-Fi instead of Matter " + "(the factory v1.0 firmware ignores them).")) + return EnvCheck(values, issues) + + +def load_template(path: Path) -> EnvConfig: + """Settings from an existing ENV.TXT (e.g. 002-SD_card_format/ENV.txt).""" + v = parse_env(path.read_bytes()).values + return EnvConfig(v.get("INFLUX_URL", ""), v.get("INFLUX_TOKEN", ""), v.get("SSID", ""), v.get("PASSWORD", "")) + + +def find_case_insensitive(root: Path, name: str) -> Path | None: + for p in root.iterdir() if root.is_dir() else (): + if p.is_file() and p.name.lower() == name.lower(): + return p + return None diff --git a/kota_tech/sd/provision.py b/kota_tech/sd/provision.py new file mode 100644 index 0000000..75f4aa6 --- /dev/null +++ b/kota_tech/sd/provision.py @@ -0,0 +1,273 @@ +"""Format an SD card for the sensorbox and write / check ENV.TXT. + +Card format required by the firmware: MBR partition table, FAT32 (its FAT +driver has no exFAT and no long file names), ENV.TXT in the root. + +Windows: formatting needs administrator rights, so a generated PowerShell +script runs elevated (UAC prompt). The script re-checks the disk (number, +size, bus, not system/boot) right before erasing it, so a card swapped +after the technician confirmed is not touched. Windows cannot format FAT32 +above 32 GB, so larger cards get a 32 GB partition (plenty for the box). +macOS: `diskutil eraseDisk FAT32 KOTA MBRFormat` (same as format_sd.sh). +""" + +from __future__ import annotations + +import json +import os +import plistlib +import subprocess +import sys +import time +import uuid +from dataclasses import dataclass, field +from pathlib import Path + +from .disks import CARD_BUSES, Disk +from .envfile import BUFFER_NAME, ENV_NAME, EnvCheck, EnvConfig, find_case_insensitive, parse_env + +VOLUME_LABEL = "KOTA" +FAT32_MAX = 32 * 1024**3 + + +@dataclass +class Outcome: + ok: bool + message: str + mount: str = "" + log: list[str] = field(default_factory=list) + + +@dataclass +class CardReport: + mount: str + filesystem: str + env: EnvCheck | None + buffered_lines: int | None # measurements waiting on the card for upload + problems: list[str] + + +# ----------------------------------------------------------------- Windows + +def windows_format_script(disk: Disk, result_file: Path) -> 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" + + +def _workdir() -> Path: + # Not tempfile.mkdtemp(): on Windows it creates an owner-only folder, and a + # file written there by the ELEVATED script is then unreadable for this + # (non-elevated) process. A normally inherited ACL works for both. + work = WORK_ROOT / f"sd_{uuid.uuid4().hex[:10]}" + work.mkdir(parents=True, exist_ok=False) + 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 _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") + 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) + try: + res = json.loads(result.read_text(encoding="utf-8-sig")) + 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(bool(res.get("ok")), res.get("message", ""), res.get("drive", ""), log) + + +def _eject_windows(mount: str) -> bool: + """Flush, lock, dismount and eject the card - what 'Safely Remove' does. + + Not Shell.Application InvokeVerb('Eject'): it fails silently because the + verb is named 'E&ject' (and is localised on non-English Windows). + """ + import ctypes + from ctypes import wintypes + + 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] + GENERIC_RW, SHARE_RW, OPEN_EXISTING = 0xC0000000, 0x3, 3 + FSCTL_LOCK_VOLUME, FSCTL_DISMOUNT_VOLUME = 0x00090018, 0x00090020 + IOCTL_STORAGE_MEDIA_REMOVAL, IOCTL_STORAGE_EJECT_MEDIA = 0x002D4804, 0x002D4808 + handle = k32.CreateFileW(f"\\\\.\\{mount[0]}:", GENERIC_RW, SHARE_RW, None, OPEN_EXISTING, 0, None) + if handle in (None, wintypes.HANDLE(-1).value): + return False + ret = wintypes.DWORD() + + def ioctl(code: int, buf=None, size: int = 0) -> bool: + return bool(k32.DeviceIoControl(handle, code, buf, size, None, 0, ctypes.byref(ret), None)) + + try: + k32.FlushFileBuffers(handle) + for _ in range(10): # Explorer/indexer may hold it briefly + if ioctl(FSCTL_LOCK_VOLUME): + break + time.sleep(0.5) + else: + return False + if not ioctl(FSCTL_DISMOUNT_VOLUME): + return False + allow = ctypes.c_bool(False) # PREVENT_MEDIA_REMOVAL{FALSE} + ioctl(IOCTL_STORAGE_MEDIA_REMOVAL, ctypes.byref(allow), 1) + ioctl(IOCTL_STORAGE_EJECT_MEDIA) # card readers may not support it; dismount suffices + return True + finally: + k32.CloseHandle(handle) + + +# ------------------------------------------------------------------- macOS + +def _format_macos(disk: Disk, log: list[str]) -> Outcome: + log.append(f"diskutil eraseDisk FAT32 {VOLUME_LABEL} MBRFormat {disk.id}") + proc = subprocess.run(["diskutil", "eraseDisk", "FAT32", VOLUME_LABEL, "MBRFormat", disk.id], + capture_output=True, text=True, timeout=600) + log += [l for l in proc.stdout.splitlines() if l.strip()] + if proc.returncode != 0: + return Outcome(False, "Formatting failed: " + (proc.stderr.strip() or proc.stdout.strip()[-200:]) + + " (is the card's write-protect switch on?)", log=log) + part = f"{disk.id}s1" + for _ in range(20): # wait for the volume to mount + info = plistlib.loads(subprocess.run(["diskutil", "info", "-plist", part], capture_output=True).stdout or b"") + if info.get("MountPoint"): + return Outcome(True, "Formatted FAT32 (MBR).", info["MountPoint"], log) + time.sleep(0.5) + return Outcome(False, "Formatted, but the card did not mount.", log=log) + + +def _eject_macos(disk_id: str) -> bool: + return subprocess.run(["diskutil", "eject", disk_id], capture_output=True, timeout=60).returncode == 0 + + +# ------------------------------------------------------------------ common + +def write_env(mount: str, env: EnvConfig) -> Outcome: + """Write ENV.TXT (8.3 name, LF, no BOM), keep other files (buffered data), read back.""" + problems = [i.text for i in env.problems() if i.level == "error"] + if problems: + return Outcome(False, "Settings are not valid: " + " ".join(problems)) + root = Path(mount) + if not root.is_dir(): + return Outcome(False, f"Card not found at {mount}.") + old = find_case_insensitive(root, ENV_NAME) + if old is not None and old.name != ENV_NAME: + old.unlink() # replace env.txt / Env.txt variants + target = root / ENV_NAME + data = env.render() + with open(target, "wb") as fh: + fh.write(data) + fh.flush() + os.fsync(fh.fileno()) + if target.read_bytes() != data: + return Outcome(False, "ENV.TXT read-back does not match - the card may be faulty.") + check = parse_env(target.read_bytes()) + if not check.ok: + return Outcome(False, "ENV.TXT was written but does not validate: " + + " ".join(i.text for i in check.issues if i.level == "error")) + return Outcome(True, f"ENV.TXT written and verified on {mount}.", mount) + + +def format_and_write(disk: Disk, env: EnvConfig, eject_card: bool = True) -> Outcome: + problems = [i.text for i in env.problems() if i.level == "error"] + if problems: + return Outcome(False, "Settings are not valid: " + " ".join(problems)) + log: list[str] = [] + if sys.platform == "win32": + fmt = _format_windows(disk, log) + elif sys.platform == "darwin": + fmt = _format_macos(disk, log) + else: + return Outcome(False, "SD provisioning is supported on Windows and macOS only.") + if not fmt.ok: + return fmt + wr = write_env(fmt.mount, env) + log.append(wr.message) + if not wr.ok: + return Outcome(False, wr.message, fmt.mount, log) + if eject_card: + log.append("Card ejected - you can remove it." if eject(fmt.mount, disk) else + "Could not eject the card automatically - use 'Safely remove' before pulling it out.") + return Outcome(True, "Card ready: formatted FAT32 and ENV.TXT verified.", fmt.mount, log) + + +def eject(mount: str, disk: Disk | None = None) -> bool: + """True if the card was dismounted and can be pulled out safely.""" + if sys.platform == "win32": + return _eject_windows(mount) + if sys.platform == "darwin" and disk is not None: + return _eject_macos(disk.id) + return False + + +def check_card(mount: str, filesystem: str = "") -> CardReport: + """What the box would find on this card (values masked in the UI).""" + root = Path(mount) + problems: list[str] = [] + fs = filesystem.upper() + if fs and "FAT32" not in fs and fs not in ("FAT", "MS-DOS FAT32", "MSDOS"): + problems.append(f"File system is {filesystem}; the box needs FAT32 - re-format the card.") + env_file = find_case_insensitive(root, ENV_NAME) + env = parse_env(env_file.read_bytes()) if env_file else None + if env_file is None: + problems.append("No ENV.TXT on the card - the box stops at start-up (white LED).") + elif len(env_file.name.split(".")[0]) > 8: + problems.append("ENV.TXT has a long name variant; use exactly ENV.TXT.") + buf = find_case_insensitive(root, BUFFER_NAME) + lines = None + if buf is not None: + with open(buf, "rb") as fh: + lines = sum(1 for ln in fh if ln.strip()) + return CardReport(mount, filesystem, env, lines, problems) diff --git a/kota_tech/service/__init__.py b/kota_tech/service/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/kota_tech/service/report.py b/kota_tech/service/report.py new file mode 100644 index 0000000..085a7b8 --- /dev/null +++ b/kota_tech/service/report.py @@ -0,0 +1,101 @@ +"""Hand-off of a diagnosis to the KOTA service system (Google Apps Script). + +The tool opens the box's service page in the technician's browser: + #kota-report= +The service page (v1.1.0+) reads the fragment and pre-fills status and note; +the technician checks it, adds the location and saves - so the report is +authored by their own Google login and the tool holds no credentials. +The fragment is never sent to any server (redirect host or Apps Script). +""" + +from __future__ import annotations + +import base64 +import json +import re + +from .. import __version__ +from ..diag.engine import Report, Severity + +DEFAULT_BASE = "https://service.kota-app.com/" +NOTE_LIMIT = 4000 # the service form accepts 5000; leave room for the technician + +# Status names exactly as in the service system (Code.gs STATUSES). +OK = "Działa poprawnie" +CHECK = "Do sprawdzenia" +IN_SERVICE = "W trakcie serwisu" +HARDWARE = "Błąd hardware" +SOFTWARE = "Błąd software" +POWER = "Problem z zasilaniem" +CONNECTIVITY = "Problem z łącznością" + +# Root-finding rule id prefix -> service status (first match wins). +_STATUS_BY_RULE = [ + ("reboot.brownout", POWER), ("reboot.power", POWER), ("tech.power.", POWER), + ("sys.crash", SOFTWARE), ("reboot.watchdog", SOFTWARE), ("sys.out_of_resources", SOFTWARE), + ("influx.bad_request", SOFTWARE), ("ota.selfcheck_failed", SOFTWARE), ("tech.quick.hung", SOFTWARE), + ("rtc.init_failed", HARDWARE), ("rtc.lost_after_sync", HARDWARE), ("i2c.", HARDWARE), ("sensor.", HARDWARE), + ("sd.", HARDWARE), ("sen66.", HARDWARE), ("bme688.", HARDWARE), ("as7341.", HARDWARE), + ("mic.read_fail", HARDWARE), ("mic.silent", HARDWARE), ("tech.rtc_i2c.", HARDWARE), ("tech.sensors.", HARDWARE), + ("tech.mic.mic_faulty", HARDWARE), ("tech.mic.mic_switch_faulty", HARDWARE), ("tech.white.sd_", HARDWARE), + ("matter.", CONNECTIVITY), ("net.", CONNECTIVITY), ("wifi.", CONNECTIVITY), ("influx.", CONNECTIVITY), + ("ota.", CONNECTIVITY), ("tech.network.", CONNECTIVITY), + ("rtc.clock_not_running", CHECK), ("reboot.", CHECK), +] +_MAC = re.compile(r"^[0-9A-F]{12}$") + + +def normalize_mac(mac: str) -> str: + return re.sub(r"[^0-9A-Fa-f]", "", mac or "").upper() + + +def service_status(report: Report) -> str: + """Suggested status; the technician can change it in the form.""" + problems = [rc for rc in report.roots if rc.severity >= Severity.WARNING] + if not problems: + return OK if any(m == "data_uploaded" and t is not None for m, _, t in report.milestones) else CHECK + rid = problems[0].finding.rule_id + for prefix, status in _STATUS_BY_RULE: + if rid.startswith(prefix): + return status + return CHECK + + +def service_note(report: Report, limit: int = NOTE_LIMIT) -> str: + lines = [f"[KOTA Technician Tool {__version__}] {report.verdict}", + f"Firmware: {report.firmware.describe()}", + f"Expected LED: {report.led.capitalize()}"] + reached = [label for _, label, t in report.milestones if t is not None] + missing = [label for _, label, t in report.milestones if t is None] + if reached: + lines.append("OK: " + ", ".join(reached)) + if missing: + lines.append("Not reached: " + ", ".join(missing)) + for rc in report.roots: + f = rc.finding + lines.append(f"- [{f.severity.label}] {f.title}") + for c in rc.consequences: + lines.append(f" -> {c.title}") + if report.technician: + lines.append("Technician checks:") + lines += [f" * {t}" for t in report.technician[-12:]] + text = "\n".join(lines) + return text if len(text) <= limit else text[:limit - 20].rstrip() + "\n...(shortened)" + + +def report_url(mac: str, report: Report, base: str = DEFAULT_BASE) -> str: + mac = normalize_mac(mac) + if not _MAC.match(mac): + raise ValueError(f"not a box MAC address: {mac!r}") + payload = {"v": 1, "mac": mac, "status": service_status(report), "note": service_note(report), + "tool": f"v{__version__}"} + raw = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + frag = base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=") + base = base if base.endswith("/") else base + "/" + return f"{base}{mac}#kota-report={frag}" + + +def decode_fragment(url: str) -> dict: + """Inverse of report_url (used by tests and for troubleshooting).""" + frag = url.split("#kota-report=", 1)[1] + return json.loads(base64.urlsafe_b64decode(frag + "=" * (-len(frag) % 4)).decode("utf-8")) diff --git a/kota_tech/ui/flash_tab.py b/kota_tech/ui/flash_tab.py index 1253c7e..9636e20 100644 --- a/kota_tech/ui/flash_tab.py +++ b/kota_tech/ui/flash_tab.py @@ -1,4 +1,4 @@ -"""Flash firmware tab: official image only, App-only by default.""" +"""Flash firmware tab: factory image or imported developer builds; App-only by default.""" from __future__ import annotations @@ -10,6 +10,7 @@ from PySide6.QtCore import QObject, Qt, QTimer, Signal from PySide6.QtWidgets import ( QButtonGroup, QCheckBox, + QFileDialog, QComboBox, QGroupBox, QHBoxLayout, @@ -24,7 +25,7 @@ from PySide6.QtWidgets import ( ) from ..core.session import SESSIONS_DIR -from ..flash.image import FirmwareImage, ImageError, official_images +from ..flash.image import FirmwareImage, FirmwareLibrary, ImageError, factory_images from ..flash.plan import Mode, build_plan from ..flash.runner import FlashRunner, Result from ..serialio.ports import PortInfo, list_serial_ports @@ -71,13 +72,26 @@ class FlashTab(QWidget): bl.addWidget(self.box_fw) # 2. firmware - fwbox = QGroupBox("2. Firmware (official images only)") + fwbox = QGroupBox("2. Firmware") self.image_combo = QComboBox(minimumWidth=360) + self.import_btn = QPushButton("Import firmware file...") + self.import_btn.setToolTip("Add a newer build the firmware team sent you (.bin)") + self.import_btn.clicked.connect(self._import) + self.remove_btn = QPushButton("Remove from list") + self.remove_btn.clicked.connect(self._remove) self.image_info = QLabel(wordWrap=True) self.image_info.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse) fl = QVBoxLayout(fwbox) fl.addWidget(self.image_combo) + brow = QHBoxLayout() + brow.addWidget(self.import_btn) + brow.addWidget(self.remove_btn) + brow.addStretch(1) + fl.addLayout(brow) fl.addWidget(self.image_info) + self._library = FirmwareLibrary() + self._reference: FirmwareImage | None = None + self._problems: list[str] = [] # 3. mode modebox = QGroupBox("3. What to write") @@ -151,17 +165,55 @@ class FlashTab(QWidget): self._show_result("", None) # ------------------------------------------------------------ setup - def _load_images(self) -> None: + def _load_images(self, select_sha: str | None = None) -> None: + self._problems = [] try: - self._images = official_images() + factory = factory_images() except ImageError as exc: - self._images = [] - self.image_info.setText(f"Firmware file problem: {exc}") + factory = [] + self._problems.append(f"factory image: {exc}") + self._reference = factory[0] if factory else None + dev, problems = self._library.images(self._reference) + 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) + self._images = dev + factory + self.image_combo.blockSignals(True) self.image_combo.clear() for img in self._images: - self.image_combo.addItem(img.name) + self.image_combo.addItem(img.label()) if not self._images: - self.image_combo.addItem("No official firmware image found in the firmware folder") + self.image_combo.addItem("No firmware image available - import one") + idx = next((i for i, im in enumerate(self._images) if im.sha256 == select_sha), None) + if idx is None: + idx = len(dev) if factory else 0 # default: the factory image + self.image_combo.setCurrentIndex(idx) + self.image_combo.blockSignals(False) + self._update() + + def _import(self) -> None: + path, _ = QFileDialog.getOpenFileName(self, "Import firmware from the firmware team", str(Path.home()), + "Firmware (*.bin);;All files (*)") + if not path: + return + try: + img = self._library.import_file(Path(path), self._reference) + except (ImageError, OSError) as exc: + QMessageBox.warning(self, "Cannot use this file", f"This file cannot be flashed:\n\n{exc}") + return + self._load_images(select_sha=img.sha256) + kind = "complete image" if img.kind == "merged" else "app only" + QMessageBox.information( + self, "Firmware imported", + f"{img.name}\nbuild {img.app.elf_sha[:9]}, {kind}\n\nFile fingerprint (SHA-256):\n{img.sha256}\n\n" + "Compare it with the fingerprint the firmware team sent you before flashing.") + + def _remove(self) -> None: + img = self._selected_image() + if img and img.is_developer and QMessageBox.question( + self, "Remove firmware", f"Remove {img.name} from this computer?") == QMessageBox.StandardButton.Yes: + self._library.remove(img.sha256) + self._load_images() def refresh_ports(self) -> None: current = self.port_combo.currentData() @@ -189,24 +241,27 @@ class FlashTab(QWidget): def _update(self) -> None: img = self._selected_image() + info = "" if img: a = img.app - self.image_info.setText( - f"{a.project} {a.version}, built {a.built}, ESP-IDF {a.idf}
" - f"build {a.elf_sha[:9]} · file SHA-256 {img.sha256[:12]}... verified
" - f"{img.notes}") - fw = self.serial_tab.engine.firmware - if fw.elf_sha and img: - same = fw.elf_sha.startswith(img.app.elf_sha[:9]) or img.app.elf_sha.startswith(fw.elf_sha) - msg = f"Box currently runs: {fw.describe()}" - if same: - msg += "
It already has this firmware. Re-flash only if you suspect the firmware is damaged." - else: - msg += ("
The box runs a different build (probably updated over OTA). " - "Flashing puts the older factory image back; the box updates itself again once online.") - else: - msg = "Firmware on the box: unknown - connect in the Serial monitor and press Reset & capture to see it." - self.box_fw.setText(msg) + origin = ("Factory image (bundled with the tool)" if not img.is_developer else + "Developer build - only flash builds you received " + "from the firmware team") + kind = "complete image (bootloader + layout + app)" if img.kind == "merged" else "app only" + info = (f"{origin}
{a.project} {a.version}, built {a.built}, ESP-IDF {a.idf}, {kind}
" + f"build {a.elf_sha[:9]} · file SHA-256 {img.sha256[:16]}... " + f"· checksum OK" + (f"
{img.notes}" if img.notes else "")) + if self._problems: + info += ("
Not offered: " + + "; ".join(self._problems) + "") + self.image_info.setText(info) + self.remove_btn.setEnabled(bool(img and img.is_developer) and self._runner is None) + # An app-only file has no bootloader/layout: Full mode is impossible. + full_ok = bool(img and img.kind == "merged") + self.rb_full.setEnabled(full_ok and self._runner is None) + if not full_ok and self.rb_full.isChecked(): + self.rb_app.setChecked(True) + self.box_fw.setText(self._box_text(img)) self.confirm_full.setVisible(self.rb_full.isChecked()) ok = bool(img and self._selected_port() and self._runner is None @@ -224,6 +279,23 @@ class FlashTab(QWidget): else: self.plan_label.setText("Select a sensorbox and a firmware image.") + def _box_text(self, img: FirmwareImage | None) -> str: + fw = self.serial_tab.engine.firmware + if not fw.elf_sha: + return "Firmware on the box: unknown - connect in the Serial monitor and press Reset & capture to see it." + msg = f"Box currently runs: {fw.describe()}" + (f", built {fw.built}" if fw.built else "") + if not img: + return msg + if fw.elf_sha.startswith(img.app.elf_sha[:9]) or img.app.elf_sha.startswith(fw.elf_sha): + return msg + "
It already has this firmware. Re-flash only if you suspect the firmware is damaged." + box_built, img_built = _parse_built(fw.built), img.app.built_at + if box_built and img_built and img_built > box_built: + return msg + f"
The selected firmware is newer (built {img.app.built})." + if box_built and img_built and img_built < box_built: + return msg + ("
The selected firmware is OLDER than the box's (the box was " + "probably updated over OTA). The box updates itself again once online.") + return msg + "
The box runs a different build." + # ------------------------------------------------------------ flash def _start(self) -> None: img, port = self._selected_image(), self._selected_port() @@ -256,7 +328,7 @@ class FlashTab(QWidget): self._runner.start() def _set_busy(self, busy: bool) -> None: - for w in (self.port_combo, self.image_combo, self.rb_app, self.rb_full, self.backup, self.confirm_full): + for w in (self.port_combo, self.image_combo, self.rb_app, self.backup, self.confirm_full, self.import_btn): w.setEnabled(not busy) self._update() @@ -307,3 +379,13 @@ class FlashTab(QWidget): border, bg = colours[kind or "info"] self.result.setStyleSheet(f"QLabel {{ border:2px solid {border}; background:{bg}; border-radius:6px; }}") self.result.setText(html) + + +def _parse_built(s: str | None) -> dt.datetime | None: + """'Jun 29 2026 11:03:32' (app_init "Compile time") -> datetime.""" + if not s: + return None + try: + return dt.datetime.strptime(" ".join(s.split()), "%b %d %Y %H:%M:%S") + except ValueError: + return None diff --git a/kota_tech/ui/main_window.py b/kota_tech/ui/main_window.py index 0076a87..ec57664 100644 --- a/kota_tech/ui/main_window.py +++ b/kota_tech/ui/main_window.py @@ -1,358 +1,408 @@ -from __future__ import annotations - -from pathlib import Path - -from PySide6.QtCore import QCoreApplication, QObject, Qt, QTimer, QUrl, Signal -from PySide6.QtGui import QColor, QDesktopServices, QFont, QGuiApplication, QTextCharFormat, QTextCursor -from PySide6.QtWidgets import ( - QCheckBox, - QComboBox, - QFileDialog, - QHBoxLayout, - QLabel, - QMainWindow, - QPlainTextEdit, - QPushButton, - QSplitter, - QTabWidget, - QTextBrowser, - QVBoxLayout, - QWidget, -) - -from .. import __version__ -from ..core.logfile import read_log_file -from ..core.logline import LineKind, LogLine, host_event -from ..core.session import SessionLog -from ..diag.engine import DiagnosticsEngine -from ..diag.guides import GuideRunner -from ..serialio.manager import SerialManager, State -from ..serialio.ports import PortInfo, list_serial_ports -from .cheatsheet import cheatsheet_html -from .diag_panel import DiagnosticsPanel -from .flash_tab import FlashTab -from .guide_panel import GuidePanel - -MAX_LINES = 20000 -LEVEL_FILTERS = {"All lines": None, "Warnings + errors": {"W", "E"}, "Errors only": {"E"}} -COLORS = { - "E": "#ff6b6b", - "W": "#f5c451", - "I": "#d4d4d4", - LineKind.BOOT: "#6fc3ff", - LineKind.RESET: "#6fc3ff", - LineKind.PANIC: "#ff4dd2", - LineKind.HOST: "#7ee07e", - LineKind.RAW: "#a8a8a8", -} - - -class _Bridge(QObject): - """Marshals SerialManager callbacks (reader thread) onto the UI thread.""" - - line = Signal(object) - state = Signal(object, str) - - -class SerialTab(QWidget): - def __init__(self, parent: QWidget | None = None): - super().__init__(parent) - self._bridge = _Bridge() - self._bridge.line.connect(self._on_line) - self._bridge.state.connect(self._on_state) - self._manager = SerialManager(self._bridge.line.emit, self._bridge.state.emit) - self._session = SessionLog() - self._ports: list[PortInfo] = [] - self._counts = {"E": 0, "W": 0, "boots": 0} - self._engine = DiagnosticsEngine() - self._diag_dirty = True - self._runner = GuideRunner(self._engine, on_action=self._guide_action, on_log=self._guide_log) - - # --- controls - self.port_combo = QComboBox(minimumWidth=320) - self.all_ports = QCheckBox("Show all ports") - self.refresh_btn = QPushButton("Refresh") - self.connect_btn = QPushButton("Connect") - self.reset_btn = QPushButton("Reset && capture") - self.reset_btn.setToolTip("Reboot the sensorbox and record the log from the first boot line") - self.filter_combo = QComboBox() - self.filter_combo.addItems(LEVEL_FILTERS) - self.clear_btn = QPushButton("Clear view") - self.save_btn = QPushButton("Save view as...") - self.folder_btn = QPushButton("Open logs folder") - self.analyze_btn = QPushButton("Analyze log file...") - self.analyze_btn.setToolTip("Load a saved session log or serial capture and diagnose it") - self.copy_btn = QPushButton("Copy report") - self.copy_btn.setToolTip("Copy the diagnosis as text (e.g. to send to the firmware team)") - - row1 = QHBoxLayout() - for w in (QLabel("Port:"), self.port_combo, self.all_ports, self.refresh_btn, - self.connect_btn, self.reset_btn): - row1.addWidget(w) - row1.addStretch(1) - row2 = QHBoxLayout() - for w in (QLabel("Show:"), self.filter_combo, self.clear_btn, self.save_btn, self.folder_btn, - self.analyze_btn, self.copy_btn): - row2.addWidget(w) - row2.addStretch(1) - - self.view = QPlainTextEdit(readOnly=True) - self.view.setMaximumBlockCount(MAX_LINES) - self.view.setLineWrapMode(QPlainTextEdit.LineWrapMode.NoWrap) - mono = QFont("Menlo" if QFont("Menlo").exactMatch() else "Consolas") - mono.setStyleHint(QFont.StyleHint.Monospace) - mono.setPointSize(10) - self.view.setFont(mono) - self.view.setStyleSheet("QPlainTextEdit { background:#1e1e1e; color:#d4d4d4; }") - - self.diag = DiagnosticsPanel() - self.guide = GuidePanel(self._runner, self._engine) - self.side_tabs = QTabWidget() - self.side_tabs.addTab(self.diag, "Diagnosis") - self.side_tabs.addTab(self.guide, "Guided troubleshooting") - split = QSplitter(Qt.Orientation.Horizontal) - split.addWidget(self.view) - split.addWidget(self.side_tabs) - split.setStretchFactor(0, 3) - split.setStretchFactor(1, 2) - split.setSizes([700, 480]) - - self.status = QLabel() - layout = QVBoxLayout(self) - layout.addLayout(row1) - layout.addLayout(row2) - layout.addWidget(split, 1) - layout.addWidget(self.status) - - # --- wiring - self.refresh_btn.clicked.connect(self.refresh_ports) - self.all_ports.toggled.connect(self.refresh_ports) - self.connect_btn.clicked.connect(self._toggle_connect) - self.reset_btn.clicked.connect(self._manager.reset_board) - self.clear_btn.clicked.connect(self.view.clear) - self.save_btn.clicked.connect(self._save_view) - self.folder_btn.clicked.connect(self._open_folder) - self.analyze_btn.clicked.connect(self._analyze_file) - self.copy_btn.clicked.connect(self._copy_report) - - # Report refresh is throttled; lines can arrive at >100/s. - self._diag_timer = QTimer(self, interval=400) - self._diag_timer.timeout.connect(self._refresh_diag) - self._diag_timer.start() - - # Auto-detect a box being plugged in while idle. - self._poll = QTimer(self, interval=2000) - self._poll.timeout.connect(self._auto_refresh) - self._poll.start() - - self.refresh_ports() - self._on_state(State.DISCONNECTED, "") - - # ---- ports ------------------------------------------------------- - def refresh_ports(self) -> None: - current = self.port_combo.currentData() - self._ports = list_serial_ports(include_all=self.all_ports.isChecked()) - self.port_combo.clear() - for p in self._ports: - self.port_combo.addItem(p.label, p.device) - if not self._ports: - self.port_combo.addItem("No sensorbox found - check USB cable and power switch", None) - idx = self.port_combo.findData(current) - self.port_combo.setCurrentIndex(max(idx, 0)) - - def _auto_refresh(self) -> None: - if self._manager.is_running: - return - devices = [p.device for p in list_serial_ports(self.all_ports.isChecked())] - if devices != [p.device for p in self._ports]: - self.refresh_ports() - - def _selected_port(self) -> PortInfo | None: - dev = self.port_combo.currentData() - return next((p for p in self._ports if p.device == dev), None) - - # ---- connection -------------------------------------------------- - def _toggle_connect(self) -> None: - if self._manager.is_running: - # Session file is closed in _on_state(DISCONNECTED), after the - # reader thread's final queued lines have been written. - self._manager.disconnect() - return - port = self._selected_port() - if port is None: - return - self._connect(port) - - def _connect(self, port: PortInfo) -> None: - self._counts = {"E": 0, "W": 0, "boots": 0} - self._engine.reset() - self._runner.cancel() - self._diag_dirty = True - self._session.start(port.mac or port.device) - self._manager.connect(port) - - # ---- hand-off with the flash tab ------------------------------------ - @property - def engine(self) -> DiagnosticsEngine: - return self._engine - - def release_port(self) -> None: - """Stop monitoring so esptool can open the port exclusively.""" - if self._manager.is_running: - self._manager.disconnect() # joins the reader thread - QCoreApplication.processEvents() # deliver its queued final lines/state now - - def connect_port(self, port: PortInfo, capture: bool = True) -> None: - """Monitor `port`; with capture, restart the box once connected to record its boot.""" - self.release_port() - idx = self.port_combo.findData(port.device) - if idx < 0: - self.port_combo.addItem(port.label, port.device) - self._ports.append(port) - idx = self.port_combo.count() - 1 - self.port_combo.setCurrentIndex(idx) - self._reset_on_connect = capture - self._connect(port) - - def _on_state(self, state: State, detail: str) -> None: - running = state != State.DISCONNECTED - if not running and self._manager.is_running: - return # stale event from a session that was already replaced - if state == State.CONNECTED and getattr(self, "_reset_on_connect", False): - self._reset_on_connect = False - QTimer.singleShot(300, self._manager.reset_board) - if not running and self._session.path is not None: - self._session.write_report(self._engine.report().to_text()) - self._session.stop() - self.connect_btn.setText("Disconnect" if running else "Connect") - self.reset_btn.setEnabled(state == State.CONNECTED) - self.port_combo.setEnabled(not running) - self.all_ports.setEnabled(not running) - self.refresh_btn.setEnabled(not running) - # Offline analysis would mix with the live session; disconnect first. - self.analyze_btn.setEnabled(not running) - self._state_text = f"{state.value}" + (f" ({detail})" if detail else "") - self._update_status() - - def _update_status(self) -> None: - sess = f" | Session log: {self._session.path}" if self._session.path else "" - c = self._counts - self.status.setText( - f"{getattr(self, '_state_text', '')} | boots seen: {c['boots']} " - f"errors: {c['E']} warnings: {c['W']}{sess}" - ) - - # ---- lines ------------------------------------------------------- - def _on_line(self, line: LogLine) -> None: - self._session.write(line) - self._engine.feed(line) - self._runner.feed(line) - self._diag_dirty = True - if line.kind == LineKind.BOOT: - self._counts["boots"] += 1 - if line.level in ("E", "W"): - self._counts[line.level] += 1 - self._update_status() - - allowed = LEVEL_FILTERS[self.filter_combo.currentText()] - if allowed is not None and line.kind != LineKind.HOST and not ( - line.level in allowed or line.kind in (LineKind.PANIC, LineKind.BOOT, LineKind.RESET) - ): - return - self._append(line) - - def _append(self, line: LogLine) -> None: - sb = self.view.verticalScrollBar() - at_bottom = sb.value() >= sb.maximum() - 4 - fmt = QTextCharFormat() - fmt.setForeground(QColor(COLORS.get(line.level) or COLORS.get(line.kind, "#d4d4d4"))) - if line.kind in (LineKind.PANIC, LineKind.BOOT, LineKind.HOST): - fmt.setFontWeight(QFont.Weight.Bold) - cur = QTextCursor(self.view.document()) - cur.movePosition(QTextCursor.MoveOperation.End) - if not self.view.document().isEmpty(): - cur.insertBlock() - cur.insertText(line.text, fmt) - if at_bottom: - sb.setValue(sb.maximum()) - - # ---- diagnostics ------------------------------------------------- - def _refresh_diag(self) -> None: - if self._diag_dirty: - self._diag_dirty = False - self.diag.show_report(self._engine.report()) - self.guide.refresh() # countdowns and timeouts need a steady tick - - # ---- guided troubleshooting hooks ------------------------------------ - def _guide_action(self, action: str) -> bool: - if action == "reset" and self._manager.state == State.CONNECTED: - self._manager.reset_board() - return True - return False - - def _guide_log(self, text: str) -> None: - # Technician answers go into the session log and the log view. - line = host_event(text) - self._session.write(line) - self._append(line) - self._diag_dirty = True - - def _copy_report(self) -> None: - QGuiApplication.clipboard().setText(self._engine.report().to_text()) - - def _analyze_file(self) -> None: - path, _ = QFileDialog.getOpenFileName( - self, "Analyze log file", str(self._session.root), "Logs (*.log *.txt);;All files (*)") - if not path: - return - lines = read_log_file(Path(path)) - self.view.clear() - self._engine.reset() - self._runner.cancel() - self._counts = {"E": 0, "W": 0, "boots": 0} - for line in lines: - self._engine.feed(line) - if line.kind == LineKind.BOOT: - self._counts["boots"] += 1 - if line.level in ("E", "W"): - self._counts[line.level] += 1 - self._append(line) - self._state_text = f"Analyzed file: {Path(path).name} ({len(lines)} lines)" - self._update_status() - self._diag_dirty = True - self._refresh_diag() - - # ---- misc -------------------------------------------------------- - def _save_view(self) -> None: - path, _ = QFileDialog.getSaveFileName(self, "Save log view", "sensorbox_log.txt", "Text (*.txt)") - if path: - with open(path, "w", encoding="utf-8") as fh: - fh.write(self.view.toPlainText()) - - def _open_folder(self) -> None: - self._session.root.mkdir(parents=True, exist_ok=True) - QDesktopServices.openUrl(QUrl.fromLocalFile(str(self._session.root))) - - def shutdown(self) -> None: - self._manager.disconnect() - self._session.stop() - - -class MainWindow(QMainWindow): - def __init__(self): - super().__init__() - self.setWindowTitle(f"KOTA Sensorbox Technician Tool v{__version__}") - self.resize(1200, 800) - tabs = QTabWidget() - guide = QTextBrowser(openExternalLinks=True) - guide.setHtml(cheatsheet_html()) - self.serial_tab = SerialTab() - self.flash_tab = FlashTab(self.serial_tab) - tabs.addTab(guide, "Quick start") - tabs.addTab(self.serial_tab, "Serial monitor") - tabs.addTab(self.flash_tab, "Flash firmware") - self.setCentralWidget(tabs) - - def closeEvent(self, event): # noqa: N802 (Qt naming) - self.serial_tab.shutdown() - super().closeEvent(event) +from __future__ import annotations + +from pathlib import Path + +from PySide6.QtCore import QCoreApplication, QObject, QSettings, Qt, QTimer, QUrl, Signal +from PySide6.QtGui import QColor, QDesktopServices, QFont, QGuiApplication, QTextCharFormat, QTextCursor +from PySide6.QtWidgets import ( + QCheckBox, + QComboBox, + QFileDialog, + QHBoxLayout, + QInputDialog, + QLabel, + QMainWindow, + QMessageBox, + QPlainTextEdit, + QPushButton, + QSplitter, + QTabWidget, + QTextBrowser, + QVBoxLayout, + QWidget, +) + +from .. import __version__ +from ..core.logfile import read_log_file +from ..core.logline import LineKind, LogLine, host_event +from ..core.session import SessionLog +from ..diag.engine import DiagnosticsEngine +from ..diag.guides import GuideRunner +from ..serialio.manager import SerialManager, State +from ..serialio.ports import PortInfo, list_serial_ports +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 .guide_panel import GuidePanel +from .sd_tab import SdTab + +MAX_LINES = 20000 +LEVEL_FILTERS = {"All lines": None, "Warnings + errors": {"W", "E"}, "Errors only": {"E"}} +COLORS = { + "E": "#ff6b6b", + "W": "#f5c451", + "I": "#d4d4d4", + LineKind.BOOT: "#6fc3ff", + LineKind.RESET: "#6fc3ff", + LineKind.PANIC: "#ff4dd2", + LineKind.HOST: "#7ee07e", + LineKind.RAW: "#a8a8a8", +} + + +class _Bridge(QObject): + """Marshals SerialManager callbacks (reader thread) onto the UI thread.""" + + line = Signal(object) + state = Signal(object, str) + + +class SerialTab(QWidget): + def __init__(self, parent: QWidget | None = None): + super().__init__(parent) + self._bridge = _Bridge() + self._bridge.line.connect(self._on_line) + self._bridge.state.connect(self._on_state) + self._manager = SerialManager(self._bridge.line.emit, self._bridge.state.emit) + self._session = SessionLog() + self._ports: list[PortInfo] = [] + self._counts = {"E": 0, "W": 0, "boots": 0} + self._engine = DiagnosticsEngine() + self._diag_dirty = True + self._runner = GuideRunner(self._engine, on_action=self._guide_action, on_log=self._guide_log) + + # --- controls + self.port_combo = QComboBox(minimumWidth=320) + self.all_ports = QCheckBox("Show all ports") + self.refresh_btn = QPushButton("Refresh") + self.connect_btn = QPushButton("Connect") + self.reset_btn = QPushButton("Reset && capture") + self.reset_btn.setToolTip("Reboot the sensorbox and record the log from the first boot line") + self.filter_combo = QComboBox() + self.filter_combo.addItems(LEVEL_FILTERS) + self.clear_btn = QPushButton("Clear view") + self.save_btn = QPushButton("Save view as...") + self.folder_btn = QPushButton("Open logs folder") + self.analyze_btn = QPushButton("Analyze log file...") + self.analyze_btn.setToolTip("Load a saved session log or serial capture and diagnose it") + self.copy_btn = QPushButton("Copy report") + self.copy_btn.setToolTip("Copy the diagnosis as text (e.g. to send to the firmware team)") + self.service_btn = QPushButton("Service report...") + self.service_btn.setToolTip("Open this box's page in the KOTA service system with the diagnosis " + "filled in; check it and press Save there") + self._settings = QSettings("KOTA", "TechnicianTool") + self.auto_service = QCheckBox("Open service report on disconnect") + self.auto_service.setChecked(self._settings.value("service/auto", False, type=bool)) + self.auto_service.toggled.connect(lambda on: self._settings.setValue("service/auto", on)) + self._active_port: PortInfo | None = None + + row1 = QHBoxLayout() + for w in (QLabel("Port:"), self.port_combo, self.all_ports, self.refresh_btn, + self.connect_btn, self.reset_btn): + row1.addWidget(w) + row1.addStretch(1) + row2 = QHBoxLayout() + for w in (QLabel("Show:"), self.filter_combo, self.clear_btn, self.save_btn, self.folder_btn, + self.analyze_btn, self.copy_btn, self.service_btn, self.auto_service): + row2.addWidget(w) + row2.addStretch(1) + + self.view = QPlainTextEdit(readOnly=True) + self.view.setMaximumBlockCount(MAX_LINES) + self.view.setLineWrapMode(QPlainTextEdit.LineWrapMode.NoWrap) + mono = QFont("Menlo" if QFont("Menlo").exactMatch() else "Consolas") + mono.setStyleHint(QFont.StyleHint.Monospace) + mono.setPointSize(10) + self.view.setFont(mono) + self.view.setStyleSheet("QPlainTextEdit { background:#1e1e1e; color:#d4d4d4; }") + + self.diag = DiagnosticsPanel() + self.guide = GuidePanel(self._runner, self._engine) + self.side_tabs = QTabWidget() + self.side_tabs.addTab(self.diag, "Diagnosis") + self.side_tabs.addTab(self.guide, "Guided troubleshooting") + split = QSplitter(Qt.Orientation.Horizontal) + split.addWidget(self.view) + split.addWidget(self.side_tabs) + split.setStretchFactor(0, 3) + split.setStretchFactor(1, 2) + split.setSizes([700, 480]) + + self.status = QLabel() + layout = QVBoxLayout(self) + layout.addLayout(row1) + layout.addLayout(row2) + layout.addWidget(split, 1) + layout.addWidget(self.status) + + # --- wiring + self.refresh_btn.clicked.connect(self.refresh_ports) + self.all_ports.toggled.connect(self.refresh_ports) + self.connect_btn.clicked.connect(self._toggle_connect) + self.reset_btn.clicked.connect(self._manager.reset_board) + self.clear_btn.clicked.connect(self.view.clear) + self.save_btn.clicked.connect(self._save_view) + self.folder_btn.clicked.connect(self._open_folder) + self.analyze_btn.clicked.connect(self._analyze_file) + self.copy_btn.clicked.connect(self._copy_report) + self.service_btn.clicked.connect(lambda: self._open_service_report(auto=False)) + + # Report refresh is throttled; lines can arrive at >100/s. + self._diag_timer = QTimer(self, interval=400) + self._diag_timer.timeout.connect(self._refresh_diag) + self._diag_timer.start() + + # Auto-detect a box being plugged in while idle. + self._poll = QTimer(self, interval=2000) + self._poll.timeout.connect(self._auto_refresh) + self._poll.start() + + self.refresh_ports() + self._on_state(State.DISCONNECTED, "") + + # ---- ports ------------------------------------------------------- + def refresh_ports(self) -> None: + current = self.port_combo.currentData() + self._ports = list_serial_ports(include_all=self.all_ports.isChecked()) + self.port_combo.clear() + for p in self._ports: + self.port_combo.addItem(p.label, p.device) + if not self._ports: + self.port_combo.addItem("No sensorbox found - check USB cable and power switch", None) + idx = self.port_combo.findData(current) + self.port_combo.setCurrentIndex(max(idx, 0)) + + def _auto_refresh(self) -> None: + if self._manager.is_running: + return + devices = [p.device for p in list_serial_ports(self.all_ports.isChecked())] + if devices != [p.device for p in self._ports]: + self.refresh_ports() + + def _selected_port(self) -> PortInfo | None: + dev = self.port_combo.currentData() + return next((p for p in self._ports if p.device == dev), None) + + # ---- connection -------------------------------------------------- + def _toggle_connect(self) -> None: + if self._manager.is_running: + # Session file is closed in _on_state(DISCONNECTED), after the + # reader thread's final queued lines have been written. + self._manager.disconnect() + return + port = self._selected_port() + if port is None: + return + self._connect(port) + + def _connect(self, port: PortInfo) -> None: + self._counts = {"E": 0, "W": 0, "boots": 0} + self._engine.reset() + self._runner.cancel() + self._diag_dirty = True + self._session.start(port.mac or port.device) + self._active_port = port + self._manager.connect(port) + + # ---- service system ------------------------------------------------- + def _service_base(self) -> str: + return str(self._settings.value("service/base", DEFAULT_BASE)) + + def _open_service_report(self, auto: bool) -> None: + mac = normalize_mac(self._engine.mac or (self._active_port.mac if self._active_port else "") or "") + if len(mac) != 12: + if not auto: + QMessageBox.information(self, "Service report", "The box's MAC address is not known yet. " + "Connect to the box and press Reset & capture first.") + return + url = report_url(mac, self._engine.report(), self._service_base()) + QDesktopServices.openUrl(QUrl(url)) + # Log that it happened, not the report itself (it is already in the session log). + self._guide_log(f"service report for {mac} opened in the browser - check it and press Save there") + + # ---- hand-off with the flash tab ------------------------------------ + @property + def engine(self) -> DiagnosticsEngine: + return self._engine + + def release_port(self) -> None: + """Stop monitoring so esptool can open the port exclusively.""" + if self._manager.is_running: + self._no_auto_service = True # a flash hand-off is not the end of a session + self._manager.disconnect() # joins the reader thread + QCoreApplication.processEvents() # deliver its queued final lines/state now + self._no_auto_service = False + + def connect_port(self, port: PortInfo, capture: bool = True) -> None: + """Monitor `port`; with capture, restart the box once connected to record its boot.""" + self.release_port() + idx = self.port_combo.findData(port.device) + if idx < 0: + self.port_combo.addItem(port.label, port.device) + self._ports.append(port) + idx = self.port_combo.count() - 1 + self.port_combo.setCurrentIndex(idx) + self._reset_on_connect = capture + self._connect(port) + + def _on_state(self, state: State, detail: str) -> None: + running = state != State.DISCONNECTED + if not running and self._manager.is_running: + return # stale event from a session that was already replaced + if state == State.CONNECTED and getattr(self, "_reset_on_connect", False): + self._reset_on_connect = False + QTimer.singleShot(300, self._manager.reset_board) + if not running and self._session.path is not None: + self._session.write_report(self._engine.report().to_text()) + if (self.auto_service.isChecked() and self._engine.lines + and not getattr(self, "_no_auto_service", False)): + self._open_service_report(auto=True) + self._session.stop() + self.connect_btn.setText("Disconnect" if running else "Connect") + self.reset_btn.setEnabled(state == State.CONNECTED) + self.port_combo.setEnabled(not running) + self.all_ports.setEnabled(not running) + self.refresh_btn.setEnabled(not running) + # Offline analysis would mix with the live session; disconnect first. + self.analyze_btn.setEnabled(not running) + self._state_text = f"{state.value}" + (f" ({detail})" if detail else "") + self._update_status() + + def _update_status(self) -> None: + sess = f" | Session log: {self._session.path}" if self._session.path else "" + c = self._counts + self.status.setText( + f"{getattr(self, '_state_text', '')} | boots seen: {c['boots']} " + f"errors: {c['E']} warnings: {c['W']}{sess}" + ) + + # ---- lines ------------------------------------------------------- + def _on_line(self, line: LogLine) -> None: + self._session.write(line) + self._engine.feed(line) + self._runner.feed(line) + self._diag_dirty = True + if line.kind == LineKind.BOOT: + self._counts["boots"] += 1 + if line.level in ("E", "W"): + self._counts[line.level] += 1 + self._update_status() + + allowed = LEVEL_FILTERS[self.filter_combo.currentText()] + if allowed is not None and line.kind != LineKind.HOST and not ( + line.level in allowed or line.kind in (LineKind.PANIC, LineKind.BOOT, LineKind.RESET) + ): + return + self._append(line) + + def _append(self, line: LogLine) -> None: + sb = self.view.verticalScrollBar() + at_bottom = sb.value() >= sb.maximum() - 4 + fmt = QTextCharFormat() + fmt.setForeground(QColor(COLORS.get(line.level) or COLORS.get(line.kind, "#d4d4d4"))) + if line.kind in (LineKind.PANIC, LineKind.BOOT, LineKind.HOST): + fmt.setFontWeight(QFont.Weight.Bold) + cur = QTextCursor(self.view.document()) + cur.movePosition(QTextCursor.MoveOperation.End) + if not self.view.document().isEmpty(): + cur.insertBlock() + cur.insertText(line.text, fmt) + if at_bottom: + sb.setValue(sb.maximum()) + + # ---- diagnostics ------------------------------------------------- + def _refresh_diag(self) -> None: + if self._diag_dirty: + self._diag_dirty = False + self.diag.show_report(self._engine.report()) + self.guide.refresh() # countdowns and timeouts need a steady tick + + # ---- guided troubleshooting hooks ------------------------------------ + def _guide_action(self, action: str) -> bool: + if action == "reset" and self._manager.state == State.CONNECTED: + self._manager.reset_board() + return True + return False + + def _guide_log(self, text: str) -> None: + # Technician answers go into the session log and the log view. + line = host_event(text) + self._session.write(line) + self._append(line) + self._diag_dirty = True + + def _copy_report(self) -> None: + QGuiApplication.clipboard().setText(self._engine.report().to_text()) + + def _analyze_file(self) -> None: + path, _ = QFileDialog.getOpenFileName( + self, "Analyze log file", str(self._session.root), "Logs (*.log *.txt);;All files (*)") + if not path: + return + lines = read_log_file(Path(path)) + self.view.clear() + self._engine.reset() + self._runner.cancel() + self._counts = {"E": 0, "W": 0, "boots": 0} + for line in lines: + self._engine.feed(line) + if line.kind == LineKind.BOOT: + self._counts["boots"] += 1 + if line.level in ("E", "W"): + self._counts[line.level] += 1 + self._append(line) + self._state_text = f"Analyzed file: {Path(path).name} ({len(lines)} lines)" + self._update_status() + self._diag_dirty = True + self._refresh_diag() + + # ---- misc -------------------------------------------------------- + def _save_view(self) -> None: + path, _ = QFileDialog.getSaveFileName(self, "Save log view", "sensorbox_log.txt", "Text (*.txt)") + if path: + with open(path, "w", encoding="utf-8") as fh: + fh.write(self.view.toPlainText()) + + def _open_folder(self) -> None: + self._session.root.mkdir(parents=True, exist_ok=True) + QDesktopServices.openUrl(QUrl.fromLocalFile(str(self._session.root))) + + def shutdown(self) -> None: + self._manager.disconnect() + self._session.stop() + + +class MainWindow(QMainWindow): + def __init__(self): + super().__init__() + self.setWindowTitle(f"KOTA Sensorbox Technician Tool v{__version__}") + self.resize(1200, 800) + tabs = QTabWidget() + guide = QTextBrowser(openExternalLinks=True) + guide.setHtml(cheatsheet_html()) + self.serial_tab = SerialTab() + self.flash_tab = FlashTab(self.serial_tab) + tabs.addTab(guide, "Quick start") + tabs.addTab(self.serial_tab, "Serial monitor") + tabs.addTab(self.flash_tab, "Flash firmware") + self.sd_tab = SdTab(self.serial_tab) + tabs.addTab(self.sd_tab, "SD card") + self.setCentralWidget(tabs) + settings_menu = self.menuBar().addMenu("Settings") + settings_menu.addAction("Service system address...", self._edit_service_base) + + def _edit_service_base(self) -> None: + st = self.serial_tab + text, ok = QInputDialog.getText( + self, "Service system address", + "Address of the KOTA service system (the box MAC is added at the end):", + text=st._service_base()) + if ok and text.strip().startswith(("https://", "http://")): + st._settings.setValue("service/base", text.strip()) + elif ok: + QMessageBox.warning(self, "Service system address", "The address must start with https://") + + def closeEvent(self, event): # noqa: N802 (Qt naming) + self.serial_tab.shutdown() + super().closeEvent(event) diff --git a/kota_tech/ui/sd_tab.py b/kota_tech/ui/sd_tab.py new file mode 100644 index 0000000..43ae8aa --- /dev/null +++ b/kota_tech/ui/sd_tab.py @@ -0,0 +1,304 @@ +"""SD card tab: format a card for the sensorbox and write / check ENV.TXT.""" + +from __future__ import annotations + +import threading +from pathlib import Path + +from PySide6.QtCore import QObject, QSettings, Qt, Signal +from PySide6.QtWidgets import ( + QApplication, + QCheckBox, + QFileDialog, + QFormLayout, + QGroupBox, + QHBoxLayout, + QLabel, + QLineEdit, + QListWidget, + QListWidgetItem, + QMessageBox, + QPlainTextEdit, + QPushButton, + QVBoxLayout, + QWidget, +) + +from ..sd.disks import Disk, list_cards +from ..sd.envfile import EnvConfig, load_template +from ..sd.provision import Outcome, check_card, format_and_write, write_env + +# The shared settings file the factory script uses, next to the tool folders. +DEFAULT_TEMPLATE = Path(__file__).resolve().parents[4] / "002-SD_card_format" / "ENV.txt" + + +class _Bridge(QObject): + done = Signal(object) + + +def _secret_field() -> tuple[QLineEdit, QCheckBox]: + edit = QLineEdit() + edit.setEchoMode(QLineEdit.EchoMode.Password) + show = QCheckBox("show") + show.toggled.connect(lambda on: edit.setEchoMode( + QLineEdit.EchoMode.Normal if on else QLineEdit.EchoMode.Password)) + return edit, show + + +class SdTab(QWidget): + def __init__(self, serial_tab, parent: QWidget | None = None): + super().__init__(parent) + self.serial_tab = serial_tab + self._settings = QSettings("KOTA", "TechnicianTool") + self._cards: list[Disk] = [] + self._busy = False + + # --- 1. settings + sbox = QGroupBox("1. Settings written to the card (ENV.TXT)") + self.template_label = QLabel(wordWrap=True) + load_btn = QPushButton("Load settings from ENV file...") + load_btn.clicked.connect(self._choose_template) + self.url = QLineEdit() + self.token, show_token = _secret_field() + form = QFormLayout() + form.addRow("INFLUX_URL", self.url) + trow = QHBoxLayout() + trow.addWidget(self.token, 1) + trow.addWidget(show_token) + form.addRow("INFLUX_TOKEN", trow) + self.wifi_box = QGroupBox("Local Wi-Fi instead of Matter (optional)") + self.wifi_box.setCheckable(True) + self.wifi_box.setChecked(False) + self.ssid = QLineEdit() + self.password, show_pw = _secret_field() + wform = QFormLayout(self.wifi_box) + wform.addRow("SSID", self.ssid) + prow = QHBoxLayout() + prow.addWidget(self.password, 1) + prow.addWidget(show_pw) + wform.addRow("PASSWORD", prow) + self.wifi_note = QLabel(wordWrap=True) + self.wifi_note.setStyleSheet("color:#555") + wform.addRow(self.wifi_note) + self.issues = QLabel(wordWrap=True) + sl = QVBoxLayout(sbox) + sl.addWidget(self.template_label) + sl.addWidget(load_btn, 0, Qt.AlignmentFlag.AlignLeft) + sl.addLayout(form) + sl.addWidget(self.wifi_box) + sl.addWidget(self.issues) + + # --- 2. card + cbox = QGroupBox("2. SD card (inside the sensorbox case)") + self.card_list = QListWidget() + self.card_list.setMinimumHeight(80) + self.others = QLabel(wordWrap=True) + self.others.setStyleSheet("color:#666") + refresh = QPushButton("Refresh cards") + refresh.clicked.connect(self.refresh_cards) + cl = QVBoxLayout(cbox) + cl.addWidget(QLabel("Insert the card into the laptop's reader. Only removable cards are listed.")) + cl.addWidget(self.card_list) + cl.addWidget(self.others) + cl.addWidget(refresh, 0, Qt.AlignmentFlag.AlignLeft) + + # --- 3. actions + abox = QGroupBox("3. Action") + self.format_btn = QPushButton("Format card and write ENV.TXT") + self.format_btn.setMinimumHeight(34) + self.format_btn.setToolTip("Erases the card: FAT32 (MBR), label KOTA, then writes and verifies ENV.TXT") + self.write_btn = QPushButton("Write ENV.TXT only (keep card contents)") + self.write_btn.setToolTip("Keeps buffered measurements (BUFFER.TXT) on the card") + self.check_btn = QPushButton("Check card") + self.check_btn.setToolTip("Show what the box would read from this card") + self.format_btn.clicked.connect(self._format) + self.write_btn.clicked.connect(self._write_only) + self.check_btn.clicked.connect(self._check) + self.result = QLabel(wordWrap=True) + self.result.setTextFormat(Qt.TextFormat.RichText) + self.result.setMargin(8) + self.log = QPlainTextEdit(readOnly=True) + self.log.setMaximumBlockCount(500) + al = QVBoxLayout(abox) + for w in (self.format_btn, self.write_btn, self.check_btn, self.result, self.log): + al.addWidget(w) + + left = QVBoxLayout() + left.addWidget(sbox) + left.addStretch(1) + right = QVBoxLayout() + right.addWidget(cbox) + right.addWidget(abox, 1) + lay = QHBoxLayout(self) + lay.addLayout(left, 1) + lay.addLayout(right, 1) + + self._bridge = _Bridge() + self._bridge.done.connect(self._on_done) + for w in (self.url, self.token, self.ssid, self.password): + w.textChanged.connect(self._update) + self.wifi_box.toggled.connect(self._update) + self.card_list.currentRowChanged.connect(self._update) + + path = Path(self._settings.value("sd/template", str(DEFAULT_TEMPLATE))) + self._load_template(path, quiet=True) + self._show_result("", "") + self._update() + + # ---------------------------------------------------------- settings + def _choose_template(self) -> None: + path, _ = QFileDialog.getOpenFileName(self, "ENV file with the InfluxDB settings", + str(DEFAULT_TEMPLATE.parent), "Text (*.txt *.env);;All files (*)") + if path: + self._load_template(Path(path)) + + def _load_template(self, path: Path, quiet: bool = False) -> None: + try: + env = load_template(path) + except OSError as exc: + self.template_label.setText("No settings file loaded - load the ENV file the firmware team gave you.") + if not quiet: + QMessageBox.warning(self, "Cannot read file", str(exc)) + return + self._settings.setValue("sd/template", str(path)) + self.template_label.setText(f"Loaded from {path}") + self.url.setText(env.influx_url) + self.token.setText(env.influx_token) + # Wi-Fi is decided per box by the technician, never copied silently. + self._update() + + def _env(self) -> EnvConfig: + wifi = self.wifi_box.isChecked() + return EnvConfig(self.url.text(), self.token.text(), + self.ssid.text() if wifi else "", self.password.text() if wifi else "") + + def _update(self) -> None: + env = self._env() + issues = env.problems() + colour = {"error": "#b03a2e", "warning": "#c77c02", "info": "#2e6da4"} + self.issues.setText("
".join(f"{i.level.upper()}: {i.text}" + for i in issues) or "Settings OK.") + fw = self.serial_tab.engine.firmware + support = fw.supports_sd_wifi + self.wifi_note.setText( + "Only for boxes that should NOT use Matter. The box then joins this Wi-Fi directly. " + + {True: "The connected box's firmware supports this.", + False: "The connected box runs the factory v1.0 firmware, which ignores these settings " + "(it only uses Matter). Newer firmware (after an OTA update) uses them.", + None: "Supported by newer firmware only; the factory v1.0 firmware ignores it."}[support]) + errors = any(i.level == "error" for i in issues) + card = self._selected_card() + mounted = bool(card and card.volumes and card.volumes[0].mount) + self.format_btn.setEnabled(not self._busy and not errors and card is not None) + self.write_btn.setEnabled(not self._busy and not errors and mounted) + self.check_btn.setEnabled(not self._busy and mounted) + + # -------------------------------------------------------------- cards + def showEvent(self, event): # noqa: N802 + self.refresh_cards() + super().showEvent(event) + + def refresh_cards(self) -> None: + if self._busy: + 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)) + if self._cards: + self.card_list.setCurrentRow(0) + empty = [d for d in others if not d.has_media] + big = [d for d in others if d.has_media] + notes = [] + if empty: + notes.append("Card reader found but no card inserted: " + ", ".join(d.name for d in empty) + ".") + if big: + notes.append("Not offered (too large to be an SD card): " + ", ".join(d.label for d in big) + ".") + if not self._cards and not empty: + notes.append("No SD card or card reader found.") + self.others.setText(" ".join(notes)) + self._update() + + def _selected_card(self) -> Disk | None: + i = self.card_list.currentRow() + return self._cards[i] if 0 <= i < len(self._cards) else None + + # ------------------------------------------------------------ actions + def _format(self) -> None: + card = self._selected_card() + if card is None: + return + box = QMessageBox(QMessageBox.Icon.Warning, "Format SD card", + f"ALL DATA ON THIS CARD WILL BE ERASED:\n\n{card.label}\n\n" + "Buffered measurements on the card are lost too. Continue?", + QMessageBox.StandardButton.Cancel, self) + erase = box.addButton("Erase and format", QMessageBox.ButtonRole.DestructiveRole) + box.exec() + if box.clickedButton() is not erase: + return + env = self._env() + self._run(lambda: format_and_write(card, env), f"Formatting {card.name}...") + + def _write_only(self) -> None: + card = self._selected_card() + if card and card.volumes: + env, mount = self._env(), card.volumes[0].mount + self._run(lambda: write_env(mount, env), f"Writing ENV.TXT to {mount}...") + + def _check(self) -> None: + card = self._selected_card() + if not (card and card.volumes): + return + vol = card.volumes[0] + rep = check_card(vol.mount, vol.filesystem) + lines = [f"Card {vol.mount} ({vol.filesystem or '?'})"] + if rep.env: + lines += [f" {k}={v}" for k, v in rep.env.masked().items()] + lines += [f" {i.level.upper()}: {i.text}" for i in rep.env.issues] + lines += [f" PROBLEM: {p}" for p in rep.problems] + if rep.buffered_lines is not None: + lines.append(f" {rep.buffered_lines} measurement lines buffered on the card, waiting for upload.") + self.log.appendPlainText("\n".join(lines)) + ok = not rep.problems and rep.env is not None and rep.env.ok + self._show_result("Card is OK for the sensorbox." if ok else "Card has problems - see below. " + "Use 'Write ENV.TXT only' or 'Format card'.", "ok" if ok else "error") + + def _run(self, job, what: str) -> None: + self._busy = True + self._update() + self._show_result(what, "info") + + def worker(): + try: + out = job() + except Exception as exc: # report, never leave the tab busy + out = Outcome(False, f"Unexpected error: {exc}") + self._bridge.done.emit(out) + + threading.Thread(target=worker, daemon=True).start() + + def _on_done(self, out: Outcome) -> None: + self._busy = False + for line in out.log: + self.log.appendPlainText(line) + self.log.appendPlainText(out.message) + if out.ok: + self._show_result(f"{out.message}
Insert the next card and press Refresh cards to " + "provision another one.", "ok") + else: + self._show_result(f"{out.message}", "error") + self.refresh_cards() + + def _show_result(self, html: str, kind: str) -> None: + colours = {"ok": ("#2e9e44", "#e3f4e7"), "info": ("#2e6da4", "#e3edf7"), "error": ("#d9534f", "#f9e0df")} + self.result.setVisible(bool(html)) + if html: + border, bg = colours[kind] + self.result.setStyleSheet(f"QLabel {{ border:2px solid {border}; background:{bg}; border-radius:6px; }}") + self.result.setText(html) + + diff --git a/pyproject.toml b/pyproject.toml index e5b2697..7327dc0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "kota-technician-tool" -version = "0.4.0" +version = "0.5.0" description = "KOTA sensorbox technician diagnostic tool (demo)" requires-python = ">=3.11,<3.14" dependencies = [ diff --git a/run_macos.command b/run_macos.command index 654deab..acd3418 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.4" +export UV_PROJECT_ENVIRONMENT="$HOME/Library/Caches/KotaTechnicianTool/venv-0.5" 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 923ad5e..a979e72 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.4" +set "UV_PROJECT_ENVIRONMENT=%LOCALAPPDATA%\KotaTechnicianTool\venv-0.5" set "PATH=%USERPROFILE%\.local\bin;%PATH%" where uv >nul 2>nul diff --git a/tests/test_flash.py b/tests/test_flash.py index 068a12d..cf6dfac 100644 --- a/tests/test_flash.py +++ b/tests/test_flash.py @@ -52,7 +52,7 @@ def test_modified_image_is_refused(tmp_path): @pytest.mark.parametrize("patch,msg", [ - ((0, 0x00), "not a merged"), + ((0, 0x00), "not an ESP32 firmware"), ((12, 0x05), "not ESP32-S3"), ]) def test_wrong_images_are_rejected(tmp_path, patch, msg): @@ -160,3 +160,87 @@ def test_runner_reports_connect_failure_with_hint(image, tmp_path): res, *_ = _run(plan, cmd) assert not res.ok and res.failed_step == 0 assert "BOOT" in res.hint + + +# ------------------------------------------------- developer builds (v0.5) + +from kota_tech.flash.image import FirmwareLibrary, app_image_length # noqa: E402 + + +def _app_only_file(image, tmp_path, name="kota-sensorboard.bin"): + p = tmp_path / name + p.write_bytes(image.app_bytes()) # what a developer build/ folder contains + return p + + +def test_integrity_check_on_factory_image(image): + d = image.data + assert app_image_length(d, 0) > 0 # bootloader + assert app_image_length(d, 0x40000) == len(d) - 0x40000 # app ends exactly at file end + + +@pytest.mark.parametrize("damage,msg", [ + (lambda b: b[:len(b) // 2], "truncated"), + (lambda b: b[:0x40000 + 5000] + bytes([b[0x40000 + 5000] ^ 1]) + b[0x40000 + 5001:], "corrupted"), +]) +def test_corrupted_or_truncated_images_refused(tmp_path, damage, msg): + p = tmp_path / "broken.bin" + p.write_bytes(damage(FW.read_bytes())) + with pytest.raises(ImageError, match=msg): + inspect_image(p, origin="developer") + + +def test_import_app_only_developer_build(image, tmp_path): + lib = FirmwareLibrary(tmp_path / "lib") + src = _app_only_file(image, tmp_path) + img = lib.import_file(src, reference=image) + assert img.is_developer and img.kind == "app" + assert img.partition("ota_0").offset == 0x40000 # layout from the factory image + assert img.app_bytes() == image.app_bytes() + listed, problems = lib.images(image) + assert [i.sha256 for i in listed] == [img.sha256] and problems == [] + assert lib.import_file(src, reference=image).sha256 == img.sha256 # re-import: no duplicate + assert len(lib.images(image)[0]) == 1 + + +def test_app_only_build_cannot_be_flashed_full(image, tmp_path): + lib = FirmwareLibrary(tmp_path / "lib") + img = lib.import_file(_app_only_file(image, tmp_path), reference=image) + plan = build_plan(img, Mode.APP_ONLY, "COM5", tmp_path / "w") + assert [s.kind for s in plan.steps] == ["read_pt", "write"] + with pytest.raises(ValueError, match="app only"): + build_plan(img, Mode.FULL, "COM5", tmp_path / "w") + + +def test_import_merged_developer_build(image, tmp_path): + lib = FirmwareLibrary(tmp_path / "lib") + src = tmp_path / "dev_full.bin" + src.write_bytes(image.data) + img = lib.import_file(src, reference=image) + assert img.kind == "merged" and img.is_developer + assert [s.kind for s in build_plan(img, Mode.FULL, "COM5", tmp_path / "w").steps] == ["write"] + + +def test_library_detects_file_changed_after_import(image, tmp_path): + lib = FirmwareLibrary(tmp_path / "lib") + img = lib.import_file(_app_only_file(image, tmp_path), reference=image) + stored = tmp_path / "lib" / img.path.name + data = bytearray(stored.read_bytes()) + data[-40] ^= 1 + stored.write_bytes(data) + listed, problems = lib.images(image) + assert listed == [] and problems + + +def test_library_remove(image, tmp_path): + lib = FirmwareLibrary(tmp_path / "lib") + img = lib.import_file(_app_only_file(image, tmp_path), reference=image) + lib.remove(img.sha256) + assert lib.images(image) == ([], []) and not img.path.exists() + + +def test_non_firmware_file_refused(tmp_path, image): + p = tmp_path / "notes.bin" + p.write_bytes(b"hello firmware team") + with pytest.raises(ImageError, match="not an ESP32 firmware"): + FirmwareLibrary(tmp_path / "lib").import_file(p, reference=image) diff --git a/tests/test_sd.py b/tests/test_sd.py new file mode 100644 index 0000000..7d93818 --- /dev/null +++ b/tests/test_sd.py @@ -0,0 +1,159 @@ +"""SD provisioning: ENV.TXT (firmware-exact parsing), disk discovery parsers, +the elevated Windows format script, and write/check on a stand-in card.""" + +import plistlib + +import pytest + +from kota_tech.sd.disks import MAX_CARD_BYTES, Disk, parse_macos, parse_windows +from kota_tech.sd.envfile import EnvConfig, parse_env +from kota_tech.sd.provision import check_card, windows_format_script, write_env + +URL = "https://influx.kota-app.com/api/v2/write?bucket=sensorbox&org=kota" +TOKEN = "T" * 86 + "==" # shape of an InfluxDB token, not a real one + + +def cfg(**kw): + base = dict(influx_url=URL, influx_token=TOKEN) + base.update(kw) + return EnvConfig(**base) + + +# ------------------------------------------------------------------ ENV.TXT + +def test_render_matches_the_factory_file_format(): + raw = cfg().render() + assert raw == f"INFLUX_TOKEN={TOKEN}\nINFLUX_URL={URL}\n".encode() + assert not raw.startswith(b"\xef\xbb\xbf") and b"\r" not in raw + assert parse_env(raw).ok + + +def test_wifi_only_written_when_both_set(): + raw = cfg(ssid="Site WiFi", password="correct horse").render().decode() + assert "SSID=Site WiFi\n" in raw and "PASSWORD=correct horse\n" in raw + assert any("both SSID and PASSWORD" in i.text for i in cfg(ssid="x").problems()) + + +@pytest.mark.parametrize("kw,fragment", [ + (dict(influx_url="influx.kota-app.com"), "https://"), + (dict(influx_url="https://influx.kota-app.com/"), "write URL"), + (dict(influx_token=""), "INFLUX_TOKEN is empty"), + (dict(influx_token="abc def" * 5), "spaces"), + (dict(influx_token=TOKEN + " "), "space"), + (dict(ssid="s" * 33, password="12345678"), "32 bytes"), + (dict(ssid="net", password="short"), "8-63"), +]) +def test_invalid_settings_are_errors(kw, fragment): + errors = [i.text for i in cfg(**kw).problems() if i.level == "error"] + assert any(fragment in e for e in errors), errors + + +def test_parser_mirrors_firmware(): + raw = b"\xef\xbb\xbfINFLUX_TOKEN=abc\r\nINFLUX_URL = x\nINFLUX_URL=u1\nINFLUX_URL=u2\nHTTP_SERVER=true\n" + chk = parse_env(raw) + texts = " ".join(i.text for i in chk.issues) + assert "BOM" in texts # first key unreadable + assert "spaces around the key 'INFLUX_URL'" in texts + assert chk.values["INFLUX_URL"] == "u1" # first match wins + assert "appears twice" in texts and "setup mode" in texts + assert "INFLUX_TOKEN is missing" in texts # hidden behind the BOM + assert not chk.ok + + +def test_masking(): + chk = parse_env(cfg(ssid="net", password="secretpassword").render()) + m = chk.masked() + assert TOKEN not in m["INFLUX_TOKEN"] and m["INFLUX_TOKEN"].startswith("TTTT") + assert "secret" not in m["PASSWORD"] + + +# ------------------------------------------------------------- disk lists + +WIN_JSON = """[{"Number":0,"FriendlyName":"ST500DM002-1BC142","Size":500107862016,"BusType":"SATA", + "Volumes":[{"DriveLetter":"D","FileSystem":"NTFS","Label":"WINRE"}]}, + {"Number":4,"FriendlyName":"Mass Storage Device","Size":0,"BusType":"USB","Volumes":[]}, + {"Number":5,"FriendlyName":"SDXC Card","Size":31914983424,"BusType":"SD", + "Volumes":{"DriveLetter":"E","FileSystem":"FAT32","Label":"KOTA"}}]""" + + +def test_parse_windows_keeps_only_card_buses(): + disks = parse_windows(WIN_JSON) + assert [d.id for d in disks] == ["4", "5"] # SATA data disk never listed + reader, card = disks + assert not reader.has_media and reader.size_text == "no card" + assert card.volumes[0].mount == "E:\\" and card.volumes[0].filesystem == "FAT32" + + +def test_parse_windows_single_object_quirk(): + one = '{"Number":5,"FriendlyName":"SD","Size":8000000000,"BusType":"USB","Volumes":null}' + assert [d.id for d in parse_windows(one)] == ["5"] + + +def test_parse_macos_excludes_system_disk(): + listing = plistlib.dumps({"AllDisksAndPartitions": [ + {"DeviceIdentifier": "disk0", "Size": 500_000_000_000, "Partitions": []}, + {"DeviceIdentifier": "disk3", "Size": 994_000_000_000, "Partitions": [], "APFSVolumes": [{"MountPoint": "/"}]}, + {"DeviceIdentifier": "disk4", "Size": 15_931_539_456, + "Partitions": [{"DeviceIdentifier": "disk4s1", "MountPoint": "/Volumes/KOTA", "Content": "DOS_FAT_32", + "VolumeName": "KOTA"}]}, + ]}) + infos = {"disk0": plistlib.dumps({"Internal": True, "BusProtocol": "PCI-Express"}), + "disk3": plistlib.dumps({"Internal": True}), + "disk4": plistlib.dumps({"Internal": True, "RemovableMedia": True, "BusProtocol": "Secure Digital", + "MediaName": "SD Card Reader"})} + disks = parse_macos(listing, infos) + assert [d.id for d in disks] == ["/dev/disk4"] and disks[0].bus == "SD" + assert disks[0].volumes[0].mount == "/Volumes/KOTA" + + +# -------------------------------------------------------- format script + +def test_windows_script_rechecks_disk_before_erasing(tmp_path): + disk = Disk("5", "SDXC Card", 63_864_569_856, "SD") + ps = windows_format_script(disk, tmp_path / "r.json") + erase = ps.index("Clear-Disk") + for guard in ("IsSystem -or $d.IsBoot", "'MMC','SD','USB'", "-ne 63864569856"): + assert guard in ps and ps.index(guard) < erase + assert "-PartitionStyle MBR" in ps and "-FileSystem FAT32" in ps + assert str(32 * 1024**3) in ps # FAT32 partition capped at 32 GB + assert "NewFileSystemLabel 'KOTA'" in ps + + +def test_large_disks_are_not_cards(): + assert MAX_CARD_BYTES == 256 * 1024**3 + + +# ----------------------------------------------- write / check on a card + +def test_write_env_replaces_variants_and_keeps_buffered_data(tmp_path): + card = tmp_path / "card" + card.mkdir() + (card / "env.txt").write_bytes(b"INFLUX_URL=old\n") + (card / "BUFFER.TXT").write_bytes(b"line1\nline2\n\n") + out = write_env(str(card), cfg()) + assert out.ok, out.message + names = sorted(p.name for p in card.iterdir()) + assert names == ["BUFFER.TXT", "ENV.TXT"] + rep = check_card(str(card), "FAT32") + assert rep.env.ok and rep.problems == [] and rep.buffered_lines == 2 + + +def test_write_env_refuses_invalid_settings(tmp_path): + assert not write_env(str(tmp_path), cfg(influx_token="")).ok + assert not (tmp_path / "ENV.TXT").exists() + + +def test_check_card_reports_missing_env_and_wrong_fs(tmp_path): + rep = check_card(str(tmp_path), "exFAT") + texts = " ".join(rep.problems) + assert "needs FAT32" in texts and "No ENV.TXT" in texts + + +def test_elevated_result_folder_is_not_owner_only(tmp_path, monkeypatch): + # Regression: tempfile.mkdtemp() made the elevated script's result unreadable. + import kota_tech.sd.provision as prov + monkeypatch.setattr(prov, "WORK_ROOT", tmp_path / "tmp") + work = prov._workdir() + assert work.parent == tmp_path / "tmp" and work.is_dir() + (work / "result.json").write_text('{"ok": true}', encoding="utf-8") + assert (work / "result.json").read_text(encoding="utf-8") diff --git a/tests/test_service.py b/tests/test_service.py new file mode 100644 index 0000000..31449f3 --- /dev/null +++ b/tests/test_service.py @@ -0,0 +1,91 @@ +"""Service-system hand-off: status mapping, note, and the #kota-report fragment +(decoded exactly as the v1.1.0 service page does: base64url + UTF-8 JSON).""" + +import base64 +import json +from pathlib import Path + +import pytest + +from kota_tech.core.logfile import read_log_file +from kota_tech.core.logline import parse_line +from kota_tech.diag.engine import DiagnosticsEngine +from kota_tech.service.report import ( + CONNECTIVITY, HARDWARE, OK, decode_fragment, normalize_mac, report_url, service_note, service_status, +) + +FIX = Path(__file__).parent / "fixtures" + + +def bench(): + e = DiagnosticsEngine() + for ln in read_log_file(FIX / "live_boot_A4CB8FC946F8_fw1.0.txt"): + e.feed(ln) + return e + + +def test_engine_reads_mac_from_boot_log(): + assert bench().mac == "A4:CB:8F:C9:46:F8" + + +def test_bench_box_is_a_connectivity_problem(): + e = bench() + assert service_status(e.report()) == CONNECTIVITY + note = service_note(e.report()) + assert note.startswith("[KOTA Technician Tool") + assert "not commissioned in Matter" in note and "-> No Internet" in note + assert "Firmware: v1.0" in note + + +def test_hardware_and_healthy_statuses(): + e = DiagnosticsEngine() + for ln in ["ESP-ROM:esp32s3-20210327", "I (1299) app_init: Project name: kota-sensorboard", + "E (3511) MAIN: Failed to initialize RTC: ESP_ERR_TIMEOUT"]: + e.feed(parse_line(ln)) + assert service_status(e.report()) == HARDWARE + ok = DiagnosticsEngine() + for ln in ["ESP-ROM:esp32s3-20210327", "I (1299) app_init: Project name: kota-sensorboard", + "I (31000) INFLUX: HTTP request performed, status code: 204"]: + ok.feed(parse_line(ln)) + assert service_status(ok.report()) == OK + + +def test_url_shape_and_payload_roundtrip(): + e = bench() + url = report_url(e.mac, e.report(), "https://service.kota-app.com") + assert url.startswith("https://service.kota-app.com/A4CB8FC946F8#kota-report=") + assert "?" not in url # nothing in the query: the fragment is never sent to a server + data = decode_fragment(url) + assert data["v"] == 1 and data["mac"] == "A4CB8FC946F8" and data["status"] == CONNECTIVITY + + +def test_fragment_decodes_like_the_service_page(): + # Mirror of index.html applyToolPrefill: base64url -> pad -> bytes -> UTF-8 -> JSON + e = bench() + e.note_technician("Zażółć gęślą jaźń -> Purple") # non-ASCII survives + url = report_url("a4:cb:8f:c9:46:f8", e.report()) + frag = url.split("#kota-report=")[1] + assert set(frag) <= set("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_") # page regex + padded = frag.replace("-", "+").replace("_", "/") + padded += "==="[(len(padded) + 3) % 4:] + data = json.loads(base64.b64decode(padded).decode("utf-8")) + assert "Zażółć gęślą jaźń" in data["note"] + + +def test_note_is_limited_and_no_secrets(): + e = bench() + for i in range(400): + e.note_technician(f"check {i} " + "x" * 50) + note = service_note(e.report(), limit=4000) + assert len(note) <= 4000 + assert "INFLUX_TOKEN" not in note + + +@pytest.mark.parametrize("bad", ["", "A4CB8F", "zz:cb:8f:c9:46:f8"]) +def test_invalid_mac_rejected(bad): + with pytest.raises(ValueError): + report_url(bad, bench().report()) + + +def test_normalize_mac(): + assert normalize_mac("a4:cb:8f:c9:46:f8") == "A4CB8FC946F8"