Technician tool v0.5

Imported from the Google Drive folder 004-technician_tool/v0.5.
This commit is contained in:
kkallas committed 2026-09-26 18:30:51 +02:00
1 parent 72adfe1cec
commit 63d83edc65
21 files changed
+1680 -86

No files matched your search

+30
View File
@@ -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/<MAC>#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).
+1 -1
View File
@@ -1,3 +1,3 @@
"""KOTA sensorbox technician tool.""" """KOTA sensorbox technician tool."""
__version__ = "0.4.0" __version__ = "0.5.0"
+4
View File
@@ -25,6 +25,7 @@ from ..core.logline import LineKind, LogLine
from .firmware import FirmwareInfo from .firmware import FirmwareInfo
RULES_PATH = Path(__file__).with_name("rules.toml") 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): class Severity(IntEnum):
@@ -278,6 +279,7 @@ class DiagnosticsEngine:
self._tool_reset_at: float | None = None self._tool_reset_at: float | None = None
self._panic_tail = 0 # RAW lines after a panic line to keep self._panic_tail = 0 # RAW lines after a panic line to keep
self.technician: list[str] = [] self.technician: list[str] = []
self.mac: str | None = None # from "ID: Device MAC address: ..." at boot
@property @property
def boot(self) -> BootRecord: def boot(self) -> BootRecord:
@@ -317,6 +319,8 @@ class DiagnosticsEngine:
self._crash_finding().evidence.append(text) self._crash_finding().evidence.append(text)
self.firmware.observe(text) self.firmware.observe(text)
if m := _MAC_LINE.search(text):
self.mac = m.group(1).upper()
matched = False matched = False
for rule in self.rules: for rule in self.rules:
m = rule.pattern.search(text) m = rule.pattern.search(text)
+1 -1
View File
@@ -379,7 +379,7 @@ severity = "critical"
title = "SD card is unreadable or has wrong settings" title = "SD card is unreadable or has wrong settings"
text = "Even after re-seating, the card or its ENV.TXT cannot be used." text = "Even after re-seating, the card or its ENV.TXT cannot be used."
actions = [ 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).", "The Diagnosis tab shows which step failed (card, file system or ENV.TXT keys).",
] ]
+185 -51
View File
@@ -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 Two origins:
offered for flashing: boxes normally get new firmware over OTA, so the tool * factory - bundled with the tool (firmware/firmware.toml), SHA-256 locked;
only ever writes the official factory image. * 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 Every image is checked structurally (ESP32-S3, kota-sensorboard app
table at 0x8000, app at ota_0). App-only flashing takes the app out of it descriptor, segment checksum and the appended SHA-256) so a truncated or
and writes it to ota_0; the layout comes from the image's own partition corrupted file is refused before anything is written.
table, not from constants.
""" """
from __future__ import annotations from __future__ import annotations
import datetime as dt
import hashlib import hashlib
import json
import shutil
import struct import struct
import tomllib import tomllib
from dataclasses import dataclass from dataclasses import dataclass
@@ -20,6 +28,8 @@ from pathlib import Path
FIRMWARE_DIR = Path(__file__).resolve().parents[2] / "firmware" FIRMWARE_DIR = Path(__file__).resolve().parents[2] / "firmware"
MANIFEST = FIRMWARE_DIR / "firmware.toml" MANIFEST = FIRMWARE_DIR / "firmware.toml"
LIBRARY_DIR = Path.home() / "KotaTechnicianTool" / "firmware"
LIBRARY_INDEX = "library.json"
PT_OFFSET = 0x8000 PT_OFFSET = 0x8000
PT_SIZE = 0xC00 PT_SIZE = 0xC00
@@ -27,6 +37,8 @@ ESP32S3_CHIP_ID = 9
IMAGE_MAGIC = 0xE9 IMAGE_MAGIC = 0xE9
APP_DESC_MAGIC = 0xABCD5432 APP_DESC_MAGIC = 0xABCD5432
FLASH_SIZES = {0: "1MB", 1: "2MB", 2: "4MB", 3: "8MB", 4: "16MB", 5: "32MB"} 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): class ImageError(ValueError):
@@ -46,9 +58,18 @@ class Partition:
class AppInfo: class AppInfo:
project: str project: str
version: str version: str
built: str built: str # "Jun 29 2026 11:03:32" as in the app descriptor
idf: str 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) @dataclass(frozen=True)
@@ -56,36 +77,43 @@ class FirmwareImage:
path: Path path: Path
name: str name: str
notes: str notes: str
origin: str # "factory" | "developer"
kind: str # "merged" | "app"
sha256: str sha256: str
flash_size: str flash_size: str
partitions: tuple[Partition, ...] 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 app: AppInfo
@property @property
def data(self) -> bytes: def data(self) -> bytes:
return self.path.read_bytes() return self.path.read_bytes()
@property
def is_developer(self) -> bool:
return self.origin == "developer"
def partition(self, label: str) -> Partition: def partition(self, label: str) -> Partition:
for p in self.partitions: for p in self.partitions:
if p.label == label: if p.label == label:
return p return p
raise ImageError(f"partition {label!r} not in image") 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: def app_bytes(self) -> bytes:
"""The application image stored in the ota_0 region of the merged image.""" return self.data[self.app_offset:self.app_offset + self.app_length]
# 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]
def blank_otadata(self) -> bytes: def blank_otadata(self) -> bytes:
"""Erased otadata: the bootloader then boots ota_0.""" """Erased otadata: the bootloader then boots ota_0."""
return b"\xff" * self.partition("otadata").size 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, ...]: def parse_partition_table(raw: bytes) -> tuple[Partition, ...]:
parts = [] parts = []
@@ -103,50 +131,156 @@ def parse_partition_table(raw: bytes) -> tuple[Partition, ...]:
return tuple(parts) return tuple(parts)
def _app_info(data: bytes, offset: int) -> AppInfo: def app_image_length(data: bytes, offset: int = 0) -> int:
if len(data) < offset + 0x20 + 256 or data[offset] != IMAGE_MAGIC: """Validate an ESP app image at `offset`; return its exact length.
raise ImageError(f"no application image at 0x{offset:x}")
d = data[offset + 0x20:offset + 0x20 + 256]
if struct.unpack_from("<I", d)[0] != APP_DESC_MAGIC:
raise ImageError("application has no ESP-IDF app descriptor")
s = lambda o, n: d[o:o + n].split(b"\0")[0].decode("ascii", "replace")
return AppInfo(project=s(48, 32), version=s(16, 32), built=f"{s(96, 16)} {s(80, 16)}",
idf=s(112, 32), elf_sha=d[144:176].hex())
Checks the segment checksum and, when the header says so, the appended
def inspect_image(path: Path, name: str = "", notes: str = "") -> FirmwareImage: SHA-256 - the same checks the bootloader performs before booting it.
data = path.read_bytes() """
if len(data) < PT_OFFSET + PT_SIZE or data[0] != IMAGE_MAGIC: if len(data) < offset + 24 or data[offset] != IMAGE_MAGIC:
raise ImageError("not a merged ESP32 image (no bootloader at offset 0)") raise ImageError(f"no firmware image at 0x{offset:x}")
chip_id = struct.unpack_from("<H", data, 12)[0] segments = data[offset + 1]
chip_id = struct.unpack_from("<H", data, offset + 12)[0]
if chip_id != ESP32S3_CHIP_ID: if chip_id != ESP32S3_CHIP_ID:
raise ImageError(f"image is for chip id {chip_id}, not ESP32-S3") raise ImageError(f"image is for chip id {chip_id}, not ESP32-S3")
flash_size = FLASH_SIZES.get(data[3] >> 4, "?") hash_appended = data[offset + 23] == 1
parts = parse_partition_table(data[PT_OFFSET:PT_OFFSET + PT_SIZE]) pos, checksum = offset + 24, 0xEF
labels = {p.label for p in parts} for _ in range(segments):
for need in ("nvs", "otadata", "ota_0"): if pos + 8 > len(data):
if need not in labels: raise ImageError("image is truncated (segment header missing)")
raise ImageError(f"partition table has no {need!r} partition") _, seg_len = struct.unpack_from("<II", data, pos)
ota0 = next(p for p in parts if p.label == "ota_0") pos += 8
app = _app_info(data, ota0.offset) if pos + seg_len > len(data):
if app.project != "kota-sensorboard": raise ImageError("image is truncated (segment data missing)")
raise ImageError(f"image is {app.project!r}, not kota-sensorboard") for b in data[pos:pos + seg_len]:
return FirmwareImage(path, name or path.name, notes, hashlib.sha256(data).hexdigest(), checksum ^= b
flash_size, parts, app) 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]: def _app_info(data: bytes, offset: int) -> AppInfo:
"""Images from the manifest whose file exists and matches its SHA-256.""" d = data[offset + 0x20:offset + 0x20 + 256]
if len(d) < 256 or struct.unpack_from("<I", d)[0] != APP_DESC_MAGIC:
raise ImageError("firmware has no ESP-IDF app descriptor")
s = lambda o, n: d[o:o + n].split(b"\0")[0].decode("ascii", "replace")
info = AppInfo(project=s(48, 32), version=s(16, 32), built=f"{s(96, 16)} {s(80, 16)}",
idf=s(112, 32), elf_sha=d[144:176].hex())
if info.project != "kota-sensorboard":
raise ImageError(f"firmware is {info.project!r}, not kota-sensorboard")
return info
def inspect_image(path: Path, name: str = "", notes: str = "", origin: str = "factory",
reference: FirmwareImage | None = None) -> 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(): if not manifest.exists():
return [] return []
entries = tomllib.loads(manifest.read_text(encoding="utf-8")).get("image", [])
out = [] out = []
for e in entries: for e in tomllib.loads(manifest.read_text(encoding="utf-8")).get("image", []):
path = manifest.parent / e["file"] path = manifest.parent / e["file"]
if not path.exists(): if not path.exists():
continue 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(): if img.sha256 != e["sha256"].lower():
raise ImageError(f"{path.name}: SHA-256 does not match the manifest - file changed or corrupt") raise ImageError(f"{path.name}: SHA-256 does not match the manifest - file changed or corrupt")
out.append(img) out.append(img)
return out 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
+4 -1
View File
@@ -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)], hex(otadata.offset), str(ota_file), hex(ota0.offset), str(app_file)],
est_s=20)) est_s=20))
else: 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( 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)], _common(port, "hard_reset") + write_opts + ["0x0", str(image.path)],
est_s=25)) est_s=25))
return plan return plan
View File
Whitespace-only changes.
+128
View File
@@ -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 <disk>` 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]
+151
View File
@@ -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
+273
View File
@@ -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"<plist><dict/></plist>")
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)
View File
Whitespace-only changes.
+101
View File
@@ -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:
<base><MAC>#kota-report=<base64url(JSON)>
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"))
+108 -26
View File
@@ -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 from __future__ import annotations
@@ -10,6 +10,7 @@ from PySide6.QtCore import QObject, Qt, QTimer, Signal
from PySide6.QtWidgets import ( from PySide6.QtWidgets import (
QButtonGroup, QButtonGroup,
QCheckBox, QCheckBox,
QFileDialog,
QComboBox, QComboBox,
QGroupBox, QGroupBox,
QHBoxLayout, QHBoxLayout,
@@ -24,7 +25,7 @@ from PySide6.QtWidgets import (
) )
from ..core.session import SESSIONS_DIR 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.plan import Mode, build_plan
from ..flash.runner import FlashRunner, Result from ..flash.runner import FlashRunner, Result
from ..serialio.ports import PortInfo, list_serial_ports from ..serialio.ports import PortInfo, list_serial_ports
@@ -71,13 +72,26 @@ class FlashTab(QWidget):
bl.addWidget(self.box_fw) bl.addWidget(self.box_fw)
# 2. firmware # 2. firmware
fwbox = QGroupBox("2. Firmware (official images only)") fwbox = QGroupBox("2. Firmware")
self.image_combo = QComboBox(minimumWidth=360) 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 = QLabel(wordWrap=True)
self.image_info.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse) self.image_info.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
fl = QVBoxLayout(fwbox) fl = QVBoxLayout(fwbox)
fl.addWidget(self.image_combo) 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) fl.addWidget(self.image_info)
self._library = FirmwareLibrary()
self._reference: FirmwareImage | None = None
self._problems: list[str] = []
# 3. mode # 3. mode
modebox = QGroupBox("3. What to write") modebox = QGroupBox("3. What to write")
@@ -151,17 +165,55 @@ class FlashTab(QWidget):
self._show_result("", None) self._show_result("", None)
# ------------------------------------------------------------ setup # ------------------------------------------------------------ setup
def _load_images(self) -> None: def _load_images(self, select_sha: str | None = None) -> None:
self._problems = []
try: try:
self._images = official_images() factory = factory_images()
except ImageError as exc: except ImageError as exc:
self._images = [] factory = []
self.image_info.setText(f"<span style='color:#b03a2e'><b>Firmware file problem:</b> {exc}</span>") 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() self.image_combo.clear()
for img in self._images: for img in self._images:
self.image_combo.addItem(img.name) self.image_combo.addItem(img.label())
if not self._images: 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: def refresh_ports(self) -> None:
current = self.port_combo.currentData() current = self.port_combo.currentData()
@@ -189,24 +241,27 @@ class FlashTab(QWidget):
def _update(self) -> None: def _update(self) -> None:
img = self._selected_image() img = self._selected_image()
info = ""
if img: if img:
a = img.app a = img.app
self.image_info.setText( origin = ("<b>Factory image</b> (bundled with the tool)" if not img.is_developer else
f"{a.project} {a.version}, built {a.built}, ESP-IDF {a.idf}<br>" "<b style='color:#c77c02'>Developer build</b> - only flash builds you received "
f"<small>build {a.elf_sha[:9]} &middot; file SHA-256 {img.sha256[:12]}... verified</small><br>" "from the firmware team")
f"<small>{img.notes}</small>") kind = "complete image (bootloader + layout + app)" if img.kind == "merged" else "app only"
fw = self.serial_tab.engine.firmware info = (f"{origin}<br>{a.project} {a.version}, built {a.built}, ESP-IDF {a.idf}, {kind}<br>"
if fw.elf_sha and img: f"<small>build {a.elf_sha[:9]} &middot; file SHA-256 {img.sha256[:16]}... "
same = fw.elf_sha.startswith(img.app.elf_sha[:9]) or img.app.elf_sha.startswith(fw.elf_sha) f"&middot; checksum OK</small>" + (f"<br><small>{img.notes}</small>" if img.notes else ""))
msg = f"Box currently runs: {fw.describe()}" if self._problems:
if same: info += ("<br><span style='color:#b03a2e'><b>Not offered:</b> "
msg += "<br><i>It already has this firmware. Re-flash only if you suspect the firmware is damaged.</i>" + "; ".join(self._problems) + "</span>")
else: self.image_info.setText(info)
msg += ("<br><b style='color:#c77c02'>The box runs a different build (probably updated over OTA). " self.remove_btn.setEnabled(bool(img and img.is_developer) and self._runner is None)
"Flashing puts the older factory image back; the box updates itself again once online.</b>") # An app-only file has no bootloader/layout: Full mode is impossible.
else: full_ok = bool(img and img.kind == "merged")
msg = "Firmware on the box: unknown - connect in the Serial monitor and press Reset & capture to see it." self.rb_full.setEnabled(full_ok and self._runner is None)
self.box_fw.setText(msg) 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()) self.confirm_full.setVisible(self.rb_full.isChecked())
ok = bool(img and self._selected_port() and self._runner is None ok = bool(img and self._selected_port() and self._runner is None
@@ -224,6 +279,23 @@ class FlashTab(QWidget):
else: else:
self.plan_label.setText("Select a sensorbox and a firmware image.") 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 + "<br><i>It already has this firmware. Re-flash only if you suspect the firmware is damaged.</i>"
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"<br><b style='color:#2e6da4'>The selected firmware is newer (built {img.app.built}).</b>"
if box_built and img_built and img_built < box_built:
return msg + ("<br><b style='color:#c77c02'>The selected firmware is OLDER than the box's (the box was "
"probably updated over OTA). The box updates itself again once online.</b>")
return msg + "<br><b style='color:#c77c02'>The box runs a different build.</b>"
# ------------------------------------------------------------ flash # ------------------------------------------------------------ flash
def _start(self) -> None: def _start(self) -> None:
img, port = self._selected_image(), self._selected_port() img, port = self._selected_image(), self._selected_port()
@@ -256,7 +328,7 @@ class FlashTab(QWidget):
self._runner.start() self._runner.start()
def _set_busy(self, busy: bool) -> None: 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) w.setEnabled(not busy)
self._update() self._update()
@@ -307,3 +379,13 @@ class FlashTab(QWidget):
border, bg = colours[kind or "info"] border, bg = colours[kind or "info"]
self.result.setStyleSheet(f"QLabel {{ border:2px solid {border}; background:{bg}; border-radius:6px; }}") self.result.setStyleSheet(f"QLabel {{ border:2px solid {border}; background:{bg}; border-radius:6px; }}")
self.result.setText(html) 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
+52 -2
View File
@@ -2,15 +2,17 @@ from __future__ import annotations
from pathlib import Path from pathlib import Path
from PySide6.QtCore import QCoreApplication, QObject, Qt, QTimer, QUrl, Signal from PySide6.QtCore import QCoreApplication, QObject, QSettings, Qt, QTimer, QUrl, Signal
from PySide6.QtGui import QColor, QDesktopServices, QFont, QGuiApplication, QTextCharFormat, QTextCursor from PySide6.QtGui import QColor, QDesktopServices, QFont, QGuiApplication, QTextCharFormat, QTextCursor
from PySide6.QtWidgets import ( from PySide6.QtWidgets import (
QCheckBox, QCheckBox,
QComboBox, QComboBox,
QFileDialog, QFileDialog,
QHBoxLayout, QHBoxLayout,
QInputDialog,
QLabel, QLabel,
QMainWindow, QMainWindow,
QMessageBox,
QPlainTextEdit, QPlainTextEdit,
QPushButton, QPushButton,
QSplitter, QSplitter,
@@ -28,10 +30,12 @@ from ..diag.engine import DiagnosticsEngine
from ..diag.guides import GuideRunner from ..diag.guides import GuideRunner
from ..serialio.manager import SerialManager, State from ..serialio.manager import SerialManager, State
from ..serialio.ports import PortInfo, list_serial_ports from ..serialio.ports import PortInfo, list_serial_ports
from ..service.report import DEFAULT_BASE, normalize_mac, report_url
from .cheatsheet import cheatsheet_html from .cheatsheet import cheatsheet_html
from .diag_panel import DiagnosticsPanel from .diag_panel import DiagnosticsPanel
from .flash_tab import FlashTab from .flash_tab import FlashTab
from .guide_panel import GuidePanel from .guide_panel import GuidePanel
from .sd_tab import SdTab
MAX_LINES = 20000 MAX_LINES = 20000
LEVEL_FILTERS = {"All lines": None, "Warnings + errors": {"W", "E"}, "Errors only": {"E"}} LEVEL_FILTERS = {"All lines": None, "Warnings + errors": {"W", "E"}, "Errors only": {"E"}}
@@ -84,6 +88,14 @@ class SerialTab(QWidget):
self.analyze_btn.setToolTip("Load a saved session log or serial capture and diagnose it") self.analyze_btn.setToolTip("Load a saved session log or serial capture and diagnose it")
self.copy_btn = QPushButton("Copy report") self.copy_btn = QPushButton("Copy report")
self.copy_btn.setToolTip("Copy the diagnosis as text (e.g. to send to the firmware team)") 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() row1 = QHBoxLayout()
for w in (QLabel("Port:"), self.port_combo, self.all_ports, self.refresh_btn, for w in (QLabel("Port:"), self.port_combo, self.all_ports, self.refresh_btn,
@@ -92,7 +104,7 @@ class SerialTab(QWidget):
row1.addStretch(1) row1.addStretch(1)
row2 = QHBoxLayout() row2 = QHBoxLayout()
for w in (QLabel("Show:"), self.filter_combo, self.clear_btn, self.save_btn, self.folder_btn, for w in (QLabel("Show:"), self.filter_combo, self.clear_btn, self.save_btn, self.folder_btn,
self.analyze_btn, self.copy_btn): self.analyze_btn, self.copy_btn, self.service_btn, self.auto_service):
row2.addWidget(w) row2.addWidget(w)
row2.addStretch(1) row2.addStretch(1)
@@ -134,6 +146,7 @@ class SerialTab(QWidget):
self.folder_btn.clicked.connect(self._open_folder) self.folder_btn.clicked.connect(self._open_folder)
self.analyze_btn.clicked.connect(self._analyze_file) self.analyze_btn.clicked.connect(self._analyze_file)
self.copy_btn.clicked.connect(self._copy_report) 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. # Report refresh is throttled; lines can arrive at >100/s.
self._diag_timer = QTimer(self, interval=400) self._diag_timer = QTimer(self, interval=400)
@@ -189,8 +202,25 @@ class SerialTab(QWidget):
self._runner.cancel() self._runner.cancel()
self._diag_dirty = True self._diag_dirty = True
self._session.start(port.mac or port.device) self._session.start(port.mac or port.device)
self._active_port = port
self._manager.connect(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 ------------------------------------ # ---- hand-off with the flash tab ------------------------------------
@property @property
def engine(self) -> DiagnosticsEngine: def engine(self) -> DiagnosticsEngine:
@@ -199,8 +229,10 @@ class SerialTab(QWidget):
def release_port(self) -> None: def release_port(self) -> None:
"""Stop monitoring so esptool can open the port exclusively.""" """Stop monitoring so esptool can open the port exclusively."""
if self._manager.is_running: 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 self._manager.disconnect() # joins the reader thread
QCoreApplication.processEvents() # deliver its queued final lines/state now QCoreApplication.processEvents() # deliver its queued final lines/state now
self._no_auto_service = False
def connect_port(self, port: PortInfo, capture: bool = True) -> None: def connect_port(self, port: PortInfo, capture: bool = True) -> None:
"""Monitor `port`; with capture, restart the box once connected to record its boot.""" """Monitor `port`; with capture, restart the box once connected to record its boot."""
@@ -223,6 +255,9 @@ class SerialTab(QWidget):
QTimer.singleShot(300, self._manager.reset_board) QTimer.singleShot(300, self._manager.reset_board)
if not running and self._session.path is not None: if not running and self._session.path is not None:
self._session.write_report(self._engine.report().to_text()) 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._session.stop()
self.connect_btn.setText("Disconnect" if running else "Connect") self.connect_btn.setText("Disconnect" if running else "Connect")
self.reset_btn.setEnabled(state == State.CONNECTED) self.reset_btn.setEnabled(state == State.CONNECTED)
@@ -351,7 +386,22 @@ class MainWindow(QMainWindow):
tabs.addTab(guide, "Quick start") tabs.addTab(guide, "Quick start")
tabs.addTab(self.serial_tab, "Serial monitor") tabs.addTab(self.serial_tab, "Serial monitor")
tabs.addTab(self.flash_tab, "Flash firmware") tabs.addTab(self.flash_tab, "Flash firmware")
self.sd_tab = SdTab(self.serial_tab)
tabs.addTab(self.sd_tab, "SD card")
self.setCentralWidget(tabs) 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) def closeEvent(self, event): # noqa: N802 (Qt naming)
self.serial_tab.shutdown() self.serial_tab.shutdown()
+304
View File
@@ -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 <code>{path}</code>")
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("<br>".join(f"<span style='color:{colour[i.level]}'>{i.level.upper()}: {i.text}</span>"
for i in issues) or "<span style='color:#2e9e44'>Settings OK.</span>")
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: "<b>The connected box runs the factory v1.0 firmware, which ignores these settings "
"(it only uses Matter).</b> 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"<b>{out.message}</b><br>Insert the next card and press Refresh cards to "
"provision another one.", "ok")
else:
self._show_result(f"<b>{out.message}</b>", "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)
+1 -1
View File
@@ -1,6 +1,6 @@
[project] [project]
name = "kota-technician-tool" name = "kota-technician-tool"
version = "0.4.0" version = "0.5.0"
description = "KOTA sensorbox technician diagnostic tool (demo)" description = "KOTA sensorbox technician diagnostic tool (demo)"
requires-python = ">=3.11,<3.14" requires-python = ">=3.11,<3.14"
dependencies = [ dependencies = [
+1 -1
View File
@@ -3,7 +3,7 @@
cd "$(dirname "$0")" || exit 1 cd "$(dirname "$0")" || exit 1
# Keep the virtual environment out of the (Google Drive) tool folder. # 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" export PATH="$HOME/.local/bin:$PATH"
if ! command -v uv >/dev/null 2>&1; then if ! command -v uv >/dev/null 2>&1; then
+1 -1
View File
@@ -3,7 +3,7 @@ setlocal
cd /d "%~dp0" cd /d "%~dp0"
rem Keep the virtual environment out of the (Google Drive) tool folder. 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%" set "PATH=%USERPROFILE%\.local\bin;%PATH%"
where uv >nul 2>nul where uv >nul 2>nul
+85 -1
View File
@@ -52,7 +52,7 @@ def test_modified_image_is_refused(tmp_path):
@pytest.mark.parametrize("patch,msg", [ @pytest.mark.parametrize("patch,msg", [
((0, 0x00), "not a merged"), ((0, 0x00), "not an ESP32 firmware"),
((12, 0x05), "not ESP32-S3"), ((12, 0x05), "not ESP32-S3"),
]) ])
def test_wrong_images_are_rejected(tmp_path, patch, msg): 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) res, *_ = _run(plan, cmd)
assert not res.ok and res.failed_step == 0 assert not res.ok and res.failed_step == 0
assert "BOOT" in res.hint 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)
+159
View File
@@ -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")
+91
View File
@@ -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"