qbuissondebon/cmp170hx_check
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Single-card flow; preflight sudo precondition + Ubuntu package install

Browse Source
This commit is contained in:
qbuissondebon
2026-09-26 17:27:35 +02:00
parent 4e53bd880c
commit 94818ebdec
3 changed files with 219 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files
check_cards.py
+65 -135
View File
@@ -1,23 +1,23 @@
#!/usr/bin/env python3
"""CMP 170HX batch card checker - guided one-card-at-a-time workflow (23 cards).
"""CMP 170HX single-card checker - ONE card per run, no queue.
Per card: detect GPU -> auto serial -> VBIOS backup -> PCIe link/AER/config checks
-> retrain + rescan tests -> dmesg scrape -> cmpunlocker verify.sh (read-only)
-> optional 5-min Vulkan VRAM stress (last step, ask yes/no) -> cards/<serial>/report.json.
Per run: wait for card -> auto serial -> VBIOS backup -> PCIe link/AER/config
checks -> retrain + rescan tests -> dmesg scrape -> cmpunlocker verify.sh
(read-only) -> optional 5-min Vulkan VRAM stress (last step, ask yes/no)
-> cards/<serial>/report.json.
Scoring (lenient): FAIL only on no-device/link-down; WARN on correctable AER/Xid/
width< x16/verify issues.
Scoring (lenient): FAIL only on no-device/link-down/lost-after-rescan;
WARN on correctable AER/Xid/width< x16/verify issues.
Exit codes: 0 = PASS/WARN, 1 = FAIL, 2 = no card, 3 = preflight blocking.
"""
import argparse
import json
import sys
import time
from pathlib import Path
import cxlib as X
from cxlib import (C, ask, detect_gpus, err, info, now_iso, ok, sanitize, step, warn)
TOTAL_DEFAULT = X.DEFAULT_TOTAL
from cxlib import C, ask, detect_gpus, err, info, now_iso, ok, sanitize, step, warn
# ---------------------------------------------------------------- scoring
@@ -56,7 +56,7 @@ def score(rep):
if vu.get("requested") and vu.get("ran"):
if vu.get("tool_errors"):
warns.append("errors reported during Vulkan VRAM stress")
if dm.get("errors") and vu.get("ran"):
if dm.get("errors"):
warns.append("kernel errors around VRAM stress window")
if not rep.get("bios", {}).get("ok"):
warns.append("VBIOS backup failed/incomplete")
@@ -68,41 +68,39 @@ def score(rep):
return result, fails, warns
# ---------------------------------------------------------------- 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:
warn(f"multiple CMP 170HX devices present: {', '.join(bdfs)} - using {bdfs[0]}")
return bdfs[0]
time.sleep(1)
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 find_gpu(timeout=45):
"""Wait for a CMP 170HX to appear, offering rescan retries."""
deadline = time.time() + timeout
while time.time() < deadline:
bdfs = detect_gpus()
if len(bdfs) == 1:
return bdfs[0]
if len(bdfs) > 1:
warn(f"multiple CMP 170HX devices present: {', '.join(bdfs)}")
return bdfs[0]
time.sleep(2)
warn("no CMP 170HX detected yet")
if ask("trigger PCI bus rescan and retry? (y/N)", "n").lower().startswith("y"):
if X.rescan_bus():
ok("bus rescanned")
else:
err("rescan not permitted - reseat the card and press Enter")
input()
return find_gpu(timeout=30)
return None
def process_card(index, total, args):
header = f"Card {index}/{total}"
step(header)
bdf = find_gpu()
def process_card(args):
step("Card check")
bdf = wait_for_gpu()
if not bdf:
err("no device - skipping this card")
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:
devid = line[line.index("[10de:") + 1:line.index("[10de:") + 10].rstrip("]")
i = line.index("[10de:")
devid = line[i + 1:i + 10].rstrip("]")
ok(f"device found at {bdf} ({devid})")
nf = X.nvflash_bin(args)
@@ -112,17 +110,17 @@ def process_card(index, total, args):
info(f"folder: {folder}")
rep = {
"card_index": index, "total": total, "serial": serial, "serial_source": serial_src,
"started_at": now_iso(), "bdf": bdf, "device_id": devid, "host": X.run(["uname", "-n"])["out"].strip(),
"serial": serial, "serial_source": serial_src, "started_at": now_iso(),
"bdf": bdf, "device_id": devid, "host": X.run(["uname", "-n"])["out"].strip(),
"kernel": X.run(["uname", "-r"])["out"].strip(),
}
step(f"[{index}/{total}] VBIOS backup")
step("VBIOS backup")
rep["bios"] = X.save_bios(nf, folder / "bios.rom", args)
(ok if rep["bios"]["ok"] else warn)(f"bios: {rep['bios']['method'] or 'FAILED'} "
f"{rep['bios']['size']} bytes - {rep['bios']['note']}")
step(f"[{index}/{total}] PCIe link + config")
step("PCIe link + config")
link, lspci_txt = X.check_link(bdf)
rep["link"] = link
(folder / "lspci.txt").write_text(lspci_txt)
@@ -130,7 +128,7 @@ def process_card(index, total, args):
info(f"LnkSta: {link.get('lnksta_raw') or 'n/a'} (setpci gen={link.get('setpci_current_gen')})")
rep["aer_before"] = X.aer_snapshot(bdf)
step(f"[{index}/{total}] retrain + rescan tests")
step("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)
@@ -141,24 +139,24 @@ def process_card(index, total, args):
if lspci2:
(folder / "lspci_after.txt").write_text(lspci2)
step(f"[{index}/{total}] kernel log scrape")
step("kernel log scrape")
dm = X.dmesg_capture(bdf)
rep["dmesg"] = dm
(folder / "dmesg.log").write_text("\n".join(dm.get("related", [])))
info(f"{dm.get('note')} (source: {dm.get('source')})")
if dm.get("errors"):
for e in dm["errors"][-5:]:
warn(e.strip()[:160])
for e in dm.get("errors", [])[-5:]:
warn(e.strip()[:160])
step(f"[{index}/{total}] nvidia-smi + cmpunlocker verify")
step("nvidia-smi + cmpunlocker verify")
rep["nvidia_smi"] = X.nvidia_smi_info()
rep["cmpunlocker_verify"] = X.run_verify_unlocker()
v = rep["cmpunlocker_verify"]
(folder / "verify_unlocker.log").write_text(
"\n".join(v.get("tail", [])) or (v.get("note") or "not run"))
info(f"verify: {'ran rc=' + str(v.get('rc')) + ' -> ' + str(v.get('status')) if v.get('ran') else 'skipped - ' + str(v.get('note'))}")
info("verify: " + (f"rc={v.get('rc')} -> {v.get('status')}" if v.get("ran")
else f"skipped - {v.get('note')}"))
step(f"[{index}/{total}] optional Vulkan VRAM test (last step)")
step("optional Vulkan VRAM test (last step)")
if args.vulkan == "never":
rep["vulkan_test"] = {"requested": False, "ran": False, "note": "disabled via --vulkan never"}
elif args.vulkan == "always":
@@ -189,37 +187,6 @@ def process_card(index, total, args):
return rep
# ---------------------------------------------------------------- progress
def load_progress():
try:
return json.loads(X.PROGRESS_FILE.read_text())
except Exception:
return {"done": []}
def save_progress(prog):
X.PROGRESS_FILE.write_text(json.dumps(prog, indent=2))
def scoreboard(results):
step("Scoreboard")
counts = {"PASS": 0, "WARN": 0, "FAIL": 0}
for r in results:
if not r:
continue
counts[r["result"]] = counts.get(r["result"], 0) + 1
color = {"PASS": C.GREEN, "WARN": C.YELLOW, "FAIL": C.RED}[r["result"]]
print(f" {color}{r['result']:<4}{C.RESET} #{r['card_index']:>2} "
f"{r['serial']:<24} {r.get('bdf') or '-':<12} "
f"{r.get('link', {}).get('lnksta_width') or '?':<4} "
f"{r.get('link', {}).get('lnksta_speed') or '?'}")
skipped = [r for r in results if r is None]
print(f"\n PASS={counts.get('PASS', 0)} WARN={counts.get('WARN', 0)} "
f"FAIL={counts.get('FAIL', 0)} skipped={len(skipped)} "
f"total_seen={len(results)}/{TOTAL_DEFAULT}")
# ---------------------------------------------------------------- demo
def demo(args):
@@ -228,16 +195,13 @@ def demo(args):
args.vulkan = "never"
args.serial = "DEMO-SERIAL-001"
args.skip_bios = True
real_find = globals()["find_gpu"]
globals()["find_gpu"] = lambda timeout=0: "0000:0a:00.0"
class FakeRun:
pass
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 = X.nvidia_smi_info, X.run_verify_unlocker
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": "x4", "lnksta_gen": 2,
"lnkcap_speed": "8GT/s", "lnkcap_width": "x16",
@@ -255,9 +219,9 @@ def demo(args):
X.nvidia_smi_info = lambda: {"available": False}
X.run_verify_unlocker = lambda: {"ran": False, "note": "demo"}
try:
rep = process_card(1, TOTAL_DEFAULT, args)
rep = process_card(args)
finally:
globals()["find_gpu"] = real_find
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
@@ -268,66 +232,32 @@ def demo(args):
# ---------------------------------------------------------------- main
def main():
global TOTAL_DEFAULT
ap = argparse.ArgumentParser(description="CMP 170HX guided 23-card checker")
ap.add_argument("--preflight", action="store_true", help="only run tool check/install/clone")
ap = argparse.ArgumentParser(description="CMP 170HX single-card checker (one card per run)")
ap.add_argument("--preflight", action="store_true",
help="tool check + package install (requires sudo) + harness clone, then exit")
ap.add_argument("--demo", action="store_true", help="run pipeline with fake hardware (self-test)")
ap.add_argument("--cards", type=int, default=TOTAL_DEFAULT, help="total cards (default 23)")
ap.add_argument("--start", type=int, default=1, help="first card index")
ap.add_argument("--nvflash", help="path to nvflash binary")
ap.add_argument("--serial", help="override serial for the next card (single-card run)")
ap.add_argument("--serial", help="override serial (skips nvflash auto-read)")
ap.add_argument("--skip-bios", action="store_true", help="skip VBIOS backup")
ap.add_argument("--vulkan", choices=("ask", "always", "never"), default="ask",
help="5-min Vulkan VRAM test policy (default ask, runs as last step)")
ap.add_argument("--no-install", action="store_true", help="do not auto-install missing packages")
ap.add_argument("--no-install", action="store_true",
help="do not auto-install packages (skips the sudo precondition)")
ap.add_argument("--no-clone", action="store_true", help="do not clone harness repos")
ap.add_argument("--restart", action="store_true", help="ignore previous progress.json")
ap.add_argument("--no-pause", action="store_true", help="do not wait for card insertion")
args = ap.parse_args()
TOTAL_DEFAULT = args.cards
pf = X.preflight(auto_install=not args.no_install, do_clone=not args.no_clone)
if args.preflight:
return 0
return 0 if pf.get("preflight_ok") else 3
if args.demo:
demo(args)
return 0
X.CARDS_DIR.mkdir(exist_ok=True)
prog = {"done": []} if args.restart else load_progress()
results = []
for i in range(args.start, args.cards + 1):
if str(i) in prog.get("done", []):
info(f"card {i} already done (progress.json) - skipping")
results.append(None)
continue
if not args.no_pause:
ans = ask(f"Insert card {i}/{args.cards}, then press Enter "
f"(s=skip this card, q=quit)", "")
if ans.lower() == "q":
break
if ans.lower() == "s":
results.append(None)
prog.setdefault("done", []).append(str(i))
prog.setdefault("skipped", []).append(str(i))
save_progress(prog)
continue
rep = process_card(i, args.cards, args)
results.append(rep)
if rep:
prog.setdefault("done", []).append(str(i))
save_progress(prog)
else:
if not ask("retry this card? (y/N)", "n").lower().startswith("y"):
break
if results:
scoreboard(results)
done = len([r for r in results if r])
info(f"{done} card(s) processed. Output in {X.CARDS_DIR}")
return 0
rep = process_card(args)
if not rep:
return 2
return 1 if rep.get("result") == "FAIL" else 0
if __name__ == "__main__":
@@ -335,5 +265,5 @@ if __name__ == "__main__":
sys.exit(main())
except KeyboardInterrupt:
print()
warn("interrupted - progress saved in progress.json")
warn("interrupted")
sys.exit(130)