230 lines
8.1 KiB
Python
230 lines
8.1 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import math
|
|
from collections.abc import Iterator
|
|
from pathlib import Path
|
|
from threading import Event
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from numpy.typing import NDArray
|
|
|
|
from k1link.perception.contracts import (
|
|
ClockBasis,
|
|
ModalityOutcome,
|
|
ModalityStatus,
|
|
SourceEnvelope,
|
|
TimestampBundle,
|
|
)
|
|
from k1link.perception.detector import FrozenYoloxDetectorProvider
|
|
from k1link.perception.detector_replay import run_detector_replay
|
|
from k1link.perception.detector_replay_cli import _loopback_triton_origin
|
|
from k1link.perception.detector_replay_result import (
|
|
DetectorReplayGate,
|
|
DetectorReplayResultError,
|
|
DetectorRuntimeIdentity,
|
|
read_detector_replay_result,
|
|
require_m4_detector_replay_acceptance,
|
|
)
|
|
from k1link.perception.providers import SourcePacket
|
|
from k1link.perception.yolox_object_detector import (
|
|
YOLOX_CONFIG_SHA256,
|
|
YOLOX_MODEL_SHA256,
|
|
YOLOX_VALID_FOV_SHA256,
|
|
)
|
|
|
|
|
|
class _Source:
|
|
provider_id = "ravnoves00-recorded-source/v1"
|
|
|
|
def __init__(self, packets: tuple[SourcePacket, ...]) -> None:
|
|
self._packets = packets
|
|
self.closed = False
|
|
|
|
def packets(self, stop_event: Event) -> Iterator[SourcePacket]:
|
|
try:
|
|
for packet in self._packets:
|
|
if stop_event.is_set():
|
|
return
|
|
yield packet
|
|
finally:
|
|
self.closed = True
|
|
|
|
|
|
class _Resizer:
|
|
def resize(
|
|
self,
|
|
image: NDArray[np.uint8],
|
|
width: int,
|
|
height: int,
|
|
) -> NDArray[np.uint8]:
|
|
assert image.shape == (600, 800, 3)
|
|
return np.zeros((height, width, 3), dtype=np.uint8)
|
|
|
|
|
|
class _Backend:
|
|
def infer(self, tensor: NDArray[np.float32]) -> NDArray[np.float32]:
|
|
assert tensor.shape == (1, 3, 640, 640)
|
|
output = np.zeros((1, 8400, 85), dtype=np.float32)
|
|
output[0, 0, :4] = [40.0, 30.0, math.log(10.0), math.log(10.0)]
|
|
output[0, 0, 4] = 0.9
|
|
output[0, 0, 5] = 0.9
|
|
return output
|
|
|
|
|
|
def _status() -> ModalityStatus:
|
|
return ModalityStatus(True, ModalityOutcome.AVAILABLE, "test-available")
|
|
|
|
|
|
def _packet(sequence: int, image: object) -> SourcePacket:
|
|
return SourcePacket(
|
|
envelope=SourceEnvelope(
|
|
source_id="RAVNOVES00",
|
|
session_id="20260720T065719Z_viewer_live",
|
|
frame_id=f"frame-{sequence:06d}",
|
|
sequence=sequence,
|
|
timestamps=TimestampBundle(
|
|
utc_ns=1_000 + sequence,
|
|
monotonic_ns=2_000 + sequence,
|
|
source_ns=3_000 + sequence,
|
|
clock_basis=ClockBasis.RECORDED_HOST,
|
|
),
|
|
source_age_ns=0,
|
|
binding_reason="test-recorded-source",
|
|
calibration_id="camera-1-kb4-test",
|
|
representation_id="registered-map-increment-v1",
|
|
image=_status(),
|
|
registered_point_increment=_status(),
|
|
pose=_status(),
|
|
),
|
|
image_payload=image,
|
|
registered_point_increment_payload=("points", sequence),
|
|
pose_payload=("pose", sequence),
|
|
)
|
|
|
|
|
|
def _runtime(
|
|
*,
|
|
source_mount_read_only: bool = True,
|
|
public_worker_port_added: bool = False,
|
|
same_host_tensor_transport: bool = True,
|
|
) -> DetectorRuntimeIdentity:
|
|
return DetectorRuntimeIdentity(
|
|
worker_id="worker-006",
|
|
worker_node="DESKTOP-OPJ8J04",
|
|
worker_container_id="1" * 64,
|
|
worker_image_id=f"sha256:{'2' * 64}",
|
|
triton_container_id="3" * 64,
|
|
triton_image_id=f"sha256:{'4' * 64}",
|
|
triton_model_sha256=YOLOX_MODEL_SHA256,
|
|
triton_model_config_sha256=YOLOX_CONFIG_SHA256,
|
|
valid_fov_mask_sha256=YOLOX_VALID_FOV_SHA256,
|
|
artifact_sha256="5" * 64,
|
|
code_revision="6" * 40,
|
|
source_mount_read_only=source_mount_read_only,
|
|
model_service_reused=True,
|
|
public_worker_port_added=public_worker_port_added,
|
|
same_host_tensor_transport=same_host_tensor_transport,
|
|
)
|
|
|
|
|
|
def _provider(clock_values: tuple[int, ...]) -> FrozenYoloxDetectorProvider:
|
|
return FrozenYoloxDetectorProvider(
|
|
mask=np.ones((600, 800), dtype=np.bool_),
|
|
backend=_Backend(),
|
|
resizer=_Resizer(),
|
|
clock_ns=iter(clock_values).__next__,
|
|
)
|
|
|
|
|
|
def test_replay_seals_and_reopens_exact_class_agnostic_capacity_receipt(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
image = np.zeros((600, 800, 3), dtype=np.uint8)
|
|
result = run_detector_replay(
|
|
source=_Source((_packet(0, image), _packet(1, image))),
|
|
provider=_provider((100, 1_000_100, 2_000_100, 3_000_100)),
|
|
runtime=_runtime(),
|
|
output_root=tmp_path,
|
|
gate=DetectorReplayGate(expected_frames=2, minimum_end_to_end_fps=10.004),
|
|
clock_ns=iter((0, 10, 20, 30, 40, 100_000_000)).__next__,
|
|
created_at_utc="2026-08-05T12:00:00.000Z",
|
|
)
|
|
|
|
assert result.accepted is True
|
|
assert result.metrics.frame_count == 2
|
|
assert result.metrics.completed_frame_count == 2
|
|
assert result.metrics.proposal_count == 2
|
|
assert result.metrics.semantic_hint_count == 2
|
|
assert result.metrics.provider_tracklet_count == 0
|
|
assert result.metrics.end_to_end_fps == 20.0
|
|
assert result.metrics.provider_core_fps == 1000.0
|
|
assert all(frame.to_dict()["class_routing_used"] is False for frame in result.frames)
|
|
assert all(frame.proposals[0].provider_tracklet is None for frame in result.frames)
|
|
|
|
reopened = read_detector_replay_result(result.result_root)
|
|
assert reopened.result_id == result.result_id
|
|
assert reopened.receipt["accepted"] is True
|
|
assert reopened.runtime.worker_id == "worker-006"
|
|
with pytest.raises(DetectorReplayResultError, match="does not close"):
|
|
require_m4_detector_replay_acceptance(reopened)
|
|
|
|
|
|
def test_replay_seals_failed_frame_without_fabricating_missing_proposals(tmp_path: Path) -> None:
|
|
image = np.zeros((600, 800, 3), dtype=np.uint8)
|
|
source = _Source((_packet(0, image), _packet(1, "opaque-image")))
|
|
result = run_detector_replay(
|
|
source=source,
|
|
provider=_provider((100, 200, 300, 400)),
|
|
runtime=_runtime(),
|
|
output_root=tmp_path,
|
|
gate=DetectorReplayGate(expected_frames=2, minimum_end_to_end_fps=1.0),
|
|
clock_ns=iter((0, 10, 20, 30, 40, 1_000_000_000)).__next__,
|
|
created_at_utc="2026-08-05T12:01:00.000Z",
|
|
)
|
|
|
|
assert result.accepted is False
|
|
assert result.metrics.completed_frame_count == 1
|
|
assert result.metrics.failed_frame_count == 1
|
|
assert result.frames[1].outcome == "failed"
|
|
assert result.frames[1].failure_code == "DetectorProviderError"
|
|
assert result.frames[1].proposals == ()
|
|
assert source.closed is True
|
|
|
|
|
|
def test_replay_result_detects_artifact_tampering(tmp_path: Path) -> None:
|
|
image = np.zeros((600, 800, 3), dtype=np.uint8)
|
|
result = run_detector_replay(
|
|
source=_Source((_packet(0, image),)),
|
|
provider=_provider((100, 200)),
|
|
runtime=_runtime(),
|
|
output_root=tmp_path,
|
|
gate=DetectorReplayGate(expected_frames=1, minimum_end_to_end_fps=1.0),
|
|
clock_ns=iter((0, 10, 20, 1_000_000)).__next__,
|
|
created_at_utc="2026-08-05T12:02:00.000Z",
|
|
)
|
|
frames_path = result.result_root / "frames.jsonl"
|
|
row = json.loads(frames_path.read_text("utf-8"))
|
|
row["class_routing_used"] = True
|
|
frames_path.write_text(json.dumps(row) + "\n", "utf-8")
|
|
|
|
with pytest.raises(DetectorReplayResultError, match="artifact changed"):
|
|
read_detector_replay_result(result.result_root)
|
|
|
|
|
|
def test_runtime_identity_rejects_noncanonical_worker_topology() -> None:
|
|
with pytest.raises(DetectorReplayResultError, match="read-only"):
|
|
_runtime(source_mount_read_only=False)
|
|
with pytest.raises(DetectorReplayResultError, match="public worker port"):
|
|
_runtime(public_worker_port_added=True)
|
|
with pytest.raises(DetectorReplayResultError, match="worker host"):
|
|
_runtime(same_host_tensor_transport=False)
|
|
|
|
|
|
def test_m4_cli_rejects_remote_triton_tensor_transport() -> None:
|
|
assert _loopback_triton_origin("http://127.0.0.1:8000") == "http://127.0.0.1:8000"
|
|
with pytest.raises(argparse.ArgumentTypeError, match="worker-local loopback"):
|
|
_loopback_triton_origin("http://192.168.68.52:8000")
|