diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_decoder_probe.py b/experiments/perception/worker/streaming_profile_stage1/pilot_decoder_probe.py new file mode 100644 index 0000000..5feddae --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_decoder_probe.py @@ -0,0 +1,124 @@ +"""CPU-only bounded decoder parity probe; paths belong only to this source adapter.""" + +# The reference decoder runs in the pinned Python 3.9 environment. +# ruff: noqa: UP017, B905 + +import argparse +import hashlib +import json +import resource +import time +from datetime import datetime, timezone +from pathlib import Path + + +def read_fragment(root, row): + path = (root / "segments" / (str(row["sequence"]) + ".m4s")).resolve(strict=True) + if path.parent != (root / "segments").resolve(strict=True): + raise ValueError("camera path confinement failed") + with path.open("rb") as stream: + payload = stream.read(1024 * 1024 + 1) + if len(payload) != row["length"] or len(payload) > 1024 * 1024: + raise ValueError("camera fragment size changed") + if hashlib.sha256(payload).hexdigest() != row["sha256"]: + raise ValueError("camera fragment hash changed") + return payload + + +def run(args): + resource.setrlimit(resource.RLIMIT_AS, (1024**3, 1024**3)) + report = { + "mode": args.mode, + "created_utc": datetime.now(timezone.utc).isoformat(), + "started_monotonic_ns": time.monotonic_ns(), + "frames": [], + "source_pace": 1 if args.mode == "fragments" else None, + } + decoder = reference = None + if args.mode == "fragments": + from k1link.perception.streaming_decoder import FragmentDecoder + + decoder = FragmentDecoder() + with (args.camera_root / "init.mp4").open("rb") as stream: + init = stream.read(65537) + begin = time.monotonic_ns() + decoder.configure(init) + report["init_ms"] = (time.monotonic_ns() - begin) / 1e6 + report["init_sha256"] = hashlib.sha256(init).hexdigest() + else: + import cv2 + + cv2.setNumThreads(1) + reference = cv2.VideoCapture(str(args.reference_video)) + if not reference.isOpened(): + raise ValueError("reference video unavailable") + report["opencv_version"] = cv2.__version__ + source_zero = wall_zero = None + with (args.camera_root / "index.jsonl").open("rb") as index: + for sequence in range(args.frames): + line = index.readline(65537) + if not line or len(line) > 65536: + raise ValueError("index absent or unbounded") + row = json.loads(line) + if row["sequence"] != sequence + 1 or row["kind"] != "media": + raise ValueError("camera source identity changed") + source_time = row["host_monotonic_ns"] + if source_zero is None: + source_zero, wall_zero = source_time, time.monotonic_ns() + due = wall_zero + source_time - source_zero + if decoder: + time.sleep(max(0, (due - time.monotonic_ns()) / 1e9)) + read_started = time.monotonic_ns() + payload = read_fragment(args.camera_root, row) if decoder else None + decode_started = time.monotonic_ns() + if decoder: + image = decoder.decode(payload) + else: + ok, image = reference.read() + if not ok: + raise ValueError("reference ended early") + decoded = time.monotonic_ns() + report["frames"].append( + { + "sequence": sequence, + "source_time_ns": source_time, + "fragment_sha256": row["sha256"], + "read_started_ns": read_started, + "decoded_ns": decoded, + "decode_ms": (decoded - decode_started) / 1e6, + "release_lag_ms": (read_started - due) / 1e6 if decoder else None, + "shape": list(image.shape), + "bgr_sha256": hashlib.sha256(image).hexdigest(), + } + ) + del image, payload + if decoder: + report["frame_before_next_fragment_read"] = all( + a["decoded_ns"] < b["read_started_ns"] + for a, b in zip(report["frames"], report["frames"][1:]) + ) + decoder.close() + if reference: + reference.release() + report["finished_monotonic_ns"] = time.monotonic_ns() + report["maxrss_kib"] = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss + args.output.write_text(json.dumps(report, indent=2) + "\n") + print( + json.dumps( + {"mode": args.mode, "frames": len(report["frames"]), "maxrss_kib": report["maxrss_kib"]} + ), + flush=True, + ) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser() + parser.add_argument("mode", choices=("fragments", "reference")) + parser.add_argument("--camera-root", type=Path, required=True) + parser.add_argument("--reference-video", type=Path) + parser.add_argument("--frames", type=int, default=128) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + if not 1 <= args.frames <= 256 or (args.mode == "reference" and not args.reference_video): + parser.error("bounded frame count and explicit reference required") + run(args) diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_stream_ingress_probe.py b/experiments/perception/worker/streaming_profile_stage1/pilot_stream_ingress_probe.py index fae8b2b..bb16351 100644 --- a/experiments/perception/worker/streaming_profile_stage1/pilot_stream_ingress_probe.py +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_stream_ingress_probe.py @@ -2,7 +2,8 @@ Recorded mode transfers original fMP4 init/media plus existing normalized point/pose increments at original 1x release times. Receiver is never told the -source length. The consumer only hashes raw input, it does not decode/infer. +source length. Optional camera decoding does not infer or bind sensors into a +full graph scene; the default consumer only hashes raw input. Truncated mode is explicitly synthetic and exercises multi-fragment failure. """ @@ -172,13 +173,14 @@ def producer(args): threading.Event().wait(30) -def run_case(base, case, generation): +def run_case(base, case, generation, *, decode_camera=False): root = base / case root.mkdir() specification = { "case": case, "transport": wire.SCHEMA, - "consumer": "raw-digest-only", + "consumer": "fragment-decode-and-raw-digest" if decode_camera else "raw-digest-only", + "full_graph": False, "camera": "original-fmp4-init-and-media", "points": "u32le-count+xyz-f64le+intensity-u8", "pose": "position-f64le3+map-from-lidar-quaternion-f64le4", @@ -217,9 +219,16 @@ def run_case(base, case, generation): heartbeat.start() left, right = socket.socketpair() receiver = None + decoder = codec_reservation = None started = time.monotonic_ns() with (root / "producer.log").open("wb") as log, (root / "received.jsonl").open("w") as journal: try: + if decode_camera: + from k1link.perception.streaming_decoder import FragmentDecoder + + # Stored init/extradata have independent ownership until close. + codec_reservation = run.mailbox.reserve_ingress(2 * 65536) + decoder = FragmentDecoder() run.spawn( lambda: subprocess.Popen( [ @@ -244,12 +253,38 @@ def run_case(base, case, generation): left.close() def consume(event): + decoded = {} + if decoder is not None and event.modality.startswith("camera"): + # Conservative raw/sample/BGR scratch, reserved BEFORE decode. + # FFmpeg codec/DPB/native allocations are additionally bounded + # by the CPU-only container, not claimed as Python payloads. + scratch = run.mailbox.reserve_ingress( + 2 * wire.MAX_FRAGMENT + 2 * 1440000 + 65536 + ) + image = None + begin = time.monotonic_ns() + try: + if event.modality == "camera-init": + decoder.configure(event.payload) + decoded["initialized"] = True + else: + image = decoder.decode(event.payload) + decoded.update( + bgr_sha256=hashlib.sha256(image).hexdigest(), + decoded_frame_index=decoder.frames - 1, + shape=list(image.shape), + ) + decoded["decode_ms"] = (time.monotonic_ns() - begin) / 1e6 + finally: + image = None + scratch.release() emit_json( journal, { **record(event), "received_monotonic_ns": str(time.monotonic_ns()), "original_received_monotonic_ns": str(event.received_monotonic_ns), + **decoded, }, ) @@ -273,6 +308,10 @@ def run_case(base, case, generation): right.close() if receiver: receiver.join() + if decoder is not None: + decoder.close() + if codec_reservation is not None: + codec_reservation.release() heartbeat_stop.set() heartbeat.join(timeout=1) released = run.close() @@ -303,6 +342,7 @@ def run_case(base, case, generation): if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--producer", action="store_true") + parser.add_argument("--decode-camera", action="store_true") parser.add_argument("--fd", type=int) parser.add_argument("--case", choices=("recorded", "truncated")) parser.add_argument("--output", default="/out") @@ -311,4 +351,4 @@ if __name__ == "__main__": producer(args) else: for generation, case in enumerate(("recorded", "truncated"), 1): - run_case(Path(args.output), case, generation) + run_case(Path(args.output), case, generation, decode_camera=args.decode_camera)