SILVER/NODEDC_MISSION_CORE
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Files
main
NODEDC_MISSION_CORE/tests/test_perception_graph.py
T

1029 lines
34 KiB
Python
Raw Permalink Blame History

from __future__ import annotations
import gc
import json
import subprocess
import sys
import threading
import weakref
from collections.abc import Callable, Iterator
from dataclasses import replace
from pathlib import Path
from queue import Queue
from threading import Event
import numpy as np
import pytest
from k1link.compute.pipeline_telemetry import PipelineTelemetryIdentity
from k1link.perception.baseline import BASELINE_PROFILE_ID
from k1link.perception.contracts import (
BoundingRegion2D,
ClockBasis,
CorridorIntersection,
EvidenceBasis,
EvidenceCurrentness,
GridCell,
HistorySample,
LocalObstacleMap,
MetricGeometry,
ModalityOutcome,
ModalityStatus,
MotionState,
ObjectProposal2D,
ObstacleObservation,
QualificationState,
SourceEnvelope,
TemporalObstacle,
TemporalState,
ThreatAssessment,
ThreatDecision,
TimestampBundle,
)
from k1link.perception.graph import (
DeliveredFrame,
GraphExecutionError,
GraphRunMode,
GraphRunResult,
GraphState,
ReferencePerceptionGraphV1,
ReferencePerceptionGraphV2,
TerminalOutcomeType,
)
from k1link.perception.providers import (
GraphAuthority,
ProviderPin,
ProviderRole,
QueuePolicy,
ReferencePerceptionGraphConfig,
ReferencePerceptionGraphConfigV2,
SourcePacket,
)
from k1link.perception.recorded_source import (
DecodedFrameTiming,
DecodedRecordedSource,
DecodePhase,
PrefetchedRecordedImageDecoder,
RecordedRavnoves00Source,
RecordedSourceError,
ReplayPacing,
SourcePacingTiming,
)
def test_graph_import_does_not_initialize_legacy_compute_dependencies() -> None:
result = subprocess.run(
[
sys.executable,
"-c",
(
"import sys; import k1link.perception.graph; "
"assert 'k1link.compute' not in sys.modules"
),
],
check=False,
capture_output=True,
text=True,
)
assert result.returncode == 0, result.stderr
def _status(available: bool = True) -> ModalityStatus:
return ModalityStatus(
available=available,
outcome=ModalityOutcome.AVAILABLE if available else ModalityOutcome.UNAVAILABLE,
reason="test-available" if available else "test-unavailable",
)
def _packet(
sequence: int,
*,
image: bool = True,
lidar: bool = True,
source_age_ns: int = 0,
) -> SourcePacket:
envelope = SourceEnvelope(
source_id="RAVNOVES00",
session_id="20260720T065719Z_viewer_live",
frame_id=f"frame-{sequence:06d}",
sequence=sequence,
timestamps=TimestampBundle(
utc_ns=1_786_000_000_000_000_000 + sequence,
monotonic_ns=1_000 + sequence,
source_ns=35_000_000_000 + sequence * 100_000_000,
clock_basis=ClockBasis.RECORDED_HOST,
),
source_age_ns=source_age_ns,
binding_reason="test-recorded-source",
calibration_id="camera-1-kb4-test",
representation_id="registered-map-increment-v1",
image=_status(image),
registered_point_increment=_status(lidar),
pose=_status(lidar),
)
return SourcePacket(
envelope=envelope,
image_payload=("image", sequence) if image else None,
registered_point_increment_payload=("points", sequence) if lidar else None,
pose_payload=("pose", sequence) if lidar else None,
)
class _Source:
provider_id = "test-source/v1"
def __init__(self, packets: tuple[SourcePacket, ...]) -> None:
self.values = packets
def packets(self, stop_event: Event) -> Iterator[SourcePacket]:
for packet in self.values:
if stop_event.is_set():
return
yield packet
class _Detector:
provider_id = "test-detector/v1"
def detect(self, packet: SourcePacket) -> tuple[ObjectProposal2D, ...]:
sequence = packet.envelope.sequence
return (
ObjectProposal2D(
proposal_id=f"proposal-{sequence}",
source_id=packet.envelope.source_id,
frame_id=packet.envelope.frame_id,
region=BoundingRegion2D(1.0, 2.0, 20.0, 30.0),
objectness=0.9,
provider_id=self.provider_id,
model_id="objectness-test/v1",
preprocess_id="kb4-test/v1",
),
)
class _Geometry:
provider_id = "test-geometry/v1"
def associate(
self,
packet: SourcePacket,
proposals: tuple[ObjectProposal2D, ...],
) -> tuple[ObstacleObservation, ...]:
sequence = packet.envelope.sequence
if not packet.envelope.registered_point_increment.available:
return (
ObstacleObservation(
observation_id=f"camera-observation-{sequence}",
occupancy_key=f"camera-uncertainty-{sequence}",
source_id=packet.envelope.source_id,
frame_id=packet.envelope.frame_id,
evidence_time_ns=packet.envelope.timestamps.source_ns,
basis=EvidenceBasis.CAMERA,
currentness=EvidenceCurrentness.CURRENT,
occupied_support=False,
source_point_ids=(),
metric_geometry=None,
proposal_ids=(proposals[0].proposal_id,),
semantic_hint=None,
reason_codes=("camera-only",),
),
)
return (
ObstacleObservation(
observation_id=f"observation-{sequence}",
occupancy_key=f"occupancy-{sequence}",
source_id=packet.envelope.source_id,
frame_id=packet.envelope.frame_id,
evidence_time_ns=packet.envelope.timestamps.source_ns,
basis=EvidenceBasis.FUSED,
currentness=EvidenceCurrentness.CURRENT,
occupied_support=True,
source_point_ids=(sequence,),
metric_geometry=MetricGeometry(
coordinate_frame="map",
centroid_xyz_m=(4.0, 1.0, 0.5),
range_m=4.15,
covariance_diagonal_m2=(0.04, 0.04, 0.09),
),
proposal_ids=(proposals[0].proposal_id,),
semantic_hint=None,
reason_codes=("current-qualified-points",),
),
)
class _Temporal:
provider_id = "test-temporal/v1"
def update(
self,
packet: SourcePacket,
observations: tuple[ObstacleObservation, ...],
) -> tuple[TemporalObstacle, ...]:
if not packet.envelope.registered_point_increment.available:
return ()
sequence = packet.envelope.sequence
return (
TemporalObstacle(
component_id=f"component-{sequence}",
identity_scope="ephemeral",
state=TemporalState.CURRENT,
ttl_ns=750_000_000,
last_hit_ns=packet.envelope.timestamps.source_ns,
age_ns=0,
association_basis="current-spatial-support",
history=(
HistorySample(
frame_id=packet.envelope.frame_id,
evidence_time_ns=packet.envelope.timestamps.source_ns,
centroid_xyz_m=(4.0, 1.0, 0.5),
),
),
cells=(GridCell(sequence, 0, 0),),
coordinate_frame="map",
last_centroid_xyz_m=(4.0, 1.0, 0.5),
motion=MotionState.UNKNOWN,
motion_confidence=0.0,
motion_reason="insufficient-history",
),
)
class _Motion:
provider_id = "test-motion/v1"
def estimate(
self,
packet: SourcePacket,
obstacles: tuple[TemporalObstacle, ...],
) -> tuple[TemporalObstacle, ...]:
return obstacles
class _Rolling:
provider_id = "test-rolling/v1"
def update(
self,
packet: SourcePacket,
obstacles: tuple[TemporalObstacle, ...],
) -> tuple[TemporalObstacle, ...]:
if packet.envelope.sequence == 0:
return ()
return (
TemporalObstacle(
component_id=f"rolling-{packet.envelope.sequence}",
identity_scope="ephemeral",
state=TemporalState.RETAINED,
ttl_ns=3_000_000_000,
last_hit_ns=packet.envelope.timestamps.source_ns - 100_000_000,
age_ns=100_000_000,
association_basis="registered-map-increment-retention",
history=(
HistorySample(
frame_id=f"frame-{packet.envelope.sequence - 1:06d}",
evidence_time_ns=(
packet.envelope.timestamps.source_ns - 100_000_000
),
centroid_xyz_m=(3.0, 0.5, 0.5),
),
),
cells=(GridCell(99, packet.envelope.sequence, 0),),
coordinate_frame="map",
last_centroid_xyz_m=(3.0, 0.5, 0.5),
motion=MotionState.UNKNOWN,
motion_confidence=0.0,
motion_reason="retained-map-increment-no-current-motion",
),
)
class _Threat:
provider_id = "test-threat/v1"
def assess(self, obstacle_map: LocalObstacleMap) -> tuple[ThreatAssessment, ...]:
metric = tuple(
ThreatAssessment(
assessment_id=f"assessment-{item.component_id}",
component_id=item.component_id,
rig_profile_id="ravnoves00-virtual-rig/v1",
corridor_profile_id="ravnoves00-virtual-corridor/v1",
qualification=QualificationState.QUALIFIED,
relative_speed_mps=0.0,
closest_approach_m=4.0,
ttc_seconds=None,
corridor_intersection=CorridorIntersection.CLEAR,
decision=ThreatDecision.NOT_THREAT,
reason_codes=("qualified-corridor-clear",),
)
for item in obstacle_map.occupied
)
unknown = tuple(
ThreatAssessment(
assessment_id=f"assessment-{item.component_id}",
component_id=item.component_id,
rig_profile_id="ravnoves00-virtual-rig/v1",
corridor_profile_id="ravnoves00-virtual-corridor/v1",
qualification=QualificationState.UNQUALIFIED,
relative_speed_mps=None,
closest_approach_m=None,
ttc_seconds=None,
corridor_intersection=CorridorIntersection.UNKNOWN,
decision=ThreatDecision.UNKNOWN,
reason_codes=("incomplete-evidence",),
)
for item in obstacle_map.unknown
)
camera = tuple(
ThreatAssessment(
assessment_id=f"assessment-{item.proposal_id}",
component_id=item.proposal_id,
rig_profile_id="ravnoves00-virtual-rig/v1",
corridor_profile_id="ravnoves00-virtual-corridor/v1",
qualification=QualificationState.UNQUALIFIED,
relative_speed_mps=None,
closest_approach_m=None,
ttc_seconds=None,
corridor_intersection=CorridorIntersection.UNKNOWN,
decision=ThreatDecision.UNKNOWN,
reason_codes=("camera-only",),
)
for item in obstacle_map.camera_uncertainty
)
return (*metric, *unknown, *camera)
def _config(
capacity: int = 8,
terminal_timeout_ns: int = 500_000_000,
) -> ReferencePerceptionGraphConfig:
ids = {
ProviderRole.SOURCE: _Source.provider_id,
ProviderRole.DETECTOR: _Detector.provider_id,
ProviderRole.GEOMETRY: _Geometry.provider_id,
ProviderRole.TEMPORAL: _Temporal.provider_id,
ProviderRole.MOTION: _Motion.provider_id,
ProviderRole.THREAT: _Threat.provider_id,
}
return ReferencePerceptionGraphConfig(
graph_id="reference-perception-graph/v1",
source_profile_id=BASELINE_PROFILE_ID,
providers=tuple(
ProviderPin(role, provider_id, "v1", "test-revision", "a" * 64)
for role, provider_id in ids.items()
),
queues=tuple(
QueuePolicy(stage, capacity, 80_000_000, terminal_timeout_ns)
for stage in ("detector", "geometry", "temporal", "threat")
),
authority=GraphAuthority(),
)
def _config_v2(
capacity: int = 8,
terminal_timeout_ns: int = 500_000_000,
) -> ReferencePerceptionGraphConfigV2:
ids = {
ProviderRole.SOURCE: _Source.provider_id,
ProviderRole.DETECTOR: _Detector.provider_id,
ProviderRole.GEOMETRY: _Geometry.provider_id,
ProviderRole.TEMPORAL: _Temporal.provider_id,
ProviderRole.MOTION: _Motion.provider_id,
ProviderRole.ROLLING: _Rolling.provider_id,
ProviderRole.THREAT: _Threat.provider_id,
}
return ReferencePerceptionGraphConfigV2(
graph_id="reference-perception-graph/v2",
source_profile_id=BASELINE_PROFILE_ID,
providers=tuple(
ProviderPin(role, provider_id, "v1", "test-revision", "b" * 64)
for role, provider_id in ids.items()
),
queues=tuple(
QueuePolicy(
stage,
capacity,
min(80_000_000, terminal_timeout_ns),
terminal_timeout_ns,
)
for stage in ("detector", "geometry", "temporal", "rolling", "threat")
),
authority=GraphAuthority(),
)
class _MemoryTelemetry:
def __init__(self) -> None:
self.records: list[dict[str, object]] = []
self._lock = threading.Lock()
def publish(self, topic: str, payload: bytes) -> None:
with self._lock:
self.records.append({"topic": topic, "payload": json.loads(payload)})
def _graph(
source: object,
*,
detector: object | None = None,
capacity: int = 8,
terminal_timeout_ns: int = 500_000_000,
telemetry: _MemoryTelemetry | None = None,
) -> ReferencePerceptionGraphV1:
identity = (
PipelineTelemetryIdentity(
contour_id="worker-006",
agent_id="mission-core-worker",
node_id="DESKTOP-OPJ8J04",
lab_id="M4",
run_id="m4-unit-run",
source_id="RAVNOVES00",
source_package_id="ravnoves00-recorded-source/v1",
method_id="reference-perception-graph/v1",
)
if telemetry is not None
else None
)
return ReferencePerceptionGraphV1(
config=_config(capacity, terminal_timeout_ns),
source=source,
detector=detector or _Detector(),
geometry=_Geometry(),
temporal=_Temporal(),
motion=_Motion(),
threat=_Threat(),
telemetry_identity=identity,
telemetry_sink=telemetry,
clock_ns=lambda: 10_000,
)
def _graph_v2(
source: object,
*,
detector: object | None = None,
capacity: int = 8,
run_mode: GraphRunMode = GraphRunMode.LOSSLESS_REPLAY,
terminal_timeout_ns: int = 500_000_000,
delivery_observer: Callable[[DeliveredFrame, int], None] | None = None,
delivery_evidence_observer: Callable[..., None] | None = None,
) -> ReferencePerceptionGraphV2:
return ReferencePerceptionGraphV2(
config=_config_v2(capacity, terminal_timeout_ns),
source=source,
detector=detector or _Detector(),
geometry=_Geometry(),
temporal=_Temporal(),
motion=_Motion(),
rolling=_Rolling(),
threat=_Threat(),
run_mode=run_mode,
delivery_observer=delivery_observer,
delivery_evidence_observer=delivery_evidence_observer,
clock_ns=lambda: 10_000,
)
def test_reference_graph_closes_accounting_telemetry_and_deterministic_digest() -> None:
telemetry = _MemoryTelemetry()
graph = _graph(_Source((_packet(0), _packet(1))), telemetry=telemetry)
first = graph.run()
second = graph.run()
assert first.state is GraphState.STOPPED
assert first.admitted_count == 2
assert [item.outcome for item in first.terminal_outcomes] == [
TerminalOutcomeType.DELIVERED,
TerminalOutcomeType.DELIVERED,
]
assert all(
delivery.obstacle_map.accounting.source_envelopes == 1
for delivery in first.deliveries
)
assert all(delivery.obstacle_map.free_space_claimed is False for delivery in first.deliveries)
assert first.canonical_payload_sha256 == second.canonical_payload_sha256
assert graph.worker_threads_alive == 0
stage_ids = {
record["payload"].get("stage_id")
for record in telemetry.records
if isinstance(record["payload"], dict)
}
assert {"detector", "geometry", "temporal", "threat"}.issubset(stage_ids)
def test_reference_graph_v2_publishes_current_and_retained_occupancy() -> None:
result = _graph_v2(_Source((_packet(0), _packet(1)))).run()
assert result.graph_id == "reference-perception-graph/v2"
assert result.run_mode is GraphRunMode.LOSSLESS_REPLAY
assert [item.outcome for item in result.terminal_outcomes] == [
TerminalOutcomeType.DELIVERED,
TerminalOutcomeType.DELIVERED,
]
second = result.deliveries[1].obstacle_map
assert [item.state for item in second.occupied] == [
TemporalState.CURRENT,
TemporalState.RETAINED,
]
assert second.unknown == ()
assert set(dict(result.queue_high_watermarks)) == {
"detector",
"geometry",
"temporal",
"rolling",
"threat",
}
def test_reference_graph_v2_observes_final_delivery_completion_age() -> None:
observed: list[tuple[int, int, int]] = []
def observe(delivery: DeliveredFrame, completion_age_ns: int) -> None:
observed.append(
(
delivery.sequence,
delivery.obstacle_map.output_age_ns,
completion_age_ns,
)
)
result = _graph_v2(
_Source((_packet(0, source_age_ns=123),)),
delivery_observer=observe,
).run()
assert result.state is GraphState.STOPPED
assert observed == [(0, 123, 123)]
def test_reference_graph_retains_source_identity_without_processed_payload() -> None:
packet = _packet(0)
graph = _graph_v2(_Source((packet,)))
result = graph.run()
assert result.state is GraphState.STOPPED
assert graph._admitted == {0: packet.envelope}
assert graph._admitted_at_ns == {}
def test_reference_graph_releases_ephemeral_source_payloads_after_run() -> None:
payloads: list[weakref.ReferenceType[np.ndarray]] = []
class EphemeralSource:
provider_id = _Source.provider_id
def packets(self, stop_event: Event) -> Iterator[SourcePacket]:
for sequence in range(32):
if stop_event.is_set():
return
image = np.empty((64, 64, 3), dtype=np.uint8)
payloads.append(weakref.ref(image))
yield replace(_packet(sequence), image_payload=image)
graph = _graph_v2(EphemeralSource())
result = graph.run()
gc.collect()
assert result.state is GraphState.STOPPED
assert not any(reference() is not None for reference in payloads)
def test_reference_graph_v2_observes_exact_delivery_evidence_inputs() -> None:
observed: list[tuple[int, int, tuple[str, ...], frozenset[str], int]] = []
def observe(delivery, packet, proposals, associated_ids, completion_age_ns) -> None:
observed.append(
(
delivery.sequence,
packet.envelope.sequence,
tuple(proposal.proposal_id for proposal in proposals),
associated_ids,
completion_age_ns,
)
)
result = _graph_v2(
_Source((_packet(0, source_age_ns=123),)),
delivery_evidence_observer=observe,
).run()
assert result.state is GraphState.STOPPED
assert observed == [(0, 0, ("proposal-0",), frozenset({"proposal-0"}), 123)]
def test_reference_graph_v2_lossless_mode_applies_bounded_backpressure() -> None:
release = Event()
class BlockingDetector(_Detector):
def detect(self, packet: SourcePacket) -> tuple[ObjectProposal2D, ...]:
if packet.envelope.sequence == 0:
assert release.wait(2)
return super().detect(packet)
BlockingDetector.provider_id = _Detector.provider_id
graph = _graph_v2(
_Source(tuple(_packet(index) for index in range(4))),
detector=BlockingDetector(),
capacity=1,
)
holder: list[GraphRunResult] = []
runner = threading.Thread(target=lambda: holder.append(graph.run()))
runner.start()
release.set()
runner.join(3)
assert not runner.is_alive()
assert len(holder[0].terminal_outcomes) == 4
assert all(
item.outcome is TerminalOutcomeType.DELIVERED
for item in holder[0].terminal_outcomes
)
def test_reference_graph_v2_restart_requires_fresh_stateful_providers() -> None:
graph = _graph_v2(_Source((_packet(0),)))
assert graph.run().state is GraphState.STOPPED
with pytest.raises(
GraphExecutionError,
match="restart requires freshly instantiated providers",
):
graph.run()
def test_reference_graph_v2_terminal_timeout_fails_stranded_packet() -> None:
packet = _packet(0)
graph = _graph_v2(
_Source(()),
capacity=1,
terminal_timeout_ns=1,
)
queue = Queue(maxsize=1)
queue.put_nowait(packet)
graph._put_stop(queue, "detector")
outcome = graph._outcomes[0]
assert outcome.outcome is TerminalOutcomeType.FAILED
assert outcome.reason == "terminal-queue-timeout"
assert queue.qsize() == 1
def test_reference_graph_marks_unavailable_stale_and_provider_failure() -> None:
class FailingDetector(_Detector):
def detect(self, packet: SourcePacket) -> tuple[ObjectProposal2D, ...]:
if packet.envelope.sequence == 2:
raise RuntimeError("deliberate provider failure")
return super().detect(packet)
FailingDetector.provider_id = _Detector.provider_id
packets = (
_packet(0, image=False),
_packet(1, source_age_ns=80_000_001),
_packet(2),
_packet(3),
)
result = _graph(_Source(packets), detector=FailingDetector()).run()
assert [item.outcome for item in result.terminal_outcomes] == [
TerminalOutcomeType.UNAVAILABLE,
TerminalOutcomeType.STALE,
TerminalOutcomeType.FAILED,
TerminalOutcomeType.DELIVERED,
]
assert len(result.deliveries) == 1
assert result.deliveries[0].sequence == 3
def test_bounded_detector_queue_has_explicit_latest_wins_supersession() -> None:
detector_started = Event()
source_emitted = Event()
release_detector = Event()
class BlockingDetector(_Detector):
def detect(self, packet: SourcePacket) -> tuple[ObjectProposal2D, ...]:
if packet.envelope.sequence == 0:
detector_started.set()
assert release_detector.wait(2)
return super().detect(packet)
BlockingDetector.provider_id = _Detector.provider_id
class CoordinatedSource(_Source):
def packets(self, stop_event: Event) -> Iterator[SourcePacket]:
yield self.values[0]
assert detector_started.wait(2)
yield from self.values[1:]
source_emitted.set()
source = CoordinatedSource(tuple(_packet(index) for index in range(4)))
graph = _graph(
source,
detector=BlockingDetector(),
capacity=1,
terminal_timeout_ns=1_000_000_000,
)
holder: list[GraphRunResult] = []
runner = threading.Thread(target=lambda: holder.append(graph.run()))
runner.start()
assert source_emitted.wait(2)
release_detector.set()
runner.join(3)
assert not runner.is_alive()
outcomes = {item.sequence: item.outcome for item in holder[0].terminal_outcomes}
assert outcomes[1] is TerminalOutcomeType.SUPERSEDED
assert outcomes[2] is TerminalOutcomeType.SUPERSEDED
assert outcomes[3] is TerminalOutcomeType.DELIVERED
assert set(outcomes) == {0, 1, 2, 3}
assert graph.worker_threads_alive == 0
def test_cancelled_graph_closes_threads_and_can_restart() -> None:
source_waiting = Event()
class RestartableSource(_Source):
block = True
def packets(self, stop_event: Event) -> Iterator[SourcePacket]:
yield self.values[0]
if self.block:
source_waiting.set()
stop_event.wait(2)
source = RestartableSource((_packet(0),))
graph = _graph(source)
holder: list[GraphRunResult] = []
runner = threading.Thread(target=lambda: holder.append(graph.run()))
runner.start()
assert source_waiting.wait(2)
graph.cancel()
runner.join(3)
assert holder[0].state is GraphState.CANCELLED
assert graph.worker_threads_alive == 0
source.block = False
restarted = graph.run()
assert restarted.state is GraphState.STOPPED
assert restarted.terminal_outcomes[0].outcome is TerminalOutcomeType.DELIVERED
assert graph.worker_threads_alive == 0
def _write_recorded_fixture(root: Path, *, mismatched: bool = False) -> tuple[Path, Path]:
camera_path = root / "camera.jsonl"
timeline_path = root / "source-pack.npz"
camera_rows = [
{
"schema_version": "missioncore.camera-recording-index/v1",
"kind": "media",
"sequence": index + 1,
"host_epoch_ns": 100 + index,
"host_monotonic_ns": 200 + index,
"sha256": f"{index + 1:064x}",
}
for index in range(2)
]
camera_path.write_text("\n".join(json.dumps(row) for row in camera_rows) + "\n", "utf-8")
np.savez(
timeline_path,
frame_indices=np.asarray([0, 2 if mismatched else 1], dtype=np.int64),
source_frame_indices=np.asarray([0, 1], dtype=np.int64),
session_seconds=np.asarray([10.0, 10.1], dtype=np.float64),
sample_available=np.asarray([True, False], dtype=np.bool_),
)
return camera_path, timeline_path
def test_recorded_source_reuses_one_timeline_for_1x_and_uncapped(tmp_path: Path) -> None:
camera_path, timeline_path = _write_recorded_fixture(tmp_path)
now = [1_000_000_000]
waits: list[float] = []
def wait(stop_event: Event, seconds: float) -> bool:
waits.append(seconds)
now[0] += round(seconds * 1_000_000_000)
return stop_event.is_set()
source = RecordedRavnoves00Source(
camera_index_path=camera_path,
source_pack_path=timeline_path,
pacing=ReplayPacing.ONE_X,
expected_frame_count=2,
expected_source_pack_sha256=None,
clock_ns=lambda: now[0],
wait=wait,
)
one_x = list(source.packets(Event()))
uncapped = list(
RecordedRavnoves00Source(
camera_index_path=camera_path,
source_pack_path=timeline_path,
pacing=ReplayPacing.UNCAPPED,
expected_frame_count=2,
expected_source_pack_sha256=None,
wait=lambda _event, _seconds: pytest.fail("uncapped replay waited"),
).packets(Event())
)
assert [packet.envelope.to_dict() for packet in one_x] == [
packet.envelope.to_dict() for packet in uncapped
]
assert waits == pytest.approx([0.1])
assert one_x[1].envelope.registered_point_increment.available is False
assert one_x[1].registered_point_increment_payload is None
def test_recorded_source_target_rate_changes_only_wall_clock_pacing(tmp_path: Path) -> None:
camera_path, timeline_path = _write_recorded_fixture(tmp_path)
now = [1_000_000_000]
waits: list[float] = []
pacing: list[SourcePacingTiming] = []
def wait(stop_event: Event, seconds: float) -> bool:
waits.append(seconds)
now[0] += round(seconds * 1_000_000_000)
return stop_event.is_set()
source = RecordedRavnoves00Source(
camera_index_path=camera_path,
source_pack_path=timeline_path,
pacing=ReplayPacing.ONE_X,
target_rate_hz=20.0,
expected_frame_count=2,
expected_source_pack_sha256=None,
pacing_observer=pacing.append,
clock_ns=lambda: now[0],
wait=wait,
)
packets = list(source.packets(Event()))
assert waits == pytest.approx([0.05])
assert [sample.sequence for sample in pacing] == [0, 1]
assert [sample.lateness_ns for sample in pacing] == [0, 0]
assert (
packets[1].envelope.timestamps.source_ns
- packets[0].envelope.timestamps.source_ns
== 100_000_000
)
def test_recorded_source_rejects_target_rate_for_uncapped_replay(tmp_path: Path) -> None:
camera_path, timeline_path = _write_recorded_fixture(tmp_path)
with pytest.raises(RecordedSourceError, match="requires paced replay"):
RecordedRavnoves00Source(
camera_index_path=camera_path,
source_pack_path=timeline_path,
pacing=ReplayPacing.UNCAPPED,
target_rate_hz=12.0,
expected_frame_count=2,
expected_source_pack_sha256=None,
)
def test_prefetched_decoder_moves_cold_decode_before_source_admission() -> None:
observed: list[DecodedFrameTiming] = []
class Decoder:
def frames(self, stop_event: Event) -> Iterator[np.ndarray]:
for index in range(4):
if stop_event.is_set():
return
yield np.full((2, 3, 3), index, dtype=np.uint8)
decoder = PrefetchedRecordedImageDecoder(
Decoder(),
capacity_frames=2,
ready_frames=2,
timing_observer=observed.append,
)
stop_event = Event()
snapshot = decoder.prepare(stop_event)
assert snapshot.buffered_frames == 2
assert snapshot.capacity_frames == 2
assert [sample.phase for sample in observed] == [
DecodePhase.PREADMISSION,
DecodePhase.PREADMISSION,
]
decoder.mark_admission_started()
frames = list(decoder.frames(stop_event))
decoder.close()
assert [int(frame[0, 0, 0]) for frame in frames] == [0, 1, 2, 3]
assert [sample.sequence for sample in observed] == [0, 1, 2, 3]
assert [sample.phase for sample in observed[2:]] == [
DecodePhase.HOT_LOOP,
DecodePhase.HOT_LOOP,
]
def test_recorded_source_rejects_timeline_mismatch(tmp_path: Path) -> None:
camera_path, timeline_path = _write_recorded_fixture(tmp_path, mismatched=True)
source = RecordedRavnoves00Source(
camera_index_path=camera_path,
source_pack_path=timeline_path,
expected_frame_count=2,
expected_source_pack_sha256=None,
)
with pytest.raises(RecordedSourceError, match="timeline frame index"):
list(source.packets(Event()))
def test_decoded_recorded_source_attaches_images_without_detector_logic(tmp_path: Path) -> None:
camera_path, timeline_path = _write_recorded_fixture(tmp_path)
source = RecordedRavnoves00Source(
camera_index_path=camera_path,
source_pack_path=timeline_path,
expected_frame_count=2,
expected_source_pack_sha256=None,
)
class Decoder:
def frames(self, stop_event: Event) -> Iterator[np.ndarray]:
for value in (3, 7):
if stop_event.is_set():
return
yield np.full((600, 800, 3), value, dtype=np.uint8)
packets = list(DecodedRecordedSource(source=source, decoder=Decoder()).packets(Event()))
assert len(packets) == 2
assert isinstance(packets[0].image_payload, np.ndarray)
assert int(packets[0].image_payload[0, 0, 0]) == 3
assert int(packets[1].image_payload[0, 0, 0]) == 7
def test_decoded_recorded_source_reports_per_frame_decode_timing(tmp_path: Path) -> None:
camera_path, timeline_path = _write_recorded_fixture(tmp_path)
source = RecordedRavnoves00Source(
camera_index_path=camera_path,
source_pack_path=timeline_path,
expected_frame_count=2,
expected_source_pack_sha256=None,
)
observed: list[DecodedFrameTiming] = []
class Decoder:
def frames(self, _stop_event: Event) -> Iterator[np.ndarray]:
yield np.zeros((600, 800, 3), dtype=np.uint8)
yield np.zeros((600, 800, 3), dtype=np.uint8)
packets = list(
DecodedRecordedSource(
source=source,
decoder=Decoder(),
timing_observer=observed.append,
clock_ns=iter((10, 30, 40, 90, 100)).__next__,
).packets(Event())
)
assert len(packets) == 2
assert observed == [DecodedFrameTiming(0, 20), DecodedFrameTiming(1, 50)]
def test_decoded_recorded_source_primes_decoder_before_source_clock() -> None:
events: list[str] = []
class Source(RecordedRavnoves00Source):
def __init__(self) -> None:
pass
def packets(self, stop_event: Event) -> Iterator[SourcePacket]:
events.append("source-clock-started")
if not stop_event.is_set():
yield _packet(0)
class Decoder:
def frames(self, stop_event: Event) -> Iterator[np.ndarray]:
events.append("decoder-primed")
if not stop_event.is_set():
yield np.zeros((600, 800, 3), dtype=np.uint8)
packets = list(
DecodedRecordedSource(source=Source(), decoder=Decoder()).packets(Event())
)
assert len(packets) == 1
assert events == ["decoder-primed", "source-clock-started"]
def test_camera_only_path_never_invents_metric_occupancy_or_free_space() -> None:
result = _graph(_Source((_packet(0, lidar=False),))).run()
delivery = result.deliveries[0]
assert delivery.obstacle_map.occupied == ()
assert delivery.obstacle_map.unknown == ()
assert delivery.obstacle_map.free_space_claimed is False
assert len(delivery.obstacle_map.camera_uncertainty) == 1
def test_graph_config_rejects_unbounded_or_missing_stage_policy() -> None:
config = _config()
with pytest.raises(ValueError, match="each reference stage"):
replace(config, queues=config.queues[:-1])
Reference in New Issue View Git Blame Copy Permalink