compare M4.7 threat parity by component identity

This commit is contained in:
DCCONSTRUCTIONS
2026-08-23 20:39:16 +03:00
parent bcda5d001a
commit 4ac07f7671
2 changed files with 65 additions and 7 deletions
+43 -3
View File
@@ -3,7 +3,7 @@ from __future__ import annotations
import json
from pathlib import Path
from k1link.perception.contracts import LocalObstacleMap, SourceAccounting
from k1link.perception.contracts import LocalObstacleMap, SourceAccounting, ThreatAssessment
from k1link.perception.graph_contracts import (
REFERENCE_GRAPH_ID_V2,
DeliveredFrame,
@@ -18,7 +18,7 @@ from k1link.perception.reference_graph_parity import (
)
def _result():
def _result(threats: tuple[ThreatAssessment, ...] = ()):
obstacle_map = LocalObstacleMap(
source_id="RAVNOVES00",
session_id="20260720T065719Z_viewer_live",
@@ -48,7 +48,7 @@ def _result():
reason="object-payload-delivered",
),
),
deliveries=(DeliveredFrame(0, obstacle_map, ()),),
deliveries=(DeliveredFrame(0, obstacle_map, threats),),
queue_high_watermarks=(("detector", 1),),
)
@@ -119,3 +119,43 @@ def test_reference_graph_parity_reports_threat_drift(tmp_path: Path) -> None:
assert report.accepted is False
assert dict(report.mismatch_counts)["threat_assessments"] == 1
def test_reference_graph_parity_treats_assessment_order_as_nonsemantic(
tmp_path: Path,
) -> None:
first = {
"schema_version": "missioncore.threat-assessment/v1",
"assessment_id": "threat-first",
"component_id": "component-first",
"rig_profile_id": "rig-test",
"corridor_profile_id": "corridor-test",
"qualification": "unqualified",
"relative_speed_mps": None,
"closest_approach_m": None,
"ttc_seconds": None,
"corridor_intersection": "unknown",
"decision": "unknown",
"reason_codes": ["test-unavailable"],
"authority": "replay-simulated",
"physical_collision_accepted": False,
"actuation_allowed": False,
}
second = {
**first,
"assessment_id": "threat-second",
"component_id": "component-second",
}
temporal, threat = _write_ledgers(
tmp_path,
threat_assessments=[second, first],
)
report = compare_reference_graph_to_accepted_ledgers(
_result((ThreatAssessment.from_dict(first), ThreatAssessment.from_dict(second))),
temporal_frames_path=temporal,
threat_frames_path=threat,
expected_frames=1,
)
assert report.accepted is True