#!/usr/bin/env python3 """Unlock one CMP 170HX, then assess its PCIe link and optional diagnostics. Exit codes: 0 PASS/WARN, 1 FAIL, 2 no card, 3 preflight blocked, 4 cold reboot required before checking the card. """ import argparse import datetime import json import shutil import sys import time from pathlib import Path import cxlib as X from cxlib import ask, detect_gpus, err, info, now_iso, ok, sanitize, step, warn # ---------------------------------------------------------------- scoring def score(rep): fails, warns = [], [] if rep.get("unlock", {}).get("status") != "unlocked": fails.append("cmpunlocker memory unlock is not active") if not rep.get("serial"): fails.append("physical card serial unavailable; diagnostics kept under unidentified/") if not rep.get("bdf"): fails.append("no CMP 170HX device detected on the bus") link = rep.get("link", {}) if rep.get("bdf"): if link.get("lspci_rc") not in (0, None): fails.append("lspci could not read the device") if not link_passes(link): generation = f"Gen{link['lnksta_gen']}" if link.get("lnksta_gen") else "unknown generation" fails.append(f"PCIe {generation} {link.get('lnksta_width') or 'unknown width'}; " "Gen2 x16 required after cmpunlocker (check unlock, mod, slot and riser)") if rep.get("rescan", {}).get("ok") is False: fails.append("device lost after bus rescan") if rep.get("retrain", {}).get("ok") is False: fails.append("link did not survive retrain") if rep.get("link_after") and not link_passes(rep["link_after"]): fails.append("link does not report Gen2 x16 after diagnostics") aer = rep.get("aer_after", {}) aer_before = rep.get("aer_before", {}) if (aer.get("ce_active") or aer.get("parent_ce_active") or (aer.get("aer_correctable_sum", 0) > aer_before.get("aer_correctable_sum", 0))): warns.append("correctable AER errors present/incremented") if aer.get("ue_active"): fails.append(f"uncorrectable AER errors active: {aer['ue_active']}") if aer.get("aer_fatal_sum", 0) > aer_before.get("aer_fatal_sum", 0): fails.append("fatal PCIe AER counter increased during the check") dm = rep.get("dmesg", {}) if dm.get("errors"): warns.append(f"{len(dm['errors'])} kernel error line(s) related to this card (see dmesg.log)") vu = rep.get("vulkan_probe", {}) if vu.get("requested") and (not vu.get("ran") or vu.get("rc") != 0): warns.append("optional Vulkan capability probe failed") if rep.get("bios", {}).get("requested") and not rep["bios"].get("ok"): warns.append("VBIOS backup failed/incomplete") ver = rep.get("cmpunlocker_verify", {}) if ver.get("status") not in ("unlocked-ok", "skipped-stale-inventory"): warns.append(f"cmpunlocker verify status: {ver.get('status')}") result = "FAIL" if fails else ("WARN" if warns else "PASS") return result, fails, warns def link_passes(link): return link.get("lnksta_width") == "x16" and link.get("lnksta_gen") == 2 def width_status(link): width = link.get("lnksta_width") return "PASS" if width == "x16" else "FAIL" if width else "UNKNOWN" def log_capacitor_width(link, phase): width = link.get("lnksta_width") speed = link.get("lnksta_speed") or "unknown speed" generation = f"Gen{link['lnksta_gen']}" if link.get("lnksta_gen") else "unknown generation" if width == "x16": ok(f"CAPACITOR MOD WIDTH CHECK PASS ({phase}): PCIe {generation} x16 ({speed}) negotiated") else: err(f"CAPACITOR MOD CHECK FAILED ({phase}): PCIe {generation} {width or 'unknown width'} " f"({speed}) negotiated; expected x16 width. Check the capacitor mod and host slot/riser.") def log_final_link(link, phase="after unlock"): width = link.get("lnksta_width") or "unknown width" generation = f"Gen{link['lnksta_gen']}" if link.get("lnksta_gen") else "unknown generation" if link_passes(link): ok(f"FINAL PCIe CHECK PASS ({phase}): Gen2 x16 negotiated") else: err(f"FINAL PCIe CHECK FAILED ({phase}): PCIe {generation} {width} negotiated; " "Gen2 x16 required after cmpunlocker.") def start_card_folder(serial, bdf): run_id = datetime.datetime.now().strftime("%Y%m%dT%H%M%S%f") folder = (X.CARDS_DIR / sanitize(serial) if serial else X.UNIDENTIFIED_DIR / f"{run_id}_{sanitize(bdf)}") run_dir = folder / "runs" / run_id run_dir.mkdir(parents=True) return folder, run_dir def promote_card_folder(folder, run_dir, serial, rep): """Move a current unidentified run once the physical serial becomes readable.""" target = X.CARDS_DIR / sanitize(serial) if target.exists(): new_run = target / "runs" / run_dir.name new_run.parent.mkdir(parents=True, exist_ok=True) shutil.move(str(run_dir), str(new_run)) extras = folder / "extras" if extras.exists(): destination = target / "extras" destination.mkdir(exist_ok=True) for item in extras.iterdir(): shutil.move(str(item), str(destination / item.name)) shutil.rmtree(folder) else: folder.rename(target) new_run = target / "runs" / run_dir.name if rep.get("unlock", {}).get("install_log"): old_log = Path(rep["unlock"]["install_log"]) rep["unlock"]["install_log"] = str(target / "extras" / old_log.name) rep["unlock"]["note"] = rep["unlock"].get("note", "").replace(str(folder), str(target)) rep["run_dir"] = str(new_run) info(f"physical serial found; moved current diagnostics to {target}") return target, new_run # ---------------------------------------------------------------- card detect def wait_for_gpu(): """Wait for a single CMP 170HX (with rescan retries). Returns BDF or None.""" while True: for _ in range(5): bdfs = detect_gpus() if bdfs: if len(bdfs) > 1: err(f"multiple CMP 170HX devices present: {', '.join(bdfs)}; connect one card") return None return bdfs[0] time.sleep(1) if not sys.stdin.isatty(): return None ans = ask("CMP 170HX not detected - insert/reseat the card, press Enter to rescan " "(q=quit)", "") if ans.lower().startswith("q"): return None if not X.rescan_bus(): warn("bus rescan not permitted (no root/sudo) - reseat the card physically") # ---------------------------------------------------------------- one card def process_card(args): step("Card check") started_at = now_iso() bdf = wait_for_gpu() if not bdf: err("no card detected - nothing to do") return None devid = "" for line in X.run(["lspci", "-D", "-s", bdf, "-nn"])["out"].splitlines(): if "[10de:" in line: i = line.index("[10de:") devid = line[i + 1:i + 10].rstrip("]") ok(f"device found at {bdf} ({devid})") nf = X.nvflash_bin(args) serial, serial_src = X.get_serial(nf, bdf) if not serial: warn("physical card serial unavailable; continuing diagnostics in unidentified/") folder, run_dir = start_card_folder(serial, bdf) info(f"diagnostics folder: {folder}") rep = { "serial": serial, "serial_source": serial_src, "started_at": started_at, "run_dir": str(run_dir), "bdf": bdf, "device_id": devid, "host": X.run(["uname", "-n"])["out"].strip(), "kernel": X.run(["uname", "-r"])["out"].strip(), } step("capacitor lane width before unlock") link_before, lspci_before = X.check_link(bdf) rep["link_before_unlock"] = link_before rep["link"] = link_before rep["capacitor_mod_before_unlock"] = width_status(link_before) rep["capacitor_mod"] = rep["capacitor_mod_before_unlock"] (run_dir / "lspci_before_unlock.txt").write_text(lspci_before) info(link_before.get("lnksta_raw") or "LnkSta: unavailable") log_capacitor_width(link_before, "before unlock") step("cmpunlocker memory unlock") identity = X.card_identity(bdf, devid, serial) rep["gpu_uuid"] = identity.get("gpu_uuid") rep["unlock"] = X.ensure_unlocked(bdf, devid, folder, identity, allow_install=not args.no_install) info(f"unlock: {rep['unlock']['status']} - {rep['unlock'].get('note', '')}") if not serial: found, source = X.get_serial(nf, bdf) if found: folder, run_dir = promote_card_folder(folder, run_dir, found, rep) serial, serial_src = found, source rep.update(serial=serial, serial_source=serial_src) if rep["unlock"]["status"] != "unlocked": rep["result"] = "PENDING_REBOOT" if rep["unlock"]["status"] == "reboot-required" else "FAIL" if rep["result"] == "FAIL": rep["failures"] = [rep["unlock"].get("note", "memory unlock failed")] if not serial: rep["failures"].append("physical card serial unavailable; diagnostics kept under unidentified/") rep["finished_at"] = now_iso() save_report(rep, folder, run_dir) return rep step("cmpunlocker verification") rep["cmpunlocker_verify"] = X.run_verify_unlocker(bdf, devid) v = rep["cmpunlocker_verify"] (run_dir / "verify_unlocker.log").write_text( "\n".join(v.get("tail", [])) or (v.get("note") or "not run")) info("verify: " + (f"rc={v.get('rc')} -> {v.get('status')}" if v.get("ran") else f"skipped - {v.get('note')}")) step("PCIe link + config") link, lspci_txt = X.check_link(bdf) rep["link"] = link rep["capacitor_mod_after_unlock"] = width_status(link) rep["capacitor_mod"] = rep["capacitor_mod_after_unlock"] (run_dir / "lspci.txt").write_text(lspci_txt) (run_dir / "config_space.hex").write_text(X.dump_config(bdf)) info(f"{link.get('lnksta_raw') or 'LnkSta: unavailable'} " f"(setpci gen={link.get('setpci_current_gen')})") log_capacitor_width(link, "after unlock") log_final_link(link) rep["aer_before"] = X.aer_snapshot(bdf) if args.link_tests: step("optional retrain + rescan tests") rep["retrain"] = X.retrain_test(bdf) info(f"retrain: ok={rep['retrain']['ok']} - {rep['retrain']['note']}") rep["rescan"] = X.rescan_test(bdf) info(f"rescan: ok={rep['rescan']['ok']} - {rep['rescan']['note']}") rep["aer_after"] = X.aer_snapshot(bdf) link2, lspci2 = X.check_link(bdf) rep["link_after"] = link2 rep["capacitor_mod_after_diagnostics"] = width_status(link2) if rep["capacitor_mod_after_diagnostics"] != "PASS": rep["capacitor_mod"] = rep["capacitor_mod_after_diagnostics"] if (link2.get("lnksta_width"), link2.get("lnksta_gen")) != (link.get("lnksta_width"), link.get("lnksta_gen")): if link2.get("lnksta_width") != link.get("lnksta_width"): log_capacitor_width(link2, "after diagnostics") log_final_link(link2, "after diagnostics") if lspci2: (run_dir / "lspci_after.txt").write_text(lspci2) step("kernel log scrape") dm = X.dmesg_capture(bdf, rep["started_at"]) rep["dmesg"] = dm (run_dir / "dmesg.log").write_text("\n".join(dm.get("related", []))) (run_dir / "dmesg_other_pcie.log").write_text("\n".join(dm.get("other_pcie", []))) info(f"{dm.get('note')} (source: {dm.get('source')})") for e in dm.get("errors", [])[-5:]: warn(e.strip()[:160]) step("nvidia-smi") rep["nvidia_smi"] = X.nvidia_smi_info() if args.bios: step("optional VBIOS backup") extras = folder / "extras" extras.mkdir(exist_ok=True) rep["bios"] = X.save_bios(nf, extras / "bios.rom", bdf) rep["bios"]["requested"] = True (ok if rep["bios"]["ok"] else warn)(f"bios: {rep['bios']['method'] or 'FAILED'} " f"{rep['bios']['size']} bytes - {rep['bios']['note']}") else: rep["bios"] = {"requested": False, "ok": None, "note": "not requested"} step("optional Vulkan capability probe") if args.vulkan == "never": rep["vulkan_probe"] = {"requested": False, "ran": False, "note": "disabled"} elif args.vulkan == "always": rep["vulkan_probe"] = X.run_vulkan_probe() else: if ask("Run optional Vulkan capability probe now? (y/N)", "n").lower().startswith("y"): rep["vulkan_probe"] = X.run_vulkan_probe() else: rep["vulkan_probe"] = {"requested": False, "ran": False, "note": "operator declined"} info("Vulkan probe skipped") result, fails, warns = score(rep) rep.update(finished_at=now_iso(), result=result, failures=fails, warnings=warns) save_report(rep, folder, run_dir) print() label = serial or f"unidentified card at {bdf}" if result == "PASS": ok(f"{label}: PASS") elif result == "WARN": warn(f"{label}: WARN") else: err(f"{label}: FAIL") for f in fails: err(" FAIL: " + f) for w in warns: warn(" WARN: " + w) info(f"report: {folder / 'report.json'}") return rep def save_report(rep, folder, run_dir): encoded = json.dumps(rep, indent=2) (run_dir / "report.json").write_text(encoded) (folder / "report.json").write_text(encoded) # ---------------------------------------------------------------- demo def demo(args): """Validate report pipeline without hardware: writes cards/DEMO-*.""" step("Demo mode (no hardware)") args.vulkan = "never" args.bios = False args.link_tests = False orig_wait = globals()["wait_for_gpu"] orig_detect, orig_check = X.detect_gpus, X.check_link orig_aer, orig_dm, orig_cfg = X.aer_snapshot, X.dmesg_capture, X.dump_config orig_retrain, orig_rescan = X.retrain_test, X.rescan_test orig_smi, orig_verify, orig_unlock = X.nvidia_smi_info, X.run_verify_unlocker, X.ensure_unlocked orig_serial, orig_identity = X.get_serial, X.card_identity globals()["wait_for_gpu"] = lambda: "0000:0a:00.0" X.detect_gpus = lambda: ["0000:0a:00.0"] X.check_link = lambda b: ({"lnksta_speed": "5GT/s", "lnksta_width": "x16", "lnksta_gen": 2, "lnkcap_speed": "8GT/s", "lnkcap_width": "x16", "lnksta_raw": "LnkSta: Speed 5GT/s, Width x16, TrErr-", "setpci_current_gen": 2, "lspci_rc": 0}, "demo lspci output\n") X.aer_snapshot = lambda b: {"aer_sysfs_correctable": None, "aer_sysfs_fatal": None, "aer_correctable_sum": 0, "ce_active": [], "ue_active": [], "parent_bdf": None, "parent_ce_active": []} X.dmesg_capture = lambda b, since: {"source": "demo", "related": ["demo line"], "errors": [], "other_pcie": [], "note": "demo"} X.dump_config = lambda b: "demo config space dump\n" X.retrain_test = lambda b: {"attempted": True, "ok": True, "note": "x16 -> x16", "width_before": "x16", "width_after": "x16"} X.rescan_test = lambda b: {"attempted": True, "ok": True, "note": "re-enumerated"} X.nvidia_smi_info = lambda: {"available": False} X.run_verify_unlocker = lambda b, d: {"ran": True, "rc": 0, "status": "unlocked-ok", "tail": ["demo"]} X.ensure_unlocked = lambda b, d, f, card=None, allow_install=True: {"status": "unlocked", "note": "demo"} X.get_serial = lambda nf, bdf: ("DEMO-SERIAL-001", "demo") X.card_identity = lambda b, d, serial=None: {"serial": serial, "gpu_uuid": "DEMO-UUID", "device_id": d, "bdf": b} try: rep = process_card(args) finally: globals()["wait_for_gpu"] = orig_wait X.detect_gpus, X.check_link = orig_detect, orig_check X.aer_snapshot, X.dmesg_capture, X.dump_config = orig_aer, orig_dm, orig_cfg X.retrain_test, X.rescan_test = orig_retrain, orig_rescan X.nvidia_smi_info, X.run_verify_unlocker, X.ensure_unlocked = orig_smi, orig_verify, orig_unlock X.get_serial, X.card_identity = orig_serial, orig_identity return rep # ---------------------------------------------------------------- main def main(): ap = argparse.ArgumentParser(description="CMP 170HX single-card checker (one card per run)") ap.add_argument("--preflight", action="store_true", help="prepare tools, driver and cmpunlocker source, then exit") ap.add_argument("--demo", action="store_true", help="run pipeline with fake hardware (self-test)") ap.add_argument("--nvflash", help="path to nvflash binary") ap.add_argument("--bios", action="store_true", help="save VBIOS in cards//extras/") ap.add_argument("--link-tests", action="store_true", help="also retrain and rescan PCIe bus") ap.add_argument("--vulkan", choices=("ask", "always", "never"), default="never", help="generic Vulkan capability probe policy (default never)") ap.add_argument("--no-install", action="store_true", help="do not install packages or cmpunlocker") ap.add_argument("--no-download", action="store_true", help="use existing harness sources only") args = ap.parse_args() if args.demo: demo(args) return 0 pf = X.preflight(auto_install=not args.no_install, do_download=not args.no_download) if args.preflight: return 4 if pf.get("reboot_required") else 0 if pf.get("preflight_ok") else 3 if pf.get("reboot_required"): return 4 if not pf.get("preflight_ok"): return 3 X.CARDS_DIR.mkdir(exist_ok=True) rep = process_card(args) if not rep: return 2 return 4 if rep.get("result") == "PENDING_REBOOT" else 1 if rep.get("result") == "FAIL" else 0 if __name__ == "__main__": try: sys.exit(main()) except KeyboardInterrupt: print() warn("interrupted") sys.exit(130)