recovery_trials.py (5746B)
1 #!/usr/bin/env python3 2 """Bounded hardware integration test: panel off/on, then DP recovery via DRST. 3 4 Requires sole ownership of the headset and the built displaywatch, hidcmd and 5 signed hpmctl tools. Observes CoreGraphics callbacks; never polls port registers. 6 Leaves the headset on after each successful trial. Stops on the first failure. 7 """ 8 import argparse 9 import json 10 import queue 11 import re 12 import subprocess 13 import threading 14 import time 15 from pathlib import Path 16 17 18 def main(): 19 parser = argparse.ArgumentParser(description=__doc__) 20 parser.add_argument("--trials", type=int, default=3) 21 parser.add_argument("--output", type=Path, required=True) 22 parser.add_argument("--app-recovery", action="store_true", help="wait for the running app to recover; do not issue DRST from this test") 23 parser.add_argument("--leave-off", action="store_true", help="let the app send screen-on as part of recovery") 24 parser.add_argument("--settle-seconds", type=float, default=10, help="observe this long after ADD, including any short self-healing drops") 25 args = parser.parse_args() 26 if not 1 <= args.trials <= 10: 27 parser.error("--trials must be between 1 and 10") 28 if not 0 <= args.settle_seconds <= 60: 29 parser.error("--settle-seconds must be between 0 and 60") 30 if args.leave_off and not args.app_recovery: 31 parser.error("--leave-off requires --app-recovery") 32 tools = Path(__file__).resolve().parent 33 events = queue.Queue() 34 report = {"started": time.time(), "app_recovery": args.app_recovery, "leave_off": args.leave_off, 35 "commands": [], "display_events": [], "trials": []} 36 watcher = subprocess.Popen([str(tools / "displaywatch")], stdout=subprocess.PIPE, 37 stderr=subprocess.STDOUT, text=True, bufsize=1) 38 39 def read_events(): 40 for line in watcher.stdout: 41 events.put({"received": time.time(), "line": line.strip()}) 42 43 threading.Thread(target=read_events, daemon=True).start() 44 45 def command(name, *arguments): 46 record = {"started": time.time(), "argv": [name, *arguments]} 47 argv = [str(tools / name), *arguments] 48 if name == "hpmctl": 49 argv = ["sudo", "-n", *argv] 50 result = subprocess.run(argv, capture_output=True, text=True, timeout=20) 51 record.update(ended=time.time(), code=result.returncode, 52 stdout=result.stdout.strip(), stderr=result.stderr.strip()) 53 report["commands"].append(record) 54 if result.returncode or (name == "hidcmd" and result.stdout.split()[-1:] != ["0"]): 55 raise RuntimeError(f"command failed: {record}") 56 return record["started"] 57 58 def await_display(online, timeout): 59 deadline = time.monotonic() + timeout 60 while time.monotonic() < deadline: 61 if watcher.poll() is not None: 62 raise RuntimeError("passive display watcher exited") 63 try: 64 event = events.get(timeout=min(0.2, max(0.001, deadline - time.monotonic()))) 65 except queue.Empty: 66 continue 67 report["display_events"].append(event) 68 if bool(re.search(r"\b\d+:8718\s", event["line"])) == online: 69 return event 70 return None 71 72 try: 73 if not await_display(True, 3): 74 raise RuntimeError("headset display must be online before the first trial") 75 for index in range(args.trials): 76 trial = {"index": index + 1} 77 report["trials"].append(trial) 78 command("hidcmd", "off") 79 removed = await_display(False, 5) 80 if not removed: 81 raise RuntimeError("panel-off did not remove the display") 82 if not args.leave_off: 83 command("hidcmd", "on") 84 restored = await_display(True, 2) 85 if restored: 86 trial["outcome"] = "returned_after_screen_on_without_DRST" 87 else: 88 reset_at = time.time() if args.app_recovery else command("hpmctl", "1", "cmd", "DRST") 89 event = await_display(True, 15) 90 if not event: 91 raise RuntimeError("DRST did not restore the display within 15 seconds") 92 restored = event 93 trial.update(outcome="recovered_by_app" if args.app_recovery else "recovered_after_DRST", 94 seconds=event["received"] - reset_at) 95 trial["outage_seconds"] = restored["received"] - removed["received"] 96 settle_until = time.monotonic() + args.settle_seconds 97 trial["post_recovery_drops"] = [] 98 while time.monotonic() < settle_until: 99 removed = await_display(False, settle_until - time.monotonic()) 100 if removed: 101 restored = await_display(True, 15) 102 if not restored: 103 raise RuntimeError("post-recovery drop did not heal within 15 seconds") 104 trial["post_recovery_drops"].append(restored["received"] - removed["received"]) 105 print(json.dumps(trial), flush=True) 106 report["result"] = "passed" 107 except Exception as error: 108 report.update(result="failed", error=str(error)) 109 print(str(error), flush=True) 110 finally: 111 watcher.terminate() 112 try: 113 watcher.wait(timeout=3) 114 except subprocess.TimeoutExpired: 115 watcher.kill() 116 watcher.wait() 117 report["ended"] = time.time() 118 args.output.parent.mkdir(parents=True, exist_ok=True) 119 args.output.write_text(json.dumps(report, indent=2) + "\n") 120 return 0 if report["result"] == "passed" else 1 121 122 123 if __name__ == "__main__": 124 raise SystemExit(main())