from __future__ import annotations import hashlib import json from pathlib import Path import pytest from k1link.compute.e46f_dashcam_bakeoff import ( E46F_RUNTIME_SCHEMA, E46FDashCamBakeoffError, build_e46f_dashcam_bakeoff, read_e46f_dashcam_bakeoff, ) def test_e46f_freezes_only_the_stock_detector_change(tmp_path: Path) -> None: source, profile = _source_and_profile(tmp_path, frame_count=3) raw = _raw_output(tmp_path, profile, frame_count=3) _observation(raw, "detections", 0, "car", (10, 20, 40, 60), 0.91) _observation(raw, "tracks", 0, "car", (10, 20, 40, 60), 0.88, track_id=7) _observation(raw, "tracks", 1, "car", (12, 20, 42, 60), 0.73, track_id=7) _observation(raw, "detections", 2, "person", (100, 80, 130, 160), 0.93) _observation(raw, "tracks", 2, "person", (100, 80, 130, 160), 0.79, track_id=8) result = build_e46f_dashcam_bakeoff( source_job_root=source, raw_root=raw, profile_path=profile, output_root=tmp_path / "results", ) metrics = result["report"]["metrics"] assert metrics["frame_count"] == 3 assert metrics["detection_observation_count"] == 2 assert metrics["track_observation_count"] == 3 assert metrics["tracker_recovered_frame_count"] == 1 assert result["frames"][0]["detections"][0]["provenance"] == ("nvidia-dashcamnet-detectnet-v2") assert result["report"]["comparison_contract"] == { "baseline_result_id": f"e46e-ready-stack-{'a' * 64}", "controlled_change": "detector-only", "held_constant": ["recorded RIGHT source", "DeepStream", "FP16", "NvDCF"], } assert result["report"]["decision"]["custom_temporal_logic_used"] is False assert result["manifest"]["authority"]["navigation_or_safety_accepted"] is False repeated = build_e46f_dashcam_bakeoff( source_job_root=source, raw_root=raw, profile_path=profile, output_root=tmp_path / "results", ) assert repeated["result_id"] == result["result_id"] def test_e46f_rejects_tampered_immutable_video(tmp_path: Path) -> None: source, profile = _source_and_profile(tmp_path, frame_count=1) raw = _raw_output(tmp_path, profile, frame_count=1) result = build_e46f_dashcam_bakeoff( source_job_root=source, raw_root=raw, profile_path=profile, output_root=tmp_path / "results", ) (result["result_root"] / "overlay.mp4").write_bytes(b"changed") with pytest.raises(E46FDashCamBakeoffError, match="artifact changed"): read_e46f_dashcam_bakeoff(result["result_root"]) def _source_and_profile(tmp_path: Path, *, frame_count: int) -> tuple[Path, Path]: source = tmp_path / "source-job" camera = source / "input" / "camera" / "sensor.camera.right" / "epoch-1" camera.mkdir(parents=True) rows = [ { "schema_version": "missioncore.camera-recording-index/v1", "kind": "media", "sequence": sequence, "session_monotonic_ns": 1_000_000_000 + (sequence - 1) * 100_000_000, "sha256": hashlib.sha256(f"frame-{sequence}".encode()).hexdigest(), } for sequence in range(1, frame_count + 1) ] index_path = camera / "index.jsonl" index_path.write_text( "".join(json.dumps(row, sort_keys=True) + "\n" for row in rows), encoding="utf-8", ) stream_sha = "1" * 64 summary_path = camera / "summary.json" summary_path.write_text( json.dumps( { "schema_version": "missioncore.camera-recording/v1", "stream_sha256": stream_sha, "segment_count": frame_count, }, indent=2, ) + "\n", encoding="utf-8", ) index_sha = _sha(index_path) summary_sha = _sha(summary_path) (source / "job.json").write_text( json.dumps( { "schema_version": "missioncore.compute-job/v1", "job_id": "recorded-test", "input": { "session_id": "test-session", "source_id": "sensor.camera.right", "segment_count": frame_count, "archive_index_sha256": index_sha, "archive_summary_sha256": summary_sha, "timeline": {"start_seconds": 10.0, "end_seconds": 11.0}, }, }, indent=2, ) + "\n", encoding="utf-8", ) profile_value = { "schema_version": "missioncore.e46f-dashcam-bakeoff-profile/v1", "profile_id": "e46f-test/v1", "comparison_contract": { "baseline_result_id": f"e46e-ready-stack-{'a' * 64}", "controlled_change": "detector-only", "held_constant": ["recorded RIGHT source", "DeepStream", "FP16", "NvDCF"], }, "source": { "camera_source_id": "sensor.camera.right", "job_id": "recorded-test", "session_id": "test-session", "segment_count": frame_count, "stream_sha256": stream_sha, "archive_index_sha256": index_sha, "archive_summary_sha256": summary_sha, }, "runtime": { "container_image": "nvcr.io/nvidia/deepstream:9.1-samples-multiarch@sha256:" + "2" * 64, "deepstream_version": "9.1", }, "detector": { "name": "NVIDIA DashCamNet", "version": "pruned_onnx_v1.0.4", "model_sha256": "3" * 64, "custom_postprocessing": False, }, "postprocessor": { "name": "NVIDIA DeepStream built-in DetectNet_v2 parser and NMS", "cluster_mode": "NMS", "reference_commit": "4" * 40, "reference_config_sha256": "5" * 64, "custom_mission_core_logic": False, }, "tracker": { "name": "NVIDIA NvDCF", "configuration": "stock-performance", "custom_association": False, "custom_hold_or_stitch": False, }, "output": {"frame_width": 800, "frame_height": 600}, "authority": { "ground_truth": False, "candidate_accepted": False, "commands_enabled": False, "navigation_or_safety_accepted": False, }, } profile = tmp_path / "profile.json" profile.write_text(json.dumps(profile_value, indent=2) + "\n", encoding="utf-8") return source, profile def _raw_output(tmp_path: Path, profile_path: Path, *, frame_count: int) -> Path: raw = tmp_path / "raw" for name in ("detections", "tracks"): directory = raw / name directory.mkdir(parents=True, exist_ok=True) for frame in range(frame_count): (directory / f"00_000_{frame:06d}.txt").write_text("", encoding="utf-8") overlay = raw / "overlay.mp4" overlay.write_bytes(b"synthetic-overlay") (raw / "deepstream.log").write_text("synthetic success\n", encoding="utf-8") profile = json.loads(profile_path.read_text(encoding="utf-8")) image = profile["runtime"]["container_image"] runtime = { "schema_version": E46F_RUNTIME_SCHEMA, "status": "completed", "worker_host": "TEST-WORKER-006", "gpu_name": "Synthetic RTX", "container_image": image, "container_image_digest": image.rsplit("@sha256:", 1)[1], "model_sha256": profile["detector"]["model_sha256"], "model_engine_sha256": "6" * 64, "deepstream_config_sha256": "7" * 64, "detector_config_sha256": "8" * 64, "tracker_config_sha256": "9" * 64, "input_stream_sha256": profile["source"]["stream_sha256"], "overlay_sha256": _sha(overlay), } (raw / "runtime.json").write_text(json.dumps(runtime, indent=2) + "\n", encoding="utf-8") return raw def _observation( raw: Path, directory: str, frame: int, label: str, box: tuple[int, int, int, int], confidence: float, *, track_id: int | None = None, ) -> None: left, top, right, bottom = box identity = "" if track_id is None else f" {track_id}" (raw / directory / f"00_000_{frame:06d}.txt").write_text( f"{label}{identity} 0.0 0 0.0 {left} {top} {right} {bottom} 0 0 0 0 0 0 0 {confidence}\n", encoding="utf-8", ) def _sha(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest()