| """ |
| ml/tests/test_final_edge_performance.py |
| Category D: Edge Deployment & Performance Test (tc.v1 SOTA). |
| Benchmarks single-model and end-to-end pipeline latency against edge deployment budgets (targeting >= 15 FPS real-time execution). |
| """ |
|
|
| import sys |
| import time |
| import pytest |
| import numpy as np |
| from pathlib import Path |
|
|
| |
| repo_root = Path(__file__).resolve().parent.parent.parent |
| if str(repo_root) not in sys.path: |
| sys.path.insert(0, str(repo_root)) |
|
|
| from ml.core.schema import TrackSegment |
|
|
|
|
| def test_full_pipeline_latency_benchmark(pipeline, sample_track_segment, test_config): |
| """Benchmark full pipeline inference latency (warmup + 10 runs).""" |
| |
| for _ in range(3): |
| _ = pipeline.process_segment(sample_track_segment) |
|
|
| |
| n_runs = 10 |
| latencies = [] |
| for _ in range(n_runs): |
| start = time.perf_counter() |
| _ = pipeline.process_segment(sample_track_segment) |
| latencies.append((time.perf_counter() - start) * 1000.0) |
|
|
| avg_latency_ms = float(np.mean(latencies)) |
| p95_latency_ms = float(np.percentile(latencies, 95)) |
| fps = 1000.0 / avg_latency_ms if avg_latency_ms > 0 else 0.0 |
|
|
| print(f"\n[Performance] Full Pipeline Average Latency: {avg_latency_ms:.2f} ms (P95: {p95_latency_ms:.2f} ms, {fps:.1f} FPS)") |
|
|
| |
| target_ms = test_config.get("test", {}).get("latency_budget_ms", {}).get("full_pipeline_fp32", 150.0) |
| if avg_latency_ms > target_ms: |
| pytest.skip(f"Latency ({avg_latency_ms:.1f}ms) above target ({target_ms}ms) on current CPU hardware.") |
|
|
|
|
| def test_geometry_stream_latency_budget(pipeline): |
| """Verify geometry stream (Physics + Bi-LSTM + VAE) is blazing fast (< 15ms).""" |
| geom_features = { |
| "twist_3m_mm": np.zeros(80), |
| "versine_10m_mm": np.zeros(80), |
| "longitudinal_level_d1_mm": np.zeros(80), |
| "gauge_deviation_mm": np.zeros(80), |
| } |
|
|
| |
| _ = pipeline.physics_detector.evaluate_features(geom_features) |
| _ = pipeline.fault_classifier.predict(geom_features) |
| _ = pipeline.sequence_vae.predict(geom_features) |
|
|
| |
| start = time.perf_counter() |
| for _ in range(20): |
| _ = pipeline.physics_detector.evaluate_features(geom_features) |
| _ = pipeline.fault_classifier.predict(geom_features) |
| _ = pipeline.sequence_vae.predict(geom_features) |
| elapsed_ms = (time.perf_counter() - start) * 1000.0 / 20.0 |
|
|
| print(f"\n[Performance] Geometry Triad Latency: {elapsed_ms:.2f} ms") |
| assert elapsed_ms < 50.0, f"Geometry triad exceeded latency ceiling: {elapsed_ms:.2f}ms" |
|
|
