From 84624ae3ea2342aee069f84b3351c246a5315c4b Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Tue, 25 Aug 2026 19:09:53 +0300 Subject: [PATCH] feat(perception): prewarm RF-DETR before source admission --- .../run_m48s_reference_graph_shadow_worker.py | 16 ++++- src/k1link/perception/detector.py | 70 +++++++++++++++++++ .../m48s_reference_graph_runtime.py | 21 +++--- tests/test_rf_detr_detector_provider.py | 41 +++++++---- 4 files changed, 124 insertions(+), 24 deletions(-) diff --git a/experiments/perception/run_m48s_reference_graph_shadow_worker.py b/experiments/perception/run_m48s_reference_graph_shadow_worker.py index e3982bf..16e8c98 100644 --- a/experiments/perception/run_m48s_reference_graph_shadow_worker.py +++ b/experiments/perception/run_m48s_reference_graph_shadow_worker.py @@ -24,6 +24,7 @@ from k1link.perception.contracts import MotionState, ObjectProposal2D from k1link.perception.detector import ( DetectorFrameTiming, DetectorProviderSnapshot, + DetectorWarmupSnapshot, RfDetrShadowDetectorProvider, ) from k1link.perception.geometry import Ravnoves00GeometryAssociationProvider @@ -51,7 +52,7 @@ 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/v2" +SCHEMA_VERSION: Final = "missioncore.m48s-reference-graph-shadow-load/v3" FRAME_EVIDENCE_SCHEMA: Final = "missioncore.m48s-reference-graph-frame-evidence/v1" PIPELINE_TIMING_SCHEMA: Final = "missioncore.m48s-frame-pipeline-timing/v0" GC_POLICY_SCHEMA: Final = "missioncore.cyclic-gc-hot-loop-policy/v0" @@ -443,6 +444,7 @@ def main() -> int: gc_telemetry, ), ) + detector_warmup = runtime.warm_up_detector() gc_policy = CyclicGcHotLoopPolicy() with gc_policy: loop_started_ns = time.monotonic_ns() @@ -496,6 +498,7 @@ def main() -> int: wall_seconds=loop_wall_seconds, setup_seconds=setup_seconds, gc_policy=gc_policy.to_dict(), + detector_warmup=detector_warmup, ) loop_documents.append(loop_document) completion_ages_ms.extend(value / 1_000_000.0 for value in loop_completion_ages_ns) @@ -578,6 +581,11 @@ def main() -> int: is cast(dict[str, object], loop["cyclic_gc_hot_loop"])["enabled_after"] for loop in loop_documents ), + "detector_warmup_completed_before_source_admission": all( + cast(dict[str, object], loop["detector_warmup"])["completed"] is True + and cast(dict[str, object], loop["detector_warmup"])["inference_passes"] == 1 + for loop in loop_documents + ), "authority_remains_false": all(value is False for value in AUTHORITY.values()), } integrated_runtime_gate_passed = all(checks.values()) @@ -604,6 +612,10 @@ def main() -> int: "terminal_outcomes": dict(sorted(accounting.items())), "queue_high_watermarks": queue_high_watermarks, "loops": loop_documents, + "startup_warmup_inference_passes": sum( + cast(int, cast(dict[str, object], loop["detector_warmup"])["inference_passes"]) + for loop in loop_documents + ), "frame_evidence": { "schema_version": FRAME_EVIDENCE_SCHEMA, "path": frame_ledger.name, @@ -690,6 +702,7 @@ def _loop_document( wall_seconds: float, setup_seconds: float, gc_policy: dict[str, object], + detector_warmup: DetectorWarmupSnapshot, ) -> dict[str, object]: outcomes = Counter(item.outcome.value for item in result.terminal_outcomes) outcome_stages = Counter( @@ -701,6 +714,7 @@ def _loop_document( "wall_seconds": round(wall_seconds, 6), "setup_seconds": round(setup_seconds, 6), "cyclic_gc_hot_loop": gc_policy, + "detector_warmup": asdict(detector_warmup), "admitted_count": result.admitted_count, "delivered_count": len(result.deliveries), "effective_world_state_fps": round(len(result.deliveries) / wall_seconds, 6), diff --git a/src/k1link/perception/detector.py b/src/k1link/perception/detector.py index 4286665..1875226 100644 --- a/src/k1link/perception/detector.py +++ b/src/k1link/perception/detector.py @@ -101,6 +101,31 @@ class DetectorFrameTiming: DetectorTimingObserver = Callable[[DetectorFrameTiming], None] +@dataclass(frozen=True, slots=True) +class DetectorWarmupSnapshot: + completed: bool + inference_passes: int + preprocess_duration_ns: int + inference_transport_duration_ns: int + postprocess_duration_ns: int + total_duration_ns: int + + def __post_init__(self) -> None: + durations = ( + self.preprocess_duration_ns, + self.inference_transport_duration_ns, + self.postprocess_duration_ns, + self.total_duration_ns, + ) + if ( + self.completed is not True + or self.inference_passes != 1 + or any(value < 0 for value in durations) + or sum(durations[:3]) != self.total_duration_ns + ): + raise DetectorProviderError("detector warmup snapshot is incompatible") + + class FrozenYoloxDetectorProvider: """One image payload produces one frozen inference request and proposal tuple.""" @@ -259,6 +284,50 @@ class RfDetrShadowDetectorProvider: self._proposal_count = 0 self._rejected: Counter[str] = Counter() self._core_duration_ns = 0 + self._warmup_started = False + self._warmup_snapshot: DetectorWarmupSnapshot | None = None + + def warm_up(self) -> DetectorWarmupSnapshot: + """Prime preprocessing, transport and postprocessing before source admission.""" + + with self._lock: + if self._warmup_snapshot is not None: + return self._warmup_snapshot + if self._warmup_started: + raise DetectorProviderError("RF-DETR warmup is already in progress") + self._warmup_started = True + started_ns = int(self._clock_ns()) + try: + image = np.zeros( + (self.config.source_height, self.config.source_width, 3), + dtype=np.uint8, + ) + tensor = preprocess_raw_kb4_rf_detr( + image, + self.mask, + config=self.config, + resizer=self.resizer, + ) + preprocessed_ns = int(self._clock_ns()) + output = self.backend.infer(tensor) + inferred_ns = int(self._clock_ns()) + postprocess_rf_detr(output, self.mask, config=self.config) + completed_ns = int(self._clock_ns()) + except Exception: + with self._lock: + self._warmup_started = False + raise + snapshot = DetectorWarmupSnapshot( + completed=True, + inference_passes=1, + 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), + ) + with self._lock: + self._warmup_snapshot = snapshot + return snapshot def detect(self, packet: SourcePacket) -> tuple[ObjectProposal2D, ...]: payload = packet.image_payload @@ -357,6 +426,7 @@ __all__ = [ "DetectorProviderSnapshot", "DetectorFrameTiming", "DetectorTimingObserver", + "DetectorWarmupSnapshot", "AllCocoYoloxDetectorProvider", "FrozenYoloxDetectorProvider", "RfDetrShadowDetectorProvider", diff --git a/src/k1link/perception/m48s_reference_graph_runtime.py b/src/k1link/perception/m48s_reference_graph_runtime.py index d8a8e1d..877256e 100644 --- a/src/k1link/perception/m48s_reference_graph_runtime.py +++ b/src/k1link/perception/m48s_reference_graph_runtime.py @@ -13,6 +13,7 @@ from .baseline import load_m4_baseline from .detector import ( RF_DETR_SHADOW_PROVIDER_ID, DetectorTimingObserver, + DetectorWarmupSnapshot, RfDetrShadowDetectorProvider, ) from .geometry import ( @@ -64,6 +65,12 @@ class M48sReferenceGraphRuntime: graph: ReferencePerceptionGraphV2 inference_backend: TritonRfDetrHttpInferenceBackend + def warm_up_detector(self) -> DetectorWarmupSnapshot: + detector = self.graph.detector + if not isinstance(detector, RfDetrShadowDetectorProvider): + raise M48sReferenceGraphRuntimeError("RF-DETR runtime detector changed before warmup") + return detector.warm_up() + def close(self) -> None: self.inference_backend.close() @@ -202,9 +209,7 @@ def _validate_provider_digests( raise M48sReferenceGraphRuntimeError("RF-DETR graph provider pins are incomplete") for role, path in pinned_files.items(): if _sha256_file(path) != pins[role].sha256: - raise M48sReferenceGraphRuntimeError( - f"{role.value} provider profile digest changed" - ) + raise M48sReferenceGraphRuntimeError(f"{role.value} provider profile digest changed") def _validate_detector_profile(path: Path) -> None: @@ -216,13 +221,11 @@ def _validate_detector_profile(path: Path) -> None: except (OSError, KeyError, TypeError, json.JSONDecodeError) as exc: raise M48sReferenceGraphRuntimeError("RF-DETR profile is incomplete") from exc if ( - document.get("schema_version") - != "missioncore.rf-detr-risk-shadow-profile/v0" + document.get("schema_version") != "missioncore.rf-detr-risk-shadow-profile/v0" or document.get("provider_id") != RF_DETR_SHADOW_PROVIDER_ID or model.get("model_id") != RF_DETR_MODEL_ID or model.get("model_version") != RF_DETR_MODEL_VERSION - or model.get("worker_006_rtx4090_tensorrt_11_engine_sha256") - != RF_DETR_ENGINE_SHA256 + or model.get("worker_006_rtx4090_tensorrt_11_engine_sha256") != RF_DETR_ENGINE_SHA256 or status.get("detector_load_gate_passed") is not True or status.get("production_accepted") is not False or any( @@ -244,9 +247,7 @@ def _sha256_file(path: Path) -> str: except OSError as exc: raise M48sReferenceGraphRuntimeError("pinned RF-DETR graph input is missing") from exc if resolved.is_symlink() or not resolved.is_file(): - raise M48sReferenceGraphRuntimeError( - "pinned RF-DETR graph input must be a regular file" - ) + raise M48sReferenceGraphRuntimeError("pinned RF-DETR graph input must be a regular file") digest = hashlib.sha256() with resolved.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): diff --git a/tests/test_rf_detr_detector_provider.py b/tests/test_rf_detr_detector_provider.py index 8104266..4a72337 100644 --- a/tests/test_rf_detr_detector_provider.py +++ b/tests/test_rf_detr_detector_provider.py @@ -74,9 +74,7 @@ class _Resizer: def __init__(self) -> None: self.source: NDArray[np.uint8] | None = None - def resize( - self, image: NDArray[np.uint8], width: int, height: int - ) -> NDArray[np.uint8]: + def resize(self, image: NDArray[np.uint8], width: int, height: int) -> NDArray[np.uint8]: self.source = image.copy() output = np.empty((height, width, 3), dtype=np.uint8) output[:, :, 0] = 255 @@ -135,9 +133,7 @@ def test_postprocess_maps_sparse_coco_slots_and_emits_only_risk_classes() -> Non assert tuple(item.label for item in result.detections) == ("person", "dog") assert result.detections[0].score == pytest.approx(0.8, abs=0.001) assert result.detections[1].score == pytest.approx(0.75, abs=0.001) - assert result.detections[1].bbox_xyxy == pytest.approx( - (300.0, 200.0, 500.0, 400.0), abs=0.03 - ) + assert result.detections[1].bbox_xyxy == pytest.approx((300.0, 200.0, 500.0, 400.0), abs=0.03) assert dict(result.rejected) == {"non-risk-class": 1, "unmapped-class-slot": 1} with pytest.raises(RfDetrDetectorError, match="tensor types"): @@ -193,6 +189,30 @@ def test_shadow_provider_reports_preprocess_transport_and_postprocess_timing() - } +def test_shadow_provider_warmup_is_idempotent_and_excluded_from_frame_counts() -> None: + backend = _Backend(_output()) + provider = RfDetrShadowDetectorProvider( + mask=np.ones((600, 800), dtype=np.bool_), + backend=backend, + resizer=_Resizer(), + clock_ns=iter((10, 20, 50, 70)).__next__, + ) + + first = provider.warm_up() + second = provider.warm_up() + + assert first is second + assert backend.calls == 1 + assert first.completed is True + assert first.inference_passes == 1 + assert first.preprocess_duration_ns == 10 + assert first.inference_transport_duration_ns == 30 + assert first.postprocess_duration_ns == 20 + assert first.total_duration_ns == 60 + assert provider.snapshot().input_frames == 0 + assert provider.snapshot().completed_frames == 0 + + 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"): @@ -209,16 +229,11 @@ def test_shadow_profile_is_fixed_and_transport_pins_model_version() -> None: def test_shadow_profile_pins_worker_engine_and_retains_false_authority() -> None: profile = json.loads( - (REPOSITORY_ROOT / "config/perception/rf-detr-large-risk-shadow-v0.json").read_text( - "utf-8" - ) + (REPOSITORY_ROOT / "config/perception/rf-detr-large-risk-shadow-v0.json").read_text("utf-8") ) assert profile["model"]["strongly_typed_fp16_onnx_sha256"] == RF_DETR_FP16_ONNX_SHA256 - assert ( - profile["model"]["worker_006_rtx4090_tensorrt_11_engine_sha256"] - == RF_DETR_ENGINE_SHA256 - ) + assert profile["model"]["worker_006_rtx4090_tensorrt_11_engine_sha256"] == RF_DETR_ENGINE_SHA256 assert profile["emission"]["single_inference_per_source_frame"] is True assert profile["emission"]["geometry_owns_static_occupancy"] is True assert profile["emission"]["unlisted_semantic_classes_emitted"] is False