From 9a956c318a56140489c50b106f480ec072c7a54d Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Tue, 25 Aug 2026 23:44:17 +0300 Subject: [PATCH] feat(perception): qualify M4.8T semantic identity --- .../m48t-risk-quality-temporal-v1.json | 75 ++ .../run_m48t_coco_risk_quality_worker.py | 335 +++++ .../run_m48t_temporal_semantic_shadow.py | 226 ++++ .../worker/Invoke-M48TRiskQuality.ps1 | 290 +++++ src/k1link/perception/m48t_risk_quality.py | 1130 +++++++++++++++++ tests/test_m48t_risk_quality.py | 168 +++ tests/test_m48t_temporal_semantic_identity.py | 104 ++ 7 files changed, 2328 insertions(+) create mode 100644 config/perception/m48t-risk-quality-temporal-v1.json create mode 100644 experiments/perception/run_m48t_coco_risk_quality_worker.py create mode 100644 experiments/perception/run_m48t_temporal_semantic_shadow.py create mode 100644 experiments/perception/worker/Invoke-M48TRiskQuality.ps1 create mode 100644 src/k1link/perception/m48t_risk_quality.py create mode 100644 tests/test_m48t_risk_quality.py create mode 100644 tests/test_m48t_temporal_semantic_identity.py diff --git a/config/perception/m48t-risk-quality-temporal-v1.json b/config/perception/m48t-risk-quality-temporal-v1.json new file mode 100644 index 0000000..261b86d --- /dev/null +++ b/config/perception/m48t-risk-quality-temporal-v1.json @@ -0,0 +1,75 @@ +{ + "schema_version": "missioncore.m48t-risk-quality-temporal-profile/v1", + "profile_id": "m48t-coco2017-risk-quality-temporal/v1", + "candidate": { + "provider_id": "triton-rf-detr-large-coco-risk-fp16-shadow/v0", + "model_id": "rf_detr_large:1", + "minimum_score": 0.25, + "source_projection": "stretch-to-800x600-then-rf-detr-704x704" + }, + "dataset": { + "dataset_id": "coco-2017-val", + "images_url": "http://images.cocodataset.org/zips/val2017.zip", + "annotations_url": "http://images.cocodataset.org/annotations/annotations_trainval2017.zip", + "annotation_document": "annotations/instances_val2017.json", + "split": "val2017", + "independent_human_annotations": true, + "include_iscrowd": false, + "minimum_projected_box_area_pixels": 64.0, + "maximum_projected_box_area_fraction": 0.5 + }, + "risk_classes": { + "person": ["person"], + "animal": ["bird", "cat", "dog", "horse", "sheep", "cow", "elephant", "bear", "zebra", "giraffe"], + "light-road-user": ["bicycle", "motorcycle", "skateboard"], + "vehicle": ["car", "bus", "truck"] + }, + "matching": { + "iou_threshold": 0.5, + "method": "score-ordered-greedy-exact-class", + "family_confusion_matching": "score-ordered-greedy-same-family", + "small_area_upper_pixels": 1024.0, + "medium_area_upper_pixels": 9216.0 + }, + "quality_gates": { + "minimum_truth_instances": 1000, + "minimum_micro_precision": 0.8, + "minimum_micro_recall": 0.75, + "minimum_medium_large_recall": 0.85, + "minimum_family_recall": { + "person": 0.8, + "animal": 0.75, + "light-road-user": 0.65, + "vehicle": 0.85 + }, + "minimum_class_truth_instances": 25, + "minimum_qualified_class_recall": 0.55, + "maximum_empty_prediction_risk_image_fraction": 0.1 + }, + "temporal": { + "history_size": 5, + "initial_confirmation_observations": 2, + "switch_confirmation_observations": 3, + "semantic_hold_seconds": 0.3, + "state_expiry_seconds": 1.0, + "maximum_active_components": 512, + "cross_family_conflict_fallback": "unknown", + "association_uses_semantic_class": false, + "occupancy_uses_semantic_class": false + }, + "scope": { + "child_adult_distinction_evaluated": false, + "unknown_moving_detection_evaluated": false, + "object_presence_evaluated": true, + "semantic_family_evaluated": true, + "temporal_stability_evaluated": true, + "risk_policy_evaluated": false + }, + "authority": { + "ground_truth_for_ravnoves00": false, + "candidate_accepted": false, + "commands_enabled": false, + "actuation_allowed": false, + "navigation_or_safety_accepted": false + } +} diff --git a/experiments/perception/run_m48t_coco_risk_quality_worker.py b/experiments/perception/run_m48t_coco_risk_quality_worker.py new file mode 100644 index 0000000..2ce5378 --- /dev/null +++ b/experiments/perception/run_m48t_coco_risk_quality_worker.py @@ -0,0 +1,335 @@ +#!/usr/bin/env python3 +"""Run the frozen RF-DETR risk contour on independent COCO 2017 val truth.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import math +import statistics +import time +from collections import Counter +from pathlib import Path + +import numpy as np +from PIL import Image, ImageDraw + +from k1link.perception.m48t_risk_quality import ( + CocoRiskImage, + RiskPrediction, + RiskTruth, + load_coco_risk_truth, + load_m48t_risk_quality_profile, + score_risk_quality, +) +from k1link.perception.rf_detr_object_detector import ( + TritonRfDetrHttpInferenceBackend, + postprocess_rf_detr, + preprocess_raw_kb4_rf_detr, +) + + +def parse_arguments() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--profile", type=Path, required=True) + parser.add_argument("--annotations", type=Path, required=True) + parser.add_argument("--images-root", type=Path, required=True) + parser.add_argument("--triton-origin", default="http://127.0.0.1:8000") + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--predictions", type=Path, required=True) + parser.add_argument("--failures", type=Path, required=True) + parser.add_argument("--progress", type=Path, required=True) + parser.add_argument("--review-root", type=Path, required=True) + parser.add_argument("--runtime-artifact-sha256", required=True) + parser.add_argument("--runner-sha256", required=True) + parser.add_argument("--images-archive-sha256", required=True) + parser.add_argument("--annotations-document-sha256", required=True) + parser.add_argument("--maximum-images", type=int, default=0) + return parser.parse_args() + + +def main() -> None: + arguments = parse_arguments() + _require_sha256(arguments.runtime_artifact_sha256, "runtime artifact") + _require_sha256(arguments.runner_sha256, "runner") + _require_sha256(arguments.images_archive_sha256, "images archive") + _require_sha256(arguments.annotations_document_sha256, "annotations document") + if arguments.maximum_images < 0: + raise RuntimeError("maximum images cannot be negative") + for target in ( + arguments.output, + arguments.predictions, + arguments.failures, + arguments.progress, + ): + if target.exists(): + raise RuntimeError(f"output already exists: {target}") + target.parent.mkdir(parents=True, exist_ok=True) + if arguments.review_root.exists(): + raise RuntimeError("review output already exists") + arguments.review_root.mkdir(parents=True) + + profile = load_m48t_risk_quality_profile(arguments.profile) + images, truth = load_coco_risk_truth(arguments.annotations, profile) + if arguments.maximum_images: + images = images[: arguments.maximum_images] + image_ids = {item.image_id for item in images} + truth = tuple(item for item in truth if item.image_id in image_ids) + if not images or not truth: + raise RuntimeError("selected COCO risk set is empty") + + mask = np.ones((600, 800), dtype=np.bool_) + class_to_family = profile.class_to_family + predictions: list[RiskPrediction] = [] + image_timings_ms: list[float] = [] + inference_timings_ms: list[float] = [] + rejected: Counter[str] = Counter() + started_at = time.time_ns() + backend = TritonRfDetrHttpInferenceBackend(arguments.triton_origin) + with arguments.progress.open("x", encoding="utf-8") as progress: + try: + for image_index, image in enumerate(images, start=1): + loop_started = time.perf_counter_ns() + image_path = (arguments.images_root / image.file_name).resolve(strict=True) + with Image.open(image_path) as opened: + rgb = np.asarray( + opened.convert("RGB").resize((800, 600), Image.Resampling.BILINEAR), + dtype=np.uint8, + ) + image_bgr = np.ascontiguousarray(rgb[:, :, ::-1]) + tensor = preprocess_raw_kb4_rf_detr(image_bgr, mask) + inference_started = time.perf_counter_ns() + output = backend.infer(tensor) + inference_ms = (time.perf_counter_ns() - inference_started) / 1_000_000 + inference_timings_ms.append(inference_ms) + processed = postprocess_rf_detr(output, mask) + rejected.update(dict(processed.rejected)) + for detection_index, detection in enumerate(processed.detections, start=1): + family = class_to_family.get(detection.label) + if family is None: + raise RuntimeError( + "RF-DETR emitted a class outside the frozen risk profile" + ) + predictions.append( + RiskPrediction( + image_id=image.image_id, + prediction_id=f"{image.image_id:012d}:{detection_index:03d}", + class_name=detection.label, + family=family, + score=detection.score, + bbox_xyxy=detection.bbox_xyxy, + ) + ) + image_ms = (time.perf_counter_ns() - loop_started) / 1_000_000 + image_timings_ms.append(image_ms) + if image_index == 1 or image_index % 100 == 0 or image_index == len(images): + progress.write( + _canonical_json( + { + "schema_version": "missioncore.m48t-risk-quality-progress/v1", + "completed_images": image_index, + "total_images": len(images), + "prediction_count": len(predictions), + "last_image_ms": image_ms, + "last_inference_ms": inference_ms, + } + ) + + "\n" + ) + progress.flush() + finally: + backend.close() + + result = score_risk_quality( + images=images, + truth=truth, + predictions=tuple(predictions), + profile=profile, + ) + completed_at = time.time_ns() + prediction_rows = tuple(_prediction_row(item) for item in predictions) + _write_jsonl(arguments.predictions, prediction_rows) + _write_jsonl(arguments.failures, result.failures) + review_files = _render_review_cases( + images_root=arguments.images_root, + review_root=arguments.review_root, + images=images, + truth=truth, + predictions=tuple(predictions), + failures=result.failures, + ) + report = dict(result.report) + report["execution"] = { + "worker": "DESKTOP-OPJ8J04", + "started_at_unix_ns": started_at, + "completed_at_unix_ns": completed_at, + "duration_seconds": (completed_at - started_at) / 1_000_000_000, + "image_timing_ms": _distribution(image_timings_ms), + "triton_inference_ms": _distribution(inference_timings_ms), + "effective_images_per_second": len(images) + / ((completed_at - started_at) / 1_000_000_000), + "timing_is_admission_evidence": False, + } + report["provenance"] = { + "profile_sha256": profile.profile_sha256, + "annotations_document_sha256": _sha256(arguments.annotations), + "images_archive_sha256": arguments.images_archive_sha256, + "annotations_document_expected_sha256": arguments.annotations_document_sha256, + "runtime_artifact_sha256": arguments.runtime_artifact_sha256, + "runner_sha256": arguments.runner_sha256, + "predictions_sha256": _sha256(arguments.predictions), + "failures_sha256": _sha256(arguments.failures), + } + report["artifacts"] = { + "predictions": arguments.predictions.name, + "failures": arguments.failures.name, + "progress": arguments.progress.name, + "review_files": review_files, + } + report["detector_rejections"] = dict(sorted(rejected.items())) + report["report_identity_sha256"] = hashlib.sha256( + _canonical_json( + { + "profile_sha256": profile.profile_sha256, + "dataset": report["dataset"], + "counts": report["counts"], + "metrics": report["metrics"], + "failure_buckets": report["failure_buckets"], + "quality_gates": report["quality_gates"], + "provenance": report["provenance"], + } + ).encode() + ).hexdigest() + arguments.output.write_text(_canonical_json(report) + "\n", "utf-8") + quality_gates = report.get("quality_gates") + if not isinstance(quality_gates, dict) or not isinstance( + quality_gates.get("passed"), bool + ): + raise RuntimeError("quality gate result is invalid") + print( + _canonical_json( + { + "result": str(arguments.output), + "risk_images": len(images), + "truth_instances": len(truth), + "predictions": len(predictions), + "quality_gate_passed": quality_gates["passed"], + "report_identity_sha256": report["report_identity_sha256"], + } + ), + flush=True, + ) + + +def _prediction_row(item: RiskPrediction) -> dict[str, object]: + return { + "image_id": item.image_id, + "prediction_id": item.prediction_id, + "class_name": item.class_name, + "family": item.family, + "score": item.score, + "bbox_xyxy": list(item.bbox_xyxy), + } + + +def _render_review_cases( + *, + images_root: Path, + review_root: Path, + images: tuple[CocoRiskImage, ...], + truth: tuple[RiskTruth, ...], + predictions: tuple[RiskPrediction, ...], + failures: tuple[dict[str, object], ...], +) -> list[str]: + image_by_id = {item.image_id: item for item in images} + selected_ids: list[int] = [] + for failure in failures: + if failure.get("kind") != "false-negative": + continue + raw_image_id = failure.get("image_id") + if not isinstance(raw_image_id, int) or isinstance(raw_image_id, bool): + raise RuntimeError("failure image id is invalid") + image_id = raw_image_id + if image_id not in selected_ids: + selected_ids.append(image_id) + if len(selected_ids) == 16: + break + result: list[str] = [] + for image_id in selected_ids: + metadata = image_by_id[image_id] + with Image.open((images_root / metadata.file_name).resolve(strict=True)) as opened: + canvas = opened.convert("RGB").resize((800, 600), Image.Resampling.BILINEAR) + draw = ImageDraw.Draw(canvas) + for truth_item in truth: + if truth_item.image_id != image_id: + continue + draw.rectangle(truth_item.bbox_xyxy, outline=(80, 230, 120), width=3) + draw.text( + (truth_item.bbox_xyxy[0] + 2, truth_item.bbox_xyxy[1] + 2), + f"GT {truth_item.class_name}", + fill=(80, 230, 120), + ) + for prediction_item in predictions: + if prediction_item.image_id != image_id: + continue + draw.rectangle(prediction_item.bbox_xyxy, outline=(255, 200, 50), width=2) + draw.text( + (prediction_item.bbox_xyxy[0] + 2, prediction_item.bbox_xyxy[3] - 13), + f"P {prediction_item.class_name} {prediction_item.score:.2f}", + fill=(255, 200, 50), + ) + name = f"review-{image_id:012d}.jpg" + canvas.save(review_root / name, format="JPEG", quality=90, optimize=True) + result.append(f"review/{name}") + return result + + +def _distribution(values: list[float]) -> dict[str, float]: + if not values: + raise RuntimeError("timing distribution is empty") + ordered = sorted(values) + return { + "count": float(len(ordered)), + "mean": statistics.fmean(ordered), + "p50": _percentile(ordered, 0.50), + "p95": _percentile(ordered, 0.95), + "p99": _percentile(ordered, 0.99), + "maximum": ordered[-1], + } + + +def _percentile(values: list[float], quantile: float) -> float: + position = (len(values) - 1) * quantile + lower = math.floor(position) + upper = math.ceil(position) + if lower == upper: + return values[lower] + return values[lower] + (values[upper] - values[lower]) * (position - lower) + + +def _write_jsonl(path: Path, rows: tuple[dict[str, object], ...]) -> None: + with path.open("x", encoding="utf-8") as handle: + for row in rows: + handle.write(_canonical_json(row) + "\n") + + +def _canonical_json(value: object) -> str: + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _require_sha256(value: str, label: str) -> None: + if len(value) != 64 or any(character not in "0123456789abcdef" for character in value): + raise RuntimeError(f"{label} SHA-256 is invalid") + + +if __name__ == "__main__": + main() diff --git a/experiments/perception/run_m48t_temporal_semantic_shadow.py b/experiments/perception/run_m48t_temporal_semantic_shadow.py new file mode 100644 index 0000000..d6e17eb --- /dev/null +++ b/experiments/perception/run_m48t_temporal_semantic_shadow.py @@ -0,0 +1,226 @@ +#!/usr/bin/env python3 +"""Replay M4.8T semantic stabilization over geometry-owned component ids.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +from collections import Counter +from pathlib import Path + +from k1link.perception.m48t_risk_quality import ( + BoundedTemporalSemanticIdentity, + TemporalSemanticObservation, + load_m48t_risk_quality_profile, +) + + +def parse_arguments() -> argparse.Namespace: + repository = Path(__file__).resolve().parents[2] + parser = argparse.ArgumentParser() + parser.add_argument( + "--profile", + type=Path, + default=repository / "config/perception/m48t-risk-quality-temporal-v1.json", + ) + parser.add_argument("--frames", type=Path, required=True) + parser.add_argument("--expected-frames-sha256", required=True) + parser.add_argument("--output", type=Path, required=True) + return parser.parse_args() + + +def main() -> None: + arguments = parse_arguments() + if arguments.output.exists(): + raise RuntimeError("temporal semantic output already exists") + if not _is_sha256(arguments.expected_frames_sha256): + raise RuntimeError("expected frame ledger SHA-256 is invalid") + profile = load_m48t_risk_quality_profile(arguments.profile) + stabilizer = BoundedTemporalSemanticIdentity(profile) + digest = hashlib.sha256() + frames = 0 + publications = 0 + semantic_current = 0 + selected_publications = 0 + raw_switches = 0 + stable_switches = 0 + raw_family_switches = 0 + stable_family_switches = 0 + rolling_retained_skipped = 0 + resolutions: Counter[str] = Counter() + stable_families: Counter[str] = Counter() + last_raw: dict[str, str] = {} + last_stable: dict[str, str] = {} + class_to_family = profile.class_to_family + + with arguments.frames.expanduser().resolve(strict=True).open("rb") as handle: + for line_number, raw_line in enumerate(handle, start=1): + digest.update(raw_line) + try: + document = json.loads(raw_line) + frame_time_ns = document["source_envelope"]["timestamps"]["source_ns"] + occupied = document["delivery"]["obstacle_map"]["occupied"] + except (KeyError, TypeError, json.JSONDecodeError) as exc: + raise RuntimeError(f"frame ledger row {line_number} is invalid") from exc + if not isinstance(frame_time_ns, int) or not isinstance(occupied, list): + raise RuntimeError(f"frame ledger row {line_number} contract changed") + frames += 1 + for value in occupied: + if not isinstance(value, dict): + raise RuntimeError("temporal obstacle is invalid") + component_id = value.get("component_id") + state = value.get("state") + if state == "retained": + rolling_retained_skipped += 1 + continue + raw_class = value.get("semantic_hint") if state == "current" else None + if not isinstance(component_id, str) or state not in { + "current", + "held", + "expired", + }: + raise RuntimeError("temporal obstacle identity or state changed") + if raw_class is not None and not isinstance(raw_class, str): + raise RuntimeError("temporal semantic hint is invalid") + if raw_class is not None: + semantic_current += 1 + previous_raw = last_raw.get(component_id) + if previous_raw is not None and previous_raw != raw_class: + raw_switches += 1 + if class_to_family[previous_raw] != class_to_family[raw_class]: + raw_family_switches += 1 + last_raw[component_id] = raw_class + result = stabilizer.update( + TemporalSemanticObservation( + component_id=component_id, + evidence_time_ns=frame_time_ns, + raw_class_name=raw_class, + currentness=state, + ) + ) + publications += 1 + resolutions[result.resolution] += 1 + if result.selected_class_name is not None: + selected_publications += 1 + if result.selected_family is None: + raise RuntimeError("selected semantic family is missing") + stable_families[result.selected_family] += 1 + previous_stable = last_stable.get(component_id) + if ( + previous_stable is not None + and previous_stable != result.selected_class_name + ): + stable_switches += 1 + if ( + class_to_family[previous_stable] + != class_to_family[result.selected_class_name] + ): + stable_family_switches += 1 + last_stable[component_id] = result.selected_class_name + + frames_sha256 = digest.hexdigest() + if frames_sha256 != arguments.expected_frames_sha256: + raise RuntimeError("temporal frame ledger SHA-256 changed") + snapshot = stabilizer.snapshot() + gate_checks = { + "all-publications-accounted": snapshot.input_observations == publications, + "bounded-active-components": ( + snapshot.peak_active_components <= profile.temporal.maximum_active_components + ), + "stable-switches-not-greater-than-raw-switches": stable_switches <= raw_switches, + "stable-family-switches-not-greater-than-raw-family-switches": ( + stable_family_switches <= raw_family_switches + ), + "association-remains-class-independent": True, + "occupancy-remains-class-independent": True, + } + report = { + "schema_version": "missioncore.m48t-temporal-semantic-shadow/v1", + "profile_id": profile.profile_id, + "profile_sha256": profile.profile_sha256, + "source": { + "source_id": "RAVNOVES00", + "frame_ledger": str(arguments.frames), + "frame_ledger_sha256": frames_sha256, + "frames": frames, + "publications": publications, + "independent_semantic_truth_available": False, + }, + "metrics": { + "semantic_current_observations": semantic_current, + "rolling_retained_publications_skipped": rolling_retained_skipped, + "selected_publications": selected_publications, + "raw_class_switches": raw_switches, + "stable_class_switches": stable_switches, + "suppressed_or_deferred_class_switches": raw_switches - stable_switches, + "raw_family_switches": raw_family_switches, + "stable_family_switches": stable_family_switches, + "suppressed_or_deferred_family_switches": ( + raw_family_switches - stable_family_switches + ), + "resolutions": dict(sorted(resolutions.items())), + "stable_family_publications": dict(sorted(stable_families.items())), + "snapshot": { + field: getattr(snapshot, field) + for field in snapshot.__dataclass_fields__ + }, + }, + "temporal_invariant_gate": { + "checks": gate_checks, + "passed": all(gate_checks.values()), + "semantic_quality_accepted": False, + }, + "policy": { + "initial_confirmation_observations": ( + profile.temporal.initial_confirmation_observations + ), + "switch_confirmation_observations": ( + profile.temporal.switch_confirmation_observations + ), + "semantic_hold_seconds": profile.temporal.semantic_hold_seconds, + "state_expiry_seconds": profile.temporal.state_expiry_seconds, + "history_size": profile.temporal.history_size, + "maximum_active_components": profile.temporal.maximum_active_components, + "cross_family_conflict_fallback": "unknown", + "association_uses_semantic_class": False, + "occupancy_uses_semantic_class": False, + }, + "authority": { + "ground_truth_for_ravnoves00": False, + "candidate_accepted": False, + "commands_enabled": False, + "actuation_allowed": False, + "navigation_or_safety_accepted": False, + }, + } + arguments.output.parent.mkdir(parents=True, exist_ok=True) + arguments.output.write_text( + json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n", + "utf-8", + ) + print( + json.dumps( + { + "output": str(arguments.output), + "frames": frames, + "publications": publications, + "raw_class_switches": raw_switches, + "stable_class_switches": stable_switches, + "invariant_gate_passed": all(gate_checks.values()), + }, + sort_keys=True, + ) + ) + + +def _is_sha256(value: object) -> bool: + return ( + isinstance(value, str) + and len(value) == 64 + and all(character in "0123456789abcdef" for character in value) + ) + + +if __name__ == "__main__": + main() diff --git a/experiments/perception/worker/Invoke-M48TRiskQuality.ps1 b/experiments/perception/worker/Invoke-M48TRiskQuality.ps1 new file mode 100644 index 0000000..e336766 --- /dev/null +++ b/experiments/perception/worker/Invoke-M48TRiskQuality.ps1 @@ -0,0 +1,290 @@ +[CmdletBinding()] +param( + [Parameter(Mandatory = $true)] + [string]$ReleaseRoot, + [Parameter(Mandatory = $true)] + [ValidatePattern("^[a-f0-9]{64}$")] + [string]$ExpectedWheelSha256, + [Parameter(Mandatory = $true)] + [ValidatePattern("^[A-Za-z0-9._-]{1,96}$")] + [string]$RunId, + [ValidateRange(0, 5000)] + [int]$MaximumImages = 0, + [string]$DatasetRoot = "D:\NDC_MISSIONCORE\datasets\coco-2017-val", + [string]$OutputRoot = "D:\NDC_MISSIONCORE\runtime\results\m48t-risk-quality" +) + +$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 Ensure-PinnedMirrorFile( + [string]$Path, + [string]$Url, + [long]$ExpectedLength, + [string]$ExpectedSha256, + [string]$Label +) { + if (Test-Path -LiteralPath $Path -PathType Leaf) { + if ( + (Get-Item -LiteralPath $Path).Length -ne $ExpectedLength -or + (Get-Sha256 $Path) -cne $ExpectedSha256 + ) { + Remove-Item -LiteralPath $Path -Force + } + } + if (-not (Test-Path -LiteralPath $Path -PathType Leaf)) { + $partial = $Path + ".partial" + if (Test-Path -LiteralPath $partial) { Remove-Item -LiteralPath $partial -Force } + & curl.exe --fail --location --retry 5 ` + --speed-limit 1048576 --speed-time 30 --output $partial $Url + Assert-LastExitCode "$Label download" + if ( + (Get-Item -LiteralPath $partial).Length -ne $ExpectedLength -or + (Get-Sha256 $partial) -cne $ExpectedSha256 + ) { + throw "$Label length or SHA-256 changed" + } + Move-Item -LiteralPath $partial -Destination $Path + } + if ( + (Get-Item -LiteralPath $Path).Length -ne $ExpectedLength -or + (Get-Sha256 $Path) -cne $ExpectedSha256 + ) { + throw "$Label length or SHA-256 changed" + } +} + +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 "M48T risk quality is pinned to DESKTOP-OPJ8J04" +} + +$release = Resolve-DDirectory $ReleaseRoot "M48T release root" $false +$dataset = Resolve-DDirectory $DatasetRoot "M48T dataset root" $true +$output = Resolve-DDirectory $OutputRoot "M48T output root" $true +$runOutput = Join-Path $output $RunId +if (Test-Path -LiteralPath $runOutput) { throw "M48T run output already exists" } +$null = New-Item -ItemType Directory -Path $runOutput +$runOutput = Resolve-DDirectory $runOutput "M48T run output" $false + +$wheel = Assert-RegularFile ( + (Join-Path $release "nodedc_mission_core-0.1.0-py3-none-any.whl") +) "M48T wheel" +if ((Get-Sha256 $wheel) -cne $ExpectedWheelSha256) { throw "M48T wheel SHA-256 changed" } +$profile = Assert-RegularFile ( + (Join-Path $release "m48t-risk-quality-temporal-v1.json") +) "M48T profile" +$runner = Assert-RegularFile ( + (Join-Path $release "run_m48t_coco_risk_quality_worker.py") +) "M48T runner" +$runnerSha256 = Get-Sha256 $runner + +$imagesArchive = Join-Path $dataset "val2017.zip" +$annotations = Join-Path $dataset "instances_val2017.json" +# These mirrors rehost the unmodified official COCO assets and publish pinned SHA-256 digests. +Ensure-PinnedMirrorFile $imagesArchive ` + "https://huggingface.co/datasets/pcuenq/coco-2017-mirror/resolve/main/val2017.zip?download=true" ` + 815585330 ` + "4f7e2ccb2866ec5041993c9cf2a952bbed69647b115d0f74da7ce8f4bef82f05" ` + "COCO val2017" +Ensure-PinnedMirrorFile $annotations ` + "https://huggingface.co/datasets/LibreYOLO/coco2017/resolve/main/instances_val2017.json?download=true" ` + 19987840 ` + "e8c7f7908f1d7278341fae127d0da654f102f11bd7b21d8aeefa635b8c810b6f" ` + "COCO instances_val2017" +$imagesArchive = Assert-RegularFile $imagesArchive "COCO images archive" +$annotations = Assert-RegularFile $annotations "COCO val2017 instances" +$imagesArchiveSha256 = Get-Sha256 $imagesArchive +$annotationsDocumentSha256 = Get-Sha256 $annotations + +$imagesRoot = Join-Path $dataset "val2017" +if (-not (Test-Path -LiteralPath $imagesRoot -PathType Container)) { + & tar.exe -xf $imagesArchive -C $dataset + Assert-LastExitCode "COCO val2017 extraction" +} +$imagesRoot = Resolve-DDirectory $imagesRoot "COCO val2017 images" $false +if (@(Get-ChildItem -LiteralPath $imagesRoot -File -Filter "*.jpg").Count -ne 5000) { + throw "COCO val2017 image count changed" +} + +$experimentRoot = Resolve-DDirectory ( + "D:\NDC_MISSIONCORE\runtime\experiments\m48t-fixed-detector-20260825T095425Z" +) "M48T RF-DETR experiment root" $false +$modelRoot = Resolve-DDirectory ( + (Join-Path $experimentRoot "triton-models") +) "M48T RF-DETR model root" $false +if ((Get-Sha256 (Assert-RegularFile ( + (Join-Path $modelRoot "rf_detr_large\1\model.plan") +) "RF-DETR TensorRT engine")) -cne "986399ce706b7380472cf5e473232249fed6e628971d8007f6609e83128d46b8") { + throw "RF-DETR TensorRT engine SHA-256 changed" +} + +$image = "nvcr.io/nvidia/tritonserver:26.06-py3@sha256:58df7489c3f2276f9591d500a012dee03e23d35543ce3c390b4c001e6bf90794" +& docker image inspect $image *> $null +Assert-LastExitCode "Pinned M48T image inspection" +$historicalTriton = Get-Container "ndc-mission-core-triton" +if (-not $historicalTriton.State.Running -or $historicalTriton.State.Health.Status -cne "healthy") { + throw "Historical Triton must remain healthy during M48T quality evaluation" +} +$historicalTritonId = [string]$historicalTriton.Id +$tritonName = "ndc-mission-core-m48t-risk-quality-triton" +$qualityName = "ndc-mission-core-m48t-risk-quality" +foreach ($name in @($tritonName, $qualityName)) { + if (& docker ps -a --format "{{.Names}}" --filter "name=^/$name$") { + throw "M48T candidate container $name already exists" + } +} + +$opencv = Resolve-DDirectory ( + "D:\NDC_MISSIONCORE\runtime\derived\perception-e3-opencv413092-v1\packages" +) "OpenCV dependency" $false +$pillow = Resolve-DDirectory ( + "D:\NDC_MISSIONCORE\runtime\derived\perception-p0-env-v1" +) "Pillow dependency" $false + +try { + & docker create ` + --name $tritonName ` + --read-only ` + --security-opt "no-new-privileges:true" ` + --cap-drop ALL ` + --pids-limit 512 ` + --shm-size 1g ` + --gpus all ` + --tmpfs "/tmp:rw,noexec,nosuid,size=2g" ` + --health-cmd "curl --fail --silent http://127.0.0.1:8000/v2/health/ready" ` + --health-interval 5s ` + --health-timeout 3s ` + --health-start-period 20s ` + --health-retries 24 ` + -v ((Convert-ToDockerPath $modelRoot) + ":/models:ro") ` + $image ` + tritonserver ` + --model-repository=/models ` + --model-control-mode=explicit ` + --load-model=rf_detr_large ` + --disable-auto-complete-config ` + --strict-readiness=true ` + --exit-on-error=true ` + --allow-http=true ` + --allow-grpc=false ` + --allow-metrics=false *> $null + Assert-LastExitCode "M48T Triton creation" + & docker start $tritonName *> $null + Assert-LastExitCode "M48T Triton start" + $ready = $false + foreach ($attempt in 1..60) { + Start-Sleep -Seconds 2 + $candidate = Get-Container $tritonName + if (-not $candidate.State.Running) { throw "M48T Triton stopped during startup" } + if ($candidate.State.Health.Status -ceq "healthy") { $ready = $true; break } + } + if (-not $ready) { throw "M48T Triton did not become healthy" } + + $maximumArguments = @() + if ($MaximumImages -gt 0) { + $maximumArguments = @("--maximum-images", ([string]$MaximumImages)) + } + $arguments = @( + "run", "--name", $qualityName, + "--network", ("container:{0}" -f $tritonName), + "--read-only", + "--security-opt", "no-new-privileges:true", + "--cap-drop", "ALL", + "--pids-limit", "256", + "--gpus", "all", + "--tmpfs", "/tmp:rw,noexec,nosuid,size=2g", + "-e", "PYTHONDONTWRITEBYTECODE=1", + "-e", "PYTHONPATH=/release/nodedc_mission_core-0.1.0-py3-none-any.whl:/opt/opencv:/opt/pillow", + "-v", ((Convert-ToDockerPath $release) + ":/release:ro"), + "-v", ((Convert-ToDockerPath $imagesRoot) + ":/dataset/val2017:ro"), + "-v", ((Convert-ToDockerPath $annotations) + ":/dataset/instances_val2017.json:ro"), + "-v", ((Convert-ToDockerPath $runOutput) + ":/output:rw"), + "-v", ((Convert-ToDockerPath $opencv) + ":/opt/opencv:ro"), + "-v", ((Convert-ToDockerPath $pillow) + ":/opt/pillow:ro"), + "--entrypoint", "python3", + $image, + "/release/run_m48t_coco_risk_quality_worker.py", + "--profile", "/release/m48t-risk-quality-temporal-v1.json", + "--annotations", "/dataset/instances_val2017.json", + "--images-root", "/dataset/val2017", + "--triton-origin", "http://127.0.0.1:8000", + "--output", "/output/result.json", + "--predictions", "/output/predictions.jsonl", + "--failures", "/output/failures.jsonl", + "--progress", "/output/progress.jsonl", + "--review-root", "/output/review", + "--runtime-artifact-sha256", $ExpectedWheelSha256, + "--runner-sha256", $runnerSha256, + "--images-archive-sha256", $imagesArchiveSha256, + "--annotations-document-sha256", $annotationsDocumentSha256 + ) + $maximumArguments + & docker @arguments + Assert-LastExitCode "M48T COCO risk quality evaluation" + if (-not (Test-Path -LiteralPath (Join-Path $runOutput "result.json") -PathType Leaf)) { + throw "M48T result was not written" + } +} finally { + foreach ($name in @($qualityName, $tritonName)) { + if (& docker ps -a --format "{{.Names}}" --filter "name=^/$name$") { + & docker rm -f $name *> $null + } + } + $historicalAfter = Get-Container "ndc-mission-core-triton" + if ( + $historicalAfter.Id -cne $historicalTritonId -or + -not $historicalAfter.State.Running -or + $historicalAfter.State.Health.Status -cne "healthy" + ) { + throw "Historical Triton changed during M48T quality evaluation" + } +} + +Write-Output ("M48T_RESULT={0}" -f (Join-Path $runOutput "result.json")) +Write-Output ("COCO_IMAGES_SHA256={0}" -f $imagesArchiveSha256) +Write-Output ("COCO_ANNOTATIONS_SHA256={0}" -f $annotationsDocumentSha256) +Write-Output "HISTORICAL_TRITON_ACTION=none" +Write-Output "PRODUCTION_ACCEPTED=false" diff --git a/src/k1link/perception/m48t_risk_quality.py b/src/k1link/perception/m48t_risk_quality.py new file mode 100644 index 0000000..3e6a206 --- /dev/null +++ b/src/k1link/perception/m48t_risk_quality.py @@ -0,0 +1,1130 @@ +"""Independent COCO risk-quality scoring and class-independent semantic identity. + +The evaluation contour scores the frozen RF-DETR candidate against public human +annotations. The temporal contour consumes an already assigned geometry-owned +``component_id`` and therefore cannot influence spatial association or occupancy. +""" + +from __future__ import annotations + +import hashlib +import json +import math +from collections import Counter, deque +from collections.abc import Mapping +from dataclasses import dataclass, field +from pathlib import Path +from typing import Final, TypeGuard + +M48T_PROFILE_SCHEMA: Final = "missioncore.m48t-risk-quality-temporal-profile/v1" +M48T_REPORT_SCHEMA: Final = "missioncore.m48t-risk-quality-report/v1" +DEFAULT_M48T_PROFILE_PATH: Final = Path( + "config/perception/m48t-risk-quality-temporal-v1.json" +) +_SOURCE_WIDTH: Final = 800 +_SOURCE_HEIGHT: Final = 600 +_SOURCE_AREA: Final = float(_SOURCE_WIDTH * _SOURCE_HEIGHT) +_FALSE_AUTHORITY: Final = { + "ground_truth_for_ravnoves00": False, + "candidate_accepted": False, + "commands_enabled": False, + "actuation_allowed": False, + "navigation_or_safety_accepted": False, +} + + +class M48TRiskQualityError(ValueError): + """The M4.8T profile, evidence, or evaluation input is incompatible.""" + + +@dataclass(frozen=True, slots=True) +class CocoDatasetProfile: + dataset_id: str + images_url: str + annotations_url: str + annotation_document: str + minimum_projected_box_area_pixels: float + maximum_projected_box_area_fraction: float + + +@dataclass(frozen=True, slots=True) +class RiskMatchingProfile: + iou_threshold: float + small_area_upper_pixels: float + medium_area_upper_pixels: float + + +@dataclass(frozen=True, slots=True) +class RiskQualityGates: + minimum_truth_instances: int + minimum_micro_precision: float + minimum_micro_recall: float + minimum_medium_large_recall: float + minimum_family_recall: tuple[tuple[str, float], ...] + minimum_class_truth_instances: int + minimum_qualified_class_recall: float + maximum_empty_prediction_risk_image_fraction: float + + +@dataclass(frozen=True, slots=True) +class TemporalSemanticProfile: + history_size: int + initial_confirmation_observations: int + switch_confirmation_observations: int + semantic_hold_seconds: float + state_expiry_seconds: float + maximum_active_components: int + + @property + def semantic_hold_ns(self) -> int: + return round(self.semantic_hold_seconds * 1_000_000_000) + + @property + def state_expiry_ns(self) -> int: + return round(self.state_expiry_seconds * 1_000_000_000) + + +@dataclass(frozen=True, slots=True) +class M48TRiskQualityProfile: + profile_id: str + provider_id: str + model_id: str + minimum_score: float + dataset: CocoDatasetProfile + risk_families: tuple[tuple[str, tuple[str, ...]], ...] + matching: RiskMatchingProfile + gates: RiskQualityGates + temporal: TemporalSemanticProfile + profile_sha256: str + + @property + def class_to_family(self) -> dict[str, str]: + return { + class_name: family + for family, classes in self.risk_families + for class_name in classes + } + + +@dataclass(frozen=True, slots=True) +class CocoRiskImage: + image_id: int + file_name: str + width: int + height: int + + +@dataclass(frozen=True, slots=True) +class RiskTruth: + image_id: int + annotation_id: int + class_name: str + family: str + bbox_xyxy: tuple[float, float, float, float] + projected_area_pixels: float + size_band: str + + +@dataclass(frozen=True, slots=True) +class RiskPrediction: + image_id: int + prediction_id: str + class_name: str + family: str + score: float + bbox_xyxy: tuple[float, float, float, float] + + def __post_init__(self) -> None: + if not self.prediction_id: + raise M48TRiskQualityError("risk prediction id is empty") + if not math.isfinite(self.score) or not 0.0 <= self.score <= 1.0: + raise M48TRiskQualityError("risk prediction score is invalid") + _validate_bbox(self.bbox_xyxy, "risk prediction") + + +@dataclass(frozen=True, slots=True) +class RiskQualityResult: + report: dict[str, object] + failures: tuple[dict[str, object], ...] + + +@dataclass(frozen=True, slots=True) +class TemporalSemanticObservation: + component_id: str + evidence_time_ns: int + raw_class_name: str | None + currentness: str = "current" + + def __post_init__(self) -> None: + if not self.component_id: + raise M48TRiskQualityError("temporal semantic component id is empty") + if self.evidence_time_ns < 0: + raise M48TRiskQualityError("temporal semantic evidence time is invalid") + if self.raw_class_name is not None and not self.raw_class_name: + raise M48TRiskQualityError("temporal semantic raw class is empty") + if self.currentness not in {"current", "held", "expired"}: + raise M48TRiskQualityError("temporal semantic currentness is invalid") + + +@dataclass(frozen=True, slots=True) +class StableSemanticIdentity: + component_id: str + evidence_time_ns: int + raw_class_name: str | None + selected_class_name: str | None + selected_family: str | None + resolution: str + confirmation_count: int + reason_codes: tuple[str, ...] + association_uses_semantic_class: bool = False + occupancy_uses_semantic_class: bool = False + + +@dataclass(frozen=True, slots=True) +class TemporalSemanticSnapshot: + input_observations: int + confirmed_publications: int + pending_publications: int + conflict_publications: int + held_publications: int + unknown_publications: int + expired_publications: int + class_switches: int + evicted_components: int + active_components: int + peak_active_components: int + + +@dataclass(slots=True) +class _SemanticComponent: + last_update_ns: int + last_raw_ns: int | None = None + stable_class: str | None = None + history: deque[str] = field(default_factory=deque) + + +class BoundedTemporalSemanticIdentity: + """Stabilize semantics without participating in geometry-owned identity.""" + + def __init__(self, profile: M48TRiskQualityProfile) -> None: + self.profile = profile + self.config = profile.temporal + self._class_to_family = profile.class_to_family + self._components: dict[str, _SemanticComponent] = {} + self._inputs = 0 + self._confirmed = 0 + self._pending = 0 + self._conflict = 0 + self._held = 0 + self._unknown = 0 + self._expired = 0 + self._switches = 0 + self._evicted = 0 + self._peak = 0 + + def update(self, observation: TemporalSemanticObservation) -> StableSemanticIdentity: + self._inputs += 1 + self._expire_stale(observation.evidence_time_ns) + if observation.currentness == "expired": + self._components.pop(observation.component_id, None) + self._expired += 1 + return self._result(observation, None, "expired", 0, ("component-expired",)) + + component = self._components.get(observation.component_id) + if component is None: + self._make_room() + component = _SemanticComponent(last_update_ns=observation.evidence_time_ns) + component.history = deque(maxlen=self.config.history_size) + self._components[observation.component_id] = component + self._peak = max(self._peak, len(self._components)) + if observation.evidence_time_ns < component.last_update_ns: + raise M48TRiskQualityError("temporal semantic evidence moved backwards") + component.last_update_ns = observation.evidence_time_ns + + raw = observation.raw_class_name + if raw == "unknown": + raw = None + if raw is not None and raw not in self._class_to_family: + raise M48TRiskQualityError("temporal semantic class is outside the risk profile") + if observation.currentness == "held" or raw is None: + return self._publish_without_current_class(observation, component) + + component.last_raw_ns = observation.evidence_time_ns + component.history.append(raw) + support = _trailing_support(component.history, raw) + if component.stable_class is None: + if support >= self.config.initial_confirmation_observations: + component.stable_class = raw + self._confirmed += 1 + return self._result( + observation, + raw, + "confirmed", + support, + ("initial-class-confirmed", "geometry-identity-unchanged"), + ) + self._pending += 1 + return self._result( + observation, + None, + "pending", + support, + ("initial-class-awaiting-confirmation",), + ) + + if raw == component.stable_class: + self._confirmed += 1 + return self._result( + observation, + component.stable_class, + "confirmed", + support, + ("stable-class-reconfirmed",), + ) + + if support >= self.config.switch_confirmation_observations: + component.stable_class = raw + self._switches += 1 + self._confirmed += 1 + return self._result( + observation, + raw, + "confirmed", + support, + ("class-switch-confirmed", "geometry-identity-unchanged"), + ) + + old_family = self._class_to_family[component.stable_class] + new_family = self._class_to_family[raw] + if old_family != new_family: + self._conflict += 1 + return self._result( + observation, + None, + "conflict", + support, + ("cross-family-conflict", "fallback-unknown"), + ) + self._pending += 1 + return self._result( + observation, + component.stable_class, + "pending", + support, + ("same-family-switch-awaiting-confirmation",), + ) + + def snapshot(self) -> TemporalSemanticSnapshot: + return TemporalSemanticSnapshot( + input_observations=self._inputs, + confirmed_publications=self._confirmed, + pending_publications=self._pending, + conflict_publications=self._conflict, + held_publications=self._held, + unknown_publications=self._unknown, + expired_publications=self._expired, + class_switches=self._switches, + evicted_components=self._evicted, + active_components=len(self._components), + peak_active_components=self._peak, + ) + + def _publish_without_current_class( + self, + observation: TemporalSemanticObservation, + component: _SemanticComponent, + ) -> StableSemanticIdentity: + if ( + component.stable_class is not None + and component.last_raw_ns is not None + and observation.evidence_time_ns - component.last_raw_ns + <= self.config.semantic_hold_ns + ): + self._held += 1 + return self._result( + observation, + component.stable_class, + "held", + 0, + ("semantic-held-within-bounded-window",), + ) + self._unknown += 1 + return self._result( + observation, + None, + "unknown", + 0, + ("current-semantic-evidence-unavailable",), + ) + + def _result( + self, + observation: TemporalSemanticObservation, + selected: str | None, + resolution: str, + support: int, + reasons: tuple[str, ...], + ) -> StableSemanticIdentity: + return StableSemanticIdentity( + component_id=observation.component_id, + evidence_time_ns=observation.evidence_time_ns, + raw_class_name=observation.raw_class_name, + selected_class_name=selected, + selected_family=None if selected is None else self._class_to_family[selected], + resolution=resolution, + confirmation_count=support, + reason_codes=reasons, + ) + + def _expire_stale(self, now_ns: int) -> None: + stale = sorted( + component_id + for component_id, component in self._components.items() + if now_ns - component.last_update_ns > self.config.state_expiry_ns + ) + for component_id in stale: + del self._components[component_id] + + def _make_room(self) -> None: + if len(self._components) < self.config.maximum_active_components: + return + component_id = min( + self._components, + key=lambda key: (self._components[key].last_update_ns, key), + ) + del self._components[component_id] + self._evicted += 1 + + +def load_m48t_risk_quality_profile(path: Path) -> M48TRiskQualityProfile: + """Load the frozen M4.8T candidate, gates, and temporal policy strictly.""" + + try: + raw = path.expanduser().resolve(strict=True).read_bytes() + root = _object(json.loads(raw), "M4.8T profile") + except (OSError, json.JSONDecodeError) as exc: + raise M48TRiskQualityError("M4.8T profile cannot be read") from exc + _exact_keys( + root, + { + "schema_version", + "profile_id", + "candidate", + "dataset", + "risk_classes", + "matching", + "quality_gates", + "temporal", + "scope", + "authority", + }, + "M4.8T profile", + ) + if root.get("schema_version") != M48T_PROFILE_SCHEMA: + raise M48TRiskQualityError("M4.8T profile schema is incompatible") + candidate = _object(root.get("candidate"), "M4.8T candidate") + _exact_keys( + candidate, + {"provider_id", "model_id", "minimum_score", "source_projection"}, + "M4.8T candidate", + ) + if candidate != { + "provider_id": "triton-rf-detr-large-coco-risk-fp16-shadow/v0", + "model_id": "rf_detr_large:1", + "minimum_score": 0.25, + "source_projection": "stretch-to-800x600-then-rf-detr-704x704", + }: + raise M48TRiskQualityError("M4.8T candidate was tuned in place") + + dataset = _object(root.get("dataset"), "M4.8T dataset") + _exact_keys( + dataset, + { + "dataset_id", + "images_url", + "annotations_url", + "annotation_document", + "split", + "independent_human_annotations", + "include_iscrowd", + "minimum_projected_box_area_pixels", + "maximum_projected_box_area_fraction", + }, + "M4.8T dataset", + ) + if ( + dataset.get("dataset_id") != "coco-2017-val" + or dataset.get("split") != "val2017" + or dataset.get("independent_human_annotations") is not True + or dataset.get("include_iscrowd") is not False + ): + raise M48TRiskQualityError("M4.8T independent dataset contract changed") + + risk_classes = _object(root.get("risk_classes"), "M4.8T risk classes") + expected_families = ("person", "animal", "light-road-user", "vehicle") + if tuple(risk_classes) != expected_families: + raise M48TRiskQualityError("M4.8T risk family order changed") + families: list[tuple[str, tuple[str, ...]]] = [] + declared: set[str] = set() + for family in expected_families: + classes = _string_tuple(risk_classes.get(family), f"M4.8T {family} classes") + if not classes or declared.intersection(classes): + raise M48TRiskQualityError("M4.8T risk classes are empty or duplicated") + declared.update(classes) + families.append((family, classes)) + + matching = _object(root.get("matching"), "M4.8T matching") + _exact_keys( + matching, + { + "iou_threshold", + "method", + "family_confusion_matching", + "small_area_upper_pixels", + "medium_area_upper_pixels", + }, + "M4.8T matching", + ) + if ( + matching.get("method") != "score-ordered-greedy-exact-class" + or matching.get("family_confusion_matching") + != "score-ordered-greedy-same-family" + ): + raise M48TRiskQualityError("M4.8T matching contract changed") + + gates = _object(root.get("quality_gates"), "M4.8T quality gates") + _exact_keys( + gates, + { + "minimum_truth_instances", + "minimum_micro_precision", + "minimum_micro_recall", + "minimum_medium_large_recall", + "minimum_family_recall", + "minimum_class_truth_instances", + "minimum_qualified_class_recall", + "maximum_empty_prediction_risk_image_fraction", + }, + "M4.8T quality gates", + ) + family_recall = _object(gates.get("minimum_family_recall"), "family recall gates") + if tuple(family_recall) != expected_families: + raise M48TRiskQualityError("M4.8T family recall gates changed") + + temporal = _object(root.get("temporal"), "M4.8T temporal profile") + _exact_keys( + temporal, + { + "history_size", + "initial_confirmation_observations", + "switch_confirmation_observations", + "semantic_hold_seconds", + "state_expiry_seconds", + "maximum_active_components", + "cross_family_conflict_fallback", + "association_uses_semantic_class", + "occupancy_uses_semantic_class", + }, + "M4.8T temporal profile", + ) + if ( + temporal.get("cross_family_conflict_fallback") != "unknown" + or temporal.get("association_uses_semantic_class") is not False + or temporal.get("occupancy_uses_semantic_class") is not False + ): + raise M48TRiskQualityError("M4.8T class-independent temporal contract changed") + scope = _object(root.get("scope"), "M4.8T scope") + if scope != { + "child_adult_distinction_evaluated": False, + "unknown_moving_detection_evaluated": False, + "object_presence_evaluated": True, + "semantic_family_evaluated": True, + "temporal_stability_evaluated": True, + "risk_policy_evaluated": False, + }: + raise M48TRiskQualityError("M4.8T scope changed") + if _object(root.get("authority"), "M4.8T authority") != _FALSE_AUTHORITY: + raise M48TRiskQualityError("M4.8T authority changed") + + loaded = M48TRiskQualityProfile( + profile_id=_string(root, "profile_id"), + provider_id=_string(candidate, "provider_id"), + model_id=_string(candidate, "model_id"), + minimum_score=_fraction(candidate, "minimum_score"), + dataset=CocoDatasetProfile( + dataset_id=_string(dataset, "dataset_id"), + images_url=_string(dataset, "images_url"), + annotations_url=_string(dataset, "annotations_url"), + annotation_document=_string(dataset, "annotation_document"), + minimum_projected_box_area_pixels=_positive_number( + dataset, "minimum_projected_box_area_pixels" + ), + maximum_projected_box_area_fraction=_fraction( + dataset, "maximum_projected_box_area_fraction" + ), + ), + risk_families=tuple(families), + matching=RiskMatchingProfile( + iou_threshold=_fraction(matching, "iou_threshold"), + small_area_upper_pixels=_positive_number(matching, "small_area_upper_pixels"), + medium_area_upper_pixels=_positive_number( + matching, "medium_area_upper_pixels" + ), + ), + gates=RiskQualityGates( + minimum_truth_instances=_positive_integer(gates, "minimum_truth_instances"), + minimum_micro_precision=_fraction(gates, "minimum_micro_precision"), + minimum_micro_recall=_fraction(gates, "minimum_micro_recall"), + minimum_medium_large_recall=_fraction( + gates, "minimum_medium_large_recall" + ), + minimum_family_recall=tuple( + (family, _fraction(family_recall, family)) for family in expected_families + ), + minimum_class_truth_instances=_positive_integer( + gates, "minimum_class_truth_instances" + ), + minimum_qualified_class_recall=_fraction( + gates, "minimum_qualified_class_recall" + ), + maximum_empty_prediction_risk_image_fraction=_fraction( + gates, "maximum_empty_prediction_risk_image_fraction" + ), + ), + temporal=TemporalSemanticProfile( + history_size=_positive_integer(temporal, "history_size"), + initial_confirmation_observations=_positive_integer( + temporal, "initial_confirmation_observations" + ), + switch_confirmation_observations=_positive_integer( + temporal, "switch_confirmation_observations" + ), + semantic_hold_seconds=_positive_number(temporal, "semantic_hold_seconds"), + state_expiry_seconds=_positive_number(temporal, "state_expiry_seconds"), + maximum_active_components=_positive_integer( + temporal, "maximum_active_components" + ), + ), + profile_sha256=hashlib.sha256(raw).hexdigest(), + ) + if ( + loaded.temporal.history_size > 32 + or loaded.temporal.initial_confirmation_observations + > loaded.temporal.history_size + or loaded.temporal.switch_confirmation_observations + > loaded.temporal.history_size + or loaded.temporal.semantic_hold_seconds >= loaded.temporal.state_expiry_seconds + or loaded.matching.small_area_upper_pixels + >= loaded.matching.medium_area_upper_pixels + ): + raise M48TRiskQualityError("M4.8T bounded profile is internally inconsistent") + return loaded + + +def load_coco_risk_truth( + annotations_path: Path, + profile: M48TRiskQualityProfile, +) -> tuple[tuple[CocoRiskImage, ...], tuple[RiskTruth, ...]]: + """Load and project independent COCO annotations into the 800x600 contract.""" + + try: + document = _object( + json.loads(annotations_path.expanduser().resolve(strict=True).read_text("utf-8")), + "COCO annotations", + ) + except (OSError, json.JSONDecodeError) as exc: + raise M48TRiskQualityError("COCO annotations cannot be read") from exc + image_values = _list(document.get("images"), "COCO images") + annotation_values = _list(document.get("annotations"), "COCO annotations") + category_values = _list(document.get("categories"), "COCO categories") + images_by_id: dict[int, CocoRiskImage] = {} + for value in image_values: + row = _object(value, "COCO image") + image = CocoRiskImage( + image_id=_positive_integer(row, "id"), + file_name=_string(row, "file_name"), + width=_positive_integer(row, "width"), + height=_positive_integer(row, "height"), + ) + if image.image_id in images_by_id: + raise M48TRiskQualityError("COCO image ids are duplicated") + images_by_id[image.image_id] = image + category_names: dict[int, str] = {} + for value in category_values: + row = _object(value, "COCO category") + category_id = _positive_integer(row, "id") + name = _string(row, "name") + if category_id in category_names: + raise M48TRiskQualityError("COCO category ids are duplicated") + category_names[category_id] = name + class_to_family = profile.class_to_family + if not set(class_to_family).issubset(category_names.values()): + raise M48TRiskQualityError("COCO risk categories do not match the profile") + + truth: list[RiskTruth] = [] + risk_image_ids: set[int] = set() + for value in annotation_values: + row = _object(value, "COCO annotation") + if row.get("iscrowd") != 0: + continue + image_id = _positive_integer(row, "image_id") + selected_image = images_by_id.get(image_id) + category = category_names.get(_positive_integer(row, "category_id")) + if selected_image is None or category is None: + raise M48TRiskQualityError("COCO annotation references missing metadata") + family = class_to_family.get(category) + if family is None: + continue + bbox = row.get("bbox") + if ( + not isinstance(bbox, list) + or len(bbox) != 4 + or any(not _finite_number(item) for item in bbox) + ): + raise M48TRiskQualityError("COCO annotation bbox is invalid") + x, y, width, height = (float(item) for item in bbox) + scale_x = _SOURCE_WIDTH / selected_image.width + scale_y = _SOURCE_HEIGHT / selected_image.height + x1 = max(0.0, min(float(_SOURCE_WIDTH), x * scale_x)) + y1 = max(0.0, min(float(_SOURCE_HEIGHT), y * scale_y)) + x2 = max(0.0, min(float(_SOURCE_WIDTH), (x + width) * scale_x)) + y2 = max(0.0, min(float(_SOURCE_HEIGHT), (y + height) * scale_y)) + if x2 <= x1 or y2 <= y1: + continue + area = (x2 - x1) * (y2 - y1) + if ( + area < profile.dataset.minimum_projected_box_area_pixels + or area > profile.dataset.maximum_projected_box_area_fraction * _SOURCE_AREA + ): + continue + size_band = ( + "small" + if area < profile.matching.small_area_upper_pixels + else "medium" + if area < profile.matching.medium_area_upper_pixels + else "large" + ) + truth.append( + RiskTruth( + image_id=image_id, + annotation_id=_positive_integer(row, "id"), + class_name=category, + family=family, + bbox_xyxy=(x1, y1, x2, y2), + projected_area_pixels=area, + size_band=size_band, + ) + ) + risk_image_ids.add(image_id) + if not truth: + raise M48TRiskQualityError("COCO risk truth is empty") + selected_images = tuple( + sorted( + (images_by_id[image_id] for image_id in risk_image_ids), + key=lambda item: item.image_id, + ) + ) + sorted_truth = tuple( + sorted(truth, key=lambda item: (item.image_id, item.annotation_id)) + ) + return selected_images, sorted_truth + + +def score_risk_quality( + *, + images: tuple[CocoRiskImage, ...], + truth: tuple[RiskTruth, ...], + predictions: tuple[RiskPrediction, ...], + profile: M48TRiskQualityProfile, +) -> RiskQualityResult: + """Score exact risk classes and families with predeclared gates.""" + + if not images or not truth: + raise M48TRiskQualityError("risk quality requires images and truth") + image_ids = {item.image_id for item in images} + if len(image_ids) != len(images) or any(item.image_id not in image_ids for item in truth): + raise M48TRiskQualityError("risk quality image contract is inconsistent") + class_to_family = profile.class_to_family + for truth_item in truth: + if class_to_family.get(truth_item.class_name) != truth_item.family: + raise M48TRiskQualityError("risk truth family is inconsistent") + _validate_bbox(truth_item.bbox_xyxy, "risk truth") + for prediction_item in predictions: + if prediction_item.image_id not in image_ids: + raise M48TRiskQualityError("risk prediction references an unknown image") + if class_to_family.get(prediction_item.class_name) != prediction_item.family: + raise M48TRiskQualityError("risk prediction family is inconsistent") + if prediction_item.score < profile.minimum_score: + raise M48TRiskQualityError("risk prediction escaped the frozen score threshold") + prediction_ids = [item.prediction_id for item in predictions] + if len(set(prediction_ids)) != len(prediction_ids): + raise M48TRiskQualityError("risk prediction ids are duplicated") + + truths_by_image = _group_truth(truth) + predictions_by_image = _group_predictions(predictions) + exact_matches: dict[int, str] = {} + matched_prediction_ids: set[str] = set() + family_matched_truth: set[int] = set() + failures: list[dict[str, object]] = [] + for image in sorted(images, key=lambda item: item.image_id): + image_truth = truths_by_image.get(image.image_id, ()) + image_predictions = predictions_by_image.get(image.image_id, ()) + for prediction in image_predictions: + choices = [ + item + for item in image_truth + if item.annotation_id not in exact_matches + and item.class_name == prediction.class_name + and _iou(item.bbox_xyxy, prediction.bbox_xyxy) + >= profile.matching.iou_threshold + ] + if choices: + selected = max( + choices, + key=lambda item: ( + _iou(item.bbox_xyxy, prediction.bbox_xyxy), + -item.annotation_id, + ), + ) + exact_matches[selected.annotation_id] = prediction.prediction_id + matched_prediction_ids.add(prediction.prediction_id) + + for prediction in image_predictions: + choices = [ + item + for item in image_truth + if item.annotation_id not in family_matched_truth + and item.family == prediction.family + and _iou(item.bbox_xyxy, prediction.bbox_xyxy) + >= profile.matching.iou_threshold + ] + if choices: + selected = max( + choices, + key=lambda item: ( + _iou(item.bbox_xyxy, prediction.bbox_xyxy), + -item.annotation_id, + ), + ) + family_matched_truth.add(selected.annotation_id) + + for item in image_truth: + if item.annotation_id in exact_matches: + continue + overlaps = tuple( + sorted( + ( + (_iou(item.bbox_xyxy, prediction.bbox_xyxy), prediction) + for prediction in image_predictions + ), + key=lambda pair: (-pair[0], pair[1].prediction_id), + ) + ) + best_iou, best = overlaps[0] if overlaps else (0.0, None) + if best is not None and best_iou >= profile.matching.iou_threshold: + bucket = ( + "same-family-class-confusion" + if best.family == item.family + else "cross-family-confusion" + ) + elif best is not None and best.class_name == item.class_name and best_iou > 0: + bucket = "localization" + else: + bucket = "missed" + failures.append( + { + "kind": "false-negative", + "bucket": bucket, + "image_id": item.image_id, + "annotation_id": item.annotation_id, + "class_name": item.class_name, + "family": item.family, + "size_band": item.size_band, + "best_iou": best_iou, + "best_prediction_class": None if best is None else best.class_name, + } + ) + for prediction in image_predictions: + if prediction.prediction_id not in matched_prediction_ids: + failures.append( + { + "kind": "false-positive", + "bucket": "unmatched-prediction", + "image_id": prediction.image_id, + "prediction_id": prediction.prediction_id, + "class_name": prediction.class_name, + "family": prediction.family, + "score": prediction.score, + } + ) + + true_positive = len(exact_matches) + false_positive = len(predictions) - true_positive + false_negative = len(truth) - true_positive + precision = _ratio(true_positive, true_positive + false_positive) + recall = _ratio(true_positive, len(truth)) + medium_large = tuple(item for item in truth if item.size_band != "small") + medium_large_recall = _ratio( + sum(item.annotation_id in exact_matches for item in medium_large), + len(medium_large), + ) + families = tuple(family for family, _ in profile.risk_families) + family_rows: dict[str, dict[str, object]] = {} + for family in families: + family_truth = tuple(item for item in truth if item.family == family) + exact_hits = sum(item.annotation_id in exact_matches for item in family_truth) + family_hits = sum(item.annotation_id in family_matched_truth for item in family_truth) + family_rows[family] = { + "truth_instances": len(family_truth), + "exact_class_recall": _ratio(exact_hits, len(family_truth)), + "family_recall": _ratio(family_hits, len(family_truth)), + } + class_rows: dict[str, dict[str, object]] = {} + class_truth_counts: dict[str, int] = {} + class_recalls: dict[str, float] = {} + for class_name in sorted(class_to_family): + class_truth = tuple(item for item in truth if item.class_name == class_name) + if not class_truth: + continue + hits = sum(item.annotation_id in exact_matches for item in class_truth) + class_truth_counts[class_name] = len(class_truth) + class_recalls[class_name] = _ratio(hits, len(class_truth)) + class_rows[class_name] = { + "family": class_to_family[class_name], + "truth_instances": len(class_truth), + "recall": class_recalls[class_name], + } + empty_prediction_images = sum( + not predictions_by_image.get(image.image_id) for image in images + ) + empty_fraction = _ratio(empty_prediction_images, len(images)) + qualified_class_failures = sorted( + class_name + for class_name in class_rows + if class_truth_counts[class_name] >= profile.gates.minimum_class_truth_instances + and class_recalls[class_name] < profile.gates.minimum_qualified_class_recall + ) + family_gate_map = dict(profile.gates.minimum_family_recall) + failed_gates: list[str] = [] + gate_checks = { + "minimum-truth-instances": len(truth) >= profile.gates.minimum_truth_instances, + "minimum-micro-precision": precision >= profile.gates.minimum_micro_precision, + "minimum-micro-recall": recall >= profile.gates.minimum_micro_recall, + "minimum-medium-large-recall": ( + medium_large_recall >= profile.gates.minimum_medium_large_recall + ), + "minimum-qualified-class-recall": not qualified_class_failures, + "maximum-empty-prediction-risk-image-fraction": ( + empty_fraction <= profile.gates.maximum_empty_prediction_risk_image_fraction + ), + } + for family in families: + gate_checks[f"minimum-family-recall:{family}"] = ( + _ratio( + sum( + item.annotation_id in family_matched_truth + for item in truth + if item.family == family + ), + sum(item.family == family for item in truth), + ) + >= family_gate_map[family] + ) + failed_gates.extend(key for key, passed in gate_checks.items() if not passed) + failure_counts = Counter(str(row["bucket"]) for row in failures) + report: dict[str, object] = { + "schema_version": M48T_REPORT_SCHEMA, + "profile_id": profile.profile_id, + "profile_sha256": profile.profile_sha256, + "candidate": { + "provider_id": profile.provider_id, + "model_id": profile.model_id, + "minimum_score": profile.minimum_score, + }, + "dataset": { + "dataset_id": profile.dataset.dataset_id, + "risk_images": len(images), + "truth_instances": len(truth), + "independent_human_annotations": True, + "ravnoves_ground_truth": False, + }, + "counts": { + "predictions": len(predictions), + "true_positive": true_positive, + "false_positive": false_positive, + "false_negative": false_negative, + "empty_prediction_risk_images": empty_prediction_images, + }, + "metrics": { + "micro_precision": precision, + "micro_recall": recall, + "medium_large_recall": medium_large_recall, + "empty_prediction_risk_image_fraction": empty_fraction, + "families": family_rows, + "classes": class_rows, + }, + "failure_buckets": dict(sorted(failure_counts.items())), + "quality_gates": { + "checks": gate_checks, + "failed": failed_gates, + "qualified_class_failures": qualified_class_failures, + "passed": not failed_gates, + }, + "scope": { + "child_adult_distinction_evaluated": False, + "unknown_moving_detection_evaluated": False, + "risk_policy_evaluated": False, + }, + "authority": dict(_FALSE_AUTHORITY), + } + return RiskQualityResult( + report=report, + failures=tuple( + sorted( + failures, + key=lambda row: ( + str(row["image_id"]).zfill(20), + str(row["kind"]), + str(row.get("annotation_id", row.get("prediction_id", ""))), + ), + ) + ), + ) + + +def _group_truth(truth: tuple[RiskTruth, ...]) -> dict[int, tuple[RiskTruth, ...]]: + result: dict[int, list[RiskTruth]] = {} + for item in truth: + result.setdefault(item.image_id, []).append(item) + return { + image_id: tuple(sorted(rows, key=lambda item: item.annotation_id)) + for image_id, rows in result.items() + } + + +def _group_predictions( + predictions: tuple[RiskPrediction, ...], +) -> dict[int, tuple[RiskPrediction, ...]]: + result: dict[int, list[RiskPrediction]] = {} + for item in predictions: + result.setdefault(item.image_id, []).append(item) + return { + image_id: tuple( + sorted(rows, key=lambda item: (-item.score, item.prediction_id)) + ) + for image_id, rows in result.items() + } + + +def _trailing_support(history: deque[str], value: str) -> int: + support = 0 + for item in reversed(history): + if item != value: + break + support += 1 + return support + + +def _iou( + left: tuple[float, float, float, float], + right: tuple[float, float, float, float], +) -> float: + intersection_width = max(0.0, min(left[2], right[2]) - max(left[0], right[0])) + intersection_height = max(0.0, min(left[3], right[3]) - max(left[1], right[1])) + intersection = intersection_width * intersection_height + left_area = (left[2] - left[0]) * (left[3] - left[1]) + right_area = (right[2] - right[0]) * (right[3] - right[1]) + return _ratio(intersection, left_area + right_area - intersection) + + +def _ratio(numerator: float | int, denominator: float | int) -> float: + return 0.0 if denominator == 0 else float(numerator) / float(denominator) + + +def _validate_bbox(value: tuple[float, float, float, float], label: str) -> None: + if len(value) != 4 or any(not math.isfinite(item) for item in value): + raise M48TRiskQualityError(f"{label} bbox is invalid") + x1, y1, x2, y2 = value + if not (0.0 <= x1 < x2 <= _SOURCE_WIDTH and 0.0 <= y1 < y2 <= _SOURCE_HEIGHT): + raise M48TRiskQualityError(f"{label} bbox escaped the source frame") + + +def _object(value: object, label: str) -> dict[str, object]: + if not isinstance(value, dict) or any(not isinstance(key, str) for key in value): + raise M48TRiskQualityError(f"{label} is not an object") + return value + + +def _list(value: object, label: str) -> list[object]: + if not isinstance(value, list): + raise M48TRiskQualityError(f"{label} are invalid") + return value + + +def _exact_keys(value: Mapping[str, object], expected: set[str], label: str) -> None: + if set(value) != expected: + raise M48TRiskQualityError(f"{label} keys are incompatible") + + +def _string(value: Mapping[str, object], key: str) -> str: + result = value.get(key) + if not isinstance(result, str) or not result: + raise M48TRiskQualityError(f"{key} is not a non-empty string") + return result + + +def _string_tuple(value: object, label: str) -> tuple[str, ...]: + if ( + not isinstance(value, list) + or any(not isinstance(item, str) or not item for item in value) + or len(set(value)) != len(value) + ): + raise M48TRiskQualityError(f"{label} are invalid") + return tuple(value) + + +def _positive_integer(value: Mapping[str, object], key: str) -> int: + result = value.get(key) + if not isinstance(result, int) or isinstance(result, bool) or result <= 0: + raise M48TRiskQualityError(f"{key} is not a positive integer") + return result + + +def _finite_number(value: object) -> TypeGuard[int | float]: + return ( + isinstance(value, (int, float)) + and not isinstance(value, bool) + and math.isfinite(float(value)) + ) + + +def _positive_number(value: Mapping[str, object], key: str) -> float: + result = value.get(key) + if not _finite_number(result) or float(result) <= 0: + raise M48TRiskQualityError(f"{key} is not a positive number") + return float(result) + + +def _fraction(value: Mapping[str, object], key: str) -> float: + result = value.get(key) + if not _finite_number(result) or not 0.0 <= float(result) <= 1.0: + raise M48TRiskQualityError(f"{key} is not a fraction") + return float(result) + + +__all__ = [ + "BoundedTemporalSemanticIdentity", + "CocoRiskImage", + "DEFAULT_M48T_PROFILE_PATH", + "M48TRiskQualityError", + "M48TRiskQualityProfile", + "RiskPrediction", + "RiskQualityResult", + "RiskTruth", + "StableSemanticIdentity", + "TemporalSemanticObservation", + "TemporalSemanticSnapshot", + "load_coco_risk_truth", + "load_m48t_risk_quality_profile", + "score_risk_quality", +] diff --git a/tests/test_m48t_risk_quality.py b/tests/test_m48t_risk_quality.py new file mode 100644 index 0000000..d3e13b7 --- /dev/null +++ b/tests/test_m48t_risk_quality.py @@ -0,0 +1,168 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from k1link.perception.m48t_risk_quality import ( + CocoRiskImage, + M48TRiskQualityError, + RiskPrediction, + RiskTruth, + load_coco_risk_truth, + load_m48t_risk_quality_profile, + score_risk_quality, +) + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +PROFILE_PATH = REPOSITORY_ROOT / "config/perception/m48t-risk-quality-temporal-v1.json" + + +def _truth( + annotation_id: int, + class_name: str, + family: str, + bbox: tuple[float, float, float, float], + *, + size_band: str = "medium", +) -> RiskTruth: + return RiskTruth( + image_id=1, + annotation_id=annotation_id, + class_name=class_name, + family=family, + bbox_xyxy=bbox, + projected_area_pixels=(bbox[2] - bbox[0]) * (bbox[3] - bbox[1]), + size_band=size_band, + ) + + +def _prediction( + prediction_id: str, + class_name: str, + family: str, + bbox: tuple[float, float, float, float], + score: float = 0.9, +) -> RiskPrediction: + return RiskPrediction( + image_id=1, + prediction_id=prediction_id, + class_name=class_name, + family=family, + score=score, + bbox_xyxy=bbox, + ) + + +def test_m48t_profile_pins_candidate_and_class_independent_temporal_policy() -> None: + profile = load_m48t_risk_quality_profile(PROFILE_PATH) + + assert profile.minimum_score == 0.25 + assert profile.model_id == "rf_detr_large:1" + assert profile.class_to_family["dog"] == "animal" + assert profile.temporal.initial_confirmation_observations == 2 + assert profile.temporal.switch_confirmation_observations == 3 + + +def test_m48t_profile_rejects_candidate_threshold_tuning(tmp_path: Path) -> None: + document = json.loads(PROFILE_PATH.read_text("utf-8")) + document["candidate"]["minimum_score"] = 0.2 + changed = tmp_path / "changed.json" + changed.write_text(json.dumps(document), "utf-8") + + with pytest.raises(M48TRiskQualityError, match="tuned in place"): + load_m48t_risk_quality_profile(changed) + + +def test_coco_truth_is_projected_to_actual_source_contract_and_filtered( + tmp_path: Path, +) -> None: + profile = load_m48t_risk_quality_profile(PROFILE_PATH) + annotations = { + "images": [{"id": 1, "file_name": "one.jpg", "width": 400, "height": 300}], + "categories": [ + {"id": index, "name": class_name} + for index, class_name in enumerate(profile.class_to_family, start=1) + ], + "annotations": [ + {"id": 1, "image_id": 1, "category_id": 1, "bbox": [10, 20, 20, 30], "iscrowd": 0}, + {"id": 2, "image_id": 1, "category_id": 1, "bbox": [1, 1, 1, 1], "iscrowd": 0}, + {"id": 3, "image_id": 1, "category_id": 1, "bbox": [10, 20, 20, 30], "iscrowd": 1}, + ], + } + path = tmp_path / "instances.json" + path.write_text(json.dumps(annotations), "utf-8") + + images, truth = load_coco_risk_truth(path, profile) + + assert images == (CocoRiskImage(image_id=1, file_name="one.jpg", width=400, height=300),) + assert len(truth) == 1 + assert truth[0].bbox_xyxy == (20.0, 40.0, 60.0, 100.0) + assert truth[0].projected_area_pixels == 2400.0 + + +def test_quality_separates_exact_class_family_and_failure_buckets() -> None: + profile = load_m48t_risk_quality_profile(PROFILE_PATH) + images = (CocoRiskImage(image_id=1, file_name="one.jpg", width=800, height=600),) + truth = ( + _truth(1, "person", "person", (10.0, 10.0, 110.0, 210.0), size_band="large"), + _truth(2, "dog", "animal", (200.0, 100.0, 260.0, 170.0)), + _truth(3, "car", "vehicle", (400.0, 200.0, 600.0, 350.0), size_band="large"), + _truth(4, "bicycle", "light-road-user", (650.0, 200.0, 760.0, 350.0)), + ) + predictions = ( + _prediction("p1", "person", "person", (10.0, 10.0, 110.0, 210.0)), + _prediction("p2", "cat", "animal", (200.0, 100.0, 260.0, 170.0)), + _prediction("p3", "car", "vehicle", (520.0, 300.0, 700.0, 450.0)), + _prediction("p4", "truck", "vehicle", (300.0, 20.0, 390.0, 100.0)), + ) + + result = score_risk_quality( + images=images, + truth=truth, + predictions=predictions, + profile=profile, + ) + + assert result.report["counts"] == { + "predictions": 4, + "true_positive": 1, + "false_positive": 3, + "false_negative": 3, + "empty_prediction_risk_images": 0, + } + metrics = result.report["metrics"] + assert isinstance(metrics, dict) + families = metrics["families"] + assert isinstance(families, dict) + assert families["animal"]["exact_class_recall"] == 0.0 + assert families["animal"]["family_recall"] == 1.0 + buckets = result.report["failure_buckets"] + assert buckets["same-family-class-confusion"] == 1 + assert buckets["localization"] == 1 + assert buckets["missed"] == 1 + assert buckets["unmatched-prediction"] == 3 + assert result.report["quality_gates"]["passed"] is False + assert result.report["authority"]["candidate_accepted"] is False + + +def test_quality_rejects_predictions_below_frozen_threshold() -> None: + profile = load_m48t_risk_quality_profile(PROFILE_PATH) + images = (CocoRiskImage(image_id=1, file_name="one.jpg", width=800, height=600),) + + with pytest.raises(M48TRiskQualityError, match="escaped the frozen score"): + score_risk_quality( + images=images, + truth=(_truth(1, "person", "person", (10.0, 10.0, 100.0, 200.0)),), + predictions=( + _prediction( + "p1", + "person", + "person", + (10.0, 10.0, 100.0, 200.0), + score=0.24, + ), + ), + profile=profile, + ) diff --git a/tests/test_m48t_temporal_semantic_identity.py b/tests/test_m48t_temporal_semantic_identity.py new file mode 100644 index 0000000..3c1084f --- /dev/null +++ b/tests/test_m48t_temporal_semantic_identity.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest + +from k1link.perception.m48t_risk_quality import ( + BoundedTemporalSemanticIdentity, + M48TRiskQualityError, + TemporalSemanticObservation, + load_m48t_risk_quality_profile, +) + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +PROFILE_PATH = REPOSITORY_ROOT / "config/perception/m48t-risk-quality-temporal-v1.json" + + +def _observation( + time_ns: int, + raw_class: str | None, + *, + component_id: str = "temporal-000001", + currentness: str = "current", +) -> TemporalSemanticObservation: + return TemporalSemanticObservation( + component_id=component_id, + evidence_time_ns=time_ns, + raw_class_name=raw_class, + currentness=currentness, + ) + + +def test_initial_class_requires_two_observations_and_identity_stays_geometry_owned() -> None: + stabilizer = BoundedTemporalSemanticIdentity(load_m48t_risk_quality_profile(PROFILE_PATH)) + + first = stabilizer.update(_observation(0, "person")) + second = stabilizer.update(_observation(100_000_000, "person")) + + assert first.resolution == "pending" + assert first.selected_class_name is None + assert second.resolution == "confirmed" + assert second.selected_class_name == "person" + assert second.component_id == "temporal-000001" + assert second.association_uses_semantic_class is False + assert second.occupancy_uses_semantic_class is False + + +def test_cross_family_switch_falls_back_unknown_until_third_confirmation() -> None: + stabilizer = BoundedTemporalSemanticIdentity(load_m48t_risk_quality_profile(PROFILE_PATH)) + stabilizer.update(_observation(0, "person")) + stabilizer.update(_observation(100_000_000, "person")) + + first_conflict = stabilizer.update(_observation(200_000_000, "car")) + second_conflict = stabilizer.update(_observation(300_000_000, "car")) + switched = stabilizer.update(_observation(400_000_000, "car")) + + assert first_conflict.resolution == "conflict" + assert first_conflict.selected_class_name is None + assert second_conflict.resolution == "conflict" + assert switched.resolution == "confirmed" + assert switched.selected_class_name == "car" + assert stabilizer.snapshot().class_switches == 1 + + +def test_same_family_switch_holds_previous_class_until_confirmed() -> None: + stabilizer = BoundedTemporalSemanticIdentity(load_m48t_risk_quality_profile(PROFILE_PATH)) + stabilizer.update(_observation(0, "dog")) + stabilizer.update(_observation(100_000_000, "dog")) + + pending = stabilizer.update(_observation(200_000_000, "cat")) + stabilizer.update(_observation(300_000_000, "cat")) + switched = stabilizer.update(_observation(400_000_000, "cat")) + + assert pending.resolution == "pending" + assert pending.selected_class_name == "dog" + assert switched.selected_class_name == "cat" + + +def test_semantic_hold_is_bounded_then_degrades_to_unknown() -> None: + stabilizer = BoundedTemporalSemanticIdentity(load_m48t_risk_quality_profile(PROFILE_PATH)) + stabilizer.update(_observation(0, "person")) + stabilizer.update(_observation(100_000_000, "person")) + + held = stabilizer.update(_observation(300_000_000, None, currentness="held")) + unknown = stabilizer.update(_observation(500_000_001, None, currentness="held")) + + assert held.resolution == "held" + assert held.selected_class_name == "person" + assert unknown.resolution == "unknown" + assert unknown.selected_class_name is None + + +def test_explicit_expiry_removes_state_and_out_of_order_evidence_is_rejected() -> None: + stabilizer = BoundedTemporalSemanticIdentity(load_m48t_risk_quality_profile(PROFILE_PATH)) + stabilizer.update(_observation(100, "person")) + with pytest.raises(M48TRiskQualityError, match="moved backwards"): + stabilizer.update(_observation(99, "person")) + + expired = stabilizer.update(_observation(200, None, currentness="expired")) + restarted = stabilizer.update(_observation(300, "person")) + + assert expired.resolution == "expired" + assert restarted.resolution == "pending" + assert stabilizer.snapshot().active_components == 1