From bfe6ef4c77ccb71b40ab4b0a4d450c25a2ccea25 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Wed, 2 Sep 2026 00:59:48 +0300 Subject: [PATCH] feat(perception): add bounded full-graph streaming and latency pilots --- .../Dockerfile.dependency-probe | 17 + .../Dockerfile.joint-pilot | 40 ++ .../baseline_ddrnet.py | 127 ++++ .../baseline_surface.py | 32 + .../streaming_profile_stage1/baseline_tgs.cpp | 61 ++ .../pilot_ddrnet_runtime.py | 138 ++++ .../streaming_profile_stage1/pilot_graph.py | 451 +++++++++++++ .../streaming_profile_stage1/pilot_ipc.py | 41 ++ .../streaming_profile_stage1/pilot_model.py | 110 ++++ .../pilot_preflight.py | 69 ++ .../streaming_profile_stage1/pilot_queue.py | 74 +++ .../pilot_scheduler.py | 61 ++ .../streaming_profile_stage1/pilot_source.py | 154 +++++ .../streaming_profile_stage1/pilot_tgs.cpp | 62 ++ .../rfdetr-config.pbtxt | 9 + .../run_joint_pilot.py | 605 ++++++++++++++++++ tests/test_perception_joint_pilot.py | 216 +++++++ 17 files changed, 2267 insertions(+) create mode 100644 experiments/perception/worker/streaming_profile_stage1/Dockerfile.dependency-probe create mode 100644 experiments/perception/worker/streaming_profile_stage1/Dockerfile.joint-pilot create mode 100644 experiments/perception/worker/streaming_profile_stage1/baseline_ddrnet.py create mode 100644 experiments/perception/worker/streaming_profile_stage1/baseline_surface.py create mode 100644 experiments/perception/worker/streaming_profile_stage1/baseline_tgs.cpp create mode 100644 experiments/perception/worker/streaming_profile_stage1/pilot_ddrnet_runtime.py create mode 100644 experiments/perception/worker/streaming_profile_stage1/pilot_graph.py create mode 100644 experiments/perception/worker/streaming_profile_stage1/pilot_ipc.py create mode 100644 experiments/perception/worker/streaming_profile_stage1/pilot_model.py create mode 100644 experiments/perception/worker/streaming_profile_stage1/pilot_preflight.py create mode 100644 experiments/perception/worker/streaming_profile_stage1/pilot_queue.py create mode 100644 experiments/perception/worker/streaming_profile_stage1/pilot_scheduler.py create mode 100644 experiments/perception/worker/streaming_profile_stage1/pilot_source.py create mode 100644 experiments/perception/worker/streaming_profile_stage1/pilot_tgs.cpp create mode 100644 experiments/perception/worker/streaming_profile_stage1/rfdetr-config.pbtxt create mode 100644 experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py create mode 100644 tests/test_perception_joint_pilot.py diff --git a/experiments/perception/worker/streaming_profile_stage1/Dockerfile.dependency-probe b/experiments/perception/worker/streaming_profile_stage1/Dockerfile.dependency-probe new file mode 100644 index 0000000..38876fb --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/Dockerfile.dependency-probe @@ -0,0 +1,17 @@ +# Stage-1 disposable ABI/process-isolation probe, NOT a released full profile. +# Before building, verify both local base tags against the IDs in the report. +FROM ndc/mission-core-installed-lab-v1-ddrnet-step:439127908dba AS goose +FROM ndc/mission-core-m49-t3-travel:20260826 AS travel +COPY baseline_tgs.cpp /tmp/baseline_tgs.cpp +RUN g++ -std=c++17 -O3 -DNDEBUG -pthread \ + -I/opt/travel/src/TRAVEL/cpp/travel/core -I/usr/include/eigen3 \ + /tmp/baseline_tgs.cpp -o /tmp/stage1-tgs + +FROM nvcr.io/nvidia/tritonserver:26.06-py3 +COPY --from=goose /opt/conda /opt/conda +COPY --from=travel /tmp/stage1-tgs /usr/local/bin/stage1-tgs +LABEL com.nodedc.product="mission-core" \ + com.nodedc.stack="observatory" \ + com.nodedc.role="stage1-dependency-probe" \ + com.nodedc.managed-by="codex-local" +ENTRYPOINT ["/bin/bash"] diff --git a/experiments/perception/worker/streaming_profile_stage1/Dockerfile.joint-pilot b/experiments/perception/worker/streaming_profile_stage1/Dockerfile.joint-pilot new file mode 100644 index 0000000..0f36f47 --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/Dockerfile.joint-pilot @@ -0,0 +1,40 @@ +# Bounded stage-1 common-graph pilot, NOT the stage-2 standalone release. +# Verify local base tags against the recorded image IDs before build. +FROM ndc/mission-core-installed-lab-v1-ddrnet-step:439127908dba AS ddr +FROM ndc/mission-core-m49-t3-travel:20260826 AS tgs +COPY pilot_tgs.cpp /tmp/pilot_tgs.cpp +RUN g++ -std=c++17 -O3 -DNDEBUG -pthread \ + -I/opt/travel/src/TRAVEL/cpp/travel/core -I/usr/include/eigen3 \ + /tmp/pilot_tgs.cpp -o /tmp/pilot-tgs + +FROM nvcr.io/nvidia/tritonserver:26.06-py3 AS pilot +COPY --from=ddr /opt/conda /opt/conda +# Exact version from repository uv.lock; only the CPU supervisor needs this. +RUN python3 -m pip install --no-cache-dir --only-binary=:all: pillow==12.3.0 +COPY --from=tgs /tmp/pilot-tgs /usr/local/bin/pilot-tgs +COPY source-code.tar.gz /tmp/source-code.tar.gz +RUN mkdir /code && tar --no-same-owner -xzf /tmp/source-code.tar.gz -C /code \ + && mkdir -p /models/rf_detr_large_native_kb4/1 /root/sg_logs +COPY *.py /probe/ +COPY run_goose_vegetation_benchmark.py /probe/run_goose_vegetation_benchmark.py +COPY rfdetr-config.pbtxt /models/rf_detr_large_native_kb4/config.pbtxt +ENV PYTHONPATH=/code/src \ + OPENBLAS_NUM_THREADS=1 OMP_NUM_THREADS=1 MKL_NUM_THREADS=1 \ + MPLCONFIGDIR=/tmp/matplotlib +LABEL com.nodedc.product="mission-core" com.nodedc.stack="observatory" \ + com.nodedc.role="stage1-joint-pilot" com.nodedc.managed-by="codex-local" +ENTRYPOINT ["python3", "-B", "/probe/run_joint_pilot.py"] + +# Bounded local fallback for a daemon with missing compressed registry blobs. +# Uses only already-built files; no new registry access or credential change. +# This overlay is a diagnostic pilot, NEVER a standalone image release. +FROM scratch AS pilot-assets +COPY --from=pilot /probe /probe +COPY --from=pilot /code /code +COPY --from=pilot /models /models +COPY --from=pilot /usr/local/bin/pilot-tgs /bin/pilot-tgs +COPY --from=pilot /usr/local/lib/python3.12/dist-packages/PIL /python/PIL +COPY --from=pilot /usr/local/lib/python3.12/dist-packages/pillow.libs /python/pillow.libs + +# Default docker build remains a runnable image, not the diagnostic overlay. +FROM pilot AS joint-pilot diff --git a/experiments/perception/worker/streaming_profile_stage1/baseline_ddrnet.py b/experiments/perception/worker/streaming_profile_stage1/baseline_ddrnet.py new file mode 100644 index 0000000..82a4555 --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/baseline_ddrnet.py @@ -0,0 +1,127 @@ +#!/usr/bin/env python3 +"""Bounded component baseline, NOT full-profile or transport qualification. + +Run only in the pinned installed DDRNet image on an exclusive Worker. The +existing video is decoded incrementally; it is not copied, hashed end-to-end, +cached as PNGs or fully decoded before the first measured frame. +""" + +from __future__ import annotations + +import argparse +import hashlib +import importlib.machinery +import importlib.util +import json +import signal +import statistics +import time +from datetime import datetime, timezone +from pathlib import Path + + +def digest(path: Path) -> str: + result = hashlib.sha256() + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + result.update(chunk) + return result.hexdigest() + + +def distribution(values: list[float]) -> dict[str, float]: + ordered = sorted(values) + return { + "mean": statistics.fmean(ordered), + "p50": ordered[int((len(ordered) - 1) * 0.50)], + "p95": ordered[int((len(ordered) - 1) * 0.95)], + "p99": ordered[int((len(ordered) - 1) * 0.99)], + "maximum": max(ordered), + } + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--video", type=Path, required=True) + parser.add_argument("--runner", type=Path, required=True) + parser.add_argument("--checkpoint", type=Path, required=True) + parser.add_argument("--samples", type=int, default=64, choices=range(8, 129)) + parser.add_argument("--warmup", type=int, default=8, choices=range(1, 17)) + args = parser.parse_args() + signal.alarm(180) + started = time.monotonic_ns() + utc = datetime.now(timezone.utc).isoformat() + runner = args.runner + checkpoint = args.checkpoint + expected_checkpoint = "b99c2838051bcd7b092fd3970aa62a77d5c0bbb809c9b9afb2ff4b0ebdaa4ee6" + if digest(checkpoint) != expected_checkpoint: + raise RuntimeError("checkpoint identity changed") + if digest(runner) != "b18ad60f277eea69a240a28f290611b94627fb9707faf1bb3e6e22102dad67c1": + raise RuntimeError("reused runner identity changed") + loader = importlib.machinery.SourceFileLoader("goose_baseline_runner", str(runner)) + spec = importlib.util.spec_from_loader(loader.name, loader) + if spec is None: + raise RuntimeError("runner is unavailable") + module = importlib.util.module_from_spec(spec) + loader.exec_module(module) + import cv2 + import torch + from PIL import Image + + capture = cv2.VideoCapture(str(args.video)) + try: + ok, bgr = capture.read() + if not ok or bgr is None or bgr.shape != (600, 800, 3): + raise RuntimeError("first native KB4 frame unavailable") + tensor, _ = module.preprocess(Image.fromarray(cv2.cvtColor(bgr, cv2.COLOR_BGR2RGB))) + model, name, _ = module.load_model("ddrnet", checkpoint) + for _ in range(args.warmup): + module.infer(model, tensor) + warm_ns = time.monotonic_ns() + torch.cuda.reset_peak_memory_stats() + rows = [] + mask_digest = hashlib.sha256() + for sequence in range(args.samples): + begin = time.monotonic_ns() + ok, bgr = capture.read() + decoded = time.monotonic_ns() + if not ok or bgr is None or bgr.shape != (600, 800, 3): + raise RuntimeError("bounded source ended or changed shape") + tensor, _ = module.preprocess(Image.fromarray(cv2.cvtColor(bgr, cv2.COLOR_BGR2RGB))) + preprocessed = time.monotonic_ns() + mask, forward_ms = module.infer(model, tensor) + completed = time.monotonic_ns() + mask_digest.update(mask.tobytes()) + rows.append({ + "sequence": sequence + 1, + "decode_ms": (decoded - begin) / 1e6, + "preprocess_ms": (preprocessed - decoded) / 1e6, + "h2d_forward_post_d2h_ms": (completed - preprocessed) / 1e6, + "forward_post_ms": forward_ms, + "component_total_ms": (completed - begin) / 1e6, + }) + result = { + "schema_version": "missioncore.perception-component-baseline/v1", + "component": "ddrnet", "started_utc": utc, + "started_monotonic_ns": started, "finished_monotonic_ns": time.monotonic_ns(), + "checkpoint_sha256": expected_checkpoint, "runner_sha256": digest(runner), + "model": name, "torch": torch.__version__, "cuda": torch.version.cuda, + "gpu": torch.cuda.get_device_name(0), "precision": "existing-fp32", + "samples": args.samples, "warmup_iterations": args.warmup, + "load_first_decode_warmup_seconds": (warm_ns - started) / 1e9, + "source_whole_file_copied_or_hashed": False, + "source_frames_decoded": args.samples + 1, + "source_paced": False, "full_profile_qualified": False, + "mask_sequence_sha256": mask_digest.hexdigest(), + "peak_allocated_vram_mib": torch.cuda.max_memory_allocated() / 1048576, + "peak_reserved_vram_mib": torch.cuda.max_memory_reserved() / 1048576, + "timing_ms": {key: distribution([row[key] for row in rows]) for key in rows[0] + if key != "sequence"}, + "frames": rows, + } + print("STAGE1_RESULT=" + json.dumps(result, sort_keys=True), flush=True) + finally: + capture.release() + + +if __name__ == "__main__": + main() diff --git a/experiments/perception/worker/streaming_profile_stage1/baseline_surface.py b/experiments/perception/worker/streaming_profile_stage1/baseline_surface.py new file mode 100644 index 0000000..2ab8b7d --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/baseline_surface.py @@ -0,0 +1,32 @@ +"""CPU component probe using synthetic, explicitly stationary geometry only.""" + +import json +import time + +import numpy as np + +from k1link.compute.lidar_local_surface_shadow import ( + K1LocalSurfaceShadowEstimator, + K1LocalSurfaceShadowInput, +) + +random = np.random.default_rng(20260901) +estimator = K1LocalSurfaceShadowEstimator() +rows = [] +for index in range(64): + xy = random.uniform(-8, 8, (4000, 2)) + z = random.normal(0, 0.015, (4000, 1)) + z[3800:] += 1.0 + value = K1LocalSurfaceShadowInput( + frame_index=index, source_frame_index=index, session_seconds=index / 10, + pose_binding_age_ms=0, points_map=np.column_stack((xy, z)), + position_map=np.zeros(3), published_monotonic_ns=time.monotonic_ns(), + ) + started = time.monotonic_ns() + result = estimator.process(value) + rows.append({"sequence": index, "wall_ms": (time.monotonic_ns() - started) / 1e6, + "state": result.state}) +print(json.dumps({"schema_version": "missioncore.perception-component-baseline/v1", + "component": "online-local-surface", "synthetic": True, + "points_per_frame": 4000, "frames": rows, + "full_profile_qualified": False}, sort_keys=True)) diff --git a/experiments/perception/worker/streaming_profile_stage1/baseline_tgs.cpp b/experiments/perception/worker/streaming_profile_stage1/baseline_tgs.cpp new file mode 100644 index 0000000..22a26e9 --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/baseline_tgs.cpp @@ -0,0 +1,61 @@ +// Component-only CPU measurement on 64 existing prepared rolling clouds. +// Does NOT measure source decoding, rolling input preparation, costmap or transport. +#include +#include +#include +#include +#include +#include +#include +#include +#include "travel/point_types.hpp" +#include "travel/tgs.hpp" + +struct QuietLibraryOutput { + std::ostringstream buffer; + std::streambuf* previous; + QuietLibraryOutput() : previous(std::cout.rdbuf(buffer.rdbuf())) {} + ~QuietLibraryOutput() { std::cout.rdbuf(previous); } +}; + +int main(int argc, char** argv) { + if (argc != 2) return 2; + using Clock = std::chrono::steady_clock; + try { + std::cout << "sequence,points,ground,nonground,wall_ms,algorithm_ms\n"; + for (int index = 0; index < 64; ++index) { + std::ostringstream path; + path << argv[1] << '/' << std::setw(6) << std::setfill('0') << index << ".bin"; + std::ifstream file(path.str(), std::ios::binary); + if (!file) throw std::runtime_error("bounded cloud unavailable"); + travel::PointCloud input; + float row[4]; + while (file.read(reinterpret_cast(row), sizeof(row))) { + if (input.size() >= 500000) throw std::runtime_error("cloud exceeds point budget"); + PointXYZILID point{}; + point.x = row[0]; point.y = row[1]; point.z = row[2]; point.intensity = row[3]; + if (!std::isfinite(point.x) || !std::isfinite(point.y) || !std::isfinite(point.z)) + throw std::runtime_error("nonfinite cloud"); + input.emplace_back(point); + } + if (file.gcount() || input.empty()) throw std::runtime_error("invalid cloud bytes"); + const auto start = Clock::now(); + travel::PointCloud ground, nonground; + double algorithm_seconds = 0; + { + QuietLibraryOutput quiet; + travel::TravelGroundSeg tgs; + tgs.setParams(80.0, 1.0, 8.0, 3, 5, 10, 0.5, 0.125, 0.3, 0.940, + 200.0, 0.03, 0.1, 1.0, true, false); + tgs.estimateGround(input, ground, nonground, algorithm_seconds); + } + const auto end = Clock::now(); + const auto wall = std::chrono::duration(end - start).count(); + std::cout << index << ',' << input.size() << ',' << ground.size() << ',' + << nonground.size() << ',' << wall << ',' << algorithm_seconds * 1000 << '\n'; + } + } catch (const std::exception& error) { + std::cerr << error.what() << '\n'; + return 1; + } +} diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_ddrnet_runtime.py b/experiments/perception/worker/streaming_profile_stage1/pilot_ddrnet_runtime.py new file mode 100644 index 0000000..d49c56b --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_ddrnet_runtime.py @@ -0,0 +1,138 @@ +"""Measured FP32 DDRNet execution; preserves sigmoid and first-index ties. + +An experimental memory-layout variant, not a new weight/precision/label model. +Python 3.9 compatible. GPU calls remain serialized by the owning supervisor. +""" + +import time + + +def layout_scores(scores, layout): + if layout == "reference": + return scores, 1 + if layout == "channels-last": + return scores.permute(0, 2, 3, 1).contiguous(), 3 + raise ValueError("unknown DDRNet postprocess layout") + + +class DdrnetRuntime: + def __init__(self, model, logits_from_output, layout, execution="eager"): + import torch + + if layout not in ("reference", "channels-last"): + raise ValueError("unknown DDRNet postprocess layout") + if execution not in ("eager", "cuda-graph"): + raise ValueError("unknown DDRNet execution mode") + self.torch = torch + self.model = model + self.logits_from_output = logits_from_output + self.layout = layout + self.execution = execution + self.graph = None + self.events = [torch.cuda.Event(enable_timing=True) for _ in range(5)] + + def validate_ties(self): + torch = self.torch + # Sigmoid saturates 100 and 101 to the same score. Argmax(logits) + # would change the original label; the layout optimization must not. + logits = torch.tensor( + [ + [100.0, 0.0, -100.0, 4.0], + [101.0, 0.0, -99.0, 4.0], + [-100.0, 0.0, 3.0, 0.0], + [-9.0, 0.0, -9.0, 0.0], + ], + dtype=torch.float32, + device="cuda", + ).reshape(1, 4, 2, 2) + scores = torch.sigmoid(logits) + reference = torch.argmax(scores, dim=1) + changed, dimension = layout_scores(scores, "channels-last") + candidate = torch.argmax(changed, dim=dimension) + valid = bool(torch.equal(reference, candidate) and reference[0, 0, 0].item() == 0) + if not valid: + raise RuntimeError("DDRNet layout changed saturated/tied-score labels") + return valid + + def infer(self, tensor): + if self.execution == "cuda-graph": + return self.infer_graph(tensor) + torch = self.torch + h2d_started = time.perf_counter_ns() + tensor = tensor.cuda(non_blocking=True) + torch.cuda.synchronize() + started = time.perf_counter_ns() + with torch.inference_mode(): + self.events[0].record() + logits = self.logits_from_output(self.model(tensor)) + self.events[1].record() + scores = torch.sigmoid(logits) + self.events[2].record() + scores, dimension = layout_scores(scores, self.layout) + self.events[3].record() + prediction = torch.argmax(scores, dim=dimension) + self.events[4].record() + torch.cuda.synchronize() + predicted = time.perf_counter_ns() + mask = prediction[0].to(device="cpu", dtype=torch.uint8).numpy() + copied = time.perf_counter_ns() + names = ("model_gpu_ms", "sigmoid_gpu_ms", "layout_gpu_ms", "argmax_gpu_ms") + stages = { + name: self.events[i].elapsed_time(self.events[i + 1]) for i, name in enumerate(names) + } + stages.update( + h2d_sync_wall_ms=(started - h2d_started) / 1e6, + d2h_wall_ms=(copied - predicted) / 1e6, + forward_post_wall_ms=(predicted - started) / 1e6, + ) + return mask, (predicted - started) / 1e6, stages + + def _compute(self, tensor): + logits = self.logits_from_output(self.model(tensor)) + scores, dimension = layout_scores(self.torch.sigmoid(logits), self.layout) + return self.torch.argmax(scores, dim=dimension) + + def _capture(self, tensor): + torch = self.torch + # Capture only fixed-shape DDRNet work, never source reads, RF-DETR, + # host decisions, or another profile. Capture is local to this GPU/run. + self.static_input = tensor.cuda().clone() + stream = torch.cuda.Stream() + stream.wait_stream(torch.cuda.current_stream()) + with torch.cuda.stream(stream), torch.inference_mode(): + for _ in range(3): + self._compute(self.static_input) + torch.cuda.current_stream().wait_stream(stream) + torch.cuda.synchronize() + self.graph = torch.cuda.CUDAGraph() + with torch.cuda.graph(self.graph), torch.inference_mode(): + self.static_prediction = self._compute(self.static_input) + torch.cuda.synchronize() + + def infer_graph(self, tensor): + torch = self.torch + if self.graph is None: + self._capture(tensor) + if tensor.shape != self.static_input.shape or tensor.dtype != self.static_input.dtype: + raise ValueError("DDRNet CUDA graph input shape/dtype changed") + h2d_started = time.perf_counter_ns() + self.static_input.copy_(tensor, non_blocking=True) + torch.cuda.synchronize() + started = time.perf_counter_ns() + self.events[0].record() + self.graph.replay() + self.events[1].record() + torch.cuda.synchronize() + predicted = time.perf_counter_ns() + mask = self.static_prediction[0].to(device="cpu", dtype=torch.uint8).numpy() + copied = time.perf_counter_ns() + return ( + mask, + (predicted - started) / 1e6, + { + "cuda_graph_gpu_ms": self.events[0].elapsed_time(self.events[1]), + "h2d_sync_wall_ms": (started - h2d_started) / 1e6, + "d2h_wall_ms": (copied - predicted) / 1e6, + "forward_post_wall_ms": (predicted - started) / 1e6, + }, + ) diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_graph.py b/experiments/perception/worker/streaming_profile_stage1/pilot_graph.py new file mode 100644 index 0000000..ae2df27 --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_graph.py @@ -0,0 +1,451 @@ +"""Joint-pilot wiring of existing algorithms to current, causal source data. + +Experiment adapter only: no installed LAB or product registry mutation. +Historical profile files supply algorithm parameters, never derived evidence. +""" + +from __future__ import annotations + +import csv +import hashlib +import importlib.util +import struct +import sys +import time +import zipfile +from collections import Counter, OrderedDict +from dataclasses import asdict +from pathlib import Path + +import numpy as np +from pilot_ipc import exact + +from k1link.compute.lidar_local_surface_shadow import ( + K1LocalSurfaceShadowEstimator, + K1LocalSurfaceShadowInput, +) +from k1link.perception.contracts import ( + ClockBasis, + LocalObstacleMap, + ModalityOutcome, + ModalityStatus, + SourceAccounting, + SourceEnvelope, + TemporalState, + TimestampBundle, +) +from k1link.perception.detector import NativeRfDetrShadowDetectorProvider +from k1link.perception.geometry import ( + GeometryFrame, + Ravnoves00GeometryAssociationProvider, + load_geometry_profile, +) +from k1link.perception.geometry_math import ( + Kb4ProjectionProfile, + project_map_points_kb4, + quaternion_xyzw_to_rotation_matrix, +) +from k1link.perception.motion import ClassIndependentMotionEstimator +from k1link.perception.providers import SourcePacket +from k1link.perception.rf_detr_native_object_detector import TritonNativeRfDetrHttpInferenceBackend +from k1link.perception.rolling_map import RollingLocalObstacleMapProvider, load_rolling_map_profile +from k1link.perception.temporal import BoundedSpatialTemporalProvider, load_temporal_motion_profile +from k1link.perception.threat import ( + DualEvidenceReplayThreatProvider, + ReplayBodyFrame, + load_replay_threat_profile, +) +from k1link.perception.yolox_object_detector import load_valid_fov_mask + + +def module_from_file(name, path): + spec = importlib.util.spec_from_file_location(name, path) + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module + spec.loader.exec_module(module) + return module + + +def calibration(path): + # Read only the three small static arrays, never the archive's point data. + with zipfile.ZipFile(path) as archive: + arrays = {} + for name in ("intrinsic_fx_fy_cx_cy", "distortion_kb4", "t_camera_from_lidar"): + if archive.getinfo(name + ".npy").file_size > 4096: + raise ValueError("calibration exceeds metadata budget") + with archive.open(name + ".npy") as stream: + arrays[name] = np.lib.format.read_array(stream, allow_pickle=False) + return Kb4ProjectionProfile( + 800, + 600, + tuple(arrays["intrinsic_fx_fy_cx_cy"]), + tuple(arrays["distortion_kb4"]), + arrays["t_camera_from_lidar"], + ) + + +def grid_indices(point_cells, grid): + """Exact equivalent of per-point tuple/dict lookup, including holes.""" + grid_xy = grid[:, :2].astype(np.int64) + lower = grid_xy.min(axis=0) + shape = grid_xy.max(axis=0) - lower + 1 + lookup = np.full(tuple(shape), -1, np.int64) + offset = grid_xy - lower + lookup[offset[:, 0], offset[:, 1]] = np.arange(len(grid)) + relative = point_cells - lower + inside = np.all((relative >= 0) & (relative < shape), axis=1) + result = np.full(len(point_cells), -1, np.int64) + result[inside] = lookup[relative[inside, 0], relative[inside, 1]] + return result + + +class CurrentStore: + """One current frame, no NPZ store, historical lookup or future trajectory.""" + + def __init__(self, profile): + self.profile = profile + self.current = None + self.pose = None + self.body = None + self.frame_id = None + self.body_history = OrderedDict() + + def frame(self, packet): + if packet.envelope.frame_id != self.frame_id: + raise ValueError("online geometry frame identity mismatch") + return self.current + + def current_points(self, packet): + frame = self.frame(packet) + return frame.points_map if frame is not None and frame.surface_valid else None + + def pose_values_for_frame(self, frame_id): + if frame_id != self.frame_id: + raise ValueError("past/future pose lookup forbidden") + return self.pose + + def body_frame_for_frame(self, frame_id): + if frame_id > self.frame_id: + raise ValueError("future body lookup forbidden") + return self.body_history.get(frame_id) + + def remember_body(self): + self.body_history[self.frame_id] = self.body + while len(self.body_history) > 64: + self.body_history.popitem(last=False) + + +def current_body(frame, surface): + if ( + not surface.valid + or surface.sensor_height_m is None + or abs(surface.sensor_height_m - 1.25) > 0.45 + or surface.slope_deg is None + or surface.slope_deg > 10.0 + ): + return None + rotation = quaternion_xyzw_to_rotation_matrix(frame.sensor_orientation_xyzw) + forward = rotation @ frame.projection.t_camera_from_lidar[2, :3] + forward[2] = 0 + norm = np.linalg.norm(forward) + normal = np.asarray(surface.plane_coefficients_map[:3]) + if norm < 1e-9 or abs(normal[2]) < 1e-6: + return None + forward /= norm + up = np.array([0.0, 0.0, 1.0]) + left = np.cross(up, forward) + origin = frame.sensor_position_map - up * (surface.sensor_height_m / abs(normal[2])) + basis = np.column_stack((forward, left, up)) + return ReplayBodyFrame( + f"frame-{frame.frame_index:06d}", + tuple(origin), + tuple(tuple(row) for row in basis), + surface.sensor_height_m, + surface.slope_deg, + "causal-camera-forward-simulation", + 0.0, + ) + + +class JointGraph: + def __init__(self, code_root, tgs, *, mask_path, mapping_path, calibration_pack): + root = Path(code_root) + config = root / "config/perception" + self.projection = calibration(calibration_pack) + self.surface = K1LocalSurfaceShadowEstimator() + self.store = CurrentStore(load_geometry_profile(config / "m4-geometry-association-v1.json")) + self.geometry = Ravnoves00GeometryAssociationProvider(store=self.store) + temporal_profile = load_temporal_motion_profile(config / "m4-temporal-motion-v1.json") + self.temporal = BoundedSpatialTemporalProvider( + point_resolver=self.store, profile=temporal_profile + ) + self.motion = ClassIndependentMotionEstimator(profile=temporal_profile) + self.rolling = RollingLocalObstacleMapProvider( + pose_resolver=self.store, + profile=load_rolling_map_profile(config / "m4-rolling-local-map-v1.json"), + ) + self.threat = DualEvidenceReplayThreatProvider( + body_frame_resolver=self.store, + profile=load_replay_threat_profile(config / "m4-replay-threat-v3.json"), + ) + self.backend = TritonNativeRfDetrHttpInferenceBackend( + "http://127.0.0.1:8000", timeout_seconds=5 + ) + self.detector = NativeRfDetrShadowDetectorProvider( + mask=load_valid_fov_mask(Path(mask_path)), backend=self.backend + ) + self.tgs = tgs + scripts = root / "experiments/perception/worker/m49_t3_travel" + sys.path.insert(0, str(scripts)) + from build_tgs_fail_closed_evidence import costmap_grid + from build_tgs_full_shadow_evidence import rasterize + + self.rasterize = rasterize + self.grid = costmap_grid(12.0, 0.45) + self.cell_lookup = {(int(row[0]), int(row[1])): i for i, row in enumerate(self.grid)} + policy_module = module_from_file( + "pilot_vegetation_policy", root / "src/k1link/laboratory/vegetation_mission_policy.py" + ) + self.policy = policy_module.load_vegetation_mission_policy( + config / "lab-v1-vegetation-mission-policy-v1.json", repository_root=root + ) + mapping = policy_module.load_vegetation_provider_label_map( + config / "lab-v1-vegetation-provider-label-map-v1.json", policy=self.policy + ) + if ( + hashlib.sha256(Path(mapping_path).read_bytes()).hexdigest() + != "88ae319ba5a3877dd3ae0773f693a6a5fdc283934140de9dfaff029108aefd7f" + ): + raise ValueError("GOOSE class mapping changed") + with Path(mapping_path).open(encoding="utf-8-sig", newline="") as stream: + labels = {int(row["label_key"]): row["class_name"] for row in csv.DictReader(stream)} + if set(labels) != set(range(64)): + raise ValueError("GOOSE mapping incomplete") + material_names = ["unknown", *self.policy["material_classes"]] + provider_map = mapping["providers"]["goose-fine-64"]["labels"] + self.material_names = material_names + self.material_lut = np.asarray( + [material_names.index(provider_map.get(labels[i], "unknown")) for i in range(64)] + ) + self.action_lut = np.full( + (len(material_names), 4), 2, np.uint8 + ) # 0=ALLOW,1=HIGH_COST,2=NO_GO + for i, material in enumerate(material_names): + decision = policy_module.resolve_terrain_policy( + self.policy, + preset_id="rural", + material_class=None if material == "unknown" else material, + evidence_state="SUPPORTED_GROUND", + ) + self.action_lut[i, 1] = {"ALLOW": 0, "HIGH_COST": 1, "NO_GO": 2}[ + decision.effective_action + ] + # Pilot intent: ONLY coarse hard_surface can be an ALLOW candidate. + # This additional conservative interlock cannot weaken legacy rules. + if material != "hard_surface": + self.action_lut[i, :] = 2 + + def packet(self, bundle): + available = bundle["available"] + status = ModalityStatus( + available, + ModalityOutcome.AVAILABLE if available else ModalityOutcome.UNAVAILABLE, + "causal-source-bound" if available else "no-fresh-past-cloud-or-pose", + ) + env = SourceEnvelope( + source_id="RAVNOVES00", + session_id="20260720T065719Z_viewer_live", + frame_id=f"frame-{bundle['sequence']:06d}", + sequence=bundle["sequence"], + timestamps=TimestampBundle( + utc_ns=bundle["utc_ns"], + monotonic_ns=bundle["time_ns"], + source_ns=bundle["source_ns"], + clock_basis=ClockBasis.RECORDED_HOST, + ), + source_age_ns=0, + binding_reason="past-only-source-arrival", + calibration_id="camera-1-kb4-05f3ad9b", + representation_id=self.rolling.profile.representation_id, + image=ModalityStatus(True, ModalityOutcome.AVAILABLE, "current-camera"), + registered_point_increment=status, + pose=status, + ) + return SourcePacket( + env, + bundle["image"], + bundle["points"] if available else None, + bundle["pose"] if available else None, + ) + + def process(self, bundle, segmentation, *, proposals=None, detector_ms=None): + packet = self.packet(bundle) + timing = {} + if proposals is None: + begin = time.monotonic_ns() + proposals = self.detector.detect(packet) + detector_ms = (time.monotonic_ns() - begin) / 1e6 + timing["detector_ms"] = detector_ms + self.store.frame_id = packet.envelope.frame_id + self.store.current = self.store.pose = self.store.body = None + surface = None + begin = time.monotonic_ns() + if bundle["available"]: + position, quaternion = bundle["pose"] + surface = self.surface.process( + K1LocalSurfaceShadowInput( + frame_index=bundle["sequence"], + source_frame_index=bundle["sequence"], + session_seconds=bundle["source_ns"] / 1e9, + pose_binding_age_ms=bundle["binding_age_ms"], + points_map=bundle["points"], + position_map=position, + published_monotonic_ns=bundle["due_ns"], + ) + ) + self.store.current = GeometryFrame( + bundle["sequence"], + bundle["points"], + surface.point_class, + position, + quaternion, + self.projection, + surface.valid, + ) + self.store.pose = (tuple(position), tuple(quaternion)) + self.store.body = current_body(self.store.current, surface) + self.store.remember_body() + timing["online_surface_ms"] = (time.monotonic_ns() - begin) / 1e6 + begin = time.monotonic_ns() + observations = self.geometry.associate(packet, proposals) + timing["association_ms"] = (time.monotonic_ns() - begin) / 1e6 + begin = time.monotonic_ns() + temporal = self.temporal.update(packet, observations) + moving = self.motion.estimate(packet, temporal) + retained = self.rolling.update(packet, moving) + current = tuple(item for item in moving if item.state is TemporalState.CURRENT) + unknown = tuple(item for item in moving if item.state is not TemporalState.CURRENT) + associated = { + pid + for item in observations + if item.metric_geometry is not None + for pid in item.proposal_ids + } + scene_map = LocalObstacleMap( + "RAVNOVES00", + "20260720T065719Z_viewer_live", + packet.envelope.frame_id, + "stage1-joint-online-pilot/v1", + time.monotonic_ns(), + max(0, time.monotonic_ns() - bundle["due_ns"]), + (*current, *retained), + unknown, + tuple(p for p in proposals if p.proposal_id not in associated), + SourceAccounting(1, 1, 0, 0), + ) + threats = self.threat.assess(scene_map) + timing["motion_rolling_threat_ms"] = (time.monotonic_ns() - begin) / 1e6 + begin = time.monotonic_ns() + costmap, actions, material, tgs_counts = self.costmap(bundle, segmentation, scene_map) + timing["tgs_costmap_policy_ms"] = (time.monotonic_ns() - begin) / 1e6 + return { + "timing_ms": timing, + "surface_state": surface.state if surface else "unavailable", + "segmentation_sha256": hashlib.sha256(segmentation.tobytes()).hexdigest(), + "proposals": [asdict(x) for x in proposals], + "observations": [asdict(x) for x in observations], + "tracks": [asdict(x) for x in moving], + "threats": [asdict(x) for x in threats], + "tgs_counts": tgs_counts, + "costmap_states": costmap.tolist(), + "costmap_material": material.tolist(), + "policy_actions": actions.tolist(), + "policy_counts": { + key: int(np.count_nonzero(actions == value)) + for key, value in (("ALLOW_candidate", 0), ("HIGH_COST", 1), ("NO_GO", 2)) + }, + "range_estimator": { + "detected": "median-camera-z", + "geometry_only": "nearest-euclidean-sensor", + }, + "commands_enabled": False, + "actuation_allowed": False, + } + + def costmap(self, bundle, mask, obstacle_map): + count = len(self.grid) + material = np.zeros(count, np.int32) + if not bundle["available"]: + return ( + np.zeros(count, np.uint8), + np.full(count, 2, np.uint8), + material, + {"unavailable": True}, + ) + position, quaternion = bundle["pose"] + points = bundle["rolling_points"] + stamps = bundle["rolling_times"] + local = points - position + keep = np.sum(local[:, :2] ** 2, axis=1) <= 144.0 + local, points, stamps = local[keep], points[keep], stamps[keep] + if not 0 < len(local) <= 64000: + raise ValueError("TGS rolling input outside pilot bound") + native = np.column_stack((local, np.zeros(len(local)))).astype(" 3)): + raise ValueError("TGS returned invalid state") + cells, _ = self.rasterize(native[:, :3], states, self.grid, 0.45) + point_cells = np.floor(local[:, :2] / 0.45).astype(np.int32) + ids = grid_indices(point_cells, self.grid) + valid = ids >= 0 + last_seen = np.full(count, -1, np.int64) + np.maximum.at(last_seen, ids[valid], stamps[valid]) + stale = (last_seen >= 0) & (bundle["time_ns"] - last_seen > 250_000_000) + cells[stale & (cells != 2)] = 3 + projected = project_map_points_kb4( + points, + position_map_xyz=position, + orientation_map_from_lidar_xyzw=quaternion, + profile=self.projection, + ) + uv = projected.pixels_xy + crop = (uv[:, 0] >= 100) & (uv[:, 0] < 700) & (uv[:, 1] >= 0) & (uv[:, 1] < 600) + point_ids = projected.source_indices[crop] + uv = uv[crop] + labels = mask[ + (uv[:, 1] * 512 / 600).astype(int), ((uv[:, 0] - 100) * 512 / 600).astype(int) + ] + votes = np.zeros((count, len(self.material_names)), np.int32) + cell_ids = ids[point_ids] + good = (cell_ids >= 0) & (states[point_ids] == 1) + np.add.at(votes, (cell_ids[good], self.material_lut[labels[good]]), 1) + material = votes.argmax(axis=1).astype(np.int32) + # Current and retained geometry can only add a prohibition, never clear TGS. + for obstacle in (*obstacle_map.occupied, *obstacle_map.unknown): + for cell in obstacle.cells: + xy = ( + np.asarray((cell.x, cell.y)) + 0.5 + ) * self.temporal.config.voxel_size_m - position[:2] + index = self.cell_lookup.get(tuple(np.floor(xy / 0.45).astype(int))) + if index is not None: + cells[index] = 2 + actions = self.action_lut[material, cells] + return ( + cells, + actions, + material, + { + "points": n, + "algorithm_ms": algorithm_ms, + "ground": int(np.count_nonzero(states == 1)), + "occupied": int(np.count_nonzero(states == 2)), + "rejected": int(np.count_nonzero(states == 3)), + "stale_cells": int(np.count_nonzero(stale)), + "state_counts": dict(Counter(int(x) for x in cells)), + }, + ) diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_ipc.py b/experiments/perception/worker/streaming_profile_stage1/pilot_ipc.py new file mode 100644 index 0000000..7273313 --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_ipc.py @@ -0,0 +1,41 @@ +"""Bounded local probe RPC, compatible with Python 3.9 and 3.12.""" + +from __future__ import annotations + +import json +import struct + +MAX_PAYLOAD = 16 * 1024 * 1024 + + +def exact(stream, size): + if not 0 <= size <= MAX_PAYLOAD: + raise ValueError("IPC read exceeds budget") + chunks = bytearray() + while len(chunks) < size: + part = stream.read(size - len(chunks)) + if not part: + raise EOFError("truncated IPC message") + chunks.extend(part) + return bytes(chunks) + + +def receive(stream): + length = struct.unpack(" MAX_PAYLOAD: + raise ValueError("IPC payload exceeds budget") + raw = json.dumps({**header, "payload_bytes": len(payload)}, allow_nan=False).encode() + if len(raw) > 65536: + raise ValueError("IPC header exceeds budget") + stream.write(struct.pack("= 1) & (states <= 3)) + graph = JointGraph( + "/code", + process, + mask_path="/valid-fov.png", + mapping_path="/goose.csv", + calibration_pack="/calibration.npz", + ) + assert np.count_nonzero(graph.action_lut == 0) == 1 + assert graph.action_lut[graph.material_names.index("hard_surface"), 1] == 0 + graph.backend.close() + print( + json.dumps( + { + "graph_initialized": True, + "tgs_identity_accounted": n, + "tgs_ms": ms, + "cells": len(graph.grid), + } + ), + flush=True, + ) +finally: + process.stdin.write(struct.pack("= self.capacity + or self.bytes + size > self.byte_limit + ): + old = self.pending.popleft() + self.bytes -= old["payload_bytes"] + self.dropped.append({"sequence": old["sequence"], "reason": "pending-overflow"}) + if len(self.pending) + self.external_pending >= self.capacity: + self.dropped.append({"sequence": bundle["sequence"], "reason": "pending-overflow"}) + return + if self.bytes + size > self.byte_limit: + self.dropped.append({"sequence": bundle["sequence"], "reason": "byte-budget"}) + return + self.pending.append(bundle) + self.bytes += size + self.peak_bytes = max(self.peak_bytes, self.bytes) + self.peak_pending = max(self.peak_pending, len(self.pending) + self.external_pending) + self.condition.notify_all() + + def reserve_completed(self): + """A finished GPU result shares the SAME pending budget as ingress.""" + with self.condition: + if self.external_pending: + raise ValueError("only one completed GPU slot is permitted") + while len(self.pending) + 1 > self.capacity: + old = self.pending.popleft() + self.bytes -= old["payload_bytes"] + self.dropped.append({"sequence": old["sequence"], "reason": "handoff-overflow"}) + self.external_pending = 1 + self.peak_pending = max(self.peak_pending, len(self.pending) + 1) + + def take_completed(self): + with self.condition: + if self.external_pending != 1: + raise ValueError("completed GPU slot accounting mismatch") + self.external_pending = 0 + + def take(self): + with self.condition: + self.condition.wait_for(lambda: self.pending or self.done) + return self.pending.popleft() if self.pending else None + + def release(self, bundle): + with self.condition: + self.bytes -= bundle["payload_bytes"] + + def finish(self, error=None): + with self.condition: + self.done = True + self.error = error + self.condition.notify_all() diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_scheduler.py b/experiments/perception/worker/streaming_profile_stage1/pilot_scheduler.py new file mode 100644 index 0000000..cd30842 --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_scheduler.py @@ -0,0 +1,61 @@ +"""One serial GPU stage can overlap one chronological CPU fusion stage. + +No GPU model concurrency. Ingress and completed-GPU results share two pending +slots. Active payloads remain in the shared mailbox byte budget. +""" + +import queue +import threading +import traceback + + +class GpuStage: + def __init__(self, mailbox, compute, stop): + self.mailbox = mailbox + self.compute = compute + self.stop = stop + self.output = queue.Queue(maxsize=1) + # Reserve the output slot BEFORE starting another GPU call. Otherwise a + # blocked put would hide a third pending frame outside the two queues. + self.output_slot = threading.Semaphore(1) + self.finished = threading.Event() + self.error = None + self.peak_pending = 0 + self.thread = threading.Thread(target=self._run, daemon=True) + self.thread.start() + + def _run(self): + try: + while not self.stop.is_set(): + if not self.output_slot.acquire(timeout=0.05): + continue + bundle = self.mailbox.take() + if bundle is None: + break + result = self.compute(bundle) + self.mailbox.reserve_completed() + self.output.put_nowait((bundle, result)) + self.peak_pending = max(self.peak_pending, self.output.qsize()) + except Exception: + self.error = traceback.format_exc() + finally: + self.finished.set() + + def take(self): + while True: + try: + result = self.output.get(timeout=0.05) + self.mailbox.take_completed() + self.output_slot.release() + return result + except queue.Empty: + if self.finished.is_set(): + if self.error: + raise RuntimeError(self.error) from None + return None + + def close(self): + self.stop.set() + self.mailbox.finish(self.mailbox.error) + self.thread.join(timeout=2) + return not self.thread.is_alive() diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_source.py b/experiments/perception/worker/streaming_profile_stage1/pilot_source.py new file mode 100644 index 0000000..ac99b10 --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_source.py @@ -0,0 +1,154 @@ +"""Incremental existing normalized sensor archive; no NPZ materialization. + +Only original capture clocks/points/pose are read. No camera-indexed E10 +nearest-neighbor association, surface NPZ, masks, TGS outputs or threat ledger. +One-row lookahead belongs to the source reader, never to perception providers. +""" + +from __future__ import annotations + +import heapq +import json +import math +import zipfile +from dataclasses import dataclass +from pathlib import Path + +import numpy as np + + +class NpyRows: + def __init__(self, archive, name, *, columns=(), dtype=None): + self.stream = archive.open(name + ".npy") + version = np.lib.format.read_magic(self.stream) + if version == (1, 0): + shape, fortran, actual = np.lib.format.read_array_header_1_0(self.stream) + elif version == (2, 0): + shape, fortran, actual = np.lib.format.read_array_header_2_0(self.stream) + else: + raise ValueError("unsupported bounded NPY header") + if fortran or actual.hasobject or tuple(shape[1:]) != columns: + raise ValueError("sensor archive layout changed") + if dtype is not None and actual != np.dtype(dtype): + raise ValueError("sensor archive dtype changed") + self.count = shape[0] + self.columns = columns + self.dtype = actual + self.row_bytes = actual.itemsize * math.prod(columns) + self.position = 0 + self.bytes_read = 0 + + def take(self, count=1): + size = count * self.row_bytes + if count < 0 or self.position + count > self.count or size > 8 * 1024 * 1024: + raise ValueError("sensor row exceeds source budget") + raw = self.stream.read(size) + if len(raw) != size: + raise EOFError("truncated sensor array") + self.position += count + self.bytes_read += size + return np.frombuffer(raw, self.dtype).reshape((count, *self.columns)) + + +@dataclass(frozen=True) +class SensorEvent: + time_ns: int + channel: str + sequence: int + value: object + + +class SensorArchive: + def __init__(self, path: Path): + self.archive = zipfile.ZipFile(path) + self.readers = {} + + def rows(self, name, **kwargs): + reader = NpyRows(self.archive, name, **kwargs) + self.readers[name] = reader + return reader + + def points(self): + times = self.rows("point_received_monotonic_ns", dtype=" 0.01 + ): + raise ValueError("invalid source pose") + yield SensorEvent(stamp, "pose", int(seqs.take()[0]), (position, quaternion)) + previous = stamp + + def counters(self): + return { + name: {"rows_read": r.position, "bytes_read": r.bytes_read, "source_rows": r.count} + for name, r in self.readers.items() + } + + def close(self): + for reader in self.readers.values(): + reader.stream.close() + self.archive.close() + + +def camera_events(path, limit): + previous = -1 + with Path(path).open("rb") as stream: + for index in range(limit): + raw = stream.readline(65537) + if not raw or len(raw) > 65536: + raise ValueError("camera index is truncated or unbounded") + row = json.loads(raw) + if ( + row.get("schema_version") != "missioncore.camera-recording-index/v1" + or row.get("kind") != "media" + or row.get("sequence") != index + 1 + ): + raise ValueError("camera source identity/order changed") + stamp = row["host_monotonic_ns"] + if type(stamp) is not int or stamp <= previous: + raise ValueError("camera source clock is not increasing") + yield SensorEvent(stamp, "camera", index, row) + previous = stamp + + +def merged_events(archive, camera_index, limit): + # Stable per-channel order; a tie admits sensor data before the camera. + return heapq.merge( + archive.poses(), + archive.points(), + camera_events(camera_index, limit), + key=lambda event: event.time_ns, + ) diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_tgs.cpp b/experiments/perception/worker/streaming_profile_stage1/pilot_tgs.cpp new file mode 100644 index 0000000..cf6e118 --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_tgs.cpp @@ -0,0 +1,62 @@ +// Persistent bounded TGS adapter. stdin: uint32 count + XYZ-intensity float32. +// Original point identity survives segmentation in PointXYZILID::id (<=64000). +// stdout: uint32 count + double algorithm_ms + uint8 state per input point. +#include +#include +#include +#include +#include +#include +#include "travel/point_types.hpp" +#include "travel/tgs.hpp" + +int main() { + FILE* protocol = fdopen(dup(STDOUT_FILENO), "wb"); + dup2(STDERR_FILENO, STDOUT_FILENO); // TGS diagnostics cannot corrupt RPC. + try { + std::uint32_t count; + while (std::fread(&count, sizeof(count), 1, stdin) == 1) { + if (count == 0) break; + if (count > 64000) throw std::runtime_error("TGS point budget exceeded"); + std::vector data(static_cast(count) * 4); + if (std::fread(data.data(), sizeof(float), data.size(), stdin) != data.size()) + throw std::runtime_error("truncated TGS observation"); + travel::PointCloud input, ground, nonground; + input.reserve(count); + for (std::uint32_t index = 0; index < count; ++index) { + PointXYZILID p{}; + p.x = data[index*4]; p.y = data[index*4+1]; p.z = data[index*4+2]; + p.intensity = data[index*4+3]; p.id = static_cast(index); + if (!std::isfinite(p.x) || !std::isfinite(p.y) || !std::isfinite(p.z)) + throw std::runtime_error("nonfinite TGS input"); + input.push_back(p); + } + double seconds = 0; + travel::TravelGroundSeg tgs; + tgs.setParams(80.0, 1.0, 8.0, 3, 5, 10, 0.5, 0.125, 0.3, 0.940, + 200.0, 0.03, 0.1, 1.0, true, false); + tgs.estimateGround(input, ground, nonground, seconds); + std::vector states(count, 3); // Rejected != unobserved/free. + for (const auto& p : ground.points) { + if (p.id >= count || states[p.id] != 3) + throw std::runtime_error("TGS ground identity collision"); + states[p.id] = 1; + } + for (const auto& p : nonground.points) { + if (p.id >= count || states[p.id] != 3) + throw std::runtime_error("TGS nonground identity collision"); + states[p.id] = 2; + } + const double ms = seconds * 1000.0; + if (std::fwrite(&count, sizeof(count), 1, protocol) != 1 || + std::fwrite(&ms, sizeof(ms), 1, protocol) != 1 || + std::fwrite(states.data(), 1, count, protocol) != count) + throw std::runtime_error("TGS result transport failed"); + std::fflush(protocol); + } + std::fclose(protocol); + } catch (const std::exception& e) { + std::fprintf(stderr, "TGS pilot: %s\n", e.what()); + return 1; + } +} diff --git a/experiments/perception/worker/streaming_profile_stage1/rfdetr-config.pbtxt b/experiments/perception/worker/streaming_profile_stage1/rfdetr-config.pbtxt new file mode 100644 index 0000000..d8de267 --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/rfdetr-config.pbtxt @@ -0,0 +1,9 @@ +name: "rf_detr_large_native_kb4" +platform: "tensorrt_plan" +max_batch_size: 0 +input [{ name: "raw_kb4_bgr" data_type: TYPE_UINT8 dims: [1,600,800,3] }] +output [ + { name: "dets" data_type: TYPE_FP16 dims: [1,300,4] }, + { name: "labels" data_type: TYPE_FP16 dims: [1,300,91] } +] +instance_group [{ count: 1 kind: KIND_GPU gpus: [0] }] diff --git a/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py b/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py new file mode 100644 index 0000000..1f01c8b --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py @@ -0,0 +1,605 @@ +"""Bounded source-paced Worker pilot, not an installed or qualified profile. + +Only persistent DDRNet and RF-DETR share the GPU; inference is serialized. +Source is paced independently of the graph, using original host arrival times. +The in-container collector is measured; network to Mission Core is NOT measured. +""" + +from __future__ import annotations + +import argparse +import hashlib +import http.client +import json +import math +import os +import signal +import subprocess +import threading +import time +import traceback +from collections import Counter, deque +from dataclasses import asdict +from datetime import UTC, datetime +from pathlib import Path + +import numpy as np +from pilot_ipc import receive, send +from pilot_queue import Mailbox +from pilot_scheduler import GpuStage +from pilot_source import SensorArchive, camera_events, merged_events + + +def distribution(values): + if not values: + return None + values = sorted(values) + return { + "count": len(values), + "mean": sum(values) / len(values), + "min": values[0], + "p50": values[math.ceil(len(values) * 0.5) - 1], + "p95": values[math.ceil(len(values) * 0.95) - 1], + "p99": values[math.ceil(len(values) * 0.99) - 1], + "max": values[-1], + } + + +def payload_size(bundle): + return ( + sum(bundle[key].nbytes for key in ("image", "points", "rolling_points", "rolling_times")) + + 4096 + ) + + +def produce(args, decoder, mailbox, stop, report): + archive = SensorArchive(Path(args.sensor_archive)) + first_camera = next(camera_events(args.camera_index, args.frames)).time_ns + source_zero = first_camera - 500_000_000 + wall_zero = time.monotonic_ns() + 50_000_000 + report.update( + source_zero_ns=source_zero, + wall_zero_ns=wall_zero, + first_camera_source_ns=first_camera, + source_clock_speed=1.0, + source_eof_required=False, + full_source_prepass=False, + ) + pose = None + rolling = deque() + fresh = [] + arrivals = Counter() + release_lags = [] + skipped_prefix = Counter() + try: + for event in merged_events(archive, args.camera_index, args.frames): + if stop.is_set(): + break + if event.time_ns < source_zero: + skipped_prefix[event.channel] += 1 + continue + due = wall_zero + event.time_ns - source_zero + if stop.wait(max(0.0, (due - time.monotonic_ns()) / 1e9)): + break + arrived = time.monotonic_ns() + release_lags.append(max(0, arrived - due) / 1e6) + arrivals[event.channel] += 1 + while rolling and event.time_ns - rolling[0].time_ns > 1_000_000_000: + rolling.popleft() + if event.channel == "pose": + pose = event + continue + if event.channel == "points": + rolling.append(event) + fresh.append(event) + if sum(len(e.value[0]) for e in rolling) > 64000: + raise ValueError("raw rolling cloud exceeds bounded 64000-point window") + continue + # Decoder reads ONE frame after its original due time, never full-decodes. + decode_start = time.monotonic_ns() + send(decoder.stdin, {"op": "next"}) + decoded, raw = receive(decoder.stdout) + image = np.frombuffer(raw, np.uint8).reshape(600, 800, 3) + points = np.concatenate([e.value[0] for e in fresh]) if fresh else np.empty((0, 3)) + rolling_points = ( + np.concatenate([e.value[0] for e in rolling]) if rolling else np.empty((0, 3)) + ) + rolling_times = ( + np.concatenate([np.full(len(e.value[0]), e.time_ns, np.int64) for e in rolling]) + if rolling + else np.empty(0, np.int64) + ) + pose_age = event.time_ns - pose.time_ns if pose else None + point_age = event.time_ns - fresh[-1].time_ns if fresh else None + oldest_age = event.time_ns - fresh[0].time_ns if fresh else None + binding_age = ( + max(abs(e.time_ns - pose.time_ns) for e in fresh) if fresh and pose else None + ) + available = bool( + len(points) + and pose + and 0 <= pose_age <= 100_000_000 + and 0 <= point_age <= 100_000_000 + and oldest_age <= 250_000_000 + and binding_age <= 100_000_000 + ) + bundle = { + "sequence": event.sequence, + "time_ns": event.time_ns, + "source_ns": event.time_ns - source_zero, + "due_ns": due, + "utc_ns": event.value["host_epoch_ns"], + "image": image, + "points": points, + "rolling_points": rolling_points, + "rolling_times": rolling_times, + "pose": pose.value if pose else None, + "available": available, + "binding_age_ms": binding_age / 1e6 if binding_age is not None else None, + "lineage": { + "camera_index_sequence": event.value["sequence"], + "camera_host_monotonic_ns": event.time_ns, + "pose_sequence": pose.sequence if pose else None, + "pose_host_monotonic_ns": pose.time_ns if pose else None, + "point_increments": [ + { + "sequence": e.sequence, + "host_monotonic_ns": e.time_ns, + "points": len(e.value[0]), + } + for e in fresh + ], + "pose_age_ms": pose_age / 1e6 if pose_age is not None else None, + "newest_point_age_ms": point_age / 1e6 if point_age is not None else None, + "oldest_point_age_ms": oldest_age / 1e6 if oldest_age is not None else None, + }, + "source_release_lag_ms": max(0, arrived - due) / 1e6, + "decode_ms": decoded["decode_ms"], + "decode_rpc_ms": (time.monotonic_ns() - decode_start) / 1e6, + "enqueued_ns": time.monotonic_ns(), + } + bundle["payload_bytes"] = payload_size(bundle) + mailbox.put(bundle) + fresh = [] + report["last_camera_due_ns"] = due + if arrivals["camera"] >= args.frames: + break + except Exception: + mailbox.finish(traceback.format_exc()) + finally: + report.update( + arrivals=dict(arrivals), + skipped_prefix=dict(skipped_prefix), + release_lag_ms=distribution(release_lags), + incremental_reads=archive.counters(), + window_end_monotonic_ns=time.monotonic_ns(), + ) + archive.close() + mailbox.finish(mailbox.error) + + +def telemetry(stop, samples): + while not stop.is_set(): + sample = {"monotonic_ns": time.monotonic_ns()} + try: + result = subprocess.run( + [ + "nvidia-smi", + "--query-gpu=memory.used,utilization.gpu", + "--format=csv,noheader,nounits", + ], + capture_output=True, + text=True, + timeout=2, + check=True, + ) + mem, util = result.stdout.strip().split(",") + sample.update(gpu_used_mib=int(mem), gpu_utilization=int(util)) + sample["cgroup_memory_mib"] = ( + int(Path("/sys/fs/cgroup/memory.current").read_text()) / 1048576 + ) + sample["cpu_stat"] = { + key: int(value) + for key, value in ( + line.split() + for line in Path("/sys/fs/cgroup/cpu.stat").read_text().splitlines() + ) + } + except Exception as error: + sample["error"] = str(error) + samples.append(sample) + stop.wait(0.5) + + +def wait_triton(process, seconds=90): + deadline = time.monotonic() + seconds + while time.monotonic() < deadline: + if process.poll() is not None: + raise RuntimeError("Triton exited during initialization") + connection = http.client.HTTPConnection("127.0.0.1", 8000, timeout=1) + try: + connection.request("GET", "/v2/models/rf_detr_large_native_kb4/ready") + if connection.getresponse().status == 200: + return + except OSError: + pass + finally: + connection.close() + time.sleep(0.1) + raise TimeoutError("Triton initialization exceeded budget") + + +def run(args): + output = Path(args.output) + output.mkdir(parents=True, exist_ok=False) + started = time.monotonic_ns() + report = { + "schema_version": "missioncore.stage1-joint-pilot/v1", + "run_id": args.run_id, + "started_utc": datetime.now(UTC).isoformat(), + "started_monotonic_ns": started, + "requested_frames": args.frames, + "profile": "K1 Perception — DDRNet-39 + RF-DETR + TGS", + "scope": "one-container-worker-graph-to-local-collector", + "external_transport_measured": False, + "standalone_release": False, + "authority": {"commands_enabled": False, "actuation_allowed": False}, + "effective_policy": "hard-surface-only-plus-geometry-TGS-stale-no-go", + "body_frame": "causal-camera-forward-simulation; no future trajectory", + "clock_quality": "original host arrival best effort; not hardware synchronization", + "ddrnet_postprocess_layout": args.ddrnet_layout, + "ddrnet_execution": args.ddrnet_execution, + "schedule": args.schedule, + } + children, logs, results, samples = [], [], [], [] + stop = threading.Event() + mailbox = Mailbox(capacity=2) + source_report = {} + producer = monitor = graph = gpu_stage = None + + def child(name, command, env=None): + log = (output / (name + ".log")).open("wb") + logs.append(log) + process = subprocess.Popen( + command, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=log, + env=env, + start_new_session=True, + ) + children.append(process) + return process + + def timeout_handler(signum, frame): + raise TimeoutError("joint pilot wall-clock watchdog fired") + + previous = signal.signal(signal.SIGALRM, timeout_handler) + signal.alarm(240) + try: + triton = child( + "triton", + [ + "tritonserver", + "--model-repository=/models", + "--model-control-mode=explicit", + "--load-model=rf_detr_large_native_kb4", + "--allow-grpc=false", + "--allow-metrics=false", + "--http-address=127.0.0.1", + "--pinned-memory-pool-byte-size=16777216", + "--cuda-memory-pool-byte-size=0:16777216", + ], + ) + # Triton logs stdout too: drain directly to the same bounded-run log. + threading.Thread(target=lambda: drain(triton.stdout, logs[0]), daemon=True).start() + wait_triton(triton) + python = "/opt/conda/envs/goose/bin/python" + ddr = child( + "ddrnet", + [ + python, + "-B", + "/probe/pilot_model.py", + "ddrnet", + "--ddrnet-layout", + args.ddrnet_layout, + "--ddrnet-execution", + args.ddrnet_execution, + ], + {**os.environ, "PYTHONPATH": "/probe"}, + ) + report["ddrnet_ready"], _ = receive(ddr.stdout) + decoder = child( + "decoder", + [python, "-B", "/probe/pilot_model.py", "camera"], + {**os.environ, "PYTHONPATH": "/probe", "CUDA_VISIBLE_DEVICES": ""}, + ) + report["decoder_ready"], _ = receive(decoder.stdout) + tgs = child("tgs", ["/usr/local/bin/pilot-tgs"]) + from pilot_graph import JointGraph + + graph = JointGraph( + "/code", + tgs, + mask_path="/valid-fov.png", + mapping_path="/goose.csv", + calibration_pack="/calibration.npz", + ) + report["rfdetr_warmup"] = asdict(graph.detector.warm_up()) + report["warmup_ms"] = (time.monotonic_ns() - started) / 1e6 + report["costmap_grid"] = graph.grid.tolist() + report["material_names"] = graph.material_names + + def compute_gpu(bundle): + begin = time.monotonic_ns() + send(ddr.stdin, {"op": "infer"}, bundle["image"].tobytes()) + ddr_result, raw = receive(ddr.stdout) + mask = np.frombuffer(raw, np.uint8).reshape(512, 512) + ddr_done = time.monotonic_ns() + proposals = graph.detector.detect(graph.packet(bundle)) + gpu_done = time.monotonic_ns() + return begin, ddr_done, gpu_done, ddr_result, mask, proposals + + monitor = threading.Thread(target=telemetry, args=(stop, samples), daemon=True) + monitor.start() + producer = threading.Thread( + target=produce, args=(args, decoder, mailbox, stop, source_report), daemon=True + ) + producer.start() + if args.schedule == "overlap-cpu": + gpu_stage = GpuStage(mailbox, compute_gpu, stop) + with (output / "scenes.jsonl").open("wb") as sink: + while True: + if gpu_stage: + item = gpu_stage.take() + if item is None: + break + bundle, computed = item + else: + bundle = mailbox.take() + if bundle is None: + break + computed = compute_gpu(bundle) + begin, ddr_done, gpu_done, ddr_result, mask, proposals = computed + cpu_started = time.monotonic_ns() + scene = graph.process( + bundle, + mask, + proposals=proposals, + detector_ms=(gpu_done - ddr_done) / 1e6, + ) + # Late evidence remains inspectable but cannot authorize terrain. + scene["oldest_required_input_age_ms"] = ( + time.monotonic_ns() - bundle["due_ns"] + ) / 1e6 + max( + bundle["lineage"]["pose_age_ms"] or 0, + bundle["lineage"]["oldest_point_age_ms"] or 0, + ) + scene["stale_at_publication"] = scene["oldest_required_input_age_ms"] > 250 + if scene["stale_at_publication"]: + scene["policy_actions"] = [2] * len(scene["policy_actions"]) + scene["policy_counts"] = { + "ALLOW_candidate": 0, + "HIGH_COST": 0, + "NO_GO": len(scene["policy_actions"]), + } + scene.update( + sequence=bundle["sequence"], + lineage=bundle["lineage"], + available=bundle["available"], + original_source_ns=bundle["time_ns"], + ) + encoded = json.dumps(scene, allow_nan=False, separators=(",", ":")).encode() + if len(encoded) > 1024 * 1024: + raise ValueError("scene exceeds bounded collector message budget") + # Actual bounded local receiver parse. Durable export excluded below. + received = json.loads(encoded) + if received["sequence"] != bundle["sequence"]: + raise ValueError("collector identity mismatch") + finished = time.monotonic_ns() + timing = { + **scene["timing_ms"], + **{"ddrnet_" + key: value for key, value in ddr_result["stages_ms"].items()}, + "ddrnet_rpc_ms": (ddr_done - begin) / 1e6, + "ddrnet_component_ms": ddr_result["component_ms"], + "ddrnet_forward_ms": ddr_result["forward_ms"], + "decode_ms": bundle["decode_ms"], + "decode_rpc_ms": bundle["decode_rpc_ms"], + "source_release_lag_ms": bundle["source_release_lag_ms"], + "queue_wait_ms": (begin - bundle["enqueued_ns"]) / 1e6, + "gpu_to_cpu_queue_wait_ms": (cpu_started - gpu_done) / 1e6, + "cpu_tail_to_receiver_ms": (finished - cpu_started) / 1e6, + "compute_to_receiver_ms": (finished - begin) / 1e6, + "source_due_to_receiver_ms": (finished - bundle["due_ns"]) / 1e6, + } + result = { + "sequence": bundle["sequence"], + "finished_monotonic_ns": finished, + "timing_ms": timing, + "available": bundle["available"], + "surface_state": scene["surface_state"], + "proposal_count": len(scene["proposals"]), + "observation_count": len(scene["observations"]), + "track_count": len(scene["tracks"]), + "metric_count": sum( + x["metric_geometry"] is not None for x in scene["observations"] + ), + "tgs_counts": scene["tgs_counts"], + "policy_counts": scene["policy_counts"], + "scene_bytes": len(encoded), + "scene_sha256": hashlib.sha256(encoded).hexdigest(), + } + results.append(result) + mailbox.release(bundle) + sink.write(encoded + b"\n") + if len(results) == 1: + report["reads_at_first_result"] = { + "results": 1, + "source_eof_reached": False, + "source_end_window_reached": mailbox.done, + } + if len(results) % 16 == 0: + print( + json.dumps( + { + "completed": len(results), + "last_sequence": bundle["sequence"], + "last_age_ms": timing["source_due_to_receiver_ms"], + "drops": len(mailbox.dropped), + } + ), + flush=True, + ) + if mailbox.error: + raise RuntimeError(mailbox.error) + report["execution_complete"] = True + except Exception: + report["execution_complete"] = False + report["error"] = traceback.format_exc() + print(report["error"], flush=True) + finally: + signal.alarm(0) + signal.signal(signal.SIGALRM, previous) + stop.set() + shutdown_start = time.monotonic_ns() + for process in reversed(children): + if process.poll() is None: + os.killpg(process.pid, signal.SIGTERM) + deadline = time.monotonic() + 4 + for process in reversed(children): + try: + process.wait(timeout=max(0.01, deadline - time.monotonic())) + except subprocess.TimeoutExpired: + os.killpg(process.pid, signal.SIGKILL) + process.wait(timeout=1) + if producer: + producer.join(timeout=2) + if monitor: + monitor.join(timeout=2) + report["gpu_stage_stopped"] = gpu_stage.close() if gpu_stage else True + if graph: + graph.backend.close() + for log in logs: + log.close() + report["stop_ms"] = (time.monotonic_ns() - shutdown_start) / 1e6 + report["children_stopped"] = all(p.poll() is not None for p in children) + report["source"] = source_report + report["frames"] = results + report["telemetry"] = samples + report["queue"] = { + "capacity": mailbox.capacity, + "gpu_completed_capacity": 1 if gpu_stage else 0, + "completed_shares_pending_limit": bool(gpu_stage), + "global_pending_limit": 2, + "gpu_peak_pending": gpu_stage.peak_pending if gpu_stage else 0, + "byte_limit": mailbox.byte_limit, + "peak_pending": mailbox.peak_pending, + "peak_bytes": mailbox.peak_bytes, + "drops": mailbox.dropped, + "residual_bytes": mailbox.bytes, + "residual_completed": mailbox.external_pending, + } + report["distributions_ms"] = ( + { + key: distribution([r["timing_ms"][key] for r in results]) + for key in results[0]["timing_ms"] + } + if results + else {} + ) + camera_count = source_report.get("arrivals", {}).get("camera", 0) + report["accounting"] = { + "released": camera_count, + "completed": len(results), + "dropped": len(mailbox.dropped), + "unaccounted": camera_count - len(results) - len(mailbox.dropped), + } + latency = report["distributions_ms"].get("source_due_to_receiver_ms") + queue_lags = [ + r["timing_ms"]["queue_wait_ms"] + r["timing_ms"]["gpu_to_cpu_queue_wait_ms"] + for r in results + ] + growth = ( + max( + 0.0, + sum(queue_lags[-8:]) / len(queue_lags[-8:]) - sum(queue_lags[:8]) / len(queue_lags[:8]), + ) + if queue_lags + else None + ) + report["backlog_growth_ms"] = growth + report["backlog_growth_basis"] = "ingress-plus-completed-GPU-wait-last8-minus-first8" + first_ms = ( + ((results[0]["finished_monotonic_ns"] - source_report["wall_zero_ns"]) / 1e6) + if results + else None + ) + report["first_result_from_stream_start_ms"] = first_ms + report["gates"] = { + "execution": report["execution_complete"], + "zero_drops_complete_accounting": camera_count == args.frames == len(results) + and not mailbox.dropped, + "worker_p95_p99_125ms": bool(latency and latency["p95"] <= 125 and latency["p99"] <= 125), + "source_release_lag_25ms": bool( + source_report.get("release_lag_ms") and source_report["release_lag_ms"]["max"] <= 25 + ), + "backlog_growth_25ms": growth is not None and growth <= 25, + "first_result_1s": first_ms is not None and first_ms <= 1000, + "warmup_120s": report.get("warmup_ms", math.inf) <= 120000, + "stop_5s": report["stop_ms"] <= 5000 + and report["children_stopped"] + and report["gpu_stage_stopped"], + "vram_22000MiB": bool(samples) + and max(s.get("gpu_used_mib", 99999) for s in samples) <= 22000, + "container_memory_8192MiB": bool(samples) + and max(s.get("cgroup_memory_mib", 99999) for s in samples) <= 8192, + "all_modalities_fresh": len(results) == args.frames + and all(r["available"] for r in results), + "network_end_to_end_qualified": False, + } + report["profile_realtime_qualified"] = False + (output / "report.json").write_text(json.dumps(report, indent=2, allow_nan=False) + "\n") + print( + json.dumps( + { + key: report[key] + for key in ( + "run_id", + "execution_complete", + "accounting", + "gates", + "distributions_ms", + ) + } + ), + flush=True, + ) + return 0 if report["execution_complete"] else 1 + + +def drain(stream, sink): + while True: + raw = stream.read(4096) + if not raw: + break + sink.write(raw) + sink.flush() + + +if __name__ == "__main__": + parser = argparse.ArgumentParser() + parser.add_argument("--frames", type=int, default=128) + parser.add_argument("--run-id", required=True) + parser.add_argument("--output", required=True) + parser.add_argument("--sensor-archive", default="/sensor-source.npz") + parser.add_argument("--camera-index", default="/camera-index.jsonl") + parser.add_argument( + "--ddrnet-layout", choices=("reference", "channels-last"), default="reference" + ) + parser.add_argument("--ddrnet-execution", choices=("eager", "cuda-graph"), default="eager") + parser.add_argument("--schedule", choices=("serial", "overlap-cpu"), default="serial") + args = parser.parse_args() + if not 1 <= args.frames <= 256: + parser.error("pilot window must be 1..256 frames") + raise SystemExit(run(args)) diff --git a/tests/test_perception_joint_pilot.py b/tests/test_perception_joint_pilot.py new file mode 100644 index 0000000..de4b5df --- /dev/null +++ b/tests/test_perception_joint_pilot.py @@ -0,0 +1,216 @@ +"""Small synthetic checks only; model/real-source runs belong on Worker 006.""" + +import importlib +import io +import json +import threading +import zipfile +from pathlib import Path + +import numpy as np +import pytest + + +@pytest.fixture +def pilot(monkeypatch): + root = Path(__file__).resolve().parents[1] + monkeypatch.syspath_prepend( + str(root / "experiments/perception/worker/streaming_profile_stage1") + ) + return lambda name: importlib.import_module(name) + + +def test_pending_overflow_is_explicit_and_does_not_evict_active(pilot): + queue = pilot("pilot_queue").Mailbox(capacity=2, byte_limit=100) + + def make(sequence): + return {"sequence": sequence, "payload_bytes": 10} + + queue.put(make(0)) + active = queue.take() + for sequence in (1, 2, 3): + queue.put(make(sequence)) + assert queue.dropped == [{"sequence": 1, "reason": "pending-overflow"}] + assert queue.bytes == 30 + assert queue.peak_pending == 2 + queue.release(active) + assert queue.take()["sequence"] == 2 + assert queue.take()["sequence"] == 3 + + +def test_active_bytes_count_towards_memory_limit(pilot): + queue = pilot("pilot_queue").Mailbox(byte_limit=20) + queue.put({"sequence": 0, "payload_bytes": 20}) + active = queue.take() + queue.put({"sequence": 1, "payload_bytes": 1}) + assert queue.dropped == [{"sequence": 1, "reason": "byte-budget"}] + assert queue.peak_bytes == 20 + queue.release(active) + queue.finish() + assert queue.take() is None + + +def test_numpy_member_is_read_incrementally_and_bounded(pilot): + data = io.BytesIO() + np.savez_compressed(data, points=np.arange(300, dtype="