"""Main entry point — the flicker hunt loop. Runs on the host PC. Cycles power, arms the scope, restarts video, captures 4-channel waveforms + bridge/SoC registers, decodes timings + Lane 0 packets, appends a row to flicker_log.csv. Repeats until KeyboardInterrupt or --max-runs is hit. If --pixel-clock differs from the default, DPHY_SPEC must be rebuilt for the new UI before checking compliance. """ from __future__ import annotations import argparse import logging import math import sys import time from typing import Optional import config from analysis import registers as reg_analysis from analysis import report from analysis import waveform as wave_analysis from hardware import PSUController, ScopeController, TargetController logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)-7s %(name)s: %(message)s", ) log = logging.getLogger("master_loop") def parse_args(argv: Optional[list[str]] = None) -> argparse.Namespace: p = argparse.ArgumentParser(description="MIPI flicker hunt loop") p.add_argument("--max-runs", type=int, default=None) p.add_argument("--timeout", type=float, default=30.0, help="Scope trigger timeout per run (s)") p.add_argument("--pixel-clock", type=int, default=config.PIXEL_CLOCK_HZ, help="Override pixel clock in Hz") p.add_argument("--note", type=str, default="", help="Note appended to every log row (e.g. 'dsi-tweak=5')") p.add_argument("--no-video", action="store_true", help="Skip PUT /video — display blank/unblank only") p.add_argument("--output-dir", type=str, default=config.CAPTURE_ROOT) return p.parse_args(argv) def _maybe_rebuild_spec(pixel_clock_hz: int) -> dict: """Recompute UI-dependent spec minimums if pixel clock was overridden.""" if pixel_clock_hz == config.PIXEL_CLOCK_HZ: return config.DPHY_SPEC derived = config.derive_clocks(pixel_clock_hz) log.info("Rebuilding DPHY_SPEC for %.3f MHz pixel clock (UI=%.3f ns)", pixel_clock_hz / 1e6, derived["UI_NS"]) return config.build_dphy_spec(derived["UI_NS"]) def run_one(idx: int, args: argparse.Namespace, spec: dict, psu: PSUController, scope: ScopeController, target: TargetController) -> None: log.info("=== run %03d ===", idx) # RESET try: target.display_off() except Exception as e: log.warning("display_off failed: %s", e) if not args.no_video: try: target.video_stop() except Exception as e: log.warning("video_stop failed: %s", e) psu.output_off() time.sleep(config.PSU_POWER_CYCLE_DELAY_S) # ARM scope.arm_single() run_dir = report.make_run_dir(root=args.output_dir, run_idx=idx) log.info("run dir: %s", run_dir) # STIMULUS psu.output_on() time.sleep(0.5) target.display_on() if not args.no_video: target.video_start(mode="static-pink") # ACQUIRE triggered = scope.wait_for_trigger(timeout_s=args.timeout) if not triggered: log.warning("TIMEOUT waiting for scope trigger (run %03d)", idx) report.save_summary(run_dir, f"run_{idx:03d}: TIMEOUT — no scope trigger\n") return # CAPTURE waveforms = scope.download_all() sn65_data = target.get_sn65_registers() dsim_data = target.get_dsim_registers() settling = target.get_sn65_settling() # ANALYSIS measurements = wave_analysis.measure_all(waveforms) spec_pass = wave_analysis.check_spec_compliance(measurements, spec) sn65_parsed = reg_analysis.parse_sn65(sn65_data) dsim_parsed = reg_analysis.parse_dsim(dsim_data) packets = wave_analysis.decode_lane0_packets(waveforms) packet_fault = wave_analysis.classify_packet_fault(packets) lane_stall = wave_analysis.detect_lane_stall(waveforms["DAT0_P"], waveforms["DAT0_N"]) flicker_detected = ( sn65_parsed.get("flicker_detected") or packet_fault.get("fault_a_detected") or lane_stall.get("fault_b_detected") ) if flicker_detected: log.warning("*** FLICKER EVENT CAPTURED (run %03d) ***", idx) # SAVE report.save_waveforms(run_dir, waveforms) report.save_registers(run_dir, dsim_parsed, sn65_parsed, settling) report.save_timing_analysis(run_dir, measurements, spec_pass, packet_fault, lane_stall) summary = report.build_summary( run_id=run_dir.name, sn65_parsed=sn65_parsed, measurements=measurements, spec_pass=spec_pass, packet_fault=packet_fault, lane_stall=lane_stall, dsim_parsed=dsim_parsed, note=args.note, ) report.save_summary(run_dir, summary) log_row = report.build_log_row( run_id=run_dir.name, sn65_parsed=sn65_parsed, measurements=measurements, spec_pass=spec_pass, dsim_parsed=dsim_parsed, note=args.note, ) report.append_flicker_log(args.output_dir, log_row) # REPORT hsp = measurements.get("t_hs_prepare") cpz = measurements.get("t_clk_prepare_plus_zero") log.info( "run %03d: flicker=%s | t_hs_prepare=%s ns | t_clk_prep+zero=%s ns", idx, "YES" if flicker_detected else "no", f"{hsp:.1f}" if hsp is not None and not math.isnan(hsp) else "nan", f"{cpz:.1f}" if cpz is not None and not math.isnan(cpz) else "nan", ) def main(argv: Optional[list[str]] = None) -> int: args = parse_args(argv) spec = _maybe_rebuild_spec(args.pixel_clock) log.info("Connecting instruments...") psu = PSUController(config.PSU_IP) scope = ScopeController(config.SCOPE_IP) target = TargetController(config.TARGET_IP, config.TARGET_PORT) try: log.info("Configuring scope...") scope.setup() psu.output_off() try: target.display_off() if not args.no_video: target.video_stop() except Exception as e: log.warning("Initial reset failed: %s", e) idx = 0 while True: idx += 1 try: run_one(idx, args, spec, psu, scope, target) except KeyboardInterrupt: raise except Exception as e: log.exception("run %03d failed: %s", idx, e) if args.max_runs is not None and idx >= args.max_runs: log.info("Reached --max-runs=%d, stopping", args.max_runs) break except KeyboardInterrupt: log.info("Interrupted by user") finally: try: psu.output_off() except Exception: pass scope.close() psu.close() return 0 if __name__ == "__main__": sys.exit(main())