From a0152f371e888aabc35cbe60c5dc34f1835ebf2f Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Tue, 25 Aug 2026 18:21:57 +0300 Subject: [PATCH] feat(perception): trace M4.8S pipeline latency --- .../run_m48s_reference_graph_shadow_worker.py | 213 +++++++++++++++++- src/k1link/perception/detector.py | 62 ++++- .../m48s_reference_graph_runtime.py | 11 +- src/k1link/perception/recorded_source.py | 41 +++- tests/test_m48s_reference_graph_timing.py | 83 +++++++ tests/test_perception_graph.py | 29 +++ tests/test_rf_detr_detector_provider.py | 23 ++ 7 files changed, 453 insertions(+), 9 deletions(-) create mode 100644 tests/test_m48s_reference_graph_timing.py diff --git a/experiments/perception/run_m48s_reference_graph_shadow_worker.py b/experiments/perception/run_m48s_reference_graph_shadow_worker.py index 58875db..3678438 100644 --- a/experiments/perception/run_m48s_reference_graph_shadow_worker.py +++ b/experiments/perception/run_m48s_reference_graph_shadow_worker.py @@ -19,9 +19,9 @@ from typing import Any, Final, TextIO, cast import numpy as np -from k1link.perception.contracts import MotionState -from k1link.perception.contracts import ObjectProposal2D +from k1link.perception.contracts import MotionState, ObjectProposal2D from k1link.perception.detector import ( + DetectorFrameTiming, DetectorProviderSnapshot, RfDetrShadowDetectorProvider, ) @@ -44,13 +44,15 @@ from k1link.perception.m48s_reference_graph_runtime import ( ) from k1link.perception.motion import ClassIndependentMotionEstimator from k1link.perception.object_understanding import AdvisoryResponse -from k1link.perception.reference_graph_runtime import ReferenceGraphRuntimePaths from k1link.perception.providers import SourcePacket +from k1link.perception.recorded_source import DecodedFrameTiming +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/v0" -FRAME_EVIDENCE_SCHEMA: Final = "missioncore.m48s-reference-graph-frame-evidence/v0" +SCHEMA_VERSION: Final = "missioncore.m48s-reference-graph-shadow-load/v1" +FRAME_EVIDENCE_SCHEMA: Final = "missioncore.m48s-reference-graph-frame-evidence/v1" +PIPELINE_TIMING_SCHEMA: Final = "missioncore.m48s-frame-pipeline-timing/v0" AUTHORITY: Final = { "ground_truth": False, "candidate_accepted": False, @@ -104,6 +106,113 @@ class GpuTelemetry: self._stop.wait(self.interval_seconds) +class FrameTimingStore: + """Join bounded decode, detector and provider timings by source sequence.""" + + def __init__(self) -> None: + self._lock = threading.Lock() + self._decode: dict[int, int] = {} + 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 + + def observe_detector(self, timing: DetectorFrameTiming) -> None: + with self._lock: + self._detector[timing.sequence] = timing + + def observe_provider(self, stage_id: str, sequence: int, duration_ns: int) -> None: + with self._lock: + stages = self._providers[sequence] + if stage_id in stages: + raise RuntimeError("provider timing stage was recorded twice") + stages[stage_id] = max(0, duration_ns) + + def take( + self, + *, + sequence: int, + source_age_ns: int, + completion_age_ns: int, + ) -> dict[str, object]: + with self._lock: + try: + decode_ns = self._decode.pop(sequence) + detector = self._detector.pop(sequence) + providers = self._providers.pop(sequence) + except KeyError as exc: + raise RuntimeError("delivered frame pipeline timing is incomplete") from exc + expected_stages = {"geometry", "temporal", "motion", "rolling", "threat"} + if set(providers) != expected_stages: + raise RuntimeError("delivered frame provider timing stages are incomplete") + admission_to_delivery_ns = max(0, completion_age_ns - source_age_ns) + provider_ns = sum(providers.values()) + attributed_graph_ns = detector.total_duration_ns + provider_ns + unattributed_ns = max(0, admission_to_delivery_ns - attributed_graph_ns) + document = { + "schema_version": PIPELINE_TIMING_SCHEMA, + "sequence": sequence, + "decode_duration_ns": decode_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, + } + with self._lock: + self._delivered.append(document) + return document + + def delivered(self) -> tuple[dict[str, object], ...]: + with self._lock: + return tuple(self._delivered) + + +class TimedProviderProxy: + """Record one provider's actual call duration without another inference pass.""" + + def __init__(self, stage_id: str, provider: object, store: FrameTimingStore) -> None: + self.stage_id = stage_id + self.provider = provider + self.store = store + self.provider_id = cast(Any, provider).provider_id + + def _call(self, sequence: int, method: str, *args: object) -> object: + started_ns = time.perf_counter_ns() + try: + return getattr(self.provider, method)(*args) + finally: + self.store.observe_provider( + self.stage_id, + sequence, + max(0, time.perf_counter_ns() - started_ns), + ) + + def associate(self, packet: SourcePacket, proposals: object) -> object: + return self._call(packet.envelope.sequence, "associate", packet, proposals) + + def update(self, packet: SourcePacket, values: object) -> object: + return self._call(packet.envelope.sequence, "update", packet, values) + + def estimate(self, packet: SourcePacket, obstacles: object) -> object: + return self._call(packet.envelope.sequence, "estimate", packet, obstacles) + + def assess(self, obstacle_map: object) -> object: + frame_id = cast(Any, obstacle_map).frame_id + try: + sequence = int(str(frame_id).rsplit("-", 1)[1]) + except (IndexError, ValueError) as exc: + raise RuntimeError("threat timing frame id is incompatible") from exc + return self._call(sequence, "assess", obstacle_map) + + def snapshot(self) -> object: + return cast(Any, self.provider).snapshot() + + def main() -> int: parser = argparse.ArgumentParser() for name in ( @@ -161,6 +270,7 @@ def main() -> int: completion_ages_ms: list[float] = [] map_output_ages_ms: list[float] = [] all_deliveries: list[DeliveredFrame] = [] + all_pipeline_timings: list[dict[str, object]] = [] with ( progress.open("x", encoding="utf-8") as progress_stream, frame_ledger.open("x", encoding="utf-8") as frame_ledger_stream, @@ -168,6 +278,7 @@ def main() -> int: ): for loop_index in range(arguments.loops): loop_completion_ages_ns: list[int] = [] + timing_store = FrameTimingStore() setup_started_ns = time.monotonic_ns() with build_m48s_reference_graph_runtime( @@ -183,9 +294,28 @@ def main() -> int: _record_delivery_evidence, frame_ledger_stream, loop_index, + timing_store, ), + decode_timing_observer=timing_store.observe_decode, + detector_timing_observer=timing_store.observe_detector, maximum_frames=arguments.maximum_frames, ) as runtime: + for stage_id, attribute in ( + ("geometry", "geometry"), + ("temporal", "temporal"), + ("motion", "motion"), + ("rolling", "rolling"), + ("threat", "threat"), + ): + setattr( + runtime.graph, + attribute, + TimedProviderProxy( + stage_id, + getattr(runtime.graph, attribute), + timing_store, + ), + ) loop_started_ns = time.monotonic_ns() result = runtime.graph.run() loop_completed_ns = time.monotonic_ns() @@ -225,6 +355,9 @@ def main() -> int: setup_seconds = (loop_started_ns - setup_started_ns) / 1_000_000_000.0 if len(loop_completion_ages_ns) != len(result.deliveries): raise RuntimeError("final delivery timing accounting did not close") + loop_pipeline_timings = timing_store.delivered() + if len(loop_pipeline_timings) != len(result.deliveries): + raise RuntimeError("pipeline timing accounting did not close") loop_document = _loop_document( loop_index=loop_index, result=result, @@ -241,6 +374,7 @@ def main() -> int: for delivery in result.deliveries ) all_deliveries.extend(result.deliveries) + all_pipeline_timings.extend(loop_pipeline_timings) frame_ledger_stream.flush() progress_row = { "loop": loop_index + 1, @@ -313,6 +447,7 @@ def main() -> int: "conservative_unknown_motion_advisory": _unknown_motion_is_conservative(advisories), "distinct_class_family_policy": len(set(advisory_policy_matrix().values())) == len(AdvisoryFamily), + "complete_pipeline_timing": len(all_pipeline_timings) == delivered, "authority_remains_false": all(value is False for value in AUTHORITY.values()), } integrated_runtime_gate_passed = all(checks.values()) @@ -351,6 +486,7 @@ def main() -> int: "local_obstacle_map_output_age_ms": _distribution(map_output_ages_ms), "identity_continuity": identity, "semantic_advisory": semantic, + "pipeline_timing": _pipeline_timing_metrics(all_pipeline_timings), "gpu": _telemetry_summary(gpu.samples), "process_peak_rss_before_mib": round(rss_before_kib / 1024.0, 6), "process_peak_rss_after_mib": round(rss_after_kib / 1024.0, 6), @@ -382,6 +518,7 @@ def _record_completion_age( def _record_delivery_evidence( stream: TextIO, loop_index: int, + timing_store: FrameTimingStore, delivery: DeliveredFrame, packet: SourcePacket, proposals: tuple[ObjectProposal2D, ...], @@ -396,6 +533,11 @@ def _record_delivery_evidence( "source_envelope": packet.envelope.to_dict(), "completion_age_ns": completion_age_ns, "local_obstacle_map_output_age_ns": delivery.obstacle_map.output_age_ns, + "pipeline_timing": timing_store.take( + sequence=delivery.sequence, + source_age_ns=packet.envelope.source_age_ns, + completion_age_ns=completion_age_ns, + ), "delivery": delivery.canonical_dict(), "detector_proposals": [proposal.to_dict() for proposal in proposals], "associated_proposal_ids": sorted(associated_proposal_ids), @@ -536,6 +678,67 @@ def _distribution(values: list[float]) -> dict[str, float]: } +def _pipeline_timing_metrics( + documents: list[dict[str, object]], +) -> dict[str, object]: + detector_fields = ( + "preprocess_duration_ns", + "inference_transport_duration_ns", + "postprocess_duration_ns", + "total_duration_ns", + ) + provider_fields = ("geometry", "temporal", "motion", "rolling", "threat") + detector_values: dict[str, list[float]] = {key: [] for key in detector_fields} + provider_values: dict[str, list[float]] = {key: [] for key in provider_fields} + top_level_fields = ( + "decode_duration_ns", + "graph_admission_to_delivery_ns", + "graph_attributed_provider_ns", + "graph_unattributed_ns", + "decode_to_delivery_processing_ns", + ) + top_level_values: dict[str, list[float]] = {key: [] for key in top_level_fields} + for document in documents: + detector = cast(Mapping[str, int], document["detector"]) + providers = cast(Mapping[str, int], document["providers"]) + for key in detector_fields: + detector_values[key].append(detector[key] / 1_000_000.0) + for key in provider_fields: + 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) + maximum = max( + documents, + key=lambda document: cast(int, document["graph_admission_to_delivery_ns"]), + default=None, + ) + return { + "sample_count": len(documents), + "decode_duration_ms": _distribution(top_level_values["decode_duration_ns"]), + "detector_ms": { + key.removesuffix("_duration_ns"): _distribution(values) + for key, values in detector_values.items() + }, + "provider_ms": { + key: _distribution(values) for key, values in provider_values.items() + }, + "graph_admission_to_delivery_ms": _distribution( + top_level_values["graph_admission_to_delivery_ns"] + ), + "graph_attributed_provider_ms": _distribution( + top_level_values["graph_attributed_provider_ns"] + ), + "graph_unattributed_ms": _distribution(top_level_values["graph_unattributed_ns"]), + "decode_to_delivery_processing_ms": _distribution( + top_level_values["decode_to_delivery_processing_ns"] + ), + "maximum_graph_sequence": ( + cast(int, maximum["sequence"]) if maximum is not None else None + ), + "additional_inference_passes": 0, + } + + 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/detector.py b/src/k1link/perception/detector.py index 292dfab..4286665 100644 --- a/src/k1link/perception/detector.py +++ b/src/k1link/perception/detector.py @@ -62,6 +62,45 @@ class DetectorProviderSnapshot: core_duration_ns: int +@dataclass(frozen=True, slots=True) +class DetectorFrameTiming: + sequence: int + preprocess_duration_ns: int + inference_transport_duration_ns: int + postprocess_duration_ns: int + total_duration_ns: int + + def __post_init__(self) -> None: + values = ( + self.sequence, + self.preprocess_duration_ns, + self.inference_transport_duration_ns, + self.postprocess_duration_ns, + self.total_duration_ns, + ) + if any(value < 0 for value in values): + raise DetectorProviderError("detector frame timing must be nonnegative") + if ( + self.preprocess_duration_ns + + self.inference_transport_duration_ns + + self.postprocess_duration_ns + != self.total_duration_ns + ): + raise DetectorProviderError("detector frame timing does not close") + + def to_dict(self) -> dict[str, int]: + return { + "sequence": self.sequence, + "preprocess_duration_ns": self.preprocess_duration_ns, + "inference_transport_duration_ns": self.inference_transport_duration_ns, + "postprocess_duration_ns": self.postprocess_duration_ns, + "total_duration_ns": self.total_duration_ns, + } + + +DetectorTimingObserver = Callable[[DetectorFrameTiming], None] + + class FrozenYoloxDetectorProvider: """One image payload produces one frozen inference request and proposal tuple.""" @@ -202,6 +241,7 @@ class RfDetrShadowDetectorProvider: resizer: ImageResizer | None = None, config: RfDetrConfig = RF_DETR_CONFIG, clock_ns: Callable[[], int] = time.perf_counter_ns, + timing_observer: DetectorTimingObserver | None = None, ) -> None: if mask.shape != (600, 800) or mask.dtype != np.bool_ or not np.any(mask): raise DetectorProviderError("RF-DETR valid-FOV mask is incompatible") @@ -210,6 +250,7 @@ class RfDetrShadowDetectorProvider: self.resizer = resizer self.config = config self._clock_ns = clock_ns + self.timing_observer = timing_observer self._lock = Lock() self._input_frames = 0 self._completed_frames = 0 @@ -236,7 +277,13 @@ class RfDetrShadowDetectorProvider: config=self.config, resizer=self.resizer, ) + preprocessed_ns = ( + int(self._clock_ns()) if self.timing_observer is not None else started_ns + ) output = self.backend.infer(tensor) + inferred_ns = ( + int(self._clock_ns()) if self.timing_observer is not None else preprocessed_ns + ) postprocessed = postprocess_rf_detr(output, self.mask, config=self.config) proposals = proposals_from_rf_detr_detections(packet, postprocessed.detections) except Exception: @@ -244,12 +291,23 @@ class RfDetrShadowDetectorProvider: self._failed_frames += 1 self._core_duration_ns += max(0, int(self._clock_ns()) - started_ns) raise + completed_ns = int(self._clock_ns()) with self._lock: self._completed_frames += 1 self._proposal_count += len(proposals) self._zero_proposal_frames += not proposals self._rejected.update(dict(postprocessed.rejected)) - self._core_duration_ns += max(0, int(self._clock_ns()) - started_ns) + self._core_duration_ns += max(0, completed_ns - started_ns) + if self.timing_observer is not None: + self.timing_observer( + DetectorFrameTiming( + sequence=packet.envelope.sequence, + preprocess_duration_ns=max(0, preprocessed_ns - started_ns), + inference_transport_duration_ns=max(0, inferred_ns - preprocessed_ns), + postprocess_duration_ns=max(0, completed_ns - inferred_ns), + total_duration_ns=max(0, completed_ns - started_ns), + ) + ) return proposals def snapshot(self) -> DetectorProviderSnapshot: @@ -297,6 +355,8 @@ __all__ = [ "RF_DETR_SHADOW_PROVIDER_ID", "DetectorProviderError", "DetectorProviderSnapshot", + "DetectorFrameTiming", + "DetectorTimingObserver", "AllCocoYoloxDetectorProvider", "FrozenYoloxDetectorProvider", "RfDetrShadowDetectorProvider", diff --git a/src/k1link/perception/m48s_reference_graph_runtime.py b/src/k1link/perception/m48s_reference_graph_runtime.py index adbcff9..d8a8e1d 100644 --- a/src/k1link/perception/m48s_reference_graph_runtime.py +++ b/src/k1link/perception/m48s_reference_graph_runtime.py @@ -10,7 +10,11 @@ from pathlib import Path from threading import Event from .baseline import load_m4_baseline -from .detector import RF_DETR_SHADOW_PROVIDER_ID, RfDetrShadowDetectorProvider +from .detector import ( + RF_DETR_SHADOW_PROVIDER_ID, + DetectorTimingObserver, + RfDetrShadowDetectorProvider, +) from .geometry import ( Ravnoves00GeometryAssociationProvider, RecordedGeometryStore, @@ -27,6 +31,7 @@ from .providers import ( ) from .recorded_source import ( DecodedRecordedSource, + DecodeTimingObserver, PyAvRecordedImageDecoder, RecordedRavnoves00Source, ReplayPacing, @@ -77,6 +82,8 @@ def build_m48s_reference_graph_runtime( run_mode: GraphRunMode, delivery_observer: Callable[[DeliveredFrame, int], None] | None = None, delivery_evidence_observer: DeliveryEvidenceObserver | None = None, + decode_timing_observer: DecodeTimingObserver | None = None, + detector_timing_observer: DetectorTimingObserver | None = None, maximum_frames: int | None = None, ) -> M48sReferenceGraphRuntime: """Instantiate the complete graph with only its detector pin replaced.""" @@ -113,6 +120,7 @@ def build_m48s_reference_graph_runtime( ), ), decoder=PyAvRecordedImageDecoder(paths.video), + timing_observer=decode_timing_observer, ) if maximum_frames is not None: source = _LimitedSource(source, maximum_frames) @@ -133,6 +141,7 @@ def build_m48s_reference_graph_runtime( detector=RfDetrShadowDetectorProvider( mask=load_valid_fov_mask(paths.valid_fov_mask), backend=backend, + timing_observer=detector_timing_observer, ), geometry=Ravnoves00GeometryAssociationProvider(store=store), temporal=BoundedSpatialTemporalProvider( diff --git a/src/k1link/perception/recorded_source.py b/src/k1link/perception/recorded_source.py index a16fde7..36703c9 100644 --- a/src/k1link/perception/recorded_source.py +++ b/src/k1link/perception/recorded_source.py @@ -85,6 +85,19 @@ class RecordedImageDecoder(Protocol): WaitFunction = Callable[[Event, float], bool] +@dataclass(frozen=True, slots=True) +class DecodedFrameTiming: + sequence: int + duration_ns: int + + def __post_init__(self) -> None: + if self.sequence < 0 or self.duration_ns < 0: + raise RecordedSourceError("decoded frame timing must be nonnegative") + + +DecodeTimingObserver = Callable[[DecodedFrameTiming], None] + + class RecordedRavnoves00Source: """Emit the admitted synchronized source timeline at 1.0x or uncapped speed.""" @@ -182,14 +195,18 @@ class DecodedRecordedSource: *, source: RecordedRavnoves00Source, decoder: RecordedImageDecoder, + timing_observer: DecodeTimingObserver | None = None, + clock_ns: Callable[[], int] = time.perf_counter_ns, ) -> None: self.source = source self.decoder = decoder + self.timing_observer = timing_observer + self._clock_ns = clock_ns def packets(self, stop_event: Event) -> Iterator[SourcePacket]: images = self.decoder.frames(stop_event) try: - image: NDArray[np.uint8] | None = next(images) + image: NDArray[np.uint8] | None = self._next_image(images, 0) except StopIteration as exc: if stop_event.is_set(): return @@ -203,7 +220,7 @@ class DecodedRecordedSource: raise RecordedSourceError("decoded image raster is incompatible") yield replace(packet, image_payload=image) try: - image = next(images) + image = self._next_image(images, packet.envelope.sequence + 1) except StopIteration: image = None if not stop_event.is_set(): @@ -211,6 +228,24 @@ class DecodedRecordedSource: return raise RecordedSourceError("decoded image stream exceeds source timeline") + def _next_image( + self, + images: Iterator[NDArray[np.uint8]], + sequence: int, + ) -> NDArray[np.uint8]: + observer = self.timing_observer + if observer is None: + return next(images) + started_ns = int(self._clock_ns()) + image = next(images) + observer( + DecodedFrameTiming( + sequence=sequence, + duration_ns=max(0, int(self._clock_ns()) - started_ns), + ) + ) + return image + class PyAvRecordedImageDecoder: """Sequential full-video decoder used by the Worker 006 recorded source adapter.""" @@ -394,6 +429,8 @@ def _event_wait(stop_event: Event, timeout_seconds: float) -> bool: __all__ = [ "BASELINE_PROFILE_ID", + "DecodeTimingObserver", + "DecodedFrameTiming", "LiveSourceAdapter", "DecodedRecordedSource", "PyAvRecordedImageDecoder", diff --git a/tests/test_m48s_reference_graph_timing.py b/tests/test_m48s_reference_graph_timing.py new file mode 100644 index 0000000..cafc713 --- /dev/null +++ b/tests/test_m48s_reference_graph_timing.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +import importlib.util +from pathlib import Path + +from k1link.perception.detector import DetectorFrameTiming +from k1link.perception.recorded_source import DecodedFrameTiming + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +RUNNER_PATH = ( + REPOSITORY_ROOT / "experiments/perception/run_m48s_reference_graph_shadow_worker.py" +) +SPEC = importlib.util.spec_from_file_location("m48s_timed_shadow_runner", RUNNER_PATH) +assert SPEC is not None and SPEC.loader is not None +RUNNER = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(RUNNER) + + +def test_frame_timing_store_closes_provider_and_unattributed_time() -> None: + store = RUNNER.FrameTimingStore() + store.observe_decode(DecodedFrameTiming(sequence=7, duration_ns=5)) + store.observe_detector( + DetectorFrameTiming( + sequence=7, + preprocess_duration_ns=10, + inference_transport_duration_ns=20, + postprocess_duration_ns=30, + total_duration_ns=60, + ) + ) + for stage_id, duration_ns in ( + ("geometry", 1), + ("temporal", 2), + ("motion", 3), + ("rolling", 4), + ("threat", 5), + ): + store.observe_provider(stage_id, 7, duration_ns) + + document = store.take(sequence=7, source_age_ns=100, completion_age_ns=300) + + assert document["schema_version"] == RUNNER.PIPELINE_TIMING_SCHEMA + assert document["graph_admission_to_delivery_ns"] == 200 + assert document["graph_attributed_provider_ns"] == 75 + assert document["graph_unattributed_ns"] == 125 + assert document["decode_to_delivery_processing_ns"] == 205 + assert store.delivered() == (document,) + + +def test_pipeline_timing_metrics_preserve_single_pass_stage_breakdown() -> None: + store = RUNNER.FrameTimingStore() + store.observe_decode(DecodedFrameTiming(sequence=3, duration_ns=1_000_000)) + store.observe_detector( + DetectorFrameTiming( + sequence=3, + preprocess_duration_ns=2_000_000, + inference_transport_duration_ns=3_000_000, + postprocess_duration_ns=4_000_000, + total_duration_ns=9_000_000, + ) + ) + for stage_id, duration_ns in ( + ("geometry", 1_000_000), + ("temporal", 2_000_000), + ("motion", 3_000_000), + ("rolling", 4_000_000), + ("threat", 5_000_000), + ): + store.observe_provider(stage_id, 3, duration_ns) + document = store.take( + sequence=3, + source_age_ns=0, + completion_age_ns=30_000_000, + ) + + metrics = RUNNER._pipeline_timing_metrics([document]) + + assert metrics["sample_count"] == 1 + assert metrics["detector_ms"]["inference_transport"]["maximum"] == 3.0 + assert metrics["provider_ms"]["threat"]["maximum"] == 5.0 + assert metrics["graph_unattributed_ms"]["maximum"] == 6.0 + assert metrics["maximum_graph_sequence"] == 3 + assert metrics["additional_inference_passes"] == 0 diff --git a/tests/test_perception_graph.py b/tests/test_perception_graph.py index f3294a6..cf84e27 100644 --- a/tests/test_perception_graph.py +++ b/tests/test_perception_graph.py @@ -60,6 +60,7 @@ from k1link.perception.providers import ( SourcePacket, ) from k1link.perception.recorded_source import ( + DecodedFrameTiming, DecodedRecordedSource, RecordedRavnoves00Source, RecordedSourceError, @@ -869,6 +870,34 @@ def test_decoded_recorded_source_attaches_images_without_detector_logic(tmp_path 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] = [] diff --git a/tests/test_rf_detr_detector_provider.py b/tests/test_rf_detr_detector_provider.py index 81e9ccd..8104266 100644 --- a/tests/test_rf_detr_detector_provider.py +++ b/tests/test_rf_detr_detector_provider.py @@ -20,6 +20,7 @@ from k1link.perception.detector import ( RF_DETR_SHADOW_MODEL_ID, RF_DETR_SHADOW_PREPROCESS_ID, RF_DETR_SHADOW_PROVIDER_ID, + DetectorFrameTiming, RfDetrShadowDetectorProvider, ) from k1link.perception.providers import SourcePacket @@ -170,6 +171,28 @@ def test_shadow_provider_uses_one_pass_and_preserves_semantic_hints() -> None: assert provider.snapshot().core_duration_ns == 20 +def test_shadow_provider_reports_preprocess_transport_and_postprocess_timing() -> None: + observed: list[DetectorFrameTiming] = [] + provider = RfDetrShadowDetectorProvider( + mask=np.ones((600, 800), dtype=np.bool_), + backend=_Backend(_output()), + resizer=_Resizer(), + clock_ns=iter((10, 20, 50, 70)).__next__, + timing_observer=observed.append, + ) + + provider.detect(_packet(7, np.zeros((600, 800, 3), dtype=np.uint8))) + + assert len(observed) == 1 + assert observed[0].to_dict() == { + "sequence": 7, + "preprocess_duration_ns": 10, + "inference_transport_duration_ns": 30, + "postprocess_duration_ns": 20, + "total_duration_ns": 60, + } + + def test_shadow_profile_is_fixed_and_transport_pins_model_version() -> None: assert RF_DETR_CONFIG.minimum_score == 0.25 with pytest.raises(RfDetrDetectorError, match="cannot be tuned"):