From c05db1fbe0ed492ae8dec7b55c4f15123017b5d6 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Wed, 26 Aug 2026 10:15:43 +0300 Subject: [PATCH] feat(perception): mine native fisheye risk cases --- .../m48q-native-risk-case-mining-v1.json | 58 ++ ...run_m48q_native_risk_case_mining_worker.py | 677 ++++++++++++++++++ .../Invoke-M48QNativeRiskCaseMining.ps1 | 207 ++++++ tests/test_m48q_native_risk_case_mining.py | 90 +++ 4 files changed, 1032 insertions(+) create mode 100644 config/perception/m48q-native-risk-case-mining-v1.json create mode 100644 experiments/perception/run_m48q_native_risk_case_mining_worker.py create mode 100644 experiments/perception/worker/Invoke-M48QNativeRiskCaseMining.ps1 create mode 100644 tests/test_m48q_native_risk_case_mining.py diff --git a/config/perception/m48q-native-risk-case-mining-v1.json b/config/perception/m48q-native-risk-case-mining-v1.json new file mode 100644 index 0000000..efbeac9 --- /dev/null +++ b/config/perception/m48q-native-risk-case-mining-v1.json @@ -0,0 +1,58 @@ +{ + "schema_version": "missioncore.m48q-native-risk-case-mining-profile/v1", + "profile_id": "m48q-ravnoves00-native-raw-kb4-risk-review/v1", + "question": "Which real RAVNOVES00 raw-fisheye frames expose the native RF-DETR risk semantics that require operator review before independent route truth exists?", + "source": { + "source_id": "RAVNOVES00", + "frame_count": 4489, + "raster_width": 800, + "raster_height": 600, + "video_sha256": "cadd1696ff000904eb78633a0a8418104b8024f178b91f3421789021ccb160e8", + "geometric_resampling": false, + "rectification": false, + "warp": false + }, + "candidate": { + "provider_id": "triton-rf-detr-large-coco-native-kb4-risk-fp16-shadow/v0", + "model_id": "rf_detr_large_native_kb4:1", + "preprocess_id": "raw-kb4-uint8-fused-mask-rgb-pad8-imagenet-trt/v0", + "engine_sha256": "b8a40b3580edff001ec9680de68707242294ff590ab296000fae371f1083f695", + "minimum_score": 0.25 + }, + "risk_families": { + "person": ["person"], + "animal": ["bird", "cat", "dog", "horse", "sheep", "cow", "elephant", "bear", "zebra", "giraffe"], + "light-road-user": ["bicycle", "motorcycle", "skateboard"], + "vehicle": ["car", "bus", "truck"] + }, + "selection": { + "case_count": 24, + "minimum_sequence_separation": 12, + "low_confidence_maximum_score": 0.4, + "fisheye_edge_margin_fraction": 0.12, + "bucket_quotas": { + "person": 3, + "animal": 3, + "light-road-user": 3, + "vehicle": 3, + "low-confidence": 3, + "fisheye-edge": 3, + "native-fewer-than-legacy": 3, + "native-more-than-legacy": 3 + } + }, + "scope": { + "quality_evaluated": false, + "ground_truth": false, + "candidate_accepted": false, + "production_accepted": false, + "purpose": "bounded-native-risk-case-review" + }, + "authority": { + "ground_truth": false, + "candidate_accepted": false, + "commands_enabled": false, + "actuation_allowed": false, + "navigation_or_safety_accepted": false + } +} diff --git a/experiments/perception/run_m48q_native_risk_case_mining_worker.py b/experiments/perception/run_m48q_native_risk_case_mining_worker.py new file mode 100644 index 0000000..2378422 --- /dev/null +++ b/experiments/perception/run_m48q_native_risk_case_mining_worker.py @@ -0,0 +1,677 @@ +#!/usr/bin/env python3 +"""Mine bounded native raw-fisheye risk cases from immutable Worker 006 ledgers.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import math +import sys +from collections import Counter +from collections.abc import Iterable, Mapping, Sequence +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Final + +PROFILE_SCHEMA: Final = "missioncore.m48q-native-risk-case-mining-profile/v1" +RESULT_SCHEMA: Final = "missioncore.m48q-native-risk-case-mining-result/v1" +CASE_SCHEMA: Final = "missioncore.m48q-native-risk-review-case/v1" +GRAPH_FRAME_SCHEMA: Final = "missioncore.m48s-reference-graph-frame-evidence/v1" +COMPARISON_FRAME_SCHEMA: Final = "missioncore.m48n-native-vs-704-frame/v0" +FALSE_AUTHORITY: Final = { + "ground_truth": False, + "candidate_accepted": False, + "commands_enabled": False, + "actuation_allowed": False, + "navigation_or_safety_accepted": False, +} + + +class M48QCaseMiningError(RuntimeError): + """Raised when immutable native review evidence is inconsistent.""" + + +@dataclass(frozen=True, slots=True) +class Proposal: + proposal_id: str + class_name: str + risk_family: str + score: float + box_xyxy: tuple[float, float, float, float] + + +@dataclass(frozen=True, slots=True) +class FrameCandidate: + sequence: int + frame_id: str + evidence_time_ns: int + proposals: tuple[Proposal, ...] + native_count: int + legacy_count: int + matched_count: int + buckets: frozenset[str] + + +def canonical_json(value: object) -> bytes: + return json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + + +def sha256_path(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _read_object(path: Path) -> dict[str, Any]: + try: + value: object = json.loads(path.read_text("utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise M48QCaseMiningError(f"cannot read JSON object: {path.name}") from exc + if not isinstance(value, dict): + raise M48QCaseMiningError(f"JSON evidence is not an object: {path.name}") + return value + + +def _verify_hash(path: Path, expected: str, label: str) -> str: + actual = sha256_path(path) + if actual != expected: + raise M48QCaseMiningError(f"{label} SHA-256 changed: {actual}") + return actual + + +def _risk_family_map(profile: Mapping[str, Any]) -> dict[str, str]: + raw = profile.get("risk_families") + if not isinstance(raw, dict): + raise M48QCaseMiningError("risk family profile is invalid") + result: dict[str, str] = {} + for family, classes in raw.items(): + if ( + family not in {"person", "animal", "light-road-user", "vehicle"} + or not isinstance(classes, list) + or not classes + ): + raise M48QCaseMiningError("risk family profile changed") + for class_name in classes: + if not isinstance(class_name, str) or class_name in result: + raise M48QCaseMiningError("risk classes are invalid or duplicated") + result[class_name] = family + return result + + +def _proposal( + value: object, + *, + sequence: int, + profile: Mapping[str, Any], + families: Mapping[str, str], +) -> Proposal: + if not isinstance(value, dict): + raise M48QCaseMiningError("detector proposal is not an object") + candidate = profile["candidate"] + expected_frame_id = f"frame-{sequence:06d}" + if ( + value.get("frame_id") != expected_frame_id + or value.get("provider_id") != candidate["provider_id"] + or value.get("model_id") != candidate["model_id"] + or value.get("preprocess_id") != candidate["preprocess_id"] + ): + raise M48QCaseMiningError("native detector proposal identity changed") + proposal_id = value.get("proposal_id") + class_name = value.get("semantic_hint") + score = value.get("objectness") + region = value.get("region") + if ( + not isinstance(proposal_id, str) + or not isinstance(class_name, str) + or class_name not in families + or not isinstance(score, (float, int)) + or isinstance(score, bool) + or not math.isfinite(float(score)) + or float(score) < candidate["minimum_score"] + or not isinstance(region, dict) + ): + raise M48QCaseMiningError("native detector proposal contract changed") + coordinates = tuple(region.get(key) for key in ("x_min", "y_min", "x_max", "y_max")) + if not all( + isinstance(item, (float, int)) and not isinstance(item, bool) and math.isfinite(float(item)) + for item in coordinates + ): + raise M48QCaseMiningError("native detector box is invalid") + left, top, right, bottom = (float(item) for item in coordinates) + source = profile["source"] + if not ( + 0 <= left < right <= source["raster_width"] and 0 <= top < bottom <= source["raster_height"] + ): + raise M48QCaseMiningError("native detector box escaped the raw source raster") + return Proposal( + proposal_id=proposal_id, + class_name=class_name, + risk_family=families[class_name], + score=float(score), + box_xyxy=(left, top, right, bottom), + ) + + +def load_comparison_rows(path: Path, expected_count: int) -> dict[int, tuple[int, int, int]]: + rows: dict[int, tuple[int, int, int]] = {} + try: + with path.open("r", encoding="utf-8") as stream: + for line_number, line in enumerate(stream, start=1): + value = json.loads(line) + if not isinstance(value, dict): + raise M48QCaseMiningError("comparison frame is not an object") + if value.get("schema_version") not in {None, COMPARISON_FRAME_SCHEMA}: + raise M48QCaseMiningError("comparison frame schema changed") + sequence = value.get("frame_index") + counts = tuple( + value.get(key) + for key in ( + "native_detection_count", + "baseline_detection_count", + "matched_detection_count_iou_at_least_0_5", + ) + ) + if ( + not isinstance(sequence, int) + or isinstance(sequence, bool) + or sequence in rows + or not all( + isinstance(item, int) and not isinstance(item, bool) and item >= 0 + for item in counts + ) + ): + raise M48QCaseMiningError(f"comparison frame {line_number} is invalid") + rows[sequence] = counts + except (OSError, json.JSONDecodeError) as exc: + raise M48QCaseMiningError("comparison frame ledger cannot be read") from exc + if set(rows) != set(range(expected_count)): + raise M48QCaseMiningError("comparison frame ledger is incomplete") + return rows + + +def _case_buckets( + proposals: Sequence[Proposal], + *, + native_count: int, + legacy_count: int, + profile: Mapping[str, Any], +) -> frozenset[str]: + selection = profile["selection"] + buckets = {item.risk_family for item in proposals} + if any(item.score <= selection["low_confidence_maximum_score"] for item in proposals): + buckets.add("low-confidence") + width = profile["source"]["raster_width"] + height = profile["source"]["raster_height"] + margin = selection["fisheye_edge_margin_fraction"] + if any( + item.box_xyxy[0] <= width * margin + or item.box_xyxy[2] >= width * (1 - margin) + or item.box_xyxy[1] <= height * margin + or item.box_xyxy[3] >= height * (1 - margin) + for item in proposals + ): + buckets.add("fisheye-edge") + if native_count < legacy_count: + buckets.add("native-fewer-than-legacy") + elif native_count > legacy_count: + buckets.add("native-more-than-legacy") + return frozenset(buckets) + + +def load_graph_candidates( + path: Path, + *, + profile: Mapping[str, Any], + comparisons: Mapping[int, tuple[int, int, int]], +) -> list[FrameCandidate]: + families = _risk_family_map(profile) + expected_count = profile["source"]["frame_count"] + candidates: list[FrameCandidate] = [] + try: + with path.open("r", encoding="utf-8") as stream: + for sequence, line in enumerate(stream): + value = json.loads(line) + if not isinstance(value, dict) or value.get("schema_version") != GRAPH_FRAME_SCHEMA: + raise M48QCaseMiningError("graph frame schema changed") + source = value.get("source_envelope") + raw_proposals = value.get("detector_proposals") + if ( + not isinstance(source, dict) + or source.get("sequence") != sequence + or source.get("frame_id") != f"frame-{sequence:06d}" + or not isinstance(raw_proposals, list) + or value.get("authority") != FALSE_AUTHORITY + ): + raise M48QCaseMiningError("graph frame identity or authority changed") + timestamps = source.get("timestamps") + evidence_time_ns = ( + timestamps.get("source_ns") if isinstance(timestamps, dict) else None + ) + if ( + not isinstance(evidence_time_ns, int) + or isinstance(evidence_time_ns, bool) + or evidence_time_ns < 0 + ): + raise M48QCaseMiningError("graph frame evidence time is invalid") + proposals = tuple( + _proposal(item, sequence=sequence, profile=profile, families=families) + for item in raw_proposals + ) + native_count, legacy_count, matched_count = comparisons[sequence] + if len(proposals) != native_count: + raise M48QCaseMiningError("native graph/comparison detector count changed") + buckets = _case_buckets( + proposals, + native_count=native_count, + legacy_count=legacy_count, + profile=profile, + ) + candidates.append( + FrameCandidate( + sequence=sequence, + frame_id=source["frame_id"], + evidence_time_ns=evidence_time_ns, + proposals=proposals, + native_count=native_count, + legacy_count=legacy_count, + matched_count=matched_count, + buckets=buckets, + ) + ) + except (OSError, json.JSONDecodeError) as exc: + raise M48QCaseMiningError("graph frame ledger cannot be read") from exc + if len(candidates) != expected_count: + raise M48QCaseMiningError("graph frame ledger is incomplete") + return candidates + + +def _quantile_order(items: Sequence[FrameCandidate]) -> Iterable[FrameCandidate]: + if not items: + return () + indexes: list[int] = [] + left = 0 + right = len(items) - 1 + while left <= right: + middle = (left + right) // 2 + indexes.append(middle) + if middle - left > 0: + indexes.append((left + middle - 1) // 2) + if right - middle > 0: + indexes.append((middle + 1 + right) // 2) + left += 1 + right -= 1 + seen: set[int] = set() + return (items[index] for index in indexes if not (index in seen or seen.add(index))) + + +def select_cases( + candidates: Sequence[FrameCandidate], + *, + bucket_quotas: Mapping[str, int], + minimum_sequence_separation: int, +) -> list[FrameCandidate]: + selected: list[FrameCandidate] = [] + selected_sequences: set[int] = set() + for bucket, quota in bucket_quotas.items(): + eligible = [item for item in candidates if bucket in item.buckets] + admitted = 0 + for item in _quantile_order(eligible): + if item.sequence in selected_sequences: + continue + if any( + abs(item.sequence - prior.sequence) < minimum_sequence_separation + for prior in selected + ): + continue + selected.append(item) + selected_sequences.add(item.sequence) + admitted += 1 + if admitted == quota: + break + if admitted != quota: + raise M48QCaseMiningError( + f"selection bucket {bucket} produced {admitted}/{quota} separated cases" + ) + expected = sum(bucket_quotas.values()) + if len(selected) != expected: + raise M48QCaseMiningError("selected case count changed") + return sorted(selected, key=lambda item: item.sequence) + + +def render_selected_frames( + *, + video_path: Path, + selected: Sequence[FrameCandidate], + cases_root: Path, + width: int, + height: int, +) -> dict[int, dict[str, object]]: + import av + from av.error import FFmpegError + from PIL import Image + + wanted = {item.sequence for item in selected} + rendered: dict[int, dict[str, object]] = {} + cases_root.mkdir(mode=0o700, parents=True, exist_ok=False) + try: + with av.open(str(video_path), mode="r") as container: + streams = list(container.streams.video) + if len(streams) != 1: + raise M48QCaseMiningError("RAVNOVES00 video stream contract changed") + for sequence, frame in enumerate(container.decode(streams[0])): + if sequence not in wanted: + continue + array = frame.to_ndarray(format="rgb24") + if array.shape != (height, width, 3): + raise M48QCaseMiningError("decoded raw-fisheye raster changed") + name = f"frame-{sequence:06d}.jpg" + path = cases_root / name + Image.fromarray(array, mode="RGB").save( + path, + format="JPEG", + quality=94, + subsampling=0, + optimize=False, + ) + rendered[sequence] = { + "path": f"cases/{name}", + "media_type": "image/jpeg", + "width": width, + "height": height, + "byte_length": path.stat().st_size, + "sha256": sha256_path(path), + "geometric_resampling": False, + } + if len(rendered) == len(wanted): + break + except (FFmpegError, OSError) as exc: + raise M48QCaseMiningError("RAVNOVES00 raw-fisheye frames cannot be decoded") from exc + if set(rendered) != wanted: + raise M48QCaseMiningError("not every selected raw-fisheye frame was decoded") + return rendered + + +def _case_document( + candidate: FrameCandidate, + *, + image: Mapping[str, object], +) -> dict[str, object]: + return { + "schema_version": CASE_SCHEMA, + "case_id": f"{candidate.sequence:06d}", + "sequence": candidate.sequence, + "frame_id": candidate.frame_id, + "evidence_time_ns": candidate.evidence_time_ns, + "selection_buckets": sorted(candidate.buckets), + "comparison": { + "native_detection_count": candidate.native_count, + "legacy_704_detection_count": candidate.legacy_count, + "matched_detection_count_iou_at_least_0_5": candidate.matched_count, + "quality_interpretation": "diagnostic-only", + }, + "image": dict(image), + "proposals": [ + { + "proposal_id": item.proposal_id, + "class_name": item.class_name, + "risk_family": item.risk_family, + "score": round(item.score, 9), + "box_xyxy": [round(value, 6) for value in item.box_xyxy], + } + for item in candidate.proposals + ], + "ground_truth": False, + "quality_evaluated": False, + "authority": dict(FALSE_AUTHORITY), + } + + +def run(args: argparse.Namespace) -> dict[str, object]: + for path in ( + args.profile, + args.graph_result, + args.graph_frames, + args.comparison_result, + args.comparison_frames, + args.video, + ): + if path.is_symlink() or not path.is_file(): + raise M48QCaseMiningError(f"required evidence is missing: {path.name}") + profile_sha256 = _verify_hash(args.profile, args.expected_profile_sha256, "profile") + graph_result_sha256 = _verify_hash( + args.graph_result, args.expected_graph_result_sha256, "graph result" + ) + graph_frames_sha256 = _verify_hash( + args.graph_frames, args.expected_graph_frames_sha256, "graph frames" + ) + comparison_result_sha256 = _verify_hash( + args.comparison_result, + args.expected_comparison_result_sha256, + "comparison result", + ) + comparison_frames_sha256 = _verify_hash( + args.comparison_frames, + args.expected_comparison_frames_sha256, + "comparison frames", + ) + video_sha256 = _verify_hash(args.video, args.expected_video_sha256, "video") + runner_sha256 = _verify_hash(Path(__file__), args.expected_runner_sha256, "runner") + + profile = _read_object(args.profile) + graph_result = _read_object(args.graph_result) + comparison_result = _read_object(args.comparison_result) + source = profile.get("source") + candidate = profile.get("candidate") + selection = profile.get("selection") + scope = profile.get("scope") + if ( + profile.get("schema_version") != PROFILE_SCHEMA + or not isinstance(source, dict) + or source.get("video_sha256") != video_sha256 + or source.get("raster_width") != 800 + or source.get("raster_height") != 600 + or source.get("frame_count") != 4489 + or source.get("geometric_resampling") is not False + or source.get("rectification") is not False + or source.get("warp") is not False + or not isinstance(candidate, dict) + or candidate.get("engine_sha256") + != "b8a40b3580edff001ec9680de68707242294ff590ab296000fae371f1083f695" + or not isinstance(selection, dict) + or selection.get("case_count") != 24 + or not isinstance(selection.get("bucket_quotas"), dict) + or sum(selection["bucket_quotas"].values()) != 24 + or scope + != { + "quality_evaluated": False, + "ground_truth": False, + "candidate_accepted": False, + "production_accepted": False, + "purpose": "bounded-native-risk-case-review", + } + or profile.get("authority") != FALSE_AUTHORITY + ): + raise M48QCaseMiningError("M4.8Q profile contract changed") + if ( + graph_result.get("schema_version") != "missioncore.m48s-reference-graph-shadow-load/v5" + or graph_result.get("completed") is not True + or graph_result.get("identity", {}).get("detector_provider_id") != candidate["provider_id"] + or graph_result.get("identity", {}).get("inputs", {}).get("video") != video_sha256 + or graph_result.get("execution", {}).get("frame_evidence", {}).get("sha256") + != graph_frames_sha256 + or graph_result.get("authority") != FALSE_AUTHORITY + or graph_result.get("production_accepted") is not False + ): + raise M48QCaseMiningError("native graph result contract changed") + if ( + comparison_result.get("schema_version") != "missioncore.m48n-native-vs-704-ravnoves00/v0" + or comparison_result.get("completed") is not True + or comparison_result.get("source", {}).get("video_sha256") != video_sha256 + or comparison_result.get("source", {}).get("frame_count") != source["frame_count"] + or comparison_result.get("execution", {}).get("frames_sha256") != comparison_frames_sha256 + or comparison_result.get("authority") != FALSE_AUTHORITY + ): + raise M48QCaseMiningError("native/legacy comparison result contract changed") + + output_root = args.output_root + if output_root.exists(): + raise M48QCaseMiningError("M4.8Q output root already exists") + output_root.mkdir(mode=0o700, parents=True) + comparisons = load_comparison_rows(args.comparison_frames, source["frame_count"]) + candidates = load_graph_candidates( + args.graph_frames, + profile=profile, + comparisons=comparisons, + ) + selected = select_cases( + candidates, + bucket_quotas=selection["bucket_quotas"], + minimum_sequence_separation=selection["minimum_sequence_separation"], + ) + rendered = render_selected_frames( + video_path=args.video, + selected=selected, + cases_root=output_root / "cases", + width=source["raster_width"], + height=source["raster_height"], + ) + cases = [ + _case_document( + item, + image=rendered[item.sequence], + ) + for item in selected + ] + cases_path = output_root / "cases.jsonl" + with cases_path.open("xb") as stream: + for case in cases: + stream.write(canonical_json(case) + b"\n") + + class_counts = Counter(proposal.class_name for item in selected for proposal in item.proposals) + selected_bucket_coverage = Counter(bucket for item in selected for bucket in item.buckets) + identity = { + "schema_version": RESULT_SCHEMA, + "profile": {"profile_id": profile["profile_id"], "sha256": profile_sha256}, + "source": { + "source_id": source["source_id"], + "video_sha256": video_sha256, + "graph_result_sha256": graph_result_sha256, + "graph_frames_sha256": graph_frames_sha256, + "comparison_result_sha256": comparison_result_sha256, + "comparison_frames_sha256": comparison_frames_sha256, + }, + "candidate": dict(candidate), + "runner_sha256": runner_sha256, + "cases_sha256": sha256_path(cases_path), + "authority": dict(FALSE_AUTHORITY), + } + result: dict[str, object] = { + "schema_version": RESULT_SCHEMA, + "status": "complete-review-ready-quality-not-adjudicated", + "completed": True, + "worker": {"worker_id": "worker-006", "node": "DESKTOP-OPJ8J04"}, + "identity": identity, + "report_identity_sha256": hashlib.sha256(canonical_json(identity)).hexdigest(), + "source": { + **dict(source), + "graph_result_sha256": graph_result_sha256, + "graph_frames_sha256": graph_frames_sha256, + "comparison_result_sha256": comparison_result_sha256, + "comparison_frames_sha256": comparison_frames_sha256, + }, + "candidate": dict(candidate), + "selection": { + "case_count": len(cases), + "minimum_sequence_separation": selection["minimum_sequence_separation"], + "configured_bucket_quotas": selection["bucket_quotas"], + "selected_bucket_coverage": dict(sorted(selected_bucket_coverage.items())), + "selected_sequences": [item.sequence for item in selected], + "selected_class_counts": dict(sorted(class_counts.items())), + }, + "artifacts": { + "cases": { + "path": "cases.jsonl", + "schema_version": CASE_SCHEMA, + "row_count": len(cases), + "sha256": sha256_path(cases_path), + }, + "images": { + "root": "cases", + "count": len(rendered), + "geometric_resampling": False, + }, + }, + "decision": { + "quality_evaluated": False, + "ground_truth": False, + "candidate_accepted": False, + "production_accepted": False, + "next_action": "operator-adjudication-in-existing-m48-review-instrument", + }, + "limitations": [ + "The selected RAVNOVES00 cases are diagnostic samples, not independent truth.", + ( + "Native-versus-legacy detection-count differences are not quality " + "verdicts because legacy 704 stretches the raw 4:3 raster." + ), + ( + "No child/adult, behavior, physical track identity, navigation or " + "actuation claim is made." + ), + ( + "JPEG review derivatives preserve the 800x600 raster without geometric " + "resampling but are not lossless source frames." + ), + ], + "authority": dict(FALSE_AUTHORITY), + } + (output_root / "result.json").write_bytes(canonical_json(result) + b"\n") + return result + + +def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + for name in ( + "profile", + "graph-result", + "graph-frames", + "comparison-result", + "comparison-frames", + "video", + "output-root", + ): + parser.add_argument(f"--{name}", type=Path, required=True) + for name in ( + "profile", + "graph-result", + "graph-frames", + "comparison-result", + "comparison-frames", + "video", + "runner", + ): + parser.add_argument(f"--expected-{name}-sha256", required=True) + return parser.parse_args(argv) + + +def main(argv: Sequence[str] | None = None) -> int: + try: + result = run(parse_args(argv)) + except M48QCaseMiningError as exc: + print(f"M4.8Q case mining refused: {exc}", file=sys.stderr) + return 2 + print(json.dumps(result, ensure_ascii=False, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/experiments/perception/worker/Invoke-M48QNativeRiskCaseMining.ps1 b/experiments/perception/worker/Invoke-M48QNativeRiskCaseMining.ps1 new file mode 100644 index 0000000..4dd5348 --- /dev/null +++ b/experiments/perception/worker/Invoke-M48QNativeRiskCaseMining.ps1 @@ -0,0 +1,207 @@ +[CmdletBinding()] +param( + [Parameter(Mandatory = $true)] + [string]$ReleaseRoot, + [Parameter(Mandatory = $true)] + [ValidatePattern("^[A-Za-z0-9._-]{1,96}$")] + [string]$RunId, + [string]$OutputRoot = "D:\NDC_MISSIONCORE\runtime\results\m48q-native-risk-case-mining" +) + +$ErrorActionPreference = "Stop" +$ProgressPreference = "SilentlyContinue" + +function Assert-LastExitCode([string]$Operation) { + if ($LASTEXITCODE -ne 0) { throw "$Operation failed with exit code $LASTEXITCODE" } +} + +function Get-Sha256([string]$Path) { + return (Get-FileHash -LiteralPath $Path -Algorithm SHA256).Hash.ToLowerInvariant() +} + +function Resolve-DDirectory([string]$Path, [string]$Label, [bool]$Create) { + if ($Create -and -not (Test-Path -LiteralPath $Path)) { + $null = New-Item -ItemType Directory -Path $Path + } + $item = Get-Item -LiteralPath (Resolve-Path -LiteralPath $Path).Path -Force + if ( + -not $item.PSIsContainer -or + ($item.Attributes -band [IO.FileAttributes]::ReparsePoint) -or + [IO.Path]::GetPathRoot($item.FullName).TrimEnd("\") -ine "D:" + ) { + throw "$Label must be a real D: directory" + } + return $item.FullName +} + +function Assert-RegularFile([string]$Path, [string]$Label) { + $item = Get-Item -LiteralPath (Resolve-Path -LiteralPath $Path).Path -Force + if ($item.PSIsContainer -or ($item.Attributes -band [IO.FileAttributes]::ReparsePoint)) { + throw "$Label must be a regular file" + } + return $item.FullName +} + +function Convert-ToDockerPath([string]$Path) { return $Path.Replace("\", "/") } + +function Get-Container([string]$Name) { + $rows = @((& docker inspect $Name) | ConvertFrom-Json) + Assert-LastExitCode "Docker inspection for $Name" + if ($rows.Count -ne 1) { throw "Container identity for $Name is not unique" } + return $rows[0] +} + +if ($env:COMPUTERNAME -cne "DESKTOP-OPJ8J04") { + throw "M4.8Q native risk case mining is pinned to DESKTOP-OPJ8J04" +} + +$release = Resolve-DDirectory $ReleaseRoot "M4.8Q release root" $false +$output = Resolve-DDirectory $OutputRoot "M4.8Q output root" $true +$runOutput = Join-Path $output $RunId +if (Test-Path -LiteralPath $runOutput) { throw "M4.8Q output already exists" } + +$runner = Assert-RegularFile ( + Join-Path $release "run_m48q_native_risk_case_mining_worker.py" +) "M4.8Q runner" +$profile = Assert-RegularFile ( + Join-Path $release "m48q-native-risk-case-mining-v1.json" +) "M4.8Q profile" +$runnerSha256 = Get-Sha256 $runner +$profileSha256 = Get-Sha256 $profile +if ($runnerSha256 -cne "39bb1d810b167d6e97dcb091505b52d4a812d69419dcb32fe23ea5d14d605d67") { + throw "M4.8Q runner SHA-256 changed" +} +if ($profileSha256 -cne "c455055e63578d505edc80b66d67e795916a3d0297cc18bfee5a6af7b032ceda") { + throw "M4.8Q profile SHA-256 changed" +} + +$graphRoot = Resolve-DDirectory ( + "D:\NDC_MISSIONCORE\runtime\results\m48n-native-reference-graph-shadow\ravnoves00-full-12fps-v0" +) "M4.8Q graph evidence root" $false +$comparisonRoot = Resolve-DDirectory ( + "D:\NDC_MISSIONCORE\runtime\results\m48n-native-vs-704\ravnoves00-full-v0" +) "M4.8Q comparison evidence root" $false +$graphResult = Assert-RegularFile (Join-Path $graphRoot "result.json") "graph result" +$graphFrames = Assert-RegularFile (Join-Path $graphRoot "frames.jsonl") "graph frames" +$comparisonResult = Assert-RegularFile (Join-Path $comparisonRoot "result.json") "comparison result" +$comparisonFrames = Assert-RegularFile (Join-Path $comparisonRoot "frames.jsonl") "comparison frames" +$video = Assert-RegularFile ( + "D:\NDC_MISSIONCORE\runtime\experiments\e46e\inputs\right-cadd1696ff000904eb78633a0a8418104b8024f178b91f3421789021ccb160e8.mp4" +) "RAVNOVES00 video" + +$graphResultSha256 = Get-Sha256 $graphResult +$graphFramesSha256 = Get-Sha256 $graphFrames +$comparisonResultSha256 = Get-Sha256 $comparisonResult +$comparisonFramesSha256 = Get-Sha256 $comparisonFrames +$videoSha256 = Get-Sha256 $video +if ($graphResultSha256 -cne "c5c3a831b1d1c3271161c91fa5c5c40533eb663ca288ed0220334a147aed6e15") { + throw "M4.8Q graph result SHA-256 changed" +} +if ($graphFramesSha256 -cne "b245af969600670d0975e89cae02b44206e3f1328eff48d5b4639b1b2ab57346") { + throw "M4.8Q graph frame evidence SHA-256 changed" +} +if ($comparisonResultSha256 -cne "fd91c2f1f477038d5af51656d510aba39d69ad2e1a387f15d98ade19dd3a0c49") { + throw "M4.8Q comparison result SHA-256 changed" +} +if ($comparisonFramesSha256 -cne "6a35125d4e003edfd8a33b4239bad0910f137fde3a770e9d5baab721cdc145b2") { + throw "M4.8Q comparison frame evidence SHA-256 changed" +} +if ($videoSha256 -cne "cadd1696ff000904eb78633a0a8418104b8024f178b91f3421789021ccb160e8") { + throw "RAVNOVES00 video SHA-256 changed" +} + +$media = Resolve-DDirectory ( + "D:\NDC_MISSIONCORE\runtime\derived\perception-e15-media-pyav180-lz445-v1" +) "PyAV dependency" $false +$pillow = Resolve-DDirectory ( + "D:\NDC_MISSIONCORE\runtime\derived\perception-p0-env-v1" +) "Pillow dependency" $false +$image = ( + "nvcr.io/nvidia/tritonserver:26.06-py3@" + + "sha256:58df7489c3f2276f9591d500a012dee03e23d35543ce3c390b4c001e6bf90794" +) +& docker image inspect $image *> $null +Assert-LastExitCode "pinned M4.8Q image inspection" + +$canonicalTriton = Get-Container "ndc-mission-core-triton" +if (-not $canonicalTriton.State.Running -or $canonicalTriton.State.Health.Status -cne "healthy") { + throw "Canonical Triton must remain healthy during M4.8Q mining" +} +$canonicalTritonId = [string]$canonicalTriton.Id +$runnerName = "ndc-mission-core-m48q-native-risk-case-mining" +if (& docker ps -a --format "{{.Names}}" --filter "name=^/$runnerName$") { + throw "M4.8Q bounded container already exists" +} + +try { + & docker run ` + --name $runnerName ` + --label "com.nodedc.product=mission-core" ` + --label "com.nodedc.stack=ndc-mission-core-compute" ` + --label "com.nodedc.role=bounded-native-risk-case-mining" ` + --label "com.nodedc.managed-by=codex-bounded-experiment" ` + --network none ` + --read-only ` + --security-opt "no-new-privileges:true" ` + --cap-drop ALL ` + --pids-limit 128 ` + --cpus 4 ` + --memory 4g ` + --tmpfs "/tmp:rw,noexec,nosuid,size=512m" ` + -e "PYTHONDONTWRITEBYTECODE=1" ` + -e "PYTHONPATH=/opt/media:/opt/pillow" ` + -v ((Convert-ToDockerPath $release) + ":/release:ro") ` + -v ((Convert-ToDockerPath $output) + ":/output:rw") ` + -v ((Convert-ToDockerPath $media) + ":/opt/media:ro") ` + -v ((Convert-ToDockerPath $pillow) + ":/opt/pillow:ro") ` + -v ((Convert-ToDockerPath $graphRoot) + ":/evidence/graph:ro") ` + -v ((Convert-ToDockerPath $comparisonRoot) + ":/evidence/comparison:ro") ` + -v ((Convert-ToDockerPath $video) + ":/source/right.mp4:ro") ` + --entrypoint python3 ` + $image ` + /release/run_m48q_native_risk_case_mining_worker.py ` + --profile /release/m48q-native-risk-case-mining-v1.json ` + --graph-result /evidence/graph/result.json ` + --graph-frames /evidence/graph/frames.jsonl ` + --comparison-result /evidence/comparison/result.json ` + --comparison-frames /evidence/comparison/frames.jsonl ` + --video /source/right.mp4 ` + --output-root ("/output/{0}" -f $RunId) ` + --expected-profile-sha256 $profileSha256 ` + --expected-graph-result-sha256 $graphResultSha256 ` + --expected-graph-frames-sha256 $graphFramesSha256 ` + --expected-comparison-result-sha256 $comparisonResultSha256 ` + --expected-comparison-frames-sha256 $comparisonFramesSha256 ` + --expected-video-sha256 $videoSha256 ` + --expected-runner-sha256 $runnerSha256 + Assert-LastExitCode "M4.8Q native risk case mining" + if ( + -not (Test-Path -LiteralPath (Join-Path $runOutput "result.json") -PathType Leaf) -or + -not (Test-Path -LiteralPath (Join-Path $runOutput "cases.jsonl") -PathType Leaf) -or + @(Get-ChildItem -LiteralPath (Join-Path $runOutput "cases") -Filter "*.jpg" -File).Count -ne 24 + ) { + throw "M4.8Q result artifacts are incomplete" + } +} finally { + if (& docker ps -a --format "{{.Names}}" --filter "name=^/$runnerName$") { + & docker rm -f $runnerName *> $null + } + $canonicalAfter = Get-Container "ndc-mission-core-triton" + if ( + [string]$canonicalAfter.Id -cne $canonicalTritonId -or + -not $canonicalAfter.State.Running -or + $canonicalAfter.State.Health.Status -cne "healthy" + ) { + throw "Canonical Triton changed during M4.8Q mining" + } +} + +$result = Get-Content -LiteralPath (Join-Path $runOutput "result.json") -Raw | ConvertFrom-Json +[pscustomobject]@{ + run_id = $RunId + output_root = $runOutput + status = $result.status + case_count = $result.selection.case_count + report_identity_sha256 = $result.report_identity_sha256 + canonical_triton = "healthy-and-unchanged" +} | ConvertTo-Json -Depth 4 diff --git a/tests/test_m48q_native_risk_case_mining.py b/tests/test_m48q_native_risk_case_mining.py new file mode 100644 index 0000000..7e6dc99 --- /dev/null +++ b/tests/test_m48q_native_risk_case_mining.py @@ -0,0 +1,90 @@ +from __future__ import annotations + +import importlib.util +import sys +from pathlib import Path + +import pytest + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +RUNNER_PATH = ( + REPOSITORY_ROOT / "experiments" / "perception" / "run_m48q_native_risk_case_mining_worker.py" +) +SPEC = importlib.util.spec_from_file_location("m48q_case_mining", RUNNER_PATH) +assert SPEC is not None and SPEC.loader is not None +MODULE = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = MODULE +SPEC.loader.exec_module(MODULE) + + +def candidate(sequence: int, *buckets: str): + return MODULE.FrameCandidate( + sequence=sequence, + frame_id=f"frame-{sequence:06d}", + evidence_time_ns=sequence * 1_000_000, + proposals=(), + native_count=0, + legacy_count=0, + matched_count=0, + buckets=frozenset(buckets), + ) + + +def test_selection_is_deterministic_balanced_and_separated() -> None: + candidates = [ + candidate(sequence, "person" if sequence % 2 == 0 else "vehicle") + for sequence in range(0, 400, 5) + ] + quotas = {"person": 3, "vehicle": 3} + + first = MODULE.select_cases( + candidates, + bucket_quotas=quotas, + minimum_sequence_separation=10, + ) + second = MODULE.select_cases( + candidates, + bucket_quotas=quotas, + minimum_sequence_separation=10, + ) + + assert [item.sequence for item in first] == [item.sequence for item in second] + assert len(first) == 6 + assert all( + abs(left.sequence - right.sequence) >= 10 + for index, left in enumerate(first) + for right in first[index + 1 :] + ) + + +def test_selection_refuses_missing_bucket_coverage() -> None: + with pytest.raises(MODULE.M48QCaseMiningError, match="animal produced 0/1"): + MODULE.select_cases( + [candidate(0, "person")], + bucket_quotas={"animal": 1}, + minimum_sequence_separation=1, + ) + + +def test_profile_freezes_raw_raster_and_false_authority() -> None: + import json + + profile = json.loads( + ( + REPOSITORY_ROOT / "config" / "perception" / "m48q-native-risk-case-mining-v1.json" + ).read_text("utf-8") + ) + + assert profile["source"] == { + "source_id": "RAVNOVES00", + "frame_count": 4489, + "raster_width": 800, + "raster_height": 600, + "video_sha256": "cadd1696ff000904eb78633a0a8418104b8024f178b91f3421789021ccb160e8", + "geometric_resampling": False, + "rectification": False, + "warp": False, + } + assert sum(profile["selection"]["bucket_quotas"].values()) == 24 + assert profile["scope"]["quality_evaluated"] is False + assert profile["authority"] == MODULE.FALSE_AUTHORITY