This commit is contained in:
david rice
2026-04-20 10:35:06 +01:00
parent e718a93667
commit 36223dbb6a

View File

@@ -1,4 +1,4 @@
ni w#!/usr/bin/env python3 #!/usr/bin/env python3
""" """
MIPI TEST APPLICATION - MIPI_TEST_INTERACTIVE.PY MIPI TEST APPLICATION - MIPI_TEST_INTERACTIVE.PY
Interactive flicker confirmation test. Interactive flicker confirmation test.
@@ -37,8 +37,7 @@ from dotenv import load_dotenv
import ai_mgmt import ai_mgmt
import rigol_scope import rigol_scope
from csv_preprocessor import (analyze_lp_file, LPMetrics, from csv_preprocessor import (analyze_lp_file, LPMetrics,
HS_BURST_AMPLITUDE_MIN_MV, FLICKER_LP_LOW_MAX_NS, HS_BURST_AMPLITUDE_MIN_MV, FLICKER_LP_LOW_MAX_NS)
analyze_int_file, CLK_LP_LOW_MIN_NS)
load_dotenv(Path(__file__).parent / ".env") load_dotenv(Path(__file__).parent / ".env")
@@ -556,85 +555,6 @@ def _restore_hs_config():
time.sleep(0.1) time.sleep(0.1)
def _arm_scope_for_clk_startup() -> None:
"""
Configure scope for CLK lane LP startup capture and arm in single-shot mode.
Must be called BEFORE display ON — the CLK+ LP-11 falling edge (start of CLK
lane SoT preamble) triggers it. The DAT0+-triggered LP pass would miss this
because CLK is already in continuous HS by the time DAT0+ falls.
"""
for ch in (1, 2, 3, 4):
scope.write(f":CHANnel{ch}:SCALe {LP_V_SCALE:.3f}")
scope.write(f":CHANnel{ch}:OFFSet {LP_V_OFFSET:.3f}")
time.sleep(0.05)
scope.write(":TRIGger:EDGE:SOURce CHANnel1") # CLK+ — fires before DAT0+
scope.write(":TRIGger:EDGE:SLOPe NEGative")
scope.write(f":TRIGger:EDGE:LEVel {LP_TRIG_LEVEL:.3f}")
scope.write(":TRIGger:SWEep NORMal")
scope.write(f":TIMebase:SCALe {LP_SCALE:.3E}")
scope.write(f":ACQuire:POINts {LP_POINTS}")
time.sleep(0.3)
scope.write(":SINGle") # arm without blocking — display ON happens next
time.sleep(0.1)
def _collect_clk_startup(ts: str, iteration: int,
timeout: float = 10.0) -> list[str]:
"""
Wait for the CLK startup trigger fired by _arm_scope_for_clk_startup(),
save CLK+ and DAT0+ channels, and return LP analysis summaries.
Restores HS config before returning.
"""
print(" CLK STARTUP: waiting for trigger...")
deadline = time.time() + timeout
triggered = False
while time.time() < deadline:
try:
status = scope.ask(":TRIGger:STATus?").strip().upper()
if status in ("STOP", "TD"):
triggered = True
break
except Exception:
pass
time.sleep(0.1)
if not triggered:
print(" CLK STARTUP: trigger timeout — CLK LP startup not captured.")
_restore_hs_config()
return []
_save_pass_channels("lp", iteration, ts)
_restore_hs_config()
try:
copied, _ = ai_mgmt.transfer_csv_files()
print(f" CLK STARTUP: {copied} file(s) transferred.")
except Exception as e:
print(f" CLK STARTUP TRANSFER ERROR: {e}")
summaries = []
for channel in ("clk", "dat"):
path = DATA_DIR / f"{ts}_lp_{iteration:04d}_{channel}.csv"
if not path.exists():
continue
try:
m = analyze_lp_file(path)
summaries.append(m.summary())
if channel == "clk":
if m.clk_lp_startup_ok is False:
print(f"\n *** CLK STARTUP SUSPECT: capture {iteration:04d} "
f"CLK LP-00={m.lp_low_duration_ns:.0f} ns "
f"< {CLK_LP_LOW_MIN_NS:.0f} ns — bridge may not lock CLK ***\n")
elif m.clk_lp_startup_ok is True:
print(f" CLK startup: LP-00={m.lp_low_duration_ns:.0f} ns ✓")
else:
print(" CLK startup: CLK already in continuous HS at trigger point.")
except Exception as e:
print(f" CLK STARTUP ANALYSIS ERROR ({channel}): {e}")
return summaries
def _fetch_registers(ts: str, iteration: int) -> None: def _fetch_registers(ts: str, iteration: int) -> None:
"""GET /registers from device server and save to data/ as JSON.""" """GET /registers from device server and save to data/ as JSON."""
try: try:
@@ -683,14 +603,7 @@ def dual_capture(iteration: int) -> str:
if n: if n:
print(f" SAVED: {v18_path.name} ({n} samples)") print(f" SAVED: {v18_path.name} ({n} samples)")
else: else:
print(" RIGOL: 1V8 waveform read failed.") print(" RIGOL: Waveform read failed — check connection and probe.")
# CH2 — INTB pin (read after CH1; scope already stopped)
int_path = DATA_DIR / f"{ts}_int_{iteration:04d}.csv"
n_int = rigol_scope.read_int_csv(int_path)
if n_int:
print(f" SAVED: {int_path.name} ({n_int} samples)")
else:
print(" RIGOL: INTB waveform read failed.")
_restore_hs_config() _restore_hs_config()
# ── Pass 2: HS signal quality ────────────────────────────────────────── # ── Pass 2: HS signal quality ──────────────────────────────────────────
@@ -861,25 +774,6 @@ def _analyze_lp_files(
return lp_summaries, suspects return lp_summaries, suspects
def _analyze_int_file(ts: str, iteration: int) -> str | None:
"""
Analyse the INTB pin CSV for this iteration.
Returns a summary string, or None if the file is missing.
"""
path = DATA_DIR / f"{ts}_int_{iteration:04d}.csv"
if not path.exists():
return None
try:
m = analyze_int_file(path)
if m.int_asserted:
print(f"\n *** INTB ASSERTED: capture {iteration:04d} "
f"bridge flagged error ({m.asserted_duration_us:.2f} µs) ***\n")
return m.summary()
except Exception as e:
print(f" INTB ANALYSIS ERROR: {e}")
return None
def _lp_followup_capture(iteration: int) -> tuple[str, list[str], list[LPMetrics]]: def _lp_followup_capture(iteration: int) -> tuple[str, list[str], list[LPMetrics]]:
""" """
LP-only follow-up capture taken immediately after a suspect is detected. LP-only follow-up capture taken immediately after a suspect is detected.
@@ -1299,22 +1193,12 @@ def run_interactive_test() -> None:
try: try:
while True: while True:
# ── Arm scope for CLK lane startup (BEFORE display ON) ─────────
# CLK+ LP-11 falls before DAT0+, so we must arm here.
# The regular LP pass (triggered on DAT0+) would miss CLK startup
# because CLK is already in continuous HS by the time DAT0+ falls.
ts_startup = datetime.now().strftime("%Y%m%d_%H%M%S")
_arm_scope_for_clk_startup()
# ── Display ON ───────────────────────────────────────────────── # ── Display ON ─────────────────────────────────────────────────
try: try:
requests.put(URL, json={"state": "on"}, timeout=2) requests.put(URL, json={"state": "on"}, timeout=2)
except requests.exceptions.RequestException as e: except requests.exceptions.RequestException as e:
print(f" WARNING: display ON failed: {e}") print(f" WARNING: display ON failed: {e}")
# ── Collect CLK startup capture ────────────────────────────────
clk_startup_summaries = _collect_clk_startup(ts_startup, iteration)
# ── Three-pass capture ───────────────────────────────────────── # ── Three-pass capture ─────────────────────────────────────────
ts = dual_capture(iteration) ts = dual_capture(iteration)
@@ -1326,9 +1210,6 @@ def run_interactive_test() -> None:
except Exception as e: except Exception as e:
print(f" TRANSFER ERROR: {e}") print(f" TRANSFER ERROR: {e}")
# ── INTB pin analysis ──────────────────────────────────────────
_analyze_int_file(ts, iteration)
# ── Rule-based LP analysis ───────────────────────────────────── # ── Rule-based LP analysis ─────────────────────────────────────
lp_summaries, suspects = _analyze_lp_files(ts, iteration) lp_summaries, suspects = _analyze_lp_files(ts, iteration)