SILVER/NODEDC_MISSION_CORE
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

feat(perception): freeze blind detector gates

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-07-29 14:20:31 +03:00
parent 8b1f109b09
commit 82a44478fb
14 changed files with 3087 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/k1link/compute/e45_binding_sensitivity.py
+988
View File
@@ -0,0 +1,988 @@
"""E45 source-scoped binding sensitivity audit over accepted E31 evidence.
E45 does not re-run projection, tune E31, or manufacture calibration target
truth. It joins accepted E31 correspondences with the exact E10 pose/timing
arrays and E30 camera geometry, then measures how the existing diagnostic
residual behaves across image radius, rig motion, and source age.
"""
from __future__ import annotations
import hashlib
import json
import math
import os
import shutil
import uuid
from collections.abc import Iterable
from dataclasses import asdict, dataclass
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Final
import numpy as np
import numpy.typing as npt
from .e31_source_qualification import (
E31_CORRESPONDENCES_NAME,
E31SourceQualificationError,
read_e31_source_qualification,
)
from .lidar_field_review import E10LidarFieldSource
E45_RESULT_SCHEMA: Final = "missioncore.e45-binding-sensitivity/v1"
E45_REPORT_SCHEMA: Final = "missioncore.e45-binding-sensitivity-report/v1"
E45_ROW_SCHEMA: Final = "missioncore.e45-binding-sensitivity-row/v1"
E45_PROFILE_SCHEMA: Final = "missioncore.e45-binding-sensitivity-profile/v1"
E45_MANIFEST_NAME: Final = "manifest.json"
E45_REPORT_NAME: Final = "binding-sensitivity-report.json"
E45_ROWS_NAME: Final = "binding-sensitivity-rows.jsonl"
_AUTHORITY: Final = {
"commands_enabled": False,
"navigation_or_safety_accepted": False,
}
class E45BindingSensitivityError(RuntimeError):
"""An E45 input, analysis, or immutable result violates the contract."""
def _valid_edges(
values: tuple[float, float],
*,
lower: float,
upper: float,
) -> bool:
return bool(
len(values) == 2
and np.isfinite(values).all()
and lower <= values[0] < values[1] <= upper
)
@dataclass(frozen=True, slots=True)
class E45BindingSensitivityProfile:
"""Frozen descriptive strata for the source-scoped E45 audit."""
profile_id: str = "e45-ravnoves00-binding-sensitivity/v1"
image_radius_edges: tuple[float, float] = (0.5, 0.85)
translation_speed_edges_mps: tuple[float, float] = (0.1, 1.0)
angular_speed_edges_deg_s: tuple[float, float] = (2.0, 12.0)
lidar_camera_age_edges_ms: tuple[float, float] = (25.0, 60.0)
pose_point_age_edges_ms: tuple[float, float] = (10.0, 25.0)
def __post_init__(self) -> None:
if (
not self.profile_id.strip()
or len(self.profile_id) > 160
or not _valid_edges(self.image_radius_edges, lower=0.0, upper=2.0)
or not _valid_edges(
self.translation_speed_edges_mps,
lower=0.0,
upper=30.0,
)
or not _valid_edges(
self.angular_speed_edges_deg_s,
lower=0.0,
upper=720.0,
)
or not _valid_edges(
self.lidar_camera_age_edges_ms,
lower=0.0,
upper=500.0,
)
or not _valid_edges(
self.pose_point_age_edges_ms,
lower=0.0,
upper=500.0,
)
):
raise E45BindingSensitivityError("E45 profile is invalid")
def to_dict(self) -> dict[str, object]:
return {
"schema_version": E45_PROFILE_SCHEMA,
**asdict(self),
"image_radius_edges": list(self.image_radius_edges),
"translation_speed_edges_mps": list(
self.translation_speed_edges_mps
),
"angular_speed_edges_deg_s": list(
self.angular_speed_edges_deg_s
),
"lidar_camera_age_edges_ms": list(
self.lidar_camera_age_edges_ms
),
"pose_point_age_edges_ms": list(self.pose_point_age_edges_ms),
"analysis_kind": "descriptive-source-scoped-sensitivity",
"threshold_tuning_allowed": False,
"calibration_target_truth_available": False,
"physical_mount_inferred": False,
}
DEFAULT_E45_BINDING_SENSITIVITY_PROFILE: Final = (
E45BindingSensitivityProfile()
)
@dataclass(frozen=True, slots=True)
class E45BindingSensitivity:
result_id: str
result_root: Path
manifest: dict[str, Any]
report: dict[str, Any]
def build_e45_binding_sensitivity(
*,
e31_result_root: Path,
source_pack_root: Path,
materialization_root: Path,
output_root: Path,
profile: E45BindingSensitivityProfile = (
DEFAULT_E45_BINDING_SENSITIVITY_PROFILE
),
) -> E45BindingSensitivity:
"""Build or verify one immutable E45 diagnostic sensitivity result."""
try:
e31 = read_e31_source_qualification(e31_result_root)
except E31SourceQualificationError as reason:
raise E45BindingSensitivityError("E31 source is invalid") from reason
if (
e31.report.get("eligible_for_e32") is not True
or e31.report.get("status") != "accepted-diagnostic-source-profile"
):
raise E45BindingSensitivityError("E31 source is not accepted")
e31_source = _object(
_object(e31.manifest.get("identity"), "E31 identity").get("source"),
"E31 source",
)
source = E10LidarFieldSource(source_pack_root)
try:
if (
source.pack_id != e31_source.get("source_pack_id")
or source.identity.get("session_id") != e31_source.get("session_id")
or source.identity.get("source_id") != e31_source.get("source_id")
):
raise E45BindingSensitivityError("E10/E31 source binding changed")
items = _load_materialized_items(
materialization_root=materialization_root,
expected_result_id=str(e31_source.get("materialization_id", "")),
)
correspondence_rows = tuple(
_read_jsonl(e31.result_root / E31_CORRESPONDENCES_NAME)
)
rows = _build_rows(
correspondence_rows=correspondence_rows,
items=items,
source=source,
profile=profile,
)
analysis = analyze_binding_sensitivity(rows, profile=profile)
profile_document = profile.to_dict()
rows_sha256 = hashlib.sha256(
b"".join(_canonical_json(row) + b"\n" for row in rows)
).hexdigest()
analysis_sha256 = hashlib.sha256(
_canonical_json(analysis)
).hexdigest()
identity = {
"schema_version": E45_RESULT_SCHEMA,
"source": {
"session_id": str(source.identity["session_id"]),
"source_id": str(source.identity["source_id"]),
"source_pack_id": source.pack_id,
"e31_result_id": e31.result_id,
"materialization_id": materialization_root.resolve(
strict=True
).name,
"calibration_content_identity_sha256": e31_source.get(
"calibration_content_identity_sha256"
),
},
"profile": profile_document,
"rows_sha256": rows_sha256,
"analysis_sha256": analysis_sha256,
"producer_sha256": _sha256(Path(__file__).resolve(strict=True)),
"authority": _AUTHORITY,
}
identity_sha256 = hashlib.sha256(
_canonical_json(identity)
).hexdigest()
result_id = f"e45-binding-sensitivity-{identity_sha256}"
destination = output_root.expanduser().absolute() / result_id
if destination.exists():
return read_e45_binding_sensitivity(destination)
report = {
"schema_version": E45_REPORT_SCHEMA,
"result_id": result_id,
"identity_sha256": identity_sha256,
"status": "completed-source-scoped-binding-sensitivity",
"analysis": analysis,
"decision": {
"e31_binding_retained": True,
"e31_thresholds_changed": False,
"p0_calibration_gate_closed": False,
"measured_calibration_target_residual_available": False,
"physical_mount_dimensions_available": False,
"next_gate": (
"collect explicit static-landmark or calibration-target "
"correspondences without changing the accepted E31 source"
),
},
"limitations": [
(
"the residual is the median occupied-support centroid "
"relative to a reviewed 2D box, not calibration-target truth"
),
(
"motion and source-age strata diagnose the recorded "
"host-arrival binding; they do not recover hardware firing time"
),
(
"physical mount transform, vehicle body dimensions and "
"cross-route transfer remain unavailable"
),
(
"correlation is descriptive on 87 accepted correspondences "
"and must not be interpreted as causal calibration error"
),
],
"authority": _AUTHORITY,
}
destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
staging = destination.parent / (
f".{result_id}.{uuid.uuid4().hex}.tmp"
)
staging.mkdir(mode=0o700, exist_ok=False)
try:
_write_jsonl(staging / E45_ROWS_NAME, rows)
_write_json(staging / E45_REPORT_NAME, report)
manifest = {
"schema_version": E45_RESULT_SCHEMA,
"result_id": result_id,
"identity_sha256": identity_sha256,
"identity": identity,
"created_at_utc": _utc_now(),
"acceptance_state": "accepted-diagnostic-measurement-only",
"artifacts": [
_artifact(staging / E45_REPORT_NAME, "sensitivity-report"),
_artifact(staging / E45_ROWS_NAME, "sensitivity-rows"),
],
"authority": _AUTHORITY,
}
_write_json(staging / E45_MANIFEST_NAME, manifest)
os.replace(staging, destination)
except BaseException:
shutil.rmtree(staging, ignore_errors=True)
raise
return read_e45_binding_sensitivity(destination)
finally:
source.close()
def read_e45_binding_sensitivity(root: Path) -> E45BindingSensitivity:
"""Read and fully validate one immutable E45 result."""
resolved = root.resolve(strict=True)
manifest = _read_json(resolved / E45_MANIFEST_NAME)
identity = _object(manifest.get("identity"), "E45 identity")
identity_sha256 = manifest.get("identity_sha256")
if (
manifest.get("schema_version") != E45_RESULT_SCHEMA
or not isinstance(identity_sha256, str)
or hashlib.sha256(_canonical_json(identity)).hexdigest()
!= identity_sha256
or manifest.get("result_id")
!= f"e45-binding-sensitivity-{identity_sha256}"
or resolved.name != manifest.get("result_id")
or manifest.get("acceptance_state")
!= "accepted-diagnostic-measurement-only"
or manifest.get("authority") != _AUTHORITY
):
raise E45BindingSensitivityError("E45 result identity is invalid")
artifact_rows = manifest.get("artifacts")
if not isinstance(artifact_rows, list) or len(artifact_rows) != 2:
raise E45BindingSensitivityError("E45 artifact catalog is invalid")
artifacts = {
str(item.get("role")): item
for item in artifact_rows
if isinstance(item, dict)
}
for role, name in (
("sensitivity-report", E45_REPORT_NAME),
("sensitivity-rows", E45_ROWS_NAME),
):
item = artifacts.get(role)
path = resolved / name
if (
item is None
or item.get("path") != name
or not path.is_file()
or item.get("byte_length") != path.stat().st_size
or item.get("sha256") != _sha256(path)
):
raise E45BindingSensitivityError("E45 artifact content changed")
report = _read_json(resolved / E45_REPORT_NAME)
analysis = _object(report.get("analysis"), "E45 analysis")
rows = tuple(_read_jsonl(resolved / E45_ROWS_NAME))
if (
report.get("schema_version") != E45_REPORT_SCHEMA
or report.get("result_id") != resolved.name
or report.get("identity_sha256") != identity_sha256
or report.get("authority") != _AUTHORITY
or hashlib.sha256(_canonical_json(analysis)).hexdigest()
!= identity.get("analysis_sha256")
or hashlib.sha256(
b"".join(_canonical_json(row) + b"\n" for row in rows)
).hexdigest()
!= identity.get("rows_sha256")
or len(rows) != analysis.get("correspondence_count")
or len({row.get("item_id") for row in rows}) != len(rows)
or any(row.get("schema_version") != E45_ROW_SCHEMA for row in rows)
or report.get("decision", {}).get("p0_calibration_gate_closed")
is not False
):
raise E45BindingSensitivityError("E45 report is invalid")
return E45BindingSensitivity(
result_id=resolved.name,
result_root=resolved,
manifest=manifest,
report=report,
)
def analyze_binding_sensitivity(
rows: Iterable[dict[str, Any]],
*,
profile: E45BindingSensitivityProfile = (
DEFAULT_E45_BINDING_SENSITIVITY_PROFILE
),
) -> dict[str, Any]:
"""Aggregate validated E45 rows into descriptive source-scoped strata."""
materialized = tuple(rows)
if not materialized:
raise E45BindingSensitivityError("E45 rows are empty")
item_ids: set[str] = set()
for row in materialized:
item_id = row.get("item_id")
values = (
row.get("image_radius_normalized"),
row.get("translation_speed_mps"),
row.get("angular_speed_deg_s"),
row.get("lidar_camera_age_ms"),
row.get("pose_point_age_ms"),
row.get("motion_exposure_translation_m"),
row.get("motion_exposure_rotation_deg"),
row.get("centroid_residual_bbox_diagonal"),
)
if (
row.get("schema_version") != E45_ROW_SCHEMA
or not isinstance(item_id, str)
or not item_id
or item_id in item_ids
or not all(
isinstance(value, (int, float))
and not isinstance(value, bool)
and math.isfinite(float(value))
and float(value) >= 0.0
for value in values
)
or not isinstance(row.get("occupied_points_in_bbox"), int)
or int(row["occupied_points_in_bbox"]) < 0
or not isinstance(row.get("supported"), bool)
):
raise E45BindingSensitivityError("E45 row is invalid")
item_ids.add(item_id)
supported = [row for row in materialized if row["supported"]]
if len(supported) != len(materialized):
raise E45BindingSensitivityError(
"E45 requires every accepted E31 correspondence to remain supported"
)
residual = _array(
materialized,
"centroid_residual_bbox_diagonal",
)
variables = {
"image_radius_normalized": _array(
materialized,
"image_radius_normalized",
),
"translation_speed_mps": _array(
materialized,
"translation_speed_mps",
),
"angular_speed_deg_s": _array(
materialized,
"angular_speed_deg_s",
),
"lidar_camera_age_ms": _array(
materialized,
"lidar_camera_age_ms",
),
"pose_point_age_ms": _array(
materialized,
"pose_point_age_ms",
),
"motion_exposure_translation_m": _array(
materialized,
"motion_exposure_translation_m",
),
"motion_exposure_rotation_deg": _array(
materialized,
"motion_exposure_rotation_deg",
),
}
return {
"correspondence_count": len(materialized),
"supported_count": len(supported),
"supported_fraction": float(len(supported) / len(materialized)),
"residual_bbox_diagonal": _distribution(residual),
"occupied_points_in_bbox": _distribution(
_array(materialized, "occupied_points_in_bbox")
),
"motion_and_age": {
name: _distribution(values)
for name, values in variables.items()
},
"strata": {
"image_radius": _stratify(
materialized,
key="image_radius_normalized",
edges=profile.image_radius_edges,
),
"translation_speed": _stratify(
materialized,
key="translation_speed_mps",
edges=profile.translation_speed_edges_mps,
),
"angular_speed": _stratify(
materialized,
key="angular_speed_deg_s",
edges=profile.angular_speed_edges_deg_s,
),
"lidar_camera_age": _stratify(
materialized,
key="lidar_camera_age_ms",
edges=profile.lidar_camera_age_edges_ms,
),
"pose_point_age": _stratify(
materialized,
key="pose_point_age_ms",
edges=profile.pose_point_age_edges_ms,
),
},
"spearman_residual_correlation": {
name: _spearman(values, residual)
for name, values in variables.items()
},
"evidence_accounting_complete": True,
"measured_calibration_target_residual_available": False,
"raw_firing_time_inferred": False,
"physical_mount_inferred": False,
}
def _build_rows(
*,
correspondence_rows: tuple[dict[str, Any], ...],
items: dict[str, dict[str, Any]],
source: E10LidarFieldSource,
profile: E45BindingSensitivityProfile,
) -> tuple[dict[str, Any], ...]:
times = np.asarray(source.arrays["session_seconds"], dtype=np.float64)
positions = np.asarray(
source.arrays["pose_positions_map"],
dtype=np.float64,
)
quaternions = np.asarray(
source.arrays["pose_quaternions_map_from_lidar"],
dtype=np.float64,
)
lidar_age = np.abs(
np.asarray(source.arrays["lidar_camera_delta_ms"], dtype=np.float64)
)
pose_age = np.abs(
np.asarray(source.arrays["pose_point_delta_ms"], dtype=np.float64)
)
available = np.asarray(
source.arrays["sample_available"],
dtype=np.bool_,
)
translation_speed, angular_speed = _rig_motion(
times=times,
positions=positions,
quaternions=quaternions,
available=available,
)
projection = _object(source.identity.get("projection"), "E10 projection")
width = float(projection["width"])
height = float(projection["height"])
rows: list[dict[str, Any]] = []
for correspondence in correspondence_rows:
item_id = str(correspondence.get("item_id", ""))
item = items.get(item_id)
if item is None:
raise E45BindingSensitivityError(
"E31 correspondence is missing from E30"
)
frame_index = _integer(
correspondence.get("frame_index"),
"E31 frame index",
)
if not 0 <= frame_index < source.frame_count:
raise E45BindingSensitivityError("E31 frame index is invalid")
scores = correspondence.get("scores")
if not isinstance(scores, list):
raise E45BindingSensitivityError("E31 scores are invalid")
baseline = next(
(
score
for score in scores
if isinstance(score, dict) and score.get("offset_ms") == 0
),
None,
)
if (
baseline is None
or baseline.get("evaluable") is not True
or not isinstance(baseline.get("occupied_points_in_bbox"), int)
or baseline.get("centroid_residual_bbox_diagonal") is None
):
raise E45BindingSensitivityError(
"E31 zero-offset correspondence is incomplete"
)
snapshot = _object(item.get("e29_snapshot"), "E30 snapshot")
bbox = np.asarray(snapshot.get("bbox_xyxy"), dtype=np.float64)
if (
bbox.shape != (4,)
or not np.isfinite(bbox).all()
or bbox[2] <= bbox[0]
or bbox[3] <= bbox[1]
):
raise E45BindingSensitivityError("E30 bbox is invalid")
center_x = float((bbox[0] + bbox[2]) * 0.5)
center_y = float((bbox[1] + bbox[3]) * 0.5)
radius = math.hypot(
(center_x - width * 0.5) / (width * 0.5),
(center_y - height * 0.5) / (height * 0.5),
)
age_seconds = (
float(lidar_age[frame_index] + pose_age[frame_index]) / 1000.0
)
row = {
"schema_version": E45_ROW_SCHEMA,
"item_id": item_id,
"review_key": str(correspondence.get("review_key", "")),
"frame_index": frame_index,
"source_frame_index": _integer(
_object(
item.get("evidence_binding"),
"E30 evidence binding",
).get("source_frame_index"),
"E30 source frame index",
),
"session_seconds": float(times[frame_index]),
"label": str(snapshot.get("label", "object")),
"bbox_xyxy": [float(value) for value in bbox],
"image_radius_normalized": radius,
"image_radius_stratum": _bin_label(
radius,
profile.image_radius_edges,
),
"translation_speed_mps": float(
translation_speed[frame_index]
),
"translation_speed_stratum": _bin_label(
float(translation_speed[frame_index]),
profile.translation_speed_edges_mps,
),
"angular_speed_deg_s": float(angular_speed[frame_index]),
"angular_speed_stratum": _bin_label(
float(angular_speed[frame_index]),
profile.angular_speed_edges_deg_s,
),
"lidar_camera_age_ms": float(lidar_age[frame_index]),
"lidar_camera_age_stratum": _bin_label(
float(lidar_age[frame_index]),
profile.lidar_camera_age_edges_ms,
),
"pose_point_age_ms": float(pose_age[frame_index]),
"pose_point_age_stratum": _bin_label(
float(pose_age[frame_index]),
profile.pose_point_age_edges_ms,
),
"motion_exposure_translation_m": float(
translation_speed[frame_index] * age_seconds
),
"motion_exposure_rotation_deg": float(
angular_speed[frame_index] * age_seconds
),
"occupied_points_in_bbox": int(
baseline["occupied_points_in_bbox"]
),
"centroid_residual_bbox_diagonal": float(
baseline["centroid_residual_bbox_diagonal"]
),
"supported": int(baseline["occupied_points_in_bbox"]) >= 2,
"residual_interpretation": (
"diagnostic-support-centroid-not-calibration-target"
),
}
if not all(
math.isfinite(_number(row[key], key))
for key in (
"image_radius_normalized",
"translation_speed_mps",
"angular_speed_deg_s",
"lidar_camera_age_ms",
"pose_point_age_ms",
"motion_exposure_translation_m",
"motion_exposure_rotation_deg",
"centroid_residual_bbox_diagonal",
)
):
raise E45BindingSensitivityError(
"E45 derived measurement is not finite"
)
rows.append(row)
rows.sort(key=lambda row: (int(row["frame_index"]), str(row["item_id"])))
if len(rows) != len({str(row["item_id"]) for row in rows}):
raise E45BindingSensitivityError("E45 item identity is duplicated")
return tuple(rows)
def _rig_motion(
*,
times: npt.NDArray[np.float64],
positions: npt.NDArray[np.float64],
quaternions: npt.NDArray[np.float64],
available: npt.NDArray[np.bool_],
) -> tuple[npt.NDArray[np.float64], npt.NDArray[np.float64]]:
count = int(times.size)
valid_indices = np.flatnonzero(available)
if (
count < 2
or positions.shape != (count, 3)
or quaternions.shape != (count, 4)
or available.shape != (count,)
or not np.isfinite(times).all()
or valid_indices.size < 2
or not np.isfinite(positions[valid_indices]).all()
or not np.isfinite(quaternions[valid_indices]).all()
or np.any(np.diff(times) <= 0.0)
):
raise E45BindingSensitivityError("E10 pose timeline is invalid")
valid_rows = np.arange(valid_indices.size)
left = valid_indices[np.maximum(valid_rows - 1, 0)]
right = valid_indices[
np.minimum(valid_rows + 1, valid_indices.size - 1)
]
duration = times[right] - times[left]
if np.any(duration <= 0.0):
raise E45BindingSensitivityError("E10 pose duration is invalid")
valid_translation = np.linalg.norm(
positions[right] - positions[left],
axis=1,
) / duration
q_left = quaternions[left].copy()
q_right = quaternions[right].copy()
q_left /= np.linalg.norm(q_left, axis=1, keepdims=True)
q_right /= np.linalg.norm(q_right, axis=1, keepdims=True)
dot = np.clip(
np.abs(np.sum(q_left * q_right, axis=1)),
0.0,
1.0,
)
valid_angular = np.degrees(2.0 * np.arccos(dot)) / duration
translation = np.full(count, np.nan, dtype=np.float64)
angular = np.full(count, np.nan, dtype=np.float64)
translation[valid_indices] = valid_translation
angular[valid_indices] = valid_angular
return (
np.asarray(translation, dtype=np.float64),
np.asarray(angular, dtype=np.float64),
)
def _load_materialized_items(
*,
materialization_root: Path,
expected_result_id: str,
) -> dict[str, dict[str, Any]]:
root = materialization_root.resolve(strict=True)
manifest = _read_json(root / "manifest.json")
if (
root.name != expected_result_id
or manifest.get("result_id") != expected_result_id
or manifest.get("schema_version")
!= "missioncore.e30-evidence-materialization/v2"
):
raise E45BindingSensitivityError(
"E30 materialization binding changed"
)
artifacts = manifest.get("artifacts")
if not isinstance(artifacts, list):
raise E45BindingSensitivityError(
"E30 materialization catalog is invalid"
)
artifact = next(
(
item
for item in artifacts
if isinstance(item, dict)
and item.get("role") == "materialized-items"
),
None,
)
if artifact is None:
raise E45BindingSensitivityError(
"E30 materialized item index is absent"
)
path = root / str(artifact.get("path", ""))
if (
not path.is_file()
or artifact.get("byte_length") != path.stat().st_size
or artifact.get("sha256") != _sha256(path)
):
raise E45BindingSensitivityError(
"E30 materialized item index changed"
)
rows = tuple(_read_jsonl(path))
result = {str(row.get("item_id", "")): row for row in rows}
if (
len(rows) != manifest.get("item_count")
or len(result) != len(rows)
or "" in result
):
raise E45BindingSensitivityError(
"E30 materialized item coverage changed"
)
return result
def _stratify(
rows: tuple[dict[str, Any], ...],
*,
key: str,
edges: tuple[float, float],
) -> list[dict[str, Any]]:
result = []
for label in ("low", "middle", "high"):
selected = [
row
for row in rows
if _bin_label(float(row[key]), edges) == label
]
result.append(
{
"stratum": label,
"minimum_inclusive": (
None
if label == "low"
else edges[0] if label == "middle" else edges[1]
),
"maximum_exclusive": (
edges[0]
if label == "low"
else edges[1] if label == "middle" else None
),
"count": len(selected),
"supported_count": sum(
1 for row in selected if row["supported"]
),
"supported_fraction": (
float(
sum(1 for row in selected if row["supported"])
/ len(selected)
)
if selected
else None
),
"centroid_residual_bbox_diagonal": (
_distribution(
_array(
selected,
"centroid_residual_bbox_diagonal",
)
)
if selected
else None
),
"occupied_points_in_bbox": (
_distribution(
_array(selected, "occupied_points_in_bbox")
)
if selected
else None
),
}
)
return result
def _spearman(
left: npt.NDArray[np.float64],
right: npt.NDArray[np.float64],
) -> float | None:
if left.size != right.size or left.size < 3:
return None
left_rank = _rank(left)
right_rank = _rank(right)
if np.std(left_rank) == 0.0 or np.std(right_rank) == 0.0:
return None
value = float(np.corrcoef(left_rank, right_rank)[0, 1])
return value if math.isfinite(value) else None
def _rank(values: npt.NDArray[np.float64]) -> npt.NDArray[np.float64]:
order = np.argsort(values, kind="mergesort")
ranks = np.empty(values.size, dtype=np.float64)
start = 0
while start < values.size:
end = start + 1
while end < values.size and values[order[end]] == values[order[start]]:
end += 1
ranks[order[start:end]] = (start + end - 1) * 0.5 + 1.0
start = end
return ranks
def _distribution(values: npt.NDArray[np.float64]) -> dict[str, object]:
if values.size == 0 or not np.isfinite(values).all():
raise E45BindingSensitivityError("E45 distribution is invalid")
return {
"count": int(values.size),
"min": float(np.min(values)),
"p05": float(np.percentile(values, 5)),
"p50": float(np.percentile(values, 50)),
"p95": float(np.percentile(values, 95)),
"max": float(np.max(values)),
"mean": float(np.mean(values)),
}
def _array(
rows: Iterable[dict[str, Any]],
key: str,
) -> npt.NDArray[np.float64]:
return np.asarray([float(row[key]) for row in rows], dtype=np.float64)
def _bin_label(value: float, edges: tuple[float, float]) -> str:
if value < edges[0]:
return "low"
if value < edges[1]:
return "middle"
return "high"
def _artifact(path: Path, role: str) -> dict[str, object]:
return {
"path": path.name,
"role": role,
"byte_length": path.stat().st_size,
"sha256": _sha256(path),
}
def _integer(value: object, label: str) -> int:
if not isinstance(value, int) or isinstance(value, bool):
raise E45BindingSensitivityError(f"{label} must be an integer")
return value
def _number(value: object, label: str) -> float:
if (
not isinstance(value, (int, float))
or isinstance(value, bool)
or not math.isfinite(float(value))
):
raise E45BindingSensitivityError(f"{label} must be finite")
return float(value)
def _object(value: object, label: str) -> dict[str, Any]:
if not isinstance(value, dict):
raise E45BindingSensitivityError(f"{label} must be an object")
return value
def _read_json(path: Path) -> dict[str, Any]:
value = json.loads(path.read_text(encoding="utf-8-sig"))
if not isinstance(value, dict):
raise E45BindingSensitivityError(
f"JSON object expected: {path.name}"
)
return value
def _read_jsonl(path: Path) -> Iterable[dict[str, Any]]:
with path.open("r", encoding="utf-8-sig") as stream:
for line_number, line in enumerate(stream, start=1):
value = json.loads(line)
if not isinstance(value, dict):
raise E45BindingSensitivityError(
f"JSON object expected at {path.name}:{line_number}"
)
yield value
def _write_json(path: Path, value: object) -> None:
with path.open("x", encoding="utf-8", newline="\n") as stream:
json.dump(value, stream, indent=2, sort_keys=True)
stream.write("\n")
stream.flush()
os.fsync(stream.fileno())
def _write_jsonl(
path: Path,
rows: Iterable[dict[str, Any]],
) -> None:
with path.open("x", encoding="utf-8", newline="\n") as stream:
for row in rows:
stream.write(
json.dumps(
row,
sort_keys=True,
separators=(",", ":"),
allow_nan=False,
)
)
stream.write("\n")
stream.flush()
os.fsync(stream.fileno())
def _canonical_json(value: object) -> bytes:
return json.dumps(
value,
sort_keys=True,
separators=(",", ":"),
allow_nan=False,
).encode()
def _sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
while chunk := stream.read(1024 * 1024):
digest.update(chunk)
return digest.hexdigest()
def _utc_now() -> str:
return datetime.now(UTC).isoformat(timespec="milliseconds").replace(
"+00:00",
"Z",
)
src/k1link/compute/e46_detector_truth_island.py
+633
View File
@@ -0,0 +1,633 @@
"""Prepare a prelabel-free detector Truth Island from immutable E2 images."""
from __future__ import annotations
import hashlib
import json
import math
import os
import shutil
import uuid
from collections import defaultdict
from collections.abc import Iterable
from dataclasses import asdict, dataclass
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Final
E46_RESULT_SCHEMA: Final = "missioncore.e46-detector-truth-island/v1"
E46_REPORT_SCHEMA: Final = "missioncore.e46-truth-island-preparation-report/v1"
E46_REFERENCE_SCHEMA: Final = "missioncore.e46-truth-island-image-reference/v1"
E46_REVIEW_SCHEMA: Final = "missioncore.e46-detector-review-template/v1"
E46_CONTRACT_SCHEMA: Final = "missioncore.e46-detector-blind-contract/v1"
E46_PROFILE_SCHEMA: Final = "missioncore.e46-truth-island-profile/v1"
E46_MANIFEST_NAME: Final = "manifest.json"
E46_REPORT_NAME: Final = "preparation-report.json"
E46_REFERENCES_NAME: Final = "image-references.jsonl"
E46_REVIEW_NAME: Final = "review-template.json"
E46_CONTRACT_NAME: Final = "blind-contract.json"
_E2_SCHEMA: Final = "missioncore.perception-evaluation-pack/v1"
_E2_IDENTITY_SCHEMA: Final = "missioncore.perception-evaluation-pack-identity/v1"
_AUTHORITY: Final = {
"commands_enabled": False,
"navigation_or_safety_accepted": False,
}
class E46DetectorTruthIslandError(RuntimeError):
"""An E46 source, selection, or blind-review package is invalid."""
@dataclass(frozen=True, slots=True)
class E46DetectorTruthIslandProfile:
profile_id: str = "e46-ravnoves00-detector-truth-island/v1"
anchor_time_bins: int = 8
anchors_per_bin: int = 2
include_all_temporal_groups: bool = True
independent_reviewers_required: int = 2
selection_seed: str = "missioncore-e46-detector-truth-island-20260729"
def __post_init__(self) -> None:
if (
not self.profile_id.strip()
or not 2 <= self.anchor_time_bins <= 32
or not 1 <= self.anchors_per_bin <= 8
or self.include_all_temporal_groups is not True
or self.independent_reviewers_required != 2
or not 16 <= len(self.selection_seed) <= 160
):
raise E46DetectorTruthIslandError("E46 profile is invalid")
def to_dict(self) -> dict[str, object]:
return {
"schema_version": E46_PROFILE_SCHEMA,
**asdict(self),
"prelabels_allowed_in_reviewer_package": False,
"predictions_visible_during_review": False,
"labels_revealed_before_prediction_freeze": False,
"selection_uses_model_output": False,
}
DEFAULT_E46_DETECTOR_TRUTH_ISLAND_PROFILE: Final = (
E46DetectorTruthIslandProfile()
)
@dataclass(frozen=True, slots=True)
class E46DetectorTruthIsland:
result_id: str
result_root: Path
manifest: dict[str, Any]
report: dict[str, Any]
def build_e46_detector_truth_island(
*,
evaluation_pack_root: Path,
output_root: Path,
profile: E46DetectorTruthIslandProfile = (
DEFAULT_E46_DETECTOR_TRUTH_ISLAND_PROFILE
),
) -> E46DetectorTruthIsland:
"""Create a references-only blind review package with no model payload."""
source_root = evaluation_pack_root.resolve(strict=True)
manifest = _read_json(source_root / "manifest.json")
identity = _object(manifest.get("identity"), "E2 identity")
identity_sha256 = manifest.get("identity_sha256")
if (
manifest.get("schema_version") != _E2_SCHEMA
or identity.get("schema_version") != _E2_IDENTITY_SCHEMA
or manifest.get("generation_id") != source_root.name
or not isinstance(identity_sha256, str)
or source_root.name != f"evaluation-pack-{identity_sha256}"
or hashlib.sha256(_canonical_json(identity)).hexdigest()
!= identity_sha256
or identity.get("preprocessing_profile") != "fixed-valid-fov-fill/v1"
or identity.get("resolution") != [800, 600]
):
raise E46DetectorTruthIslandError("E2 evaluation pack is invalid")
frames = identity.get("frames")
artifacts = manifest.get("artifacts")
if not isinstance(frames, list) or not isinstance(artifacts, list):
raise E46DetectorTruthIslandError("E2 frame catalog is invalid")
artifact_by_path = {
str(item.get("path")): item
for item in artifacts
if isinstance(item, dict) and isinstance(item.get("path"), str)
}
selected = select_truth_island_frames(frames, profile=profile)
references: list[dict[str, Any]] = []
for sequence, frame in enumerate(selected, start=1):
image_id = _integer(frame.get("image_id"), "E2 image id")
frame_index = _integer(frame.get("frame_index"), "E2 frame index")
relative = (
"images/valid-fov-fill/"
f"image-{image_id:03d}-frame-{frame_index:06d}.png"
)
artifact = artifact_by_path.get(relative)
path = source_root / relative
if (
artifact is None
or not path.is_file()
or artifact.get("byte_length") != path.stat().st_size
or artifact.get("sha256") != _sha256(path)
):
raise E46DetectorTruthIslandError(
"selected E2 image content changed"
)
references.append(
{
"schema_version": E46_REFERENCE_SCHEMA,
"truth_island_sequence": sequence,
"image_id": image_id,
"frame_index": frame_index,
"source_sequence": _integer(
frame.get("sequence"),
"E2 source sequence",
),
"session_seconds": _number(
frame.get("session_seconds"),
"E2 session time",
),
"role": str(frame.get("role")),
"group_id": str(frame.get("group_id")),
"source_path": relative,
"byte_length": int(artifact["byte_length"]),
"sha256": str(artifact["sha256"]),
"pixel_sha256": str(
frame.get("valid_fov_fill_rgb_sha256", "")
),
}
)
references_sha256 = hashlib.sha256(
b"".join(_canonical_json(row) + b"\n" for row in references)
).hexdigest()
profile_document = profile.to_dict()
package_identity = {
"schema_version": E46_RESULT_SCHEMA,
"source": {
"evaluation_pack_id": source_root.name,
"evaluation_identity_sha256": identity_sha256,
"job_id": identity.get("job_id"),
"input_sha256": identity.get("input_sha256"),
"session_id": identity.get("session_id"),
"source_id": identity.get("source_id"),
"calibration_sha256": identity.get("calibration_sha256"),
"calibration_slot": identity.get("calibration_slot"),
"preprocessing_profile": identity.get("preprocessing_profile"),
},
"profile": profile_document,
"references_sha256": references_sha256,
"producer_sha256": _sha256(Path(__file__).resolve(strict=True)),
"authority": _AUTHORITY,
}
identity_digest = hashlib.sha256(
_canonical_json(package_identity)
).hexdigest()
result_id = f"e46-detector-truth-island-{identity_digest}"
destination = output_root.expanduser().absolute() / result_id
if destination.exists():
return read_e46_detector_truth_island(destination)
temporal_groups = sorted(
{
str(row["group_id"])
for row in references
if row["role"] == "temporal"
}
)
contract = {
"schema_version": E46_CONTRACT_SCHEMA,
"task": "task-relevant-2d-object-detection",
"truth_state": "labels-unavailable",
"reviewer_package": {
"source_images": "references-only-to-immutable-e2",
"model_prelabels_included": False,
"model_predictions_included": False,
"model_scores_included": False,
"candidate_identity_included": False,
},
"annotation": {
"classes": [
"person",
"bicycle",
"motorcycle",
"car",
"heavy_vehicle",
"static_obstacle",
"animal",
],
"box_format": "xyxy-pixels-800x600",
"inside_valid_fov_only": True,
"all_identifiable_instances_required": True,
"hard_negative_frame_flag_required": True,
"occluded_and_truncated_flags_required": True,
},
"review": {
"independent_reviewers_required": (
profile.independent_reviewers_required
),
"reviewer_identity_must_differ": True,
"adjudication_required_on_disagreement": True,
"review_order_must_not_reveal_predictions": True,
},
"prediction_freeze": {
"required_before_label_reveal": True,
"candidate_profile_hash_required": True,
"per-image prediction_hash_required": True,
"source-trained_e38_e40_candidates_eligible": False,
},
"metrics_after_truth_seal": [
"coco_ap_50_95",
"ap50",
"ap75",
"ar100",
"per_class_recall",
"person_vehicle_miss_rate",
"false_large_box_rate",
"valid_fov_boundary_leakage",
"temporal_detection_flicker",
],
"authority": _AUTHORITY,
}
review_template = {
"schema_version": E46_REVIEW_SCHEMA,
"truth_island_id": result_id,
"state": "prepared-unreviewed-no-prelabels",
"reviewer_id": None,
"review_round": None,
"images": [
{
"truth_island_sequence": row["truth_island_sequence"],
"image_id": row["image_id"],
"frame_index": row["frame_index"],
"session_seconds": row["session_seconds"],
"role": row["role"],
"group_id": row["group_id"],
"source_path": row["source_path"],
"source_sha256": row["sha256"],
"review_state": "pending",
"hard_negative": None,
"objects": [],
"notes": None,
}
for row in references
],
"acceptance": None,
}
report = {
"schema_version": E46_REPORT_SCHEMA,
"result_id": result_id,
"identity_sha256": identity_digest,
"status": "prepared-awaiting-independent-human-review",
"selection": {
"frame_count": len(references),
"anchor_count": sum(
1 for row in references if row["role"] == "anchor"
),
"temporal_frame_count": sum(
1 for row in references if row["role"] == "temporal"
),
"temporal_group_count": len(temporal_groups),
"temporal_groups": temporal_groups,
"minimum_frame_index": min(
int(row["frame_index"]) for row in references
),
"maximum_frame_index": max(
int(row["frame_index"]) for row in references
),
"source_images_copied": 0,
"source_reference_bytes": sum(
int(row["byte_length"]) for row in references
),
},
"blindness": {
"model_prelabels_included": False,
"predictions_included": False,
"truth_labels_available": False,
"candidate_comparison_authorized": False,
},
"decision": {
"truth_island_preparation_complete": True,
"truth_island_sealed": False,
"human_review_required": True,
"next_gate": (
"complete two independent reviews and adjudication, while "
"freezing E47 candidate predictions before label reveal"
),
},
"limitations": [
(
"the island is an independent blind review generation on the "
"known RAVNOVES00 source, not cross-route truth"
),
(
"E37 reviewed frames are dense across the route; E38-E40 "
"source-trained predictors are therefore explicitly ineligible"
),
(
"no accuracy metric exists until two human reviews are sealed "
"and adjudicated"
),
],
"authority": _AUTHORITY,
}
destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
staging = destination.parent / f".{result_id}.{uuid.uuid4().hex}.tmp"
staging.mkdir(mode=0o700, exist_ok=False)
try:
_write_jsonl(staging / E46_REFERENCES_NAME, references)
_write_json(staging / E46_CONTRACT_NAME, contract)
_write_json(staging / E46_REVIEW_NAME, review_template)
_write_json(staging / E46_REPORT_NAME, report)
output_manifest = {
"schema_version": E46_RESULT_SCHEMA,
"result_id": result_id,
"identity_sha256": identity_digest,
"identity": package_identity,
"created_at_utc": _utc_now(),
"acceptance_state": "prepared-not-truth",
"artifacts": [
_artifact(staging / E46_REPORT_NAME, "preparation-report"),
_artifact(staging / E46_REFERENCES_NAME, "image-references"),
_artifact(staging / E46_CONTRACT_NAME, "blind-contract"),
_artifact(staging / E46_REVIEW_NAME, "review-template"),
],
"authority": _AUTHORITY,
}
_write_json(staging / E46_MANIFEST_NAME, output_manifest)
os.replace(staging, destination)
except BaseException:
shutil.rmtree(staging, ignore_errors=True)
raise
return read_e46_detector_truth_island(destination)
def select_truth_island_frames(
frames: list[object],
*,
profile: E46DetectorTruthIslandProfile = (
DEFAULT_E46_DETECTOR_TRUTH_ISLAND_PROFILE
),
) -> tuple[dict[str, Any], ...]:
"""Select temporal groups whole plus evenly distributed anchor frames."""
normalized = []
previous = -1
image_ids: set[int] = set()
for value in frames:
row = _object(value, "E2 frame")
image_id = _integer(row.get("image_id"), "E2 image id")
frame_index = _integer(row.get("frame_index"), "E2 frame index")
if (
image_id in image_ids
or frame_index <= previous
or row.get("role") not in {"anchor", "temporal"}
or not isinstance(row.get("group_id"), str)
or not str(row["group_id"])
):
raise E46DetectorTruthIslandError("E2 frame ordering is invalid")
image_ids.add(image_id)
previous = frame_index
normalized.append(row)
anchors = [row for row in normalized if row["role"] == "anchor"]
temporal = [row for row in normalized if row["role"] == "temporal"]
if (
len(anchors)
< profile.anchor_time_bins * profile.anchors_per_bin
or not temporal
):
raise E46DetectorTruthIslandError(
"E2 does not cover the frozen E46 selection"
)
bins = _partition(anchors, profile.anchor_time_bins)
selected_anchors = []
for values in bins:
ranked = sorted(
values,
key=lambda row: hashlib.sha256(
(
f"{profile.selection_seed}:"
f"{row['group_id']}:{row['frame_index']}"
).encode()
).hexdigest(),
)
selected_anchors.extend(ranked[: profile.anchors_per_bin])
by_group: dict[str, list[dict[str, Any]]] = defaultdict(list)
for row in temporal:
by_group[str(row["group_id"])].append(row)
for group_rows in by_group.values():
indices = [int(row["frame_index"]) for row in group_rows]
if len(group_rows) < 2 or indices != list(
range(indices[0], indices[0] + len(indices))
):
raise E46DetectorTruthIslandError(
"E2 temporal group is not consecutive"
)
selected = selected_anchors + [
row for group in sorted(by_group) for row in by_group[group]
]
selected.sort(key=lambda row: int(row["frame_index"]))
expected = (
profile.anchor_time_bins * profile.anchors_per_bin + len(temporal)
)
if len(selected) != expected or len(
{int(row["image_id"]) for row in selected}
) != expected:
raise E46DetectorTruthIslandError("E46 selection is inconsistent")
return tuple(selected)
def read_e46_detector_truth_island(root: Path) -> E46DetectorTruthIsland:
"""Read and validate a prepared, explicitly non-truth E46 generation."""
resolved = root.resolve(strict=True)
manifest = _read_json(resolved / E46_MANIFEST_NAME)
identity = _object(manifest.get("identity"), "E46 identity")
identity_sha256 = manifest.get("identity_sha256")
if (
manifest.get("schema_version") != E46_RESULT_SCHEMA
or not isinstance(identity_sha256, str)
or hashlib.sha256(_canonical_json(identity)).hexdigest()
!= identity_sha256
or manifest.get("result_id")
!= f"e46-detector-truth-island-{identity_sha256}"
or resolved.name != manifest.get("result_id")
or manifest.get("acceptance_state") != "prepared-not-truth"
or manifest.get("authority") != _AUTHORITY
):
raise E46DetectorTruthIslandError("E46 identity is invalid")
artifacts = manifest.get("artifacts")
if not isinstance(artifacts, list) or len(artifacts) != 4:
raise E46DetectorTruthIslandError("E46 artifacts are invalid")
for item in artifacts:
if not isinstance(item, dict) or not isinstance(item.get("path"), str):
raise E46DetectorTruthIslandError("E46 artifact row is invalid")
path = resolved / str(item["path"])
if (
not path.is_file()
or item.get("byte_length") != path.stat().st_size
or item.get("sha256") != _sha256(path)
):
raise E46DetectorTruthIslandError("E46 artifact changed")
report = _read_json(resolved / E46_REPORT_NAME)
contract = _read_json(resolved / E46_CONTRACT_NAME)
review = _read_json(resolved / E46_REVIEW_NAME)
references = tuple(_read_jsonl(resolved / E46_REFERENCES_NAME))
blindness = _object(report.get("blindness"), "E46 blindness")
reviewer_package = _object(
contract.get("reviewer_package"),
"E46 reviewer package",
)
if (
report.get("schema_version") != E46_REPORT_SCHEMA
or report.get("result_id") != resolved.name
or report.get("identity_sha256") != identity_sha256
or contract.get("schema_version") != E46_CONTRACT_SCHEMA
or contract.get("truth_state") != "labels-unavailable"
or review.get("schema_version") != E46_REVIEW_SCHEMA
or review.get("state") != "prepared-unreviewed-no-prelabels"
or blindness
!= {
"candidate_comparison_authorized": False,
"model_prelabels_included": False,
"predictions_included": False,
"truth_labels_available": False,
}
or reviewer_package.get("model_prelabels_included") is not False
or reviewer_package.get("model_predictions_included") is not False
or any(row.get("schema_version") != E46_REFERENCE_SCHEMA for row in references)
or len(references) != report.get("selection", {}).get("frame_count")
or any(
image.get("objects") != []
or image.get("hard_negative") is not None
or image.get("review_state") != "pending"
for image in review.get("images", [])
if isinstance(image, dict)
)
):
raise E46DetectorTruthIslandError("E46 blind package is invalid")
return E46DetectorTruthIsland(
result_id=resolved.name,
result_root=resolved,
manifest=manifest,
report=report,
)
def _partition(
rows: list[dict[str, Any]],
count: int,
) -> tuple[list[dict[str, Any]], ...]:
quotient, remainder = divmod(len(rows), count)
result = []
start = 0
for index in range(count):
size = quotient + (1 if index < remainder else 0)
result.append(rows[start : start + size])
start += size
return tuple(result)
def _artifact(path: Path, role: str) -> dict[str, object]:
return {
"path": path.name,
"role": role,
"byte_length": path.stat().st_size,
"sha256": _sha256(path),
}
def _integer(value: object, label: str) -> int:
if not isinstance(value, int) or isinstance(value, bool):
raise E46DetectorTruthIslandError(f"{label} must be an integer")
return value
def _number(value: object, label: str) -> float:
if not isinstance(value, (int, float)) or isinstance(value, bool):
raise E46DetectorTruthIslandError(f"{label} must be numeric")
number = float(value)
if not math.isfinite(number):
raise E46DetectorTruthIslandError(f"{label} must be finite")
return number
def _object(value: object, label: str) -> dict[str, Any]:
if not isinstance(value, dict):
raise E46DetectorTruthIslandError(f"{label} must be an object")
return value
def _read_json(path: Path) -> dict[str, Any]:
value = json.loads(path.read_text(encoding="utf-8-sig"))
if not isinstance(value, dict):
raise E46DetectorTruthIslandError(
f"JSON object expected: {path.name}"
)
return value
def _read_jsonl(path: Path) -> Iterable[dict[str, Any]]:
with path.open("r", encoding="utf-8-sig") as stream:
for line_number, line in enumerate(stream, start=1):
value = json.loads(line)
if not isinstance(value, dict):
raise E46DetectorTruthIslandError(
f"JSON object expected at {path.name}:{line_number}"
)
yield value
def _write_json(path: Path, value: object) -> None:
with path.open("x", encoding="utf-8", newline="\n") as stream:
json.dump(value, stream, indent=2, sort_keys=True)
stream.write("\n")
stream.flush()
os.fsync(stream.fileno())
def _write_jsonl(path: Path, rows: Iterable[dict[str, Any]]) -> None:
with path.open("x", encoding="utf-8", newline="\n") as stream:
for row in rows:
stream.write(
json.dumps(
row,
sort_keys=True,
separators=(",", ":"),
allow_nan=False,
)
)
stream.write("\n")
stream.flush()
os.fsync(stream.fileno())
def _canonical_json(value: object) -> bytes:
return json.dumps(
value,
sort_keys=True,
separators=(",", ":"),
allow_nan=False,
).encode()
def _sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
while chunk := stream.read(1024 * 1024):
digest.update(chunk)
return digest.hexdigest()
def _utc_now() -> str:
return datetime.now(UTC).isoformat(timespec="milliseconds").replace(
"+00:00",
"Z",
)
src/k1link/compute/e47_detector_candidate_freeze.py
+740
View File
@@ -0,0 +1,740 @@
"""Freeze detector candidates before E46 Truth Island label reveal."""
from __future__ import annotations
import hashlib
import json
import math
import os
import shutil
import uuid
from collections import Counter
from collections.abc import Iterable
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Final
from .e46_detector_truth_island import (
E46_REFERENCES_NAME,
E46DetectorTruthIslandError,
read_e46_detector_truth_island,
)
E47_RESULT_SCHEMA: Final = "missioncore.e47-detector-candidate-freeze/v1"
E47_REPORT_SCHEMA: Final = "missioncore.e47-detector-candidate-report/v1"
E47_PREDICTION_SCHEMA: Final = "missioncore.e47-detector-prediction-row/v1"
E47_MANIFEST_NAME: Final = "manifest.json"
E47_REPORT_NAME: Final = "candidate-freeze-report.json"
E47_PREDICTIONS_NAME: Final = "candidate-predictions.jsonl"
_RAW_RESULT_SCHEMA: Final = "missioncore.recorded-perception-result/v2"
_FILL_RESULT_SCHEMA: Final = "missioncore.perception-evaluation-prelabels/v1"
_RAW_CANDIDATE: Final = "maskrcnn-kb4-raw"
_FILL_CANDIDATE: Final = "maskrcnn-kb4-valid-fov-fill"
_AUTHORITY: Final = {
"commands_enabled": False,
"navigation_or_safety_accepted": False,
}
class E47DetectorCandidateFreezeError(RuntimeError):
"""An E47 candidate input or frozen prediction set is invalid."""
def build_e47_detector_candidate_freeze(
*,
truth_island_root: Path,
raw_result_root: Path,
valid_fov_result_root: Path,
output_root: Path,
) -> dict[str, Any]:
"""Freeze two exact-checkpoint preprocessing candidates without truth."""
try:
truth_island = read_e46_detector_truth_island(truth_island_root)
except E46DetectorTruthIslandError as reason:
raise E47DetectorCandidateFreezeError(
"E46 truth island is invalid"
) from reason
if (
truth_island.report.get("status")
!= "prepared-awaiting-independent-human-review"
or truth_island.report.get("blindness", {}).get(
"truth_labels_available"
)
is not False
):
raise E47DetectorCandidateFreezeError(
"E46 truth state is incompatible"
)
references = tuple(
_read_jsonl(
truth_island.result_root / E46_REFERENCES_NAME
)
)
selected_frames = {
_integer(row.get("frame_index"), "E46 frame index"): row
for row in references
}
selected_images = {
_integer(row.get("image_id"), "E46 image id"): row
for row in references
}
if len(selected_frames) != len(references) or len(selected_images) != len(
references
):
raise E47DetectorCandidateFreezeError(
"E46 selected identity is duplicated"
)
raw_root = raw_result_root.resolve(strict=True)
raw_result = _read_json(raw_root / "result.json")
raw_identity = _object(raw_result.get("identity"), "raw identity")
raw_frames_path = _verified_result_artifact(
root=raw_root,
result=raw_result,
expected_schema=_RAW_RESULT_SCHEMA,
artifact_path="frames.jsonl",
)
fill_root = valid_fov_result_root.resolve(strict=True)
fill_result = _read_json(fill_root / "result.json")
fill_identity = _object(fill_result.get("identity"), "fill identity")
fill_frames_path = _verified_result_artifact(
root=fill_root,
result=fill_result,
expected_schema=_FILL_RESULT_SCHEMA,
artifact_path="frames.jsonl",
)
truth_source = _object(
truth_island.manifest["identity"].get("source"),
"E46 source identity",
)
if (
raw_identity.get("input_sha256")
!= truth_source.get("input_sha256")
or raw_identity.get("job_id") != truth_source.get("job_id")
):
raise E47DetectorCandidateFreezeError(
"raw candidate source changed"
)
if (
fill_identity.get("evaluation_pack_id")
!= truth_source.get("evaluation_pack_id")
or fill_identity.get("evaluation_identity_sha256")
!= truth_source.get("evaluation_identity_sha256")
):
raise E47DetectorCandidateFreezeError(
"valid-FOV candidate source changed"
)
raw_weight = _instance_weight_sha256(raw_identity)
fill_weight = _instance_weight_sha256(fill_identity)
if raw_weight != fill_weight:
raise E47DetectorCandidateFreezeError(
"candidate checkpoint identity differs"
)
instance_mapping = _object(
fill_identity.get("instance_mapping"),
"candidate instance mapping",
)
target_categories = _target_categories(fill_identity)
raw_source = {
_integer(row.get("frame_index"), "raw frame index"): row
for row in _read_jsonl(raw_frames_path)
if row.get("frame_index") in selected_frames
}
fill_source = {
_integer(row.get("image_id"), "fill image id"): row
for row in _read_jsonl(fill_frames_path)
if row.get("image_id") in selected_images
}
if (
set(raw_source) != set(selected_frames)
or set(fill_source) != set(selected_images)
):
raise E47DetectorCandidateFreezeError(
"candidate frame coverage is incomplete"
)
rows: list[dict[str, Any]] = []
for reference in references:
frame_index = int(reference["frame_index"])
image_id = int(reference["image_id"])
rows.append(
_normalize_raw_row(
source=raw_source[frame_index],
reference=reference,
instance_mapping=instance_mapping,
target_categories=target_categories,
)
)
rows.append(
_normalize_fill_row(
source=fill_source[image_id],
reference=reference,
target_categories=target_categories,
)
)
rows.sort(
key=lambda row: (
int(row["truth_island_sequence"]),
str(row["candidate_id"]),
)
)
candidate_summary = {
candidate_id: _candidate_summary(
tuple(row for row in rows if row["candidate_id"] == candidate_id)
)
for candidate_id in (_RAW_CANDIDATE, _FILL_CANDIDATE)
}
predictions_sha256 = hashlib.sha256(
b"".join(_canonical_json(row) + b"\n" for row in rows)
).hexdigest()
identity = {
"schema_version": E47_RESULT_SCHEMA,
"truth_island": {
"result_id": truth_island.result_id,
"state": "prepared-unreviewed-no-prelabels",
"truth_labels_available": False,
},
"candidates": [
{
"candidate_id": _RAW_CANDIDATE,
"preprocessing": "raw-kb4-800x600",
"source_result_id": raw_result.get("result_id"),
"source_result_sha256": _sha256(raw_root / "result.json"),
"source_frames_sha256": _sha256(raw_frames_path),
"checkpoint_sha256": raw_weight,
},
{
"candidate_id": _FILL_CANDIDATE,
"preprocessing": "fixed-valid-fov-fill-800x600",
"source_result_id": fill_root.name,
"source_result_sha256": _sha256(fill_root / "result.json"),
"source_frames_sha256": _sha256(fill_frames_path),
"checkpoint_sha256": fill_weight,
},
],
"prediction_rows_sha256": predictions_sha256,
"producer_sha256": _sha256(Path(__file__).resolve(strict=True)),
"authority": _AUTHORITY,
}
identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest()
result_id = f"e47-detector-candidate-freeze-{identity_sha256}"
destination = output_root.expanduser().absolute() / result_id
if destination.exists():
return read_e47_detector_candidate_freeze(destination)
report = {
"schema_version": E47_REPORT_SCHEMA,
"result_id": result_id,
"identity_sha256": identity_sha256,
"status": "predictions-frozen-awaiting-truth-reveal",
"frame_count": len(references),
"prediction_row_count": len(rows),
"checkpoint_relation": "same-exact-maskrcnn-checkpoint",
"comparison_variable": "raw-kb4-vs-fixed-valid-fov-fill",
"candidates": candidate_summary,
"blindness": {
"truth_labels_available": False,
"truth_join_performed": False,
"accuracy_metrics_available": False,
"reviewer_package_modified": False,
},
"decision": {
"candidate_predictions_frozen": True,
"candidate_winner_selected": False,
"model_retraining_authorized": False,
"next_gate": (
"seal E46 independent reviews, reveal truth only after this "
"prediction generation, then compute detection metrics"
),
},
"limitations": [
(
"truth-free prediction counts are descriptive and cannot rank "
"candidate accuracy"
),
(
"both candidates use the same generic COCO Mask R-CNN weights; "
"E47 currently isolates only calibrated valid-FOV preprocessing"
),
(
"the comparison remains source-scoped to the known "
"RAVNOVES00 camera"
),
],
"authority": _AUTHORITY,
}
destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
staging = destination.parent / f".{result_id}.{uuid.uuid4().hex}.tmp"
staging.mkdir(mode=0o700, exist_ok=False)
try:
_write_jsonl(staging / E47_PREDICTIONS_NAME, rows)
_write_json(staging / E47_REPORT_NAME, report)
manifest = {
"schema_version": E47_RESULT_SCHEMA,
"result_id": result_id,
"identity_sha256": identity_sha256,
"identity": identity,
"created_at_utc": _utc_now(),
"acceptance_state": "accepted-prediction-freeze-only",
"artifacts": [
_artifact(staging / E47_REPORT_NAME, "candidate-report"),
_artifact(
staging / E47_PREDICTIONS_NAME,
"candidate-predictions",
),
],
"authority": _AUTHORITY,
}
_write_json(staging / E47_MANIFEST_NAME, manifest)
os.replace(staging, destination)
except BaseException:
shutil.rmtree(staging, ignore_errors=True)
raise
return read_e47_detector_candidate_freeze(destination)
def read_e47_detector_candidate_freeze(root: Path) -> dict[str, Any]:
"""Read and validate an E47 prediction-only generation."""
resolved = root.resolve(strict=True)
manifest = _read_json(resolved / E47_MANIFEST_NAME)
identity = _object(manifest.get("identity"), "E47 identity")
identity_sha256 = manifest.get("identity_sha256")
if (
manifest.get("schema_version") != E47_RESULT_SCHEMA
or not isinstance(identity_sha256, str)
or hashlib.sha256(_canonical_json(identity)).hexdigest()
!= identity_sha256
or manifest.get("result_id")
!= f"e47-detector-candidate-freeze-{identity_sha256}"
or resolved.name != manifest.get("result_id")
or manifest.get("acceptance_state")
!= "accepted-prediction-freeze-only"
or manifest.get("authority") != _AUTHORITY
):
raise E47DetectorCandidateFreezeError("E47 identity is invalid")
artifacts = manifest.get("artifacts")
if not isinstance(artifacts, list) or len(artifacts) != 2:
raise E47DetectorCandidateFreezeError("E47 artifacts are invalid")
for item in artifacts:
if not isinstance(item, dict) or not isinstance(item.get("path"), str):
raise E47DetectorCandidateFreezeError("E47 artifact row is invalid")
path = resolved / str(item["path"])
if (
not path.is_file()
or item.get("byte_length") != path.stat().st_size
or item.get("sha256") != _sha256(path)
):
raise E47DetectorCandidateFreezeError("E47 artifact changed")
report = _read_json(resolved / E47_REPORT_NAME)
rows = tuple(_read_jsonl(resolved / E47_PREDICTIONS_NAME))
if (
report.get("schema_version") != E47_REPORT_SCHEMA
or report.get("result_id") != resolved.name
or report.get("identity_sha256") != identity_sha256
or report.get("status")
!= "predictions-frozen-awaiting-truth-reveal"
or report.get("blindness")
!= {
"accuracy_metrics_available": False,
"reviewer_package_modified": False,
"truth_join_performed": False,
"truth_labels_available": False,
}
or len(rows) != report.get("prediction_row_count")
or any(row.get("schema_version") != E47_PREDICTION_SCHEMA for row in rows)
or hashlib.sha256(
b"".join(_canonical_json(row) + b"\n" for row in rows)
).hexdigest()
!= identity.get("prediction_rows_sha256")
):
raise E47DetectorCandidateFreezeError("E47 report is invalid")
return {
"result_id": resolved.name,
"result_root": resolved,
"manifest": manifest,
"report": report,
}
def normalize_candidate_predictions(
*,
instances: object,
label_mapping: dict[str, int],
target_categories: dict[int, str],
source_kind: str,
) -> tuple[dict[str, object], ...]:
"""Normalize raw or E2 draft instances into one detection contract."""
if not isinstance(instances, list):
raise E47DetectorCandidateFreezeError(
"candidate instances must be a list"
)
normalized: list[dict[str, Any]] = []
for instance in instances:
item = _object(instance, "candidate instance")
if source_kind == "raw":
source_label = str(item.get("label", ""))
category_id = label_mapping.get(source_label)
score = _number(item.get("score"), "candidate score")
bbox = item.get("box_xyxy")
elif source_kind == "fill":
category_id = _integer(
item.get("draft_category_id"),
"candidate category",
)
source_label = str(item.get("source_model_category", ""))
score = _number(item.get("score"), "candidate score")
bbox = item.get("box_xyxy")
else:
raise E47DetectorCandidateFreezeError(
"candidate source kind is invalid"
)
if category_id is None:
continue
target_label = target_categories.get(category_id)
if target_label is None:
raise E47DetectorCandidateFreezeError(
"candidate category is outside the target ontology"
)
if (
not isinstance(bbox, list)
or len(bbox) != 4
or not all(
isinstance(value, (int, float))
and not isinstance(value, bool)
and math.isfinite(float(value))
for value in bbox
)
or float(bbox[2]) <= float(bbox[0])
or float(bbox[3]) <= float(bbox[1])
or not 0.0 <= score <= 1.0
):
raise E47DetectorCandidateFreezeError(
"candidate box is invalid"
)
normalized.append(
{
"category_id": category_id,
"category": target_label,
"source_category": source_label,
"score": score,
"box_xyxy": [float(value) for value in bbox],
}
)
normalized.sort(
key=lambda row: (
-float(row["score"]),
int(row["category_id"]),
tuple(float(value) for value in row["box_xyxy"]),
)
)
return tuple(normalized)
def _normalize_raw_row(
*,
source: dict[str, Any],
reference: dict[str, Any],
instance_mapping: dict[str, Any],
target_categories: dict[int, str],
) -> dict[str, Any]:
mapping = {
str(label): _integer(category, "raw category mapping")
for label, category in instance_mapping.items()
}
predictions = normalize_candidate_predictions(
instances=source.get("instances"),
label_mapping=mapping,
target_categories=target_categories,
source_kind="raw",
)
source_instances = source.get("instances")
if not isinstance(source_instances, list):
raise E47DetectorCandidateFreezeError(
"raw source instances are invalid"
)
return _prediction_row(
candidate_id=_RAW_CANDIDATE,
reference=reference,
source_instance_count=len(source_instances),
predictions=predictions,
)
def _normalize_fill_row(
*,
source: dict[str, Any],
reference: dict[str, Any],
target_categories: dict[int, str],
) -> dict[str, Any]:
predictions = normalize_candidate_predictions(
instances=source.get("instances"),
label_mapping={},
target_categories=target_categories,
source_kind="fill",
)
source_instances = source.get("instances")
if not isinstance(source_instances, list):
raise E47DetectorCandidateFreezeError(
"fill source instances are invalid"
)
return _prediction_row(
candidate_id=_FILL_CANDIDATE,
reference=reference,
source_instance_count=len(source_instances),
predictions=predictions,
)
def _prediction_row(
*,
candidate_id: str,
reference: dict[str, Any],
source_instance_count: int,
predictions: tuple[dict[str, object], ...],
) -> dict[str, Any]:
return {
"schema_version": E47_PREDICTION_SCHEMA,
"candidate_id": candidate_id,
"truth_island_sequence": int(reference["truth_island_sequence"]),
"image_id": int(reference["image_id"]),
"frame_index": int(reference["frame_index"]),
"session_seconds": float(reference["session_seconds"]),
"source_image_sha256": str(reference["sha256"]),
"source_instance_count": source_instance_count,
"admitted_prediction_count": len(predictions),
"ignored_source_instance_count": source_instance_count - len(predictions),
"predictions": list(predictions),
"truth_joined": False,
}
def _candidate_summary(rows: tuple[dict[str, Any], ...]) -> dict[str, Any]:
if not rows:
raise E47DetectorCandidateFreezeError(
"candidate prediction rows are empty"
)
class_counts: Counter[str] = Counter()
total_predictions = 0
source_instances = 0
ignored = 0
per_frame = []
for row in rows:
predictions = row.get("predictions")
if not isinstance(predictions, list):
raise E47DetectorCandidateFreezeError(
"candidate predictions are invalid"
)
total_predictions += len(predictions)
source_instances += int(row["source_instance_count"])
ignored += int(row["ignored_source_instance_count"])
per_frame.append(len(predictions))
for item in predictions:
if isinstance(item, dict):
class_counts[str(item.get("category"))] += 1
return {
"frame_count": len(rows),
"source_instance_count": source_instances,
"admitted_prediction_count": total_predictions,
"ignored_source_instance_count": ignored,
"frames_without_predictions": sum(1 for value in per_frame if value == 0),
"predictions_per_frame": {
"min": min(per_frame),
"p50": _percentile(per_frame, 50),
"p95": _percentile(per_frame, 95),
"max": max(per_frame),
"mean": float(sum(per_frame) / len(per_frame)),
},
"class_counts": dict(sorted(class_counts.items())),
"accuracy_metrics_available": False,
}
def _verified_result_artifact(
*,
root: Path,
result: dict[str, Any],
expected_schema: str,
artifact_path: str,
) -> Path:
if result.get("schema_version") != expected_schema:
raise E47DetectorCandidateFreezeError(
"candidate result schema changed"
)
artifacts = result.get("artifacts")
if not isinstance(artifacts, list):
raise E47DetectorCandidateFreezeError(
"candidate artifact catalog is invalid"
)
artifact = next(
(
item
for item in artifacts
if isinstance(item, dict) and item.get("path") == artifact_path
),
None,
)
path = root / artifact_path
if (
artifact is None
or not path.is_file()
or artifact.get("byte_length") != path.stat().st_size
or artifact.get("sha256") != _sha256(path)
):
raise E47DetectorCandidateFreezeError(
"candidate prediction artifact changed"
)
return path
def _instance_weight_sha256(identity: dict[str, Any]) -> str:
models = _object(identity.get("models"), "candidate models")
files = models.get("files")
if not isinstance(files, list):
raise E47DetectorCandidateFreezeError(
"candidate model files are invalid"
)
matches = [
str(item.get("sha256"))
for item in files
if isinstance(item, dict)
and "maskrcnn_resnet50_fpn_v2" in str(item.get("name"))
]
if len(matches) != 1 or len(matches[0]) != 64:
raise E47DetectorCandidateFreezeError(
"candidate Mask R-CNN weight identity is invalid"
)
return matches[0]
def _target_categories(identity: dict[str, Any]) -> dict[int, str]:
value = _object(identity.get("target_categories"), "target categories")
result = {}
for category_id, label in value.items():
try:
numeric = int(category_id)
except ValueError as reason:
raise E47DetectorCandidateFreezeError(
"target category id is invalid"
) from reason
if not isinstance(label, str) or not label:
raise E47DetectorCandidateFreezeError(
"target category label is invalid"
)
result[numeric] = label
return result
def _percentile(values: list[int], percentile: int) -> float:
ordered = sorted(values)
position = (len(ordered) - 1) * percentile / 100.0
lower = int(math.floor(position))
upper = int(math.ceil(position))
if lower == upper:
return float(ordered[lower])
fraction = position - lower
return float(
ordered[lower] * (1.0 - fraction) + ordered[upper] * fraction
)
def _artifact(path: Path, role: str) -> dict[str, object]:
return {
"path": path.name,
"role": role,
"byte_length": path.stat().st_size,
"sha256": _sha256(path),
}
def _integer(value: object, label: str) -> int:
if not isinstance(value, int) or isinstance(value, bool):
raise E47DetectorCandidateFreezeError(f"{label} must be an integer")
return value
def _number(value: object, label: str) -> float:
if not isinstance(value, (int, float)) or isinstance(value, bool):
raise E47DetectorCandidateFreezeError(f"{label} must be numeric")
result = float(value)
if not math.isfinite(result):
raise E47DetectorCandidateFreezeError(f"{label} must be finite")
return result
def _object(value: object, label: str) -> dict[str, Any]:
if not isinstance(value, dict):
raise E47DetectorCandidateFreezeError(f"{label} must be an object")
return value
def _read_json(path: Path) -> dict[str, Any]:
value = json.loads(path.read_text(encoding="utf-8-sig"))
if not isinstance(value, dict):
raise E47DetectorCandidateFreezeError(
f"JSON object expected: {path.name}"
)
return value
def _read_jsonl(path: Path) -> Iterable[dict[str, Any]]:
with path.open("r", encoding="utf-8-sig") as stream:
for line_number, line in enumerate(stream, start=1):
value = json.loads(line)
if not isinstance(value, dict):
raise E47DetectorCandidateFreezeError(
f"JSON object expected at {path.name}:{line_number}"
)
yield value
def _write_json(path: Path, value: object) -> None:
with path.open("x", encoding="utf-8", newline="\n") as stream:
json.dump(value, stream, indent=2, sort_keys=True)
stream.write("\n")
stream.flush()
os.fsync(stream.fileno())
def _write_jsonl(path: Path, rows: Iterable[dict[str, Any]]) -> None:
with path.open("x", encoding="utf-8", newline="\n") as stream:
for row in rows:
stream.write(
json.dumps(
row,
sort_keys=True,
separators=(",", ":"),
allow_nan=False,
)
)
stream.write("\n")
stream.flush()
os.fsync(stream.fileno())
def _canonical_json(value: object) -> bytes:
return json.dumps(
value,
sort_keys=True,
separators=(",", ":"),
allow_nan=False,
).encode()
def _sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
while chunk := stream.read(1024 * 1024):
digest.update(chunk)
return digest.hexdigest()
def _utc_now() -> str:
return datetime.now(UTC).isoformat(timespec="milliseconds").replace(
"+00:00",
"Z",
)