From fb5bf943c96575f029466bd51d0509b6536cfa2a Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Tue, 25 Aug 2026 21:58:12 +0300 Subject: [PATCH] fix(perception): isolate recorded decode from realtime admission --- .../run_m48s_reference_graph_shadow_worker.py | 115 ++++++++- .../m48s_reference_graph_runtime.py | 43 +++- src/k1link/perception/recorded_source.py | 228 +++++++++++++++++- tests/test_perception_graph.py | 45 ++++ 4 files changed, 416 insertions(+), 15 deletions(-) diff --git a/experiments/perception/run_m48s_reference_graph_shadow_worker.py b/experiments/perception/run_m48s_reference_graph_shadow_worker.py index 3275323..cd18726 100644 --- a/experiments/perception/run_m48s_reference_graph_shadow_worker.py +++ b/experiments/perception/run_m48s_reference_graph_shadow_worker.py @@ -47,14 +47,14 @@ from k1link.perception.m48s_reference_graph_runtime import ( from k1link.perception.motion import ClassIndependentMotionEstimator from k1link.perception.object_understanding import AdvisoryResponse from k1link.perception.providers import SourcePacket -from k1link.perception.recorded_source import DecodedFrameTiming +from k1link.perception.recorded_source import DecodedFrameTiming, SourcePacingTiming from k1link.perception.reference_graph_runtime import ReferenceGraphRuntimePaths from k1link.perception.rolling_map import RollingLocalObstacleMapProvider from k1link.perception.temporal import BoundedSpatialTemporalProvider -SCHEMA_VERSION: Final = "missioncore.m48s-reference-graph-shadow-load/v4" +SCHEMA_VERSION: Final = "missioncore.m48s-reference-graph-shadow-load/v5" FRAME_EVIDENCE_SCHEMA: Final = "missioncore.m48s-reference-graph-frame-evidence/v1" -PIPELINE_TIMING_SCHEMA: Final = "missioncore.m48s-frame-pipeline-timing/v0" +PIPELINE_TIMING_SCHEMA: Final = "missioncore.m48s-frame-pipeline-timing/v1" GC_POLICY_SCHEMA: Final = "missioncore.cyclic-gc-hot-loop-policy/v0" AUTHORITY: Final = { "ground_truth": False, @@ -228,14 +228,23 @@ class FrameTimingStore: def __init__(self) -> None: self._lock = threading.Lock() - self._decode: dict[int, int] = {} + self._decode: dict[int, DecodedFrameTiming] = {} + self._pacing: dict[int, SourcePacingTiming] = {} + self._all_decode: list[DecodedFrameTiming] = [] + self._all_pacing: list[SourcePacingTiming] = [] self._detector: dict[int, DetectorFrameTiming] = {} self._providers: dict[int, dict[str, int]] = defaultdict(dict) self._delivered: list[dict[str, object]] = [] def observe_decode(self, timing: DecodedFrameTiming) -> None: with self._lock: - self._decode[timing.sequence] = timing.duration_ns + self._decode[timing.sequence] = timing + self._all_decode.append(timing) + + def observe_pacing(self, timing: SourcePacingTiming) -> None: + with self._lock: + self._pacing[timing.sequence] = timing + self._all_pacing.append(timing) def observe_detector(self, timing: DetectorFrameTiming) -> None: with self._lock: @@ -257,7 +266,8 @@ class FrameTimingStore: ) -> dict[str, object]: with self._lock: try: - decode_ns = self._decode.pop(sequence) + decode = self._decode.pop(sequence) + pacing = self._pacing.pop(sequence) detector = self._detector.pop(sequence) providers = self._providers.pop(sequence) except KeyError as exc: @@ -272,13 +282,19 @@ class FrameTimingStore: document = { "schema_version": PIPELINE_TIMING_SCHEMA, "sequence": sequence, - "decode_duration_ns": decode_ns, + "decode_duration_ns": decode.duration_ns, + "decode_phase": decode.phase.value, + "source_pacing": { + "scheduled_monotonic_ns": pacing.scheduled_monotonic_ns, + "emitted_monotonic_ns": pacing.emitted_monotonic_ns, + "lateness_ns": pacing.lateness_ns, + }, "detector": detector.to_dict(), "providers": dict(sorted(providers.items())), "graph_admission_to_delivery_ns": admission_to_delivery_ns, "graph_attributed_provider_ns": attributed_graph_ns, "graph_unattributed_ns": unattributed_ns, - "decode_to_delivery_processing_ns": decode_ns + admission_to_delivery_ns, + "decode_to_delivery_processing_ns": decode.duration_ns + admission_to_delivery_ns, } with self._lock: self._delivered.append(document) @@ -288,6 +304,14 @@ class FrameTimingStore: with self._lock: return tuple(self._delivered) + def all_decode(self) -> tuple[DecodedFrameTiming, ...]: + with self._lock: + return tuple(self._all_decode) + + def all_pacing(self) -> tuple[SourcePacingTiming, ...]: + with self._lock: + return tuple(self._all_pacing) + class TimedProviderProxy: """Record one provider's actual call duration without another inference pass.""" @@ -420,6 +444,8 @@ def main() -> int: map_output_ages_ms: list[float] = [] all_deliveries: list[DeliveredFrame] = [] all_pipeline_timings: list[dict[str, object]] = [] + all_decode_timings: list[DecodedFrameTiming] = [] + all_pacing_timings: list[SourcePacingTiming] = [] with ( progress.open("x", encoding="utf-8") as progress_stream, frame_ledger.open("x", encoding="utf-8") as frame_ledger_stream, @@ -447,6 +473,7 @@ def main() -> int: timing_store, ), decode_timing_observer=timing_store.observe_decode, + source_pacing_observer=timing_store.observe_pacing, detector_timing_observer=timing_store.observe_detector, maximum_frames=arguments.maximum_frames, source_rate_hz=arguments.source_rate_hz, @@ -469,8 +496,10 @@ def main() -> int: ), ) detector_warmup = runtime.warm_up_detector() + source_prefetch = runtime.prepare_source() gc_policy = CyclicGcHotLoopPolicy() with gc_policy: + runtime.mark_source_admission_started() loop_started_ns = time.monotonic_ns() result = runtime.graph.run() loop_completed_ns = time.monotonic_ns() @@ -523,6 +552,7 @@ def main() -> int: setup_seconds=setup_seconds, gc_policy=gc_policy.to_dict(), detector_warmup=detector_warmup, + source_prefetch=asdict(source_prefetch), ) loop_documents.append(loop_document) completion_ages_ms.extend(value / 1_000_000.0 for value in loop_completion_ages_ns) @@ -531,6 +561,8 @@ def main() -> int: ) all_deliveries.extend(result.deliveries) all_pipeline_timings.extend(loop_pipeline_timings) + all_decode_timings.extend(timing_store.all_decode()) + all_pacing_timings.extend(timing_store.all_pacing()) frame_ledger_stream.flush() progress_row = { "loop": loop_index + 1, @@ -608,6 +640,12 @@ def main() -> int: and cast(dict[str, object], loop["detector_warmup"])["inference_passes"] == 1 for loop in loop_documents ), + "source_prefetch_completed_before_source_admission": all( + cast(dict[str, object], loop["source_prefetch"])["buffered_frames"] + == cast(dict[str, object], loop["source_prefetch"])["ready_frames"] + for loop in loop_documents + ), + "source_pacing_attribution_complete": len(all_pacing_timings) == admitted, "authority_remains_false": all(value is False for value in AUTHORITY.values()), } operating_target_checks = { @@ -672,6 +710,8 @@ def main() -> int: "identity_continuity": identity, "semantic_advisory": semantic, "pipeline_timing": _pipeline_timing_metrics(all_pipeline_timings), + "source_decode": _source_decode_metrics(all_decode_timings), + "source_pacing": _source_pacing_metrics(all_pacing_timings), "python_gc": _gc_telemetry_summary(gc_telemetry.events), "gpu": _telemetry_summary(gpu.samples), "process_peak_rss_before_mib": round(rss_before_kib / 1024.0, 6), @@ -762,6 +802,7 @@ def _loop_document( setup_seconds: float, gc_policy: dict[str, object], detector_warmup: DetectorWarmupSnapshot, + source_prefetch: dict[str, object], ) -> dict[str, object]: outcomes = Counter(item.outcome.value for item in result.terminal_outcomes) outcome_stages = Counter( @@ -774,6 +815,7 @@ def _loop_document( "setup_seconds": round(setup_seconds, 6), "cyclic_gc_hot_loop": gc_policy, "detector_warmup": asdict(detector_warmup), + "source_prefetch": source_prefetch, "admitted_count": result.admitted_count, "delivered_count": len(result.deliveries), "effective_world_state_fps": round(len(result.deliveries) / wall_seconds, 6), @@ -899,6 +941,8 @@ def _pipeline_timing_metrics( "decode_to_delivery_processing_ns", ) top_level_values: dict[str, list[float]] = {key: [] for key in top_level_fields} + decode_by_phase: dict[str, list[float]] = defaultdict(list) + delivered_pacing_lateness_ms: list[float] = [] for document in documents: detector = cast(Mapping[str, int], document["detector"]) providers = cast(Mapping[str, int], document["providers"]) @@ -908,6 +952,11 @@ def _pipeline_timing_metrics( provider_values[key].append(providers[key] / 1_000_000.0) for key in top_level_fields: top_level_values[key].append(cast(int, document[key]) / 1_000_000.0) + decode_by_phase[cast(str, document["decode_phase"])].append( + cast(int, document["decode_duration_ns"]) / 1_000_000.0 + ) + source_pacing = cast(Mapping[str, int], document["source_pacing"]) + delivered_pacing_lateness_ms.append(source_pacing["lateness_ns"] / 1_000_000.0) maximum = max( documents, key=lambda document: cast(int, document["graph_admission_to_delivery_ns"]), @@ -916,6 +965,12 @@ def _pipeline_timing_metrics( return { "sample_count": len(documents), "decode_duration_ms": _distribution(top_level_values["decode_duration_ns"]), + "decode_duration_by_phase_ms": { + phase: _distribution(values) for phase, values in sorted(decode_by_phase.items()) + }, + "delivered_source_pacing_lateness_ms": _distribution( + delivered_pacing_lateness_ms + ), "detector_ms": { key.removesuffix("_duration_ns"): _distribution(values) for key, values in detector_values.items() @@ -936,6 +991,50 @@ def _pipeline_timing_metrics( } +def _source_decode_metrics(samples: list[DecodedFrameTiming]) -> dict[str, object]: + by_phase: dict[str, list[float]] = defaultdict(list) + for sample in samples: + by_phase[sample.phase.value].append(sample.duration_ns / 1_000_000.0) + return { + "sample_count": len(samples), + "phase_counts": { + phase: len(values) for phase, values in sorted(by_phase.items()) + }, + "duration_by_phase_ms": { + phase: _distribution(values) for phase, values in sorted(by_phase.items()) + }, + } + + +def _source_pacing_metrics(samples: list[SourcePacingTiming]) -> dict[str, object]: + ordered = sorted(samples, key=lambda sample: sample.sequence) + lateness_ms = [sample.lateness_ns / 1_000_000.0 for sample in ordered] + scheduled_intervals_ms = [ + (current.scheduled_monotonic_ns - previous.scheduled_monotonic_ns) / 1_000_000.0 + for previous, current in zip(ordered, ordered[1:], strict=False) + ] + emitted_intervals_ms = [ + (current.emitted_monotonic_ns - previous.emitted_monotonic_ns) / 1_000_000.0 + for previous, current in zip(ordered, ordered[1:], strict=False) + ] + catch_up_emissions = sum( + emitted < scheduled * 0.5 + for scheduled, emitted in zip( + scheduled_intervals_ms, + emitted_intervals_ms, + strict=True, + ) + if scheduled > 0 + ) + return { + "sample_count": len(ordered), + "lateness_ms": _distribution(lateness_ms), + "scheduled_interval_ms": _distribution(scheduled_intervals_ms), + "emitted_interval_ms": _distribution(emitted_intervals_ms), + "catch_up_emission_count": catch_up_emissions, + } + + def _telemetry_summary(samples: list[dict[str, float]]) -> dict[str, Any]: result: dict[str, Any] = {"sample_count": len(samples)} for key in ( diff --git a/src/k1link/perception/m48s_reference_graph_runtime.py b/src/k1link/perception/m48s_reference_graph_runtime.py index 3b551ae..1346ae8 100644 --- a/src/k1link/perception/m48s_reference_graph_runtime.py +++ b/src/k1link/perception/m48s_reference_graph_runtime.py @@ -5,7 +5,7 @@ from __future__ import annotations import hashlib import json from collections.abc import Callable, Iterator -from dataclasses import dataclass +from dataclasses import dataclass, field from pathlib import Path from threading import Event @@ -32,10 +32,13 @@ from .providers import ( ) from .recorded_source import ( DecodedRecordedSource, + DecodePrefetchSnapshot, DecodeTimingObserver, + PrefetchedRecordedImageDecoder, PyAvRecordedImageDecoder, RecordedRavnoves00Source, ReplayPacing, + SourcePacingObserver, ) from .reference_graph_runtime import ReferenceGraphRuntimePaths from .rf_detr_object_detector import ( @@ -64,6 +67,8 @@ class M48sReferenceGraphRuntime: graph: ReferencePerceptionGraphV2 inference_backend: TritonRfDetrHttpInferenceBackend + source_prefetch: PrefetchedRecordedImageDecoder + _preparation_stop_event: Event = field(default_factory=Event) def warm_up_detector(self) -> DetectorWarmupSnapshot: detector = self.graph.detector @@ -71,8 +76,18 @@ class M48sReferenceGraphRuntime: raise M48sReferenceGraphRuntimeError("RF-DETR runtime detector changed before warmup") return detector.warm_up() + def prepare_source(self) -> DecodePrefetchSnapshot: + return self.source_prefetch.prepare(self._preparation_stop_event) + + def mark_source_admission_started(self) -> None: + self.source_prefetch.mark_admission_started() + def close(self) -> None: - self.inference_backend.close() + self._preparation_stop_event.set() + try: + self.source_prefetch.close() + finally: + self.inference_backend.close() def __enter__(self) -> M48sReferenceGraphRuntime: return self @@ -90,9 +105,12 @@ def build_m48s_reference_graph_runtime( delivery_observer: Callable[[DeliveredFrame, int], None] | None = None, delivery_evidence_observer: DeliveryEvidenceObserver | None = None, decode_timing_observer: DecodeTimingObserver | None = None, + source_pacing_observer: SourcePacingObserver | None = None, detector_timing_observer: DetectorTimingObserver | None = None, maximum_frames: int | None = None, source_rate_hz: float | None = None, + source_prefetch_capacity_frames: int = 64, + source_prefetch_ready_frames: int = 64, ) -> M48sReferenceGraphRuntime: """Instantiate the complete graph with only its detector pin replaced.""" @@ -117,6 +135,12 @@ def build_m48s_reference_graph_runtime( if maximum_frames is not None and maximum_frames < 1: raise M48sReferenceGraphRuntimeError("maximum frame count must be positive") + prefetch = PrefetchedRecordedImageDecoder( + PyAvRecordedImageDecoder(paths.video), + capacity_frames=source_prefetch_capacity_frames, + ready_frames=source_prefetch_ready_frames, + timing_observer=decode_timing_observer, + ) source: SourceProvider = DecodedRecordedSource( source=RecordedRavnoves00Source( camera_index_path=paths.camera_index, @@ -131,9 +155,9 @@ def build_m48s_reference_graph_runtime( if run_mode is GraphRunMode.SOURCE_PACED_LATEST_WINS else None ), + pacing_observer=source_pacing_observer, ), - decoder=PyAvRecordedImageDecoder(paths.video), - timing_observer=decode_timing_observer, + decoder=prefetch, ) if maximum_frames is not None: source = _LimitedSource(source, maximum_frames) @@ -175,9 +199,16 @@ def build_m48s_reference_graph_runtime( delivery_evidence_observer=delivery_evidence_observer, ) except Exception: - backend.close() + try: + prefetch.close() + finally: + backend.close() raise - return M48sReferenceGraphRuntime(graph=graph, inference_backend=backend) + return M48sReferenceGraphRuntime( + graph=graph, + inference_backend=backend, + source_prefetch=prefetch, + ) class _LimitedSource: diff --git a/src/k1link/perception/recorded_source.py b/src/k1link/perception/recorded_source.py index aef37b9..d2bfdfe 100644 --- a/src/k1link/perception/recorded_source.py +++ b/src/k1link/perception/recorded_source.py @@ -10,7 +10,8 @@ from collections.abc import Callable, Iterator from dataclasses import dataclass, replace from enum import StrEnum from pathlib import Path -from threading import Event +from queue import Empty, Full, Queue +from threading import Event, Lock, Thread from typing import Any, Final, Protocol, cast import numpy as np @@ -53,6 +54,11 @@ class ReplayPacing(StrEnum): UNCAPPED = "uncapped" +class DecodePhase(StrEnum): + PREADMISSION = "preadmission" + HOT_LOOP = "hot-loop" + + @dataclass(frozen=True, slots=True) class RecordedFrameReference: """Opaque reference passed to a provider without decoding sensor data.""" @@ -89,6 +95,7 @@ WaitFunction = Callable[[Event, float], bool] class DecodedFrameTiming: sequence: int duration_ns: int + phase: DecodePhase = DecodePhase.HOT_LOOP def __post_init__(self) -> None: if self.sequence < 0 or self.duration_ns < 0: @@ -98,6 +105,43 @@ class DecodedFrameTiming: DecodeTimingObserver = Callable[[DecodedFrameTiming], None] +@dataclass(frozen=True, slots=True) +class SourcePacingTiming: + sequence: int + scheduled_monotonic_ns: int + emitted_monotonic_ns: int + lateness_ns: int + + def __post_init__(self) -> None: + if ( + self.sequence < 0 + or self.scheduled_monotonic_ns < 0 + or self.emitted_monotonic_ns < 0 + or self.lateness_ns < 0 + ): + raise RecordedSourceError("source pacing timing must be nonnegative") + + +SourcePacingObserver = Callable[[SourcePacingTiming], None] + + +@dataclass(frozen=True, slots=True) +class DecodePrefetchSnapshot: + capacity_frames: int + ready_frames: int + buffered_frames: int + preparation_duration_ns: int + producer_alive: bool + + +@dataclass(frozen=True, slots=True) +class _DecodeFailure: + error: BaseException + + +_DECODE_END: Final = object() + + class RecordedRavnoves00Source: """Emit the admitted timeline at its recorded or an explicit replay rate. @@ -116,6 +160,7 @@ class RecordedRavnoves00Source: expected_frame_count: int = DEFAULT_FRAME_COUNT, expected_source_pack_sha256: str | None = RECORDED_SOURCE_PACK_SHA256, target_rate_hz: float | None = None, + pacing_observer: SourcePacingObserver | None = None, clock_ns: Callable[[], int] = time.monotonic_ns, wait: WaitFunction | None = None, ) -> None: @@ -133,6 +178,7 @@ class RecordedRavnoves00Source: self.expected_frame_count = expected_frame_count self.expected_source_pack_sha256 = expected_source_pack_sha256 self.target_rate_hz = target_rate_hz + self.pacing_observer = pacing_observer self._clock_ns = clock_ns self._wait = wait or _event_wait @@ -189,6 +235,16 @@ class RecordedRavnoves00Source: target_ns = started_ns + round(source_elapsed_ns * pacing_scale) if not self._pace_until(stop_event, target_ns): return + emitted_ns = int(self._clock_ns()) + if self.pacing_observer is not None: + self.pacing_observer( + SourcePacingTiming( + sequence=packet.envelope.sequence, + scheduled_monotonic_ns=target_ns, + emitted_monotonic_ns=emitted_ns, + lateness_ns=max(0, emitted_ns - target_ns), + ) + ) yield packet def _pace_until(self, stop_event: Event, target_ns: int) -> bool: @@ -295,6 +351,176 @@ class PyAvRecordedImageDecoder: container.close() +class PrefetchedRecordedImageDecoder: + """Decode into a bounded queue before source admission and during replay. + + The queue absorbs ordinary storage/codec jitter without retaining the whole + recording in RAM. Decode work remains single-pass and source order remains + exact. The explicit phase boundary keeps cold codec initialization out of + hot-loop latency attribution. + """ + + def __init__( + self, + decoder: RecordedImageDecoder, + *, + capacity_frames: int = 64, + ready_frames: int = 64, + timing_observer: DecodeTimingObserver | None = None, + clock_ns: Callable[[], int] = time.perf_counter_ns, + ) -> None: + if capacity_frames < 1: + raise RecordedSourceError("decode prefetch capacity must be positive") + if ready_frames < 1 or ready_frames > capacity_frames: + raise RecordedSourceError("decode prefetch readiness must fit its capacity") + self.decoder = decoder + self.capacity_frames = capacity_frames + self.ready_frames = ready_frames + self.timing_observer = timing_observer + self._clock_ns = clock_ns + self._queue: Queue[NDArray[np.uint8] | _DecodeFailure | object] = Queue( + capacity_frames + ) + self._stop = Event() + self._ready = Event() + self._guard = Lock() + self._thread: Thread | None = None + self._phase = DecodePhase.PREADMISSION + self._produced = 0 + self._started_ns: int | None = None + self._ready_ns: int | None = None + self._closed = False + + def prepare(self, stop_event: Event) -> DecodePrefetchSnapshot: + with self._guard: + if self._closed: + raise RecordedSourceError("decode prefetch is closed") + if self._thread is None: + self._started_ns = int(self._clock_ns()) + self._thread = Thread( + target=self._produce, + name="m48s-recorded-decode-prefetch", + daemon=True, + ) + self._thread.start() + while not self._ready.wait(0.05): + if stop_event.is_set(): + raise RecordedSourceError("decode prefetch preparation was cancelled") + with self._guard: + started_ns = self._started_ns + ready_ns = self._ready_ns + thread = self._thread + produced = self._produced + if started_ns is None or ready_ns is None or thread is None: + raise RecordedSourceError("decode prefetch readiness is incomplete") + first = self._peek() + if isinstance(first, _DecodeFailure): + raise RecordedSourceError("recorded camera prefetch failed") from first.error + return DecodePrefetchSnapshot( + capacity_frames=self.capacity_frames, + ready_frames=self.ready_frames, + buffered_frames=min(produced, self.capacity_frames), + preparation_duration_ns=max(0, ready_ns - started_ns), + producer_alive=thread.is_alive(), + ) + + def mark_admission_started(self) -> None: + with self._guard: + if not self._ready.is_set(): + raise RecordedSourceError("source admission started before decode prefetch") + self._phase = DecodePhase.HOT_LOOP + + def frames(self, stop_event: Event) -> Iterator[NDArray[np.uint8]]: + self.prepare(stop_event) + while not stop_event.is_set() and not self._stop.is_set(): + try: + item = self._queue.get(timeout=0.05) + except Empty: + continue + try: + if item is _DECODE_END: + return + if isinstance(item, _DecodeFailure): + raise RecordedSourceError("recorded camera decode failed") from item.error + yield cast(NDArray[np.uint8], item) + finally: + self._queue.task_done() + + def close(self, *, timeout_seconds: float = 5.0) -> None: + with self._guard: + if self._closed: + return + self._closed = True + self._stop.set() + thread = self._thread + if thread is not None: + thread.join(timeout=max(0.0, timeout_seconds)) + if thread.is_alive(): + raise RecordedSourceError("decode prefetch worker did not stop") + + def _produce(self) -> None: + sequence = 0 + images = self.decoder.frames(self._stop) + try: + while not self._stop.is_set(): + while self._queue.full() and not self._stop.wait(0.01): + pass + if self._stop.is_set(): + return + started_ns = int(self._clock_ns()) + try: + image = next(images) + except StopIteration: + self._signal_ready() + self._put(_DECODE_END) + return + completed_ns = int(self._clock_ns()) + with self._guard: + phase = self._phase + if self.timing_observer is not None: + self.timing_observer( + DecodedFrameTiming( + sequence=sequence, + duration_ns=max(0, completed_ns - started_ns), + phase=phase, + ) + ) + if not self._put(np.asarray(image, dtype=np.uint8)): + return + sequence += 1 + with self._guard: + self._produced = sequence + ready = sequence >= self.ready_frames + if ready: + self._signal_ready() + except BaseException as exc: + self._put(_DecodeFailure(exc)) + self._signal_ready() + finally: + close = getattr(images, "close", None) + if callable(close): + close() + + def _put(self, item: NDArray[np.uint8] | _DecodeFailure | object) -> bool: + while not self._stop.is_set(): + try: + self._queue.put(item, timeout=0.05) + return True + except Full: + continue + return False + + def _signal_ready(self) -> None: + with self._guard: + if self._ready_ns is None: + self._ready_ns = int(self._clock_ns()) + self._ready.set() + + def _peek(self) -> NDArray[np.uint8] | _DecodeFailure | object | None: + with self._queue.mutex: + return self._queue.queue[0] if self._queue.queue else None + + def _packet( frame_index: int, camera: dict[str, object], diff --git a/tests/test_perception_graph.py b/tests/test_perception_graph.py index f49d1ae..80a0e13 100644 --- a/tests/test_perception_graph.py +++ b/tests/test_perception_graph.py @@ -62,9 +62,12 @@ from k1link.perception.providers import ( from k1link.perception.recorded_source import ( DecodedFrameTiming, DecodedRecordedSource, + DecodePhase, + PrefetchedRecordedImageDecoder, RecordedRavnoves00Source, RecordedSourceError, ReplayPacing, + SourcePacingTiming, ) @@ -839,6 +842,7 @@ def test_recorded_source_target_rate_changes_only_wall_clock_pacing(tmp_path: Pa 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) @@ -852,6 +856,7 @@ def test_recorded_source_target_rate_changes_only_wall_clock_pacing(tmp_path: Pa 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, ) @@ -859,6 +864,8 @@ def test_recorded_source_target_rate_changes_only_wall_clock_pacing(tmp_path: Pa 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 @@ -879,6 +886,44 @@ def test_recorded_source_rejects_target_rate_for_uncapped_replay(tmp_path: Path) ) +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(