# Master Spatial Synchronization Integration Test between Vision (2.2) and Physics (2.3) (tc.v1 SOTA). import pytest import numpy as np import pandas as pd from PIL import Image from ml.core.schema import ( ChainageWindow, TrackSegment, SegmentDecision, CalibratedSignal, SignalType, DefectClass, DecisionType, ) from ml.core.chainage import ChainageResampler from ml.features.en13848 import EN13848PhysicsCalculator from ml.models.geometry.physics_detector import EN13848PhysicsThresholdDetector from ml.models.vision.anomaly import PatchCoreAnomalyDetector from ml.models.vision.detector import YOLOv8DefectDetector from ml.fusion.rules import PersistenceRuleFusion from ml.inference.pipeline import EndToEndInferencePipeline from ml.scripts.generate_trc_telemetry import generate_trc_telemetry_csv def create_metallic_surface_frame(has_anomaly: bool = False) -> np.ndarray: """Creates a 224x224 synthetic steel surface image matching NEU-DET characteristics.""" arr = np.random.normal(120, 8, (224, 224, 3)).astype(np.uint8) if has_anomaly: arr[90:130, 80:140] = np.random.normal(30, 5, (40, 60, 3)).astype(np.uint8) return arr def test_vision_physics_spatial_sync(monkeypatch): """ Test that a 100m run with an injected visual anomaly and a 5mm twist at 50m lands in the EXACT SAME TrackSegment bin and triggers both models with exact calibration. """ # 1. Setup: 100m run @ 20 m/s (5.0 seconds total @ 100 Hz = 501 samples) sample_rate_hz = 100.0 speed_mps = 20.0 total_time_s = 5.0 timestamps = np.linspace(0, total_time_s, int(total_time_s * sample_rate_hz) + 1) n_samples = len(timestamps) # Base track roll roll_arr = np.zeros(n_samples) # Inject 5.0mm physical cant step at exactly 50m (idx = 250) # Cant = G * sin(roll) -> roll = arcsin(5.0 / 1676.0) start_idx = 250 roll_arr[start_idx:] = np.arcsin(5.0 / 1676.0) telemetry_df = pd.DataFrame({ "timestamp": timestamps, "speed_mps": np.full(n_samples, speed_mps), "roll_rad": roll_arr, "lat_accel_g": np.random.normal(0, 0.01, n_samples), "vert_accel_g": np.random.normal(1.0, 0.01, n_samples), "gauge_mm": np.full(n_samples, 1676.0), }) # Prepare 5 frames tagged at distances [0, 25, 50, 75, 100]m. Frame at 50m has visual defect frames = [ (0.0, create_metallic_surface_frame(has_anomaly=False)), (25.0, create_metallic_surface_frame(has_anomaly=False)), (50.0, create_metallic_surface_frame(has_anomaly=True)), (75.0, create_metallic_surface_frame(has_anomaly=False)), (100.0, create_metallic_surface_frame(has_anomaly=False)), ] # 2. Resample to Chainage and Segment into 2.0m physical segments resampler = ChainageResampler(bin_size_m=0.25) resampled_data = resampler.process(telemetry_df, frames, segment_length_m=2.0) # 3. Extract the Segment at 50m target_segments = [s for s in resampled_data.segments if s.chainage_start_m == 50.0] assert len(target_segments) == 1, "Target segment at 50.0m not found!" target_segment = target_segments[0] assert target_segment.chainage_start_m == 50.0 assert target_segment.chainage_end_m == 52.0 assert len(target_segment.frames) >= 1 # 4. Run Physics over continuous resampled stream with full 3m multi-chord context physics_calc = EN13848PhysicsCalculator(nominal_gauge_mm=1676.0) all_features = physics_calc.compute(resampled_data.resampled_telemetry, step_m=0.25) # Slice features for target segment (50.0m to 52.0m) grid_m = resampled_data.grid_chainage_m seg_mask = (grid_m >= 50.0) & (grid_m <= 52.0) seg_features = { k: (v[seg_mask] if isinstance(v, np.ndarray) else v) for k, v in all_features.items() } detector = EN13848PhysicsThresholdDetector(twist_limit_mm=4.0) phys_signals = detector.evaluate_features(seg_features) # 5. Run Vision (Mocked for deterministic novelty test) patchcore = PatchCoreAnomalyDetector() def mock_patch_predict(frame: np.ndarray): return [ CalibratedSignal( stream_name="patchcore_anomaly", raw_score=15.8, calibrated_prob=0.82, predicted_class=DefectClass.VISUAL_ANOMALY, is_anomaly=True, signal_type=SignalType.VISUAL_NOVEL, threshold=0.50, bbox=(80.0, 90.0, 140.0, 130.0), explanation="PatchCore surface anomaly detected", ) ] monkeypatch.setattr(patchcore, "predict", mock_patch_predict) vis_signals = patchcore.predict(target_segment.frames[0]) # 6. THE ASSERTIONS (The Spatial & Confidence Sync Proof) # Proof 1: Both models identified the fault assert any(s.fired for s in phys_signals), "Physics failed to detect the 5mm twist!" assert any(s.fired for s in vis_signals), "PatchCore failed to detect the visual anomaly!" # Proof 2: Both models output the EXACT SAME signal type mapping phys_sig = [s for s in phys_signals if s.fired][0] vis_sig = [s for s in vis_signals if s.fired][0] assert phys_sig.signal_type == SignalType.GEOMETRY_KNOWN assert vis_sig.signal_type == SignalType.VISUAL_NOVEL # Proof 3: Both are perfectly calibrated to the [0.0, 1.0] contract # 5mm twist on a 4mm limit = 5 / (2*4) = 0.625 assert abs(phys_sig.value - 0.625) < 0.01, f"Physics calibration math error: {phys_sig.value}" assert vis_sig.value > 0.50, "PatchCore sigmoid calibration failed to cross threshold!" def test_spatial_sync_telemetry_and_vision_binning(): """ Test that asynchronous 100Hz IMU and 15 FPS Vision streams are correctly resampled and aligned to the exact same uniform 0.25m distance bins. """ resampler = ChainageResampler(step_size_m=0.25) speed_mps = 20.0 duration_s = 5.0 imu_ts = np.linspace(0, duration_s, 500) imu_speeds = np.full(500, speed_mps) roll_rad = np.sin(imu_ts * 2 * np.pi * 0.5) * 0.005 grid_m, resampled = resampler.resample_telemetry_batch( timestamps_s=imu_ts, speeds_mps=imu_speeds, sensor_streams={"roll_rad": roll_rad}, ) diffs = np.diff(grid_m) assert np.allclose(diffs, 0.25, atol=1e-3) assert grid_m[0] == 0.0 assert np.isclose(grid_m[-1], 100.0, atol=0.5) assert len(grid_m) == 401 def test_clean_track_false_positive_suppression(tmp_path): """ Test that synthetic TRC telemetry without defects produces zero false alarms. """ csv_path = tmp_path / "clean_trc.csv" generate_trc_telemetry_csv( output_path=str(csv_path), length_m=200.0, defect_mm=0.0, ) df = pd.read_csv(csv_path) detector = EN13848PhysicsThresholdDetector() signals = detector.predict(df) fired_signals = [s for s in signals if s.fired] assert len(fired_signals) == 0, f"Expected 0 false alarms on clean track, got {len(fired_signals)}"