ml/tests/test_integration_sync.py

# 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)}"