diff --git a/scripts/analysis/radar_interference_check.py b/scripts/analysis/radar_interference_check.py new file mode 100644 index 000000000..a748e35c7 --- /dev/null +++ b/scripts/analysis/radar_interference_check.py @@ -0,0 +1,403 @@ +#!/usr/bin/env python3 +"""Radar interference check analysis tool (A/B/AB Protocol). + +Quantifies RF interference and degradation between OpenFlight (OPS243-A 24 GHz CW Doppler) +and a reference launch monitor radar (e.g., Rapsodo MLM2 Pro K-band radar) operating +in the same hitting enclosure. + +Protocol Phases: + - Phase A: OpenFlight only (baseline spin SNR, read rate, speed spread) + - Phase B: Reference instrument only (standalone reference baseline) + - Phase AB: Simultaneous joint operation (both radars transmitting) + +Degradation Gates: + - Spin SNR drop > 1.5 dB -> Warning + - Spin SNR drop > 3.0 dB -> Severe Interference + - Spin read rate drop > 10% -> Warning + - Ball speed jitter increase > 2.0x -> Radar Mutual Jamming + +Usage:: + + uv run python scripts/analysis/radar_interference_check.py \\ + --phase-a session_logs/session_phase_a.jsonl \\ + --phase-b session_logs/session_phase_b.csv \\ + --phase-ab session_logs/session_phase_ab.jsonl \\ + --output notes/radar_interference_report.md + +Or test with synthetic session data:: + + uv run python scripts/analysis/radar_interference_check.py --synthetic +""" + +from __future__ import annotations + +import argparse +import csv +import json +import math +import random +import statistics +import sys +from dataclasses import asdict, dataclass, field +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + + +@dataclass +class PhaseStats: + """Summary statistics for one phase of the interference check protocol.""" + + phase_name: str + total_shots: int + spin_read_count: int + spin_read_rate_pct: float + spin_snr_mean_db: Optional[float] + spin_snr_median_db: Optional[float] + spin_snr_std_db: Optional[float] + ball_speed_mean_mph: float + ball_speed_std_mph: float + trigger_latency_mean_ms: Optional[float] + trigger_latency_std_ms: Optional[float] + + +@dataclass +class InterferenceComparison: + """Comparison metrics between baseline and joint phases.""" + + spin_read_rate_delta_pct: float + spin_snr_delta_db: float + ball_speed_std_ratio: float + is_snr_degraded: bool + is_read_rate_degraded: bool + severity: str # "clean", "mild", "moderate", "severe" + recommendation: str + + +@dataclass +class InterferenceReport: + """Complete multi-phase interference protocol report.""" + + phase_a: PhaseStats + phase_b: Optional[PhaseStats] + phase_ab: PhaseStats + comparison: InterferenceComparison + notes: List[str] = field(default_factory=list) + + +def compute_phase_stats( + shots: List[Dict[str, Any]], + phase_name: str, +) -> PhaseStats: + """Compute summary statistics for a collection of shots in a protocol phase.""" + if not shots: + return PhaseStats( + phase_name=phase_name, + total_shots=0, + spin_read_count=0, + spin_read_rate_pct=0.0, + spin_snr_mean_db=None, + spin_snr_median_db=None, + spin_snr_std_db=None, + ball_speed_mean_mph=0.0, + ball_speed_std_mph=0.0, + trigger_latency_mean_ms=None, + trigger_latency_std_ms=None, + ) + + ball_speeds = [ + float(s["ball_speed_mph"]) + for s in shots + if s.get("ball_speed_mph") is not None and float(s["ball_speed_mph"]) > 0 + ] + + spin_snrs = [ + float(s["spin_snr"]) + for s in shots + if s.get("spin_snr") is not None and not math.isnan(float(s["spin_snr"])) + ] + + valid_spins = [s for s in shots if s.get("spin_rpm") is not None and float(s["spin_rpm"]) > 0] + + latencies = [ + float(s["trigger_latency_ms"]) for s in shots if s.get("trigger_latency_ms") is not None + ] + + speed_mean = statistics.mean(ball_speeds) if ball_speeds else 0.0 + speed_std = statistics.stdev(ball_speeds) if len(ball_speeds) > 1 else 0.0 + + snr_mean = statistics.mean(spin_snrs) if spin_snrs else None + snr_median = statistics.median(spin_snrs) if spin_snrs else None + snr_std = statistics.stdev(spin_snrs) if len(spin_snrs) > 1 else 0.0 if spin_snrs else None + + lat_mean = statistics.mean(latencies) if latencies else None + lat_std = statistics.stdev(latencies) if len(latencies) > 1 else 0.0 if latencies else None + + read_rate = (len(valid_spins) / len(shots) * 100.0) if shots else 0.0 + + return PhaseStats( + phase_name=phase_name, + total_shots=len(shots), + spin_read_count=len(valid_spins), + spin_read_rate_pct=round(read_rate, 1), + spin_snr_mean_db=round(snr_mean, 2) if snr_mean is not None else None, + spin_snr_median_db=round(snr_median, 2) if snr_median is not None else None, + spin_snr_std_db=round(snr_std, 2) if snr_std is not None else None, + ball_speed_mean_mph=round(speed_mean, 2), + ball_speed_std_mph=round(speed_std, 2), + trigger_latency_mean_ms=round(lat_mean, 2) if lat_mean is not None else None, + trigger_latency_std_ms=round(lat_std, 2) if lat_std is not None else None, + ) + + +def compare_phases( + baseline_a: PhaseStats, + joint_ab: PhaseStats, +) -> InterferenceComparison: + """Compare Phase AB against baseline Phase A to determine RF degradation.""" + read_rate_delta = joint_ab.spin_read_rate_pct - baseline_a.spin_read_rate_pct + + snr_a = baseline_a.spin_snr_mean_db or 0.0 + snr_ab = joint_ab.spin_snr_mean_db or 0.0 + snr_delta = snr_ab - snr_a + + speed_std_a = baseline_a.ball_speed_std_mph if baseline_a.ball_speed_std_mph > 0 else 1.0 + speed_std_ratio = joint_ab.ball_speed_std_mph / speed_std_a + + is_snr_degraded = snr_delta < -1.5 + is_read_rate_degraded = read_rate_delta < -10.0 + + # Determine severity + if snr_delta < -3.0 or read_rate_delta < -25.0: + severity = "severe" + recommendation = ( + "High mutual RF jamming detected. Switch to alternating-shot capture protocol." + ) + elif snr_delta < -1.5 or read_rate_delta < -10.0 or speed_std_ratio > 1.8: + severity = "moderate" + recommendation = ( + "Measurable degradation observed. Increase lateral separation between units to >= 1.5m." + ) + elif snr_delta < -0.8 or read_rate_delta < -8.0: + severity = "mild" + recommendation = ( + "Minor noise floor elevation. Clear for joint validation with recorded SNR baseline." + ) + else: + severity = "clean" + recommendation = ( + "Zero significant RF interference. Fully approved for paired cross-validation." + ) + + return InterferenceComparison( + spin_read_rate_delta_pct=round(read_rate_delta, 1), + spin_snr_delta_db=round(snr_delta, 2), + ball_speed_std_ratio=round(speed_std_ratio, 2), + is_snr_degraded=is_snr_degraded, + is_read_rate_degraded=is_read_rate_degraded, + severity=severity, + recommendation=recommendation, + ) + + +def analyze_interference_session( + phase_a_shots: List[Dict[str, Any]], + phase_b_shots: Optional[List[Dict[str, Any]]], + phase_ab_shots: List[Dict[str, Any]], +) -> InterferenceReport: + """Run end-to-end interference protocol analysis.""" + stats_a = compute_phase_stats(phase_a_shots, "Phase A (OpenFlight Only)") + stats_b = ( + compute_phase_stats(phase_b_shots, "Phase B (Reference Only)") if phase_b_shots else None + ) + stats_ab = compute_phase_stats(phase_ab_shots, "Phase AB (Simultaneous Joint)") + + comparison = compare_phases(stats_a, stats_ab) + + notes = [] + if stats_a.total_shots < 10 or stats_ab.total_shots < 10: + notes.append( + "Protocol recommendation: At least 10 shots per phase are recommended for robust statistics." + ) + if comparison.severity == "clean": + notes.append( + "OPS243-A Doppler radar and reference unit operate without measurable mutual desensitization." + ) + + return InterferenceReport( + phase_a=stats_a, + phase_b=stats_b, + phase_ab=stats_ab, + comparison=comparison, + notes=notes, + ) + + +def generate_synthetic_interference_session( + simulated_interference_level: str = "clean", + n_shots: int = 15, + seed: int = 101, +) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]], List[Dict[str, Any]]]: + """Generate synthetic shot records for phases A, B, and AB.""" + rng = random.Random(seed) + + def gen_shots(snr_center: float, read_prob: float, speed_jitter: float) -> List[Dict[str, Any]]: + res = [] + for i in range(1, n_shots + 1): + speed = rng.gauss(155.0, speed_jitter) + has_spin = rng.random() < read_prob + snr = rng.gauss(snr_center, 0.8) if has_spin else rng.uniform(4.0, 7.0) + spin = rng.gauss(2600.0, 150.0) if has_spin else None + res.append( + { + "shot_id": i, + "ball_speed_mph": round(speed, 1), + "spin_rpm": round(spin, 0) if spin else None, + "spin_snr": round(snr, 1), + "trigger_latency_ms": round(rng.gauss(0.12, 0.02), 3), + } + ) + return res + + # Phase A: pristine baseline + shots_a = gen_shots(snr_center=18.5, read_prob=0.95, speed_jitter=1.5) + + # Phase B: reference baseline + shots_b = gen_shots(snr_center=18.5, read_prob=0.95, speed_jitter=1.5) + + # Phase AB + if simulated_interference_level == "severe": + shots_ab = gen_shots(snr_center=14.0, read_prob=0.60, speed_jitter=3.2) + elif simulated_interference_level == "moderate": + shots_ab = gen_shots(snr_center=16.5, read_prob=0.80, speed_jitter=2.2) + else: # clean + shots_ab = gen_shots(snr_center=18.5, read_prob=0.95, speed_jitter=1.5) + + return shots_a, shots_b, shots_ab + + +def format_markdown_report(report: InterferenceReport) -> str: + """Format interference analysis into Markdown.""" + lines = [ + "# Radar Interference Check: A/B/AB Protocol Report", + "", + f"**Interference Status:** `{report.comparison.severity.upper()}`", + f"**Recommendation:** {report.comparison.recommendation}", + "", + "## Phase Summary Statistics", + "", + "| Metric | Phase A (OpenFlight Only) | Phase B (Reference Only) | Phase AB (Joint Simultaneous) | Delta (AB vs A) |", + "| --- | --- | --- | --- | --- |", + f"| **Shots Recorded** | {report.phase_a.total_shots} | {report.phase_b.total_shots if report.phase_b else 'N/A'} | {report.phase_ab.total_shots} | — |", + f"| **Spin Read Rate** | {report.phase_a.spin_read_rate_pct}% | {report.phase_b.spin_read_rate_pct if report.phase_b else 'N/A'}% | {report.phase_ab.spin_read_rate_pct}% | **{report.comparison.spin_read_rate_delta_pct:+.1f}%** |", + f"| **Spin SNR Mean** | {report.phase_a.spin_snr_mean_db or 'N/A'} dB | {report.phase_b.spin_snr_mean_db if report.phase_b else 'N/A'} dB | {report.phase_ab.spin_snr_mean_db or 'N/A'} dB | **{report.comparison.spin_snr_delta_db:+.2f} dB** |", + f"| **Ball Speed Mean** | {report.phase_a.ball_speed_mean_mph} mph | {report.phase_b.ball_speed_mean_mph if report.phase_b else 'N/A'} mph | {report.phase_ab.ball_speed_mean_mph} mph | {report.phase_ab.ball_speed_mean_mph - report.phase_a.ball_speed_mean_mph:+.2f} mph |", + f"| **Ball Speed StdDev** | {report.phase_a.ball_speed_std_mph} mph | {report.phase_b.ball_speed_std_mph if report.phase_b else 'N/A'} mph | {report.phase_ab.ball_speed_std_mph} mph | {report.comparison.ball_speed_std_ratio:.2f}x ratio |", + "", + "## Diagnostic Evaluation", + "", + f"- **Spin SNR Degradation Threshold (> 1.5 dB drop):** {'FAIL (Degraded)' if report.comparison.is_snr_degraded else 'PASS (Normal)'}", + f"- **Spin Read Rate Degradation Threshold (> 10% drop):** {'FAIL (Degraded)' if report.comparison.is_read_rate_degraded else 'PASS (Normal)'}", + "", + ] + + if report.notes: + lines.append("## Protocol Notes") + for n in report.notes: + lines.append(f"- {n}") + lines.append("") + + return "\n".join(lines) + + +def load_shots(file_path: str | Path) -> List[Dict[str, Any]]: + """Load shots from a JSONL or CSV file.""" + path = Path(file_path) + if not path.exists(): + raise FileNotFoundError(f"File not found: {path}") + + shots: List[Dict[str, Any]] = [] + if path.suffix.lower() == ".jsonl": + with path.open("r", encoding="utf-8") as f: + for line in f: + line_s = line.strip() + if line_s: + data = json.loads(line_s) + if data.get("event") in ("shot_detected", "shot", None): + shots.append(data) + elif path.suffix.lower() == ".csv": + with path.open("r", encoding="utf-8") as f: + reader = csv.DictReader(f) + for row in reader: + shots.append( + { + "ball_speed_mph": float(row["ball_speed"]) + if row.get("ball_speed") + else None, + "spin_rpm": float(row["spin_rpm"]) if row.get("spin_rpm") else None, + "spin_snr": float(row["spin_snr"]) if row.get("spin_snr") else None, + "trigger_latency_ms": float(row["trigger_latency_ms"]) + if row.get("trigger_latency_ms") + else None, + } + ) + return shots + + +def main() -> None: + """CLI entrypoint.""" + parser = argparse.ArgumentParser( + description="Radar interference check between OpenFlight and reference launch monitor (A/B/AB protocol).", + ) + parser.add_argument("--phase-a", help="Path to Phase A (OpenFlight only) session JSONL.") + parser.add_argument("--phase-b", help="Path to Phase B (Reference only) session CSV/JSONL.") + parser.add_argument("--phase-ab", help="Path to Phase AB (Joint simultaneous) session JSONL.") + parser.add_argument( + "--synthetic", action="store_true", help="Run with synthetic protocol data." + ) + parser.add_argument( + "--simulated-level", + choices=["clean", "moderate", "severe"], + default="clean", + help="Simulated interference severity when using --synthetic (default: clean).", + ) + parser.add_argument("--output", "-o", help="Path to output Markdown/JSON report.") + + args = parser.parse_args() + + if args.synthetic or (not args.phase_a and not args.phase_ab): + shots_a, shots_b, shots_ab = generate_synthetic_interference_session( + simulated_interference_level=args.simulated_level, + ) + else: + if not args.phase_a or not args.phase_ab: + print( + "Error: Both --phase-a and --phase-ab are required unless using --synthetic.", + file=sys.stderr, + ) + sys.exit(1) + shots_a = load_shots(args.phase_a) + shots_b = load_shots(args.phase_b) if args.phase_b else None + shots_ab = load_shots(args.phase_ab) + + report = analyze_interference_session( + phase_a_shots=shots_a, + phase_b_shots=shots_b, + phase_ab_shots=shots_ab, + ) + + md_report = format_markdown_report(report) + + if args.output: + out_path = Path(args.output) + out_path.parent.mkdir(parents=True, exist_ok=True) + if out_path.suffix.lower() == ".json": + out_path.write_text(json.dumps(asdict(report), indent=2), encoding="utf-8") + else: + out_path.write_text(md_report, encoding="utf-8") + print(f"Interference report saved to {out_path}") + else: + print(md_report) + + +if __name__ == "__main__": + main() diff --git a/scripts/analysis/validate_ghost_track_gate.py b/scripts/analysis/validate_ghost_track_gate.py new file mode 100644 index 000000000..0a563b8e2 --- /dev/null +++ b/scripts/analysis/validate_ghost_track_gate.py @@ -0,0 +1,476 @@ +#!/usr/bin/env python3 +"""Driver ghost-track gate validator and recovery analysis tool. + +Evaluates the ghost-track rejection gate and OPS-guided fast-track recovery +algorithm for TI IWR6843 driver launch angle measurements against truth data. + +Background: + On high-speed driver shots (140-180 mph ball speed on OPS243), the 60 GHz + radar tracker can occasionally lock onto slower clubhead or reflection + artifacts (50-65 mph), yielding severe launch angle bias (+3.39 deg bias, + 3.55 deg MAE). The ghost-track gate rejects candidates where: + track_speed < min_speed_ratio * ops_ball_speed (default 0.65) + and recovers the true ball track from candidate tracks within a fractional + window around the OPS ball speed. + +Usage:: + + uv run python scripts/analysis/validate_ghost_track_gate.py \\ + --dataset session_logs/driver_truth_session.jsonl \\ + --min-speed-ratio 0.65 \\ + --output validation_report.md + +Or test with synthetic truth data:: + + uv run python scripts/analysis/validate_ghost_track_gate.py --synthetic +""" + +from __future__ import annotations + +import argparse +import csv +import json +import random +import statistics +from dataclasses import asdict, dataclass, field +from pathlib import Path +from typing import Any, Dict, Iterable, List, Optional + + +@dataclass +class CandidateTrack: + """Individual radar track candidate.""" + + track_id: int + speed_mph: float + launch_angle_deg: float + point_count: int = 10 + snr_db: float = 18.0 + explained_fraction: float = 0.85 + + +@dataclass +class DriverShot: + """Driver shot record containing OPS measurement, reference truth, and radar tracks.""" + + shot_id: int + ops_ball_speed_mph: float + true_launch_angle_deg: float + primary_track_speed_mph: float + primary_launch_angle_deg: float + candidate_tracks: List[CandidateTrack] = field(default_factory=list) + club: str = "driver" + + +@dataclass +class GateEvaluationResult: + """Statistical summary of ghost-track gate validation.""" + + total_shots: int + ghost_tracks_detected: int + ghost_tracks_recovered: int + ungated_mae_deg: float + ungated_bias_deg: float + gated_mae_deg: float + gated_bias_deg: float + improvement_deg: float + recovery_rate_pct: float + details: List[Dict[str, Any]] = field(default_factory=list) + + +def is_ghost_track( + track_speed_mph: float, + ops_ball_speed_mph: float, + min_speed_ratio: float = 0.65, +) -> bool: + """Check if a radar track is a slow ghost track relative to OPS ball speed.""" + if ops_ball_speed_mph <= 0: + return False + ratio = track_speed_mph / ops_ball_speed_mph + return ratio < min_speed_ratio + + +def recover_fast_track( + candidate_tracks: List[CandidateTrack], + ops_ball_speed_mph: float, + recovery_window: float = 0.15, + min_points: int = 6, + min_snr_db: float = 10.0, +) -> Optional[CandidateTrack]: + """Find a candidate track matching the OPS ball speed within the recovery window.""" + if not candidate_tracks or ops_ball_speed_mph <= 0: + return None + + min_speed = ops_ball_speed_mph * (1.0 - recovery_window) + max_speed = ops_ball_speed_mph * (1.0 + recovery_window) + + valid_candidates = [ + t + for t in candidate_tracks + if min_speed <= t.speed_mph <= max_speed + and t.point_count >= min_points + and t.snr_db >= min_snr_db + ] + + if not valid_candidates: + return None + + # Pick candidate closest to OPS ball speed with highest point count tie-breaker + valid_candidates.sort( + key=lambda t: (abs(t.speed_mph - ops_ball_speed_mph), -t.point_count), + ) + return valid_candidates[0] + + +def evaluate_ghost_track_gate( + shots: Iterable[DriverShot | Dict[str, Any]], + min_speed_ratio: float = 0.65, + recovery_window: float = 0.15, +) -> GateEvaluationResult: + """Evaluate ghost-track gating and recovery across a collection of shots.""" + parsed_shots: List[DriverShot] = [] + for s in shots: + if isinstance(s, DriverShot): + parsed_shots.append(s) + elif isinstance(s, dict): + # Parse dict representation + raw_candidates = s.get("candidate_tracks", []) + cands = [] + for c in raw_candidates: + if isinstance(c, CandidateTrack): + cands.append(c) + elif isinstance(c, dict): + cands.append( + CandidateTrack( + track_id=int(c.get("track_id", 0)), + speed_mph=float(c.get("speed_mph", 0.0)), + launch_angle_deg=float(c.get("launch_angle_deg", 0.0)), + point_count=int(c.get("point_count", 10)), + snr_db=float(c.get("snr_db", 18.0)), + explained_fraction=float(c.get("explained_fraction", 0.85)), + ), + ) + parsed_shots.append( + DriverShot( + shot_id=int(s.get("shot_id", len(parsed_shots) + 1)), + ops_ball_speed_mph=float(s.get("ops_ball_speed_mph", 0.0)), + true_launch_angle_deg=float(s.get("true_launch_angle_deg", 0.0)), + primary_track_speed_mph=float(s.get("primary_track_speed_mph", 0.0)), + primary_launch_angle_deg=float(s.get("primary_launch_angle_deg", 0.0)), + candidate_tracks=cands, + club=str(s.get("club", "driver")), + ), + ) + + if not parsed_shots: + return GateEvaluationResult( + total_shots=0, + ghost_tracks_detected=0, + ghost_tracks_recovered=0, + ungated_mae_deg=0.0, + ungated_bias_deg=0.0, + gated_mae_deg=0.0, + gated_bias_deg=0.0, + improvement_deg=0.0, + recovery_rate_pct=0.0, + details=[], + ) + + ungated_errors: List[float] = [] + gated_errors: List[float] = [] + ghost_count = 0 + recovered_count = 0 + shot_details: List[Dict[str, Any]] = [] + + for shot in parsed_shots: + ungated_err = shot.primary_launch_angle_deg - shot.true_launch_angle_deg + ungated_errors.append(ungated_err) + + is_ghost = is_ghost_track( + shot.primary_track_speed_mph, + shot.ops_ball_speed_mph, + min_speed_ratio=min_speed_ratio, + ) + + resolved_angle = shot.primary_launch_angle_deg + resolved_source = "primary" + + if is_ghost: + ghost_count += 1 + recovered = recover_fast_track( + shot.candidate_tracks, + shot.ops_ball_speed_mph, + recovery_window=recovery_window, + ) + if recovered is not None: + recovered_count += 1 + resolved_angle = recovered.launch_angle_deg + resolved_source = "recovered_candidate" + else: + # Fallback to model estimate when no valid fast candidate exists + # Driver baseline launch model: ~11.5 deg + resolved_angle = 11.5 + resolved_source = "model_fallback" + + gated_err = resolved_angle - shot.true_launch_angle_deg + gated_errors.append(gated_err) + + shot_details.append( + { + "shot_id": shot.shot_id, + "ops_ball_speed": shot.ops_ball_speed_mph, + "primary_speed": shot.primary_track_speed_mph, + "true_angle": shot.true_launch_angle_deg, + "ungated_angle": shot.primary_launch_angle_deg, + "resolved_angle": resolved_angle, + "resolved_source": resolved_source, + "is_ghost": is_ghost, + "ungated_error": round(ungated_err, 2), + "gated_error": round(gated_err, 2), + }, + ) + + ungated_mae = statistics.mean(abs(e) for e in ungated_errors) + ungated_bias = statistics.mean(ungated_errors) + gated_mae = statistics.mean(abs(e) for e in gated_errors) + gated_bias = statistics.mean(gated_errors) + improvement = ungated_mae - gated_mae + recovery_rate = (recovered_count / ghost_count * 100.0) if ghost_count > 0 else 100.0 + + return GateEvaluationResult( + total_shots=len(parsed_shots), + ghost_tracks_detected=ghost_count, + ghost_tracks_recovered=recovered_count, + ungated_mae_deg=round(ungated_mae, 3), + ungated_bias_deg=round(ungated_bias, 3), + gated_mae_deg=round(gated_mae, 3), + gated_bias_deg=round(gated_bias, 3), + improvement_deg=round(improvement, 3), + recovery_rate_pct=round(recovery_rate, 1), + details=shot_details, + ) + + +def generate_synthetic_driver_dataset( + n_shots: int = 40, + ghost_fraction: float = 0.25, + seed: int = 42, +) -> List[DriverShot]: + """Generate realistic driver shots with known truth launch angles and ghost tracks.""" + rng = random.Random(seed) + shots: List[DriverShot] = [] + + for i in range(1, n_shots + 1): + ops_speed = rng.uniform(145.0, 175.0) + true_angle = rng.uniform(9.5, 14.5) + + is_ghost_shot = rng.random() < ghost_fraction + if is_ghost_shot: + # Ghost track: 48-65 mph with high launch angle artifact (18-28 deg) + primary_speed = rng.uniform(48.0, 65.0) + primary_angle = rng.uniform(18.0, 26.0) + + # In 80% of ghost shots, a valid fast ball track exists among candidates + has_recoverable_track = rng.random() < 0.80 + candidates = [ + CandidateTrack( + track_id=1, + speed_mph=primary_speed, + launch_angle_deg=primary_angle, + point_count=12, + snr_db=19.0, + ), + ] + if has_recoverable_track: + ball_cand_speed = ops_speed * rng.uniform(0.97, 1.03) + ball_cand_angle = true_angle + rng.gauss(0.0, 0.6) + candidates.append( + CandidateTrack( + track_id=2, + speed_mph=ball_cand_speed, + launch_angle_deg=ball_cand_angle, + point_count=rng.randint(8, 14), + snr_db=rng.uniform(14.0, 22.0), + ), + ) + else: + # Clean track: matches OPS speed within 3%, launch angle close to truth + primary_speed = ops_speed * rng.uniform(0.98, 1.02) + primary_angle = true_angle + rng.gauss(0.0, 0.5) + candidates = [ + CandidateTrack( + track_id=1, + speed_mph=primary_speed, + launch_angle_deg=primary_angle, + point_count=rng.randint(10, 16), + snr_db=rng.uniform(16.0, 24.0), + ), + ] + + shots.append( + DriverShot( + shot_id=i, + ops_ball_speed_mph=round(ops_speed, 1), + true_launch_angle_deg=round(true_angle, 1), + primary_track_speed_mph=round(primary_speed, 1), + primary_launch_angle_deg=round(primary_angle, 1), + candidate_tracks=candidates, + ), + ) + + return shots + + +def format_markdown_report(result: GateEvaluationResult) -> str: + """Format gate evaluation results into GitHub-flavored Markdown.""" + lines = [ + "# Driver Ghost-Track Gate & Fast-Track Recovery Validation Report", + "", + "## Summary Metrics", + "", + "| Metric | Ungated Baseline | Gated + Recovery | Delta / Status |", + "| --- | --- | --- | --- |", + f"| **Total Driver Shots** | {result.total_shots} | {result.total_shots} | — |", + f"| **Ghost Tracks Detected** | — | {result.ghost_tracks_detected} ({result.ghost_tracks_detected / max(result.total_shots, 1) * 100:.1f}%) | Flagged (< 65% OPS speed) |", + f"| **Fast Tracks Recovered** | — | {result.ghost_tracks_recovered} | {result.recovery_rate_pct}% recovery rate |", + f"| **Launch Angle MAE** | {result.ungated_mae_deg:.2f}° | {result.gated_mae_deg:.2f}° | **-{result.improvement_deg:.2f}° MAE** |", + f"| **Launch Angle Bias** | {result.ungated_bias_deg:+.2f}° | {result.gated_bias_deg:+.2f}° | **{abs(result.gated_bias_deg) - abs(result.ungated_bias_deg):+.2f}° Bias** |", + "", + "## Shot Log Breakdown", + "", + "| Shot # | OPS Speed (mph) | Primary Speed (mph) | True V.LA | Ungated V.LA | Resolved V.LA | Resolution | Ungated Err | Gated Err |", + "| --- | --- | --- | --- | --- | --- | --- | --- | --- |", + ] + + for d in result.details: + lines.append( + f"| #{d['shot_id']} | {d['ops_ball_speed']} | {d['primary_speed']} | " + f"{d['true_angle']}° | {d['ungated_angle']}° | {d['resolved_angle']}° | " + f"{d['resolved_source']} | {d['ungated_error']:+0.2f}° | {d['gated_error']:+0.2f}° |" + ) + + return "\n".join(lines) + + +def load_dataset_from_file(file_path: str | Path) -> List[DriverShot]: + """Load driver shots from JSONL or CSV.""" + path = Path(file_path) + if not path.exists(): + raise FileNotFoundError(f"Dataset file not found: {path}") + + shots: List[DriverShot] = [] + if path.suffix.lower() == ".jsonl": + with path.open("r", encoding="utf-8") as f: + for idx, line in enumerate(f, 1): + line_s = line.strip() + if not line_s: + continue + data = json.loads(line_s) + cands = [ + CandidateTrack(**c) if isinstance(c, dict) else c + for c in data.get("candidate_tracks", []) + ] + shots.append( + DriverShot( + shot_id=data.get("shot_id", idx), + ops_ball_speed_mph=float(data.get("ops_ball_speed_mph", 0.0)), + true_launch_angle_deg=float(data.get("true_launch_angle_deg", 0.0)), + primary_track_speed_mph=float(data.get("primary_track_speed_mph", 0.0)), + primary_launch_angle_deg=float(data.get("primary_launch_angle_deg", 0.0)), + candidate_tracks=cands, + club=data.get("club", "driver"), + ), + ) + elif path.suffix.lower() == ".csv": + with path.open("r", encoding="utf-8") as f: + reader = csv.DictReader(f) + for idx, row in enumerate(reader, 1): + shots.append( + DriverShot( + shot_id=int(row.get("shot_number", row.get("shot_id", idx))), + ops_ball_speed_mph=float( + row.get("ops_ball_speed", row.get("ball_speed", 0.0)) + ), + true_launch_angle_deg=float( + row.get("true_launch_angle", row.get("reference_launch_angle", 0.0)) + ), + primary_track_speed_mph=float( + row.get("primary_track_speed", row.get("radar_speed", 0.0)) + ), + primary_launch_angle_deg=float( + row.get("primary_launch_angle", row.get("radar_launch_angle", 0.0)) + ), + candidate_tracks=[], + club=row.get("club", "driver"), + ), + ) + else: + raise ValueError(f"Unsupported file format: {path.suffix}") + + return shots + + +def main() -> None: + """CLI entrypoint.""" + parser = argparse.ArgumentParser( + description="Validate driver ghost-track gating and recovery against truth datasets.", + ) + parser.add_argument( + "--dataset", + "-d", + default=None, + help="Path to session JSONL or comparison CSV with driver truth data.", + ) + parser.add_argument( + "--synthetic", + action="store_true", + help="Generate synthetic driver dataset for testing.", + ) + parser.add_argument( + "--min-speed-ratio", + type=float, + default=0.65, + help="Minimum speed ratio relative to OPS ball speed (default: 0.65).", + ) + parser.add_argument( + "--recovery-window", + type=float, + default=0.15, + help="Search window around OPS ball speed for candidate recovery (default: 0.15).", + ) + parser.add_argument( + "--output", + "-o", + default=None, + help="Path to write validation report (.md or .json).", + ) + + args = parser.parse_args() + + if args.dataset: + shots = load_dataset_from_file(args.dataset) + else: + shots = generate_synthetic_driver_dataset(n_shots=40, ghost_fraction=0.25) + + result = evaluate_ghost_track_gate( + shots=shots, + min_speed_ratio=args.min_speed_ratio, + recovery_window=args.recovery_window, + ) + + report_md = format_markdown_report(result) + + if args.output: + out_path = Path(args.output) + out_path.parent.mkdir(parents=True, exist_ok=True) + if out_path.suffix.lower() == ".json": + out_path.write_text(json.dumps(asdict(result), indent=2), encoding="utf-8") + else: + out_path.write_text(report_md, encoding="utf-8") + print(f"Validation report saved to {out_path}") + else: + print(report_md) + + +if __name__ == "__main__": + main() diff --git a/tests/test_radar_interference_check.py b/tests/test_radar_interference_check.py new file mode 100644 index 000000000..64792d0ec --- /dev/null +++ b/tests/test_radar_interference_check.py @@ -0,0 +1,185 @@ +"""Unit tests for radar_interference_check script.""" + +import importlib.util +import json +import sys +import tempfile +from pathlib import Path + +import pytest + +_script_path = ( + Path(__file__).resolve().parents[1] / "scripts" / "analysis" / "radar_interference_check.py" +) +_spec = importlib.util.spec_from_file_location("radar_interference_check", _script_path) +radar_interference_check = importlib.util.module_from_spec(_spec) +sys.modules["radar_interference_check"] = radar_interference_check +_spec.loader.exec_module(radar_interference_check) + +InterferenceComparison = radar_interference_check.InterferenceComparison +InterferenceReport = radar_interference_check.InterferenceReport +PhaseStats = radar_interference_check.PhaseStats +analyze_interference_session = radar_interference_check.analyze_interference_session +compare_phases = radar_interference_check.compare_phases +compute_phase_stats = radar_interference_check.compute_phase_stats +format_markdown_report = radar_interference_check.format_markdown_report +generate_synthetic_interference_session = ( + radar_interference_check.generate_synthetic_interference_session +) +load_shots = radar_interference_check.load_shots + + +class TestPhaseStatsComputation: + """Test phase summary metrics extraction.""" + + def test_compute_stats_from_clean_shots(self): + shots = [ + { + "ball_speed_mph": 150.0, + "spin_rpm": 2500.0, + "spin_snr": 18.0, + "trigger_latency_ms": 0.12, + }, + { + "ball_speed_mph": 152.0, + "spin_rpm": 2550.0, + "spin_snr": 19.0, + "trigger_latency_ms": 0.11, + }, + { + "ball_speed_mph": 149.0, + "spin_rpm": None, + "spin_snr": 5.0, + "trigger_latency_ms": 0.13, + }, + ] + stats = compute_phase_stats(shots, "Phase A") + assert stats.total_shots == 3 + assert stats.spin_read_count == 2 + assert stats.spin_read_rate_pct == pytest.approx(66.7, 0.1) + assert stats.spin_snr_mean_db == 14.0 + assert stats.ball_speed_mean_mph == pytest.approx(150.33, 0.01) + + def test_compute_stats_empty_shots(self): + stats = compute_phase_stats([], "Phase Empty") + assert stats.total_shots == 0 + assert stats.spin_read_rate_pct == 0.0 + assert stats.spin_snr_mean_db is None + + +class TestPhaseComparisonAndSeverity: + """Test degradation calculations and recommendation classification.""" + + def test_clean_interference_classification(self): + stats_a = PhaseStats( + phase_name="A", + total_shots=15, + spin_read_count=14, + spin_read_rate_pct=93.3, + spin_snr_mean_db=18.5, + spin_snr_median_db=18.5, + spin_snr_std_db=1.2, + ball_speed_mean_mph=155.0, + ball_speed_std_mph=1.5, + trigger_latency_mean_ms=0.12, + trigger_latency_std_ms=0.02, + ) + stats_ab = PhaseStats( + phase_name="AB", + total_shots=15, + spin_read_count=14, + spin_read_rate_pct=93.3, + spin_snr_mean_db=18.3, + spin_snr_median_db=18.3, + spin_snr_std_db=1.3, + ball_speed_mean_mph=155.1, + ball_speed_std_mph=1.6, + trigger_latency_mean_ms=0.12, + trigger_latency_std_ms=0.02, + ) + comp = compare_phases(stats_a, stats_ab) + assert comp.severity == "clean" + assert comp.is_snr_degraded is False + assert comp.is_read_rate_degraded is False + assert "Zero significant RF interference" in comp.recommendation + + def test_severe_interference_classification(self): + stats_a = PhaseStats( + phase_name="A", + total_shots=15, + spin_read_count=14, + spin_read_rate_pct=93.3, + spin_snr_mean_db=18.5, + spin_snr_median_db=18.5, + spin_snr_std_db=1.2, + ball_speed_mean_mph=155.0, + ball_speed_std_mph=1.5, + trigger_latency_mean_ms=0.12, + trigger_latency_std_ms=0.02, + ) + stats_ab = PhaseStats( + phase_name="AB", + total_shots=15, + spin_read_count=8, + spin_read_rate_pct=53.3, + spin_snr_mean_db=13.0, + spin_snr_median_db=13.0, + spin_snr_std_db=2.5, + ball_speed_mean_mph=154.5, + ball_speed_std_mph=3.5, + trigger_latency_mean_ms=0.15, + trigger_latency_std_ms=0.04, + ) + comp = compare_phases(stats_a, stats_ab) + assert comp.severity == "severe" + assert comp.is_snr_degraded is True + assert comp.is_read_rate_degraded is True + assert "alternating-shot capture protocol" in comp.recommendation + + +class TestEndToEndReportGeneration: + """Test full session analysis and report generation.""" + + def test_synthetic_session_generation_and_report(self): + shots_a, shots_b, shots_ab = generate_synthetic_interference_session( + simulated_interference_level="clean", + n_shots=15, + seed=42, + ) + report = analyze_interference_session(shots_a, shots_b, shots_ab) + assert report.phase_a.total_shots == 15 + assert report.phase_ab.total_shots == 15 + assert report.comparison.severity == "clean" + + md = format_markdown_report(report) + assert "# Radar Interference Check: A/B/AB Protocol Report" in md + assert "Phase Summary Statistics" in md + assert "Diagnostic Evaluation" in md + + def test_file_loader_jsonl_and_csv(self): + with tempfile.TemporaryDirectory() as tmpdir: + jsonl_path = Path(tmpdir) / "phase_a.jsonl" + jsonl_path.write_text( + json.dumps( + { + "event": "shot_detected", + "ball_speed_mph": 150.0, + "spin_rpm": 2500.0, + "spin_snr": 18.0, + } + ) + + "\n", + encoding="utf-8", + ) + shots_jsonl = load_shots(jsonl_path) + assert len(shots_jsonl) == 1 + assert shots_jsonl[0]["ball_speed_mph"] == 150.0 + + csv_path = Path(tmpdir) / "phase_b.csv" + csv_path.write_text( + "ball_speed,spin_rpm,spin_snr\n152.0,2600.0,19.0\n", + encoding="utf-8", + ) + shots_csv = load_shots(csv_path) + assert len(shots_csv) == 1 + assert shots_csv[0]["ball_speed_mph"] == 152.0 diff --git a/tests/test_validate_ghost_track_gate.py b/tests/test_validate_ghost_track_gate.py new file mode 100644 index 000000000..47aacd86b --- /dev/null +++ b/tests/test_validate_ghost_track_gate.py @@ -0,0 +1,145 @@ +"""Unit tests for validate_ghost_track_gate script.""" + +import importlib.util +import json +import sys +import tempfile +from pathlib import Path + +_script_path = ( + Path(__file__).resolve().parents[1] / "scripts" / "analysis" / "validate_ghost_track_gate.py" +) +_spec = importlib.util.spec_from_file_location("validate_ghost_track_gate", _script_path) +validate_ghost_track_gate = importlib.util.module_from_spec(_spec) +sys.modules["validate_ghost_track_gate"] = validate_ghost_track_gate +_spec.loader.exec_module(validate_ghost_track_gate) + +CandidateTrack = validate_ghost_track_gate.CandidateTrack +DriverShot = validate_ghost_track_gate.DriverShot +GateEvaluationResult = validate_ghost_track_gate.GateEvaluationResult +evaluate_ghost_track_gate = validate_ghost_track_gate.evaluate_ghost_track_gate +format_markdown_report = validate_ghost_track_gate.format_markdown_report +generate_synthetic_driver_dataset = validate_ghost_track_gate.generate_synthetic_driver_dataset +is_ghost_track = validate_ghost_track_gate.is_ghost_track +load_dataset_from_file = validate_ghost_track_gate.load_dataset_from_file +recover_fast_track = validate_ghost_track_gate.recover_fast_track + + +class TestGhostTrackDetection: + """Test ghost track identification logic.""" + + def test_ghost_track_detected_when_speed_ratio_below_threshold(self): + # 55 mph track on 160 mph OPS shot = 34.3% speed ratio (< 65%) + assert ( + is_ghost_track(track_speed_mph=55.0, ops_ball_speed_mph=160.0, min_speed_ratio=0.65) + is True + ) + + def test_valid_track_not_flagged_as_ghost(self): + # 158 mph track on 160 mph OPS shot = 98.7% speed ratio + assert ( + is_ghost_track(track_speed_mph=158.0, ops_ball_speed_mph=160.0, min_speed_ratio=0.65) + is False + ) + + def test_zero_or_negative_ops_speed_handled(self): + assert is_ghost_track(track_speed_mph=55.0, ops_ball_speed_mph=0.0) is False + + +class TestFastTrackRecovery: + """Test candidate recovery logic.""" + + def test_recover_valid_fast_track(self): + candidates = [ + CandidateTrack( + track_id=1, speed_mph=55.0, launch_angle_deg=22.0, point_count=10, snr_db=15.0 + ), + CandidateTrack( + track_id=2, speed_mph=159.0, launch_angle_deg=11.2, point_count=12, snr_db=20.0 + ), + ] + recovered = recover_fast_track( + candidate_tracks=candidates, + ops_ball_speed_mph=160.0, + recovery_window=0.15, + ) + assert recovered is not None + assert recovered.track_id == 2 + assert recovered.launch_angle_deg == 11.2 + + def test_no_recovery_when_no_candidates_in_window(self): + candidates = [ + CandidateTrack(track_id=1, speed_mph=55.0, launch_angle_deg=22.0), + CandidateTrack(track_id=2, speed_mph=80.0, launch_angle_deg=18.0), + ] + recovered = recover_fast_track( + candidate_tracks=candidates, + ops_ball_speed_mph=160.0, + recovery_window=0.15, + ) + assert recovered is None + + +class TestGateEvaluation: + """Test full evaluation pipeline on synthetic driver shots.""" + + def test_evaluation_reduces_mae_and_bias(self): + shots = generate_synthetic_driver_dataset(n_shots=40, ghost_fraction=0.30, seed=42) + result = evaluate_ghost_track_gate(shots, min_speed_ratio=0.65, recovery_window=0.15) + + assert result.total_shots == 40 + assert result.ghost_tracks_detected > 0 + assert result.ghost_tracks_recovered > 0 + assert result.gated_mae_deg < result.ungated_mae_deg + assert result.improvement_deg > 0.5 + assert len(result.details) == 40 + + def test_empty_shots_returns_zero_summary(self): + result = evaluate_ghost_track_gate([]) + assert result.total_shots == 0 + assert result.ungated_mae_deg == 0.0 + + def test_markdown_report_formatting(self): + shots = generate_synthetic_driver_dataset(n_shots=10, ghost_fraction=0.30, seed=42) + result = evaluate_ghost_track_gate(shots) + report = format_markdown_report(result) + assert "# Driver Ghost-Track Gate & Fast-Track Recovery Validation Report" in report + assert "Launch Angle MAE" in report + assert "Shot Log Breakdown" in report + + +class TestDatasetLoaders: + """Test loading driver datasets from files.""" + + def test_load_jsonl_dataset(self): + with tempfile.TemporaryDirectory() as tmpdir: + path = Path(tmpdir) / "test.jsonl" + sample_data = { + "shot_id": 1, + "ops_ball_speed_mph": 165.2, + "true_launch_angle_deg": 11.5, + "primary_track_speed_mph": 54.0, + "primary_launch_angle_deg": 23.4, + "candidate_tracks": [ + {"track_id": 1, "speed_mph": 54.0, "launch_angle_deg": 23.4}, + {"track_id": 2, "speed_mph": 164.0, "launch_angle_deg": 11.8}, + ], + } + path.write_text(json.dumps(sample_data) + "\n", encoding="utf-8") + loaded = load_dataset_from_file(path) + assert len(loaded) == 1 + assert loaded[0].ops_ball_speed_mph == 165.2 + assert len(loaded[0].candidate_tracks) == 2 + + def test_load_csv_dataset(self): + with tempfile.TemporaryDirectory() as tmpdir: + path = Path(tmpdir) / "test.csv" + path.write_text( + "shot_number,ops_ball_speed,true_launch_angle,primary_track_speed,primary_launch_angle\n" + "1,155.0,12.0,154.5,12.2\n", + encoding="utf-8", + ) + loaded = load_dataset_from_file(path) + assert len(loaded) == 1 + assert loaded[0].ops_ball_speed_mph == 155.0 + assert loaded[0].true_launch_angle_deg == 12.0