diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/harness/port_new_device.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/port_new_device.py | 726 |
1 files changed, 726 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/harness/port_new_device.py b/debug/risc-v-gsp-rm-vm/harness/port_new_device.py new file mode 100644 index 0000000..56522ec --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/port_new_device.py @@ -0,0 +1,726 @@ +#!/usr/bin/env python3 +"""New-device port scaffold (roadmap H4.1/R6, docs/07 P2, docs/22 §3, docs/28). + +``--device <name>`` creates or uses a profile JSON holding the user-supplied +device facts (PCI IDs, BARs, FB geometry, engine inventory, interrupt topology, +ROM/VBIOS notes) and then drives, on the local QEMU backend only: + +1. probe-first-touch on the device's firmware (first unmapped access, bounded); +2. a symbol inventory (ELF .symtab + optional Ghidra/green-eye DB sources); +3. a portability matrix run of the device against a reference firmware; +4. a port checklist report against the docs/07 evidence checklist. + +``--dry-run`` is the default: it prints the plan (profile action, missing +profile fields, steps, expected artifacts) and exits 0 without creating files. +``--run`` writes an evidence bundle under ``--out`` with a sha256 manifest. +The RM 610 gate from ``version_diff.py`` (docs/23) is enforced before any file +is opened, in dry-run too. + +This tool never touches hardware: no /dev, no kernel module, no mdev, no +guest, no physical GPU. Physical validation is a separate supervised step +(docs/06-portability/hardware-software-matrix.md §5.1). + +Run from the tool root: + + python3 harness/port_new_device.py --device rtx4090 --profile profiles/rtx4090.json + python3 harness/port_new_device.py --run --device rtx4090 \\ + --firmware ../../firmware/57X-GSP/570.144/gsp-570.144.fwimage.bin \\ + --reference firmware/uproc_demo/build/uproc_demo.elf --out dumps/port-rtx4090 +""" +from __future__ import annotations + +import argparse +import json +import os +import re +import sys +from dataclasses import dataclass + +try: + from .common import ( + MAX_LOG_BYTES, + SymbolDB, + ensure_dir, + qemu_available, + sha256_file, + write_json, + write_text, + ) + from .portability_matrix import ( + MatrixOptions, + TargetSpec, + render_markdown as render_matrix_markdown, + run_matrix, + ) + from .version_diff import ( + RM610_TOKEN, + Rm610GateError, + VersionDiffOptions, + check_rm610_gate, + load_any_target, + load_symbol_sources, + run_probe, + ) +except ImportError: # direct script execution (python3 harness/port_new_device.py) + from common import ( # type: ignore[no-redef] + MAX_LOG_BYTES, + SymbolDB, + ensure_dir, + qemu_available, + sha256_file, + write_json, + write_text, + ) + from portability_matrix import ( # type: ignore[no-redef] + MatrixOptions, + TargetSpec, + render_markdown as render_matrix_markdown, + run_matrix, + ) + from version_diff import ( # type: ignore[no-redef] + RM610_TOKEN, + Rm610GateError, + VersionDiffOptions, + check_rm610_gate, + load_any_target, + load_symbol_sources, + run_probe, + ) + +FORMAT = "greeneye-device-port/1" +PLAN_FORMAT = "greeneye-device-port-plan/1" +PROFILE_FORMAT = "greeneye-device-profile/1" +BUNDLE_FORMAT = "greeneye-release-bundle/1" +DEVICE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$") + +#: Where each user-supplied profile section comes from (docs/22 §3.1). +PROFILE_HELP = { + "pci": "lspci / vendor docs, config/hardware/*.yaml is a design target only", + "bars": "guest lspci -vvv, host sysfs", + "fb": "host oracle: rmtool mem stats --format=json", + "engines": "host oracle: rmtool engine list --format=json; validate per driver profile", + "interrupts": "MSI-X tables, guest lspci -vvv", + "rom": "ROM/VBIOS read; laptop/Surface overrides", + "firmware": "extracted GSP .fwimage or re-hosted ELF (user-supplied; never downloaded)", + "pairing": "verify image family + FMC manifest + version string before debugging (docs/07 P2.2)", + "physical": "supervised physical validation only; this tool is simulator-only (docs/22 §3.5)", +} + + +@dataclass +class DeviceOptions: + device: str + profile: str | None = None + firmware: str | None = None + reference: str | None = None + symbols: str | None = None + ghidra_exports: str | None = None + ghidra_bias: int = 0 + out: str | None = None + run: bool = False + approval_token: str | None = None + probe_instructions: int = 2000 + probe_duration: float = 0.5 + scenario_duration: float = 0.4 + max_stops: int = 5000 + max_rows: int = 40 + step_timeout: float = 600.0 + log_bytes: int = MAX_LOG_BYTES + + @property + def dry_run(self) -> bool: + return not self.run + + +# --------------------------------------------------------------------------- +# profile +# --------------------------------------------------------------------------- + + +def scaffold_profile(name: str) -> dict: + """Empty user-supplied profile; every field documents where it comes from.""" + return { + "format": PROFILE_FORMAT, + "name": name, + "vendor": "", + "device": "", + "pci": {"vendor_id": "0x10de", "device_id": "", "subsystem_vendor_id": "", + "subsystem_device_id": ""}, + "bars": [], + "fb": {"vram_mib": None, "bar1_mib": None, "ecc": None, "notes": ""}, + "engines": [], + "interrupts": {"msix_vectors": None, "notes": ""}, + "rom": {"vblos_sha256": "", "notes": ""}, + "firmware": "", + "firmware_sha256": "", + "firmware_family": "", + "firmware_version": "", + "reference": "", + "physical_validation": {"status": "pending", "date": "", "by": "", "evidence": "", "notes": ""}, + "notes": "", + "where_to_fill": dict(PROFILE_HELP), + "references": [ + "docs/07-PORTABILITY-PLAYBOOKS.md P2", + "docs/22-PORTING-CUDA-AND-DEVICES.md 3", + "docs/06-portability/hardware-software-matrix.md 5.1", + ], + } + + +def load_profile(path: str) -> tuple[dict, bool]: + """Return ``(profile, created)``; an unreadable profile is a clean error.""" + if not os.path.isfile(path): + return {}, False + with open(path, "r", encoding="utf-8") as fh: + data = json.load(fh) + if not isinstance(data, dict): + raise ValueError(f"profile {path}: expected a JSON object") + return data, True + + +def profile_missing_fields(profile: dict) -> list[str]: + missing: list[str] = [] + pci = profile.get("pci") or {} + if not any(str(pci.get(key, "")).strip() for key in ("vendor_id", "device_id", "subsystem_device_id")): + missing.append("pci") + if not profile.get("bars"): + missing.append("bars") + fb = profile.get("fb") or {} + if not any(fb.get(key) not in (None, "", 0) for key in ("vram_mib", "bar1_mib", "notes")): + missing.append("fb") + if not profile.get("engines"): + missing.append("engines") + interrupts = profile.get("interrupts") or {} + if not any(interrupts.get(key) not in (None, "", 0) for key in ("msix_vectors", "notes")): + missing.append("interrupts") + rom = profile.get("rom") or {} + if not any(str(rom.get(key, "")).strip() for key in ("vblos_sha256", "notes")): + missing.append("rom") + return missing + + +def profile_labels(profile: dict, device: str, profile_path: str | None) -> list[str]: + labels = [device, os.path.basename(profile_path or "")] + for key in ("name", "device", "firmware", "reference", "firmware_family", "firmware_version"): + value = profile.get(key) + if isinstance(value, str) and value: + labels.append(os.path.basename(value)) + return labels + + +def merge_profile(profile: dict, options: DeviceOptions) -> dict: + merged = json.loads(json.dumps(profile)) + if options.firmware: + merged["firmware"] = options.firmware + if options.reference: + merged["reference"] = options.reference + return merged + + +# --------------------------------------------------------------------------- +# plan (dry run) +# --------------------------------------------------------------------------- + + +def _expected_artifacts(out: str, device: str) -> list[str]: + base = os.path.join(out, device) + return [ + os.path.join(base, "device-port.json"), + os.path.join(base, "device-port.md"), + os.path.join(base, "port-checklist.json"), + os.path.join(base, "port-checklist.md"), + os.path.join(base, "symbols", f"{device}.json"), + os.path.join(base, "sessions"), + os.path.join(base, "portability-matrix", "portability-report.json"), + os.path.join(base, "portability-matrix", "portability-report.md"), + os.path.join(base, "bundle-manifest.json"), + ] + + +def build_plan(options: DeviceOptions) -> dict: + profile_path = options.profile or (os.path.join(options.out or "<out>", "profiles", + f"{options.device}.json")) + profile, created = load_profile(profile_path) if os.path.isfile(profile_path) else ({}, False) + merged = merge_profile(profile if created else scaffold_profile(options.device), options) + firmware = merged.get("firmware") or "" + reference = merged.get("reference") or "" + check_rm610_gate( + profile_labels(profile if created else {}, options.device, profile_path) + + [os.path.basename(firmware), os.path.basename(reference)], + approval_token=options.approval_token, + ) + out = os.path.abspath(options.out) if options.out else "<out>" + return { + "format": PLAN_FORMAT, + "tool": "harness/port_new_device.py", + "dry_run": True, + "rm610_gate": {"status": "passed", "labels": sorted(set( + profile_labels(profile if created else {}, options.device, profile_path) + + [os.path.basename(firmware), os.path.basename(reference)] + ))}, + "device": options.device, + "profile": { + "path": profile_path, + "action": "use" if created else "would-create-scaffold", + "missing_fields": profile_missing_fields(merged), + "where_to_fill": PROFILE_HELP, + }, + "firmware": {"path": firmware or None, "found": bool(firmware) and os.path.isfile(firmware)}, + "reference": {"path": reference or None, + "found": bool(reference) and os.path.isfile(reference)}, + "out": out, + "steps": [ + {"name": "probe_first_touch", "target": firmware or None, + "out": os.path.join(out, options.device, "sessions", f"{options.device}-probe")}, + {"name": "symbol_inventory", "sources": [firmware or None, options.symbols]}, + {"name": "portability_matrix", "targets": [reference or None, firmware or None]}, + {"name": "port_checklist", "out": os.path.join(out, options.device, "port-checklist.json")}, + ], + "expected_artifacts": _expected_artifacts(out, options.device), + "interpretation": ( + "dry-run: fill the profile fields, then pass --run; physical validation " + "is a separate supervised step and is never done by this tool" + ), + } + + +# --------------------------------------------------------------------------- +# steps +# --------------------------------------------------------------------------- + + +def _run_step(name: str, fn, *args, **kwargs) -> dict: + try: + detail = fn(*args, **kwargs) + except Exception as exc: # a failed step must not hide the others + return {"name": name, "status": "error", "reason": f"{type(exc).__name__}: {exc}"} + entry = {"name": name} + entry.update(detail) + return entry + + +def step_probe(loaded, db: SymbolDB, options: DeviceOptions, out_dir: str, qemu_ok: bool) -> dict: + vd = VersionDiffOptions( + old=loaded.path, + new=loaded.path, + out=out_dir, + probe_instructions=options.probe_instructions, + probe_duration=options.probe_duration, + log_bytes=options.log_bytes, + approval_token=options.approval_token, + ) + result = run_probe(loaded, db, vd, out_dir, qemu_ok) + return { + "status": result["status"], + "reason": result.get("reason"), + "session": result.get("session"), + "stopped_pc": result.get("stopped_pc"), + "stopped_symbol": result.get("stopped_symbol"), + "fault_count": result.get("fault_count"), + "first_fault": result.get("first_fault"), + } + + +def step_symbols(loaded, db: SymbolDB, sources: list[str], notes: list[str], + options: DeviceOptions, out_dir: str) -> dict: + records = [ + {"name": symbol.name, "addr": symbol.addr, "size": symbol.size, "source": symbol.source} + for symbol in db + ] + rel = None + if records: + rel = os.path.join("symbols", f"{options.device}.json") + path = os.path.join(out_dir, rel) + ensure_dir(os.path.dirname(path)) + db.save_json(path, meta={ + "device": options.device, + "firmware": os.path.basename(loaded.path), + "bias": loaded.bias, + "sources": sources, + "notes": notes, + }) + return { + "status": "ok" if records else "skipped", + "reason": None if records else "no symbols: no ELF .symtab and no Ghidra/JSON sources supplied", + "count": len(records), + "sources": sources, + "notes": notes, + "db": rel, + } + + +def step_matrix(firmware: str, reference: str | None, options: DeviceOptions, out_dir: str) -> dict: + specs = [] + if reference and os.path.isfile(reference): + specs.append(TargetSpec(path=os.path.abspath(reference), name="reference")) + specs.append(TargetSpec(path=os.path.abspath(firmware), name=options.device)) + matrix = run_matrix( + specs, + MatrixOptions( + out=out_dir, + probe_instructions=options.probe_instructions, + probe_duration=options.probe_duration, + scenario_duration=options.scenario_duration, + max_stops=options.max_stops, + max_rows=options.max_rows, + log_bytes=options.log_bytes, + ), + ) + write_json(os.path.join(out_dir, "portability-report.json"), matrix) + write_text( + os.path.join(out_dir, "portability-report.md"), + render_matrix_markdown(matrix, MatrixOptions(out=out_dir, max_rows=options.max_rows)), + ) + statuses = {row["name"]: row["status"] for row in matrix["targets"]} + return { + "status": "error" if "error" in statuses.values() else "ok", + "report": "portability-matrix/portability-report.json", + "targets": statuses, + "comparisons": [ + {"a": row["a"], "b": row["b"], "status": row["status"], "all_zero": row["all_zero"]} + for row in matrix["comparisons"] + ], + } + + +def build_checklist(profile: dict, firmware: str, reference: str | None, results: dict) -> dict: + """docs/07 evidence checklist + docs/22 §3.1 profile items.""" + missing = profile_missing_fields(profile) + pairing_ok = bool(str(profile.get("firmware_family", "")).strip() + and str(profile.get("firmware_version", "")).strip()) + physical = profile.get("physical_validation") or {} + physical_status = str(physical.get("status", "pending")).lower() + probe = results.get("probe") or {} + symbols = results.get("symbols") or {} + matrix = results.get("matrix") or {} + items = [ + { + "id": "profile", + "item": "device profile fields recorded (PCI IDs, BARs, FB, engines, interrupts, ROM)", + "status": "ok" if not missing else "pending", + "detail": "missing: " + ", ".join(missing) if missing else "all profile sections populated", + }, + { + "id": "pairing", + "item": "firmware hash + family/version pairing recorded before debugging (docs/07 P2.2)", + "status": "ok" if pairing_ok else "pending", + "detail": f"family={profile.get('firmware_family') or '-'} version={profile.get('firmware_version') or '-'}", + }, + { + "id": "hash", + "item": "firmware sha256 recorded", + "status": "ok" if firmware and os.path.isfile(firmware) else "missing", + "detail": sha256_file(firmware) if firmware and os.path.isfile(firmware) else "firmware not found", + }, + { + "id": "probe", + "item": "probe first-touch captured and explained as profile vs logic", + "status": "ok" if probe.get("status") == "ok" else "pending", + "detail": f"stopped_pc={probe.get('stopped_pc')} faults={probe.get('fault_count')}", + }, + { + "id": "symbols", + "item": "symbol inventory built with provenance", + "status": "ok" if symbols.get("status") == "ok" else "pending", + "detail": f"count={symbols.get('count')} sources={symbols.get('sources')}", + }, + { + "id": "matrix", + "item": "portability matrix run vs the reference firmware", + "status": "ok" if matrix.get("status") == "ok" and reference else "pending", + "detail": f"targets={matrix.get('targets')}", + }, + { + "id": "docs07", + "item": "docs/07 evidence checklist attachable (hash, probe, symbols, sessions, compare, M1-M6)", + "status": "ok" if firmware else "missing", + "detail": "bundle-manifest.json hashes every artifact", + }, + { + "id": "physical", + "item": "physical/supervised validation done (outside this simulator-only tool)", + "status": "ok" if physical_status in ("passed", "ok", "done") else "pending", + "detail": f"status={physical_status}; docs/22 §3.5: simulator-only evidence never justifies the support claim", + }, + ] + required_pending = [row["id"] for row in items + if row["status"] != "ok" and row["id"] != "physical"] + if required_pending: + recommendation = "hold: " + ", ".join(required_pending) + " not complete" + elif physical_status in ("passed", "ok", "done"): + recommendation = "go: simulator evidence complete and physical validation recorded" + else: + recommendation = "simulator-evidence-complete; physical validation pending (supervised step)" + return { + "format": "greeneye-device-checklist/1", + "tool": "harness/port_new_device.py", + "device": profile.get("name") or "-", + "firmware": firmware, + "reference": reference, + "profile_missing_fields": missing, + "items": items, + "recommendation": recommendation, + "safety": "simulator-only: no /dev, no kernel module, no mdev, no guest, no physical GPU", + } + + +def render_checklist_markdown(checklist: dict) -> str: + lines = [ + f"# green-eye port checklist: {checklist['device']}", + "", + "> TOOL IN DEVELOPMENT - NOT READY YET! Generated by `harness/port_new_device.py`.", + "", + f"* firmware: `{checklist['firmware']}`", + f"* reference: `{checklist['reference']}`", + f"* recommendation: **{checklist['recommendation']}**", + f"* safety: {checklist['safety']}", + "", + "| id | item | status | detail |", + "|---|---|---|---|", + ] + for row in checklist["items"]: + lines.append(f"| {row['id']} | {row['item']} | {row['status']} | {row['detail']} |") + lines += ["", "Evidence: `port-checklist.json`, `bundle-manifest.json`, `portability-matrix/`.", ""] + return "\n".join(lines) + + +def build_bundle(out_dir: str) -> dict: + files: list[dict] = [] + for root, dirs, names in os.walk(out_dir): + dirs.sort() + for name in sorted(names): + path = os.path.join(root, name) + rel = os.path.relpath(path, out_dir) + if rel == "bundle-manifest.json": + continue + files.append({"path": rel, "bytes": os.path.getsize(path), "sha256": sha256_file(path)}) + files.sort(key=lambda row: row["path"]) + return {"format": BUNDLE_FORMAT, "tool": "harness/port_new_device.py", "files": files} + + +def run_device_port(options: DeviceOptions) -> dict: + if options.out is None: + raise ValueError("--out is required unless --dry-run is used") + out_root = os.path.abspath(options.out) + device_dir = os.path.join(out_root, options.device) + profile_path = options.profile or os.path.join(out_root, "profiles", f"{options.device}.json") + profile, created = load_profile(profile_path) + created_now = not created + if not created: + profile = scaffold_profile(options.device) + profile = merge_profile(profile, options) + labels = profile_labels(profile, options.device, profile_path) + firmware = profile.get("firmware") or "" + reference = profile.get("reference") or "" + check_rm610_gate(labels + [os.path.basename(firmware), os.path.basename(reference)], + approval_token=options.approval_token) + if created_now: + ensure_dir(os.path.dirname(profile_path)) + write_json(profile_path, profile) + if not firmware: + raise ValueError( + "no device firmware: pass --firmware PATH or set \"firmware\" in the profile " + f"({profile_path}); see docs/07 P2" + ) + loaded = load_any_target(firmware) + if not loaded.ok: + raise ValueError(f"firmware not usable: {loaded.path}: {loaded.reason or loaded.kind}") + if created_now: + profile["firmware_sha256"] = loaded.sha256 + write_json(profile_path, profile) + + ensure_dir(device_dir) + db, sources, notes = load_symbol_sources( + loaded, + symbols_json=options.symbols, + ghidra_exports=options.ghidra_exports, + ghidra_bias=options.ghidra_bias, + ) + qemu_ok = qemu_available() + results: dict = {} + steps = [ + _run_step("probe_first_touch", step_probe, loaded, db, options, device_dir, qemu_ok), + _run_step("symbol_inventory", step_symbols, loaded, db, sources, notes, options, device_dir), + _run_step("portability_matrix", step_matrix, firmware, reference, options, + os.path.join(device_dir, "portability-matrix")), + ] + results["probe"] = steps[0] + results["symbols"] = steps[1] + results["matrix"] = steps[2] + checklist = build_checklist(profile, firmware, reference, results) + write_json(os.path.join(device_dir, "port-checklist.json"), checklist) + write_text(os.path.join(device_dir, "port-checklist.md"), render_checklist_markdown(checklist)) + checklist_step = { + "name": "port_checklist", + "status": "ok", + "report": "port-checklist.json", + "recommendation": checklist["recommendation"], + } + steps.append(checklist_step) + failed = [step["name"] for step in steps if step["status"] == "error"] + report = { + "format": FORMAT, + "tool": "harness/port_new_device.py", + "dry_run": False, + "rm610_gate": {"status": "passed", "labels": sorted(set(labels))}, + "device": options.device, + "profile": { + "path": os.path.relpath(profile_path, out_root), + "created": created_now, + "missing_fields": profile_missing_fields(profile), + }, + "firmware": {"path": firmware, "kind": loaded.kind, "sha256": loaded.sha256}, + "reference": {"path": reference, "found": bool(reference) and os.path.isfile(reference)}, + "qemu_available": qemu_ok, + "steps": steps, + "checklist": checklist, + "failed": failed, + "manifest": "bundle-manifest.json", + } + write_json(os.path.join(device_dir, "device-port.json"), report) + write_text(os.path.join(device_dir, "device-port.md"), render_report_markdown(report)) + write_json(os.path.join(device_dir, "bundle-manifest.json"), build_bundle(device_dir)) + return report + + +def render_report_markdown(report: dict) -> str: + lines = [ + f"# green-eye device port: {report['device']}", + "", + "> TOOL IN DEVELOPMENT - NOT READY YET! Generated by `harness/port_new_device.py`.", + "", + f"* format: `{report['format']}`", + f"* firmware: `{report['firmware']['path']}` ({report['firmware']['kind']}, " + f"sha256 `{report['firmware']['sha256'][:16]}`)", + f"* reference: `{report['reference']['path'] or '-'}`", + f"* qemu available: {report['qemu_available']}", + "", + "| step | status | report | reason |", + "|---|---|---|---|", + ] + for step in report["steps"]: + lines.append(f"| {step['name']} | {step['status']} | {step.get('report') or '-'} " + f"| {step.get('reason') or '-'} |") + lines += [ + "", + f"* checklist recommendation: **{report['checklist']['recommendation']}**", + f"* manifest: `{report['manifest']}`", + "", + ] + return "\n".join(lines) + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +_HELP_EPILOG = """\ +invocation: + dry-run first (default; nothing is created or touched): + python3 harness/port_new_device.py --device rtx4090 --profile profiles/rtx4090.json + execute (scaffolds the profile if absent; simulator only): + python3 harness/port_new_device.py --run --device rtx4090 \\ + --firmware ../../firmware/57X-GSP/570.144/gsp-570.144.fwimage.bin \\ + --reference firmware/uproc_demo/build/uproc_demo.elf --out dumps/port-rtx4090 + +agent (MCP, docs/20): read the dry-run plan, fill the profile, then --run; +read device-port.json and port-checklist.json from the bundle. +RM 610 (docs/23): rm610-labelled device/firmware labels are refused without +--approval-token RM610-APPROVED or GREENEYE_RM610_APPROVED=1. +Safety: no /dev, no kernel module, no mdev, no guest, no physical GPU. +""" + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="port_new_device", + description=__doc__.splitlines()[0], + epilog=_HELP_EPILOG, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("--device", required=True, help="device name (profile key, e.g. rtx4090)") + parser.add_argument("--profile", help="profile JSON path (default: <out>/profiles/<device>.json)") + parser.add_argument("--firmware", help="device GSP firmware image or re-hosted ELF (overrides profile)") + parser.add_argument("--reference", help="reference firmware for the portability matrix") + parser.add_argument("--symbols", help="green-eye symbol DB JSON for the device firmware") + parser.add_argument("--ghidra-exports", help="Ghidra exports TSV for the device firmware") + parser.add_argument("--ghidra-bias", type=lambda value: int(value, 0), default=0) + parser.add_argument("--out", help="evidence bundle directory (required for --run)") + parser.add_argument("--run", dest="run", action="store_true", + help="execute (default is the dry-run plan)") + parser.add_argument("--dry-run", dest="run", action="store_false", + help="print the plan and exit 0 (default)") + parser.set_defaults(run=False) + parser.add_argument("--approval-token", help=f"explicit RM610 token ({RM610_TOKEN})") + parser.add_argument("--probe-instructions", type=int, default=2000) + parser.add_argument("--probe-duration", type=float, default=0.5) + parser.add_argument("--scenario-duration", type=float, default=0.4) + parser.add_argument("--max-stops", type=int, default=5000) + parser.add_argument("--max-rows", type=int, default=40) + parser.add_argument("--step-timeout", type=float, default=600.0) + parser.add_argument("--log-bytes", type=int, default=MAX_LOG_BYTES) + return parser + + +def options_from_args(args: argparse.Namespace) -> DeviceOptions: + if not DEVICE_RE.match(args.device): + raise ValueError(f"invalid device name: {args.device!r}") + return DeviceOptions( + device=args.device, + profile=args.profile, + firmware=args.firmware, + reference=args.reference, + symbols=args.symbols, + ghidra_exports=args.ghidra_exports, + ghidra_bias=args.ghidra_bias, + out=args.out, + run=args.run, + approval_token=args.approval_token, + probe_instructions=args.probe_instructions, + probe_duration=args.probe_duration, + scenario_duration=args.scenario_duration, + max_stops=args.max_stops, + max_rows=args.max_rows, + step_timeout=args.step_timeout, + log_bytes=args.log_bytes, + ) + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + try: + options = options_from_args(args) + except ValueError as exc: + parser.error(str(exc)) + try: + check_rm610_gate( + [options.device, os.path.basename(options.profile or ""), + os.path.basename(options.firmware or ""), os.path.basename(options.reference or "")], + approval_token=options.approval_token, + ) + if options.dry_run: + plan = build_plan(options) + print(json.dumps(plan, indent=2, sort_keys=True)) + print(f"[device] dry-run: profile {plan['profile']['action']}, " + f"firmware found={plan['firmware']['found']}; nothing created or touched") + return 0 + report = run_device_port(options) + except Rm610GateError as exc: + print(f"[device] RM610 gate refused: {exc}", file=sys.stderr) + return 3 + except (ValueError, FileNotFoundError, json.JSONDecodeError) as exc: + print(f"[device] error: {exc}", file=sys.stderr) + return 2 + for step in report["steps"]: + reason = f" reason={step['reason']}" if step.get("reason") else "" + print(f"[device] {step['name']}: status={step['status']}{reason}") + print(f"[device] checklist: {report['checklist']['recommendation']}") + print(f"[device] wrote {os.path.join(os.path.abspath(options.out), options.device, 'device-port.json')}") + print(f"[device] wrote {os.path.join(os.path.abspath(options.out), options.device, 'bundle-manifest.json')}") + return 1 if report["failed"] else 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) |