kkallasandClaude Opus 5.5
0d21c5c205
Some files had been saved with CRLF on Windows, so a one-line change showed as a whole-file diff. .gitattributes now stores text as LF (.bat stays CRLF for cmd, .command/.sh LF for macOS). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
262 lines
12 KiB
Python
262 lines
12 KiB
Python
"""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 kota_tech.sd.win_format
|
|
runs elevated (UAC prompt). It re-checks the disk (number, size, bus, not the
|
|
Windows drive) right before erasing it, so a card swapped after the
|
|
technician confirmed is not touched, and it writes MBR + FAT32 itself
|
|
(no Storage cmdlets, which stalled on empty card-reader slots). Larger cards
|
|
get a 32 GB FAT32 partition (Windows convention; 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 Disk
|
|
from .envfile import BUFFER_NAME, ENV_NAME, EnvCheck, EnvConfig, find_case_insensitive, parse_env
|
|
|
|
VOLUME_LABEL = "KOTA"
|
|
|
|
|
|
@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
|
|
|
|
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 win_format_command(disk: Disk, result: Path) -> tuple[str, str, str]:
|
|
"""(program, parameters, working dir) for the elevated helper."""
|
|
root = Path(__file__).resolve().parents[2] # the version folder holding kota_tech/
|
|
params = f'-m kota_tech.sd.win_format --disk {int(disk.id)} --size {int(disk.size)} ' \
|
|
f'--label {VOLUME_LABEL} --result "{result}"'
|
|
return sys.executable, params, str(root)
|
|
|
|
|
|
def _run_elevated(program: str, params: str, cwd: str, timeout_s: int = 300) -> int | None:
|
|
"""ShellExecuteEx 'runas' + wait. None if the technician refused the UAC prompt."""
|
|
import ctypes
|
|
from ctypes import wintypes
|
|
|
|
class SHELLEXECUTEINFOW(ctypes.Structure):
|
|
_fields_ = [("cbSize", wintypes.DWORD), ("fMask", ctypes.c_ulong), ("hwnd", wintypes.HWND),
|
|
("lpVerb", wintypes.LPCWSTR), ("lpFile", wintypes.LPCWSTR), ("lpParameters", wintypes.LPCWSTR),
|
|
("lpDirectory", wintypes.LPCWSTR), ("nShow", ctypes.c_int), ("hInstApp", wintypes.HINSTANCE),
|
|
("lpIDList", ctypes.c_void_p), ("lpClass", wintypes.LPCWSTR), ("hkeyClass", wintypes.HKEY),
|
|
("dwHotKey", wintypes.DWORD), ("hIconOrMonitor", wintypes.HANDLE), ("hProcess", wintypes.HANDLE)]
|
|
|
|
SEE_MASK_NOCLOSEPROCESS, SW_HIDE = 0x40, 0
|
|
info = SHELLEXECUTEINFOW(cbSize=ctypes.sizeof(SHELLEXECUTEINFOW), fMask=SEE_MASK_NOCLOSEPROCESS,
|
|
lpVerb="runas", lpFile=program, lpParameters=params, lpDirectory=cwd, nShow=SW_HIDE)
|
|
shell32 = ctypes.WinDLL("shell32", use_last_error=True)
|
|
k32 = ctypes.WinDLL("kernel32", use_last_error=True)
|
|
if not shell32.ShellExecuteExW(ctypes.byref(info)):
|
|
return None # ERROR_CANCELLED (1223) or failure
|
|
k32.WaitForSingleObject(info.hProcess, timeout_s * 1000)
|
|
code = wintypes.DWORD()
|
|
k32.GetExitCodeProcess(info.hProcess, ctypes.byref(code))
|
|
k32.CloseHandle(info.hProcess)
|
|
return code.value
|
|
|
|
|
|
def _format_windows(disk: Disk, log: list[str]) -> Outcome:
|
|
result = _workdir() / "result.json"
|
|
log.append("Asking Windows for administrator rights to format the card (UAC prompt)...")
|
|
t = time.perf_counter()
|
|
code = _run_elevated(*win_format_command(disk, result))
|
|
if code is None:
|
|
return Outcome(False, "Administrator permission was refused - the card was not formatted.", log=log)
|
|
try:
|
|
res = json.loads(result.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError) as exc:
|
|
return Outcome(False, f"The formatter did not report back ({exc.__class__.__name__}, exit {code}). "
|
|
"Check the card with 'Check card'.", log=log)
|
|
steps = ", ".join(f"{k} {v:.1f} s" for k, v in res.get("timings", {}).items())
|
|
log.append(f"{res.get('message', '')} ({time.perf_counter() - t:.1f} s incl. permission prompt; {steps})")
|
|
return Outcome(bool(res.get("ok")), res.get("message", ""), res.get("drive", ""), log)
|
|
|
|
|
|
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] = []
|
|
# Windows says "FAT32"; macOS diskutil says "DOS_FAT_32" / "Windows_FAT_32" / "MS-DOS FAT32".
|
|
fs = filesystem.upper().replace("_", "").replace(" ", "")
|
|
if fs and "FAT32" not in fs and fs not in ("FAT", "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)
|