diff --git a/examples/camera_streamer/README.md b/examples/camera_streamer/README.md index 4685b51c6e..6a1edb9fae 100644 --- a/examples/camera_streamer/README.md +++ b/examples/camera_streamer/README.md @@ -56,6 +56,22 @@ The build downloads the pinned NVIDIA Video Codec SDK. CPU-side Python tests: uv run --project tests --extra dev python -m pytest tests/python -q ``` +## Workstation receiver + +Run `ip -brief -4 address` on the robot and choose the address reachable from the +workstation. From `examples/camera_streamer`, with the decoder already built: + +```bash +export ROBOT_IP="" +uv pip install --python .venv/bin/python -r camera_receiver/requirements.txt +PROFILE=s56-stereo-2560x1984-30 # Choose your resolution +.venv/bin/python camera_receiver/camera_receiver.py \ + --config "configs/${PROFILE}.yaml" --host "${ROBOT_IP:?Set ROBOT_IP first}" +``` + +Use the same profile on both computers. Decoded GPU frames are published through +local CUDA IPC; the receiver does not open a window. Stop it with Ctrl+C. + ## Licensing This example is Apache-2.0 licensed. Existing upstream notices and external SDK @@ -72,3 +88,9 @@ For opt-in hardware smoke tests, install PyYAML for the Python interpreter selec by CMake. Set `ARGUS_SENDER_HARDWARE_CONFIG` to your stereo profile; the smoke receiver reads `network.port` from that same file (default 5000), so there is no separate receiver-port override. + +CUDA IPC protocol 3 requires consumers to hold each slot's shared file lock and +verify `slot_generation` before reading pixels, then finish GPU reads before +unlocking. The receiver holds an exclusive lock while writing. Delayed notifications +are discarded; a slow active read backpressures slot reuse rather than corrupting +pixels. Update receiver and consumer together; protocol 2 consumers are rejected. diff --git a/examples/camera_streamer/camera_receiver/camera_receiver.py b/examples/camera_streamer/camera_receiver/camera_receiver.py new file mode 100755 index 0000000000..df04461704 --- /dev/null +++ b/examples/camera_streamer/camera_receiver/camera_receiver.py @@ -0,0 +1,1467 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Standalone AVS1 receiver and CUDA IPC publisher. + +The sender emits one AVS1 bundle per TCP record: + + uint32_be_record_bytes || AVS1 bundle + +This process connects to the sender, decodes or copies configured streams into +GPU-resident rings, and publishes those rings over CUDA IPC for same-host +consumers. +""" + +from __future__ import annotations + +import argparse +import logging +import math +import os +import signal +import socket +import struct +import threading +import time +from concurrent.futures import Future, ThreadPoolExecutor, wait +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Callable, Iterable, Optional + +import yaml + +from avs_wire import ( + CODEC_HEVC, + CODEC_RAW_NV12, + HEADER as AVS1_HEADER, + Avs1Bundle, + Avs1Sample, + Avs1Stream, + parse_avs1_record, + sensor_sof_lag_ns, + sensor_sof_ns, +) +from cuda_ipc import CudaIpcFramePublisher +from nv_video_decode import NvVideoDecoder + +logger = logging.getLogger("isaaccapture.camera_streamer.camera_receiver") + +_SOCK_TIMEOUT_S = 0.5 +# If no bytes arrive for this long the link is treated as dead and torn down so +# the produce loop reconnects. A half-open peer -- e.g. the AGX rebooted -- leaves +# the PC socket ESTABLISHED with data that never comes; without this bound recv +# just keeps timing out on that dead socket forever and the camera never returns +# until the receiver is killed by hand. Far above a 60 fps inter-frame gap (~17ms) +# so a healthy stream never trips it. +_STALL_TIMEOUT_S = 3.0 +# Grace for a FRESH connection before the stall watchdog arms. The AGX sender's +# NVENC/capture-domain cold start takes ~2-3 s to emit its first frame bundle; +# a 3 s stall timeout applied from connect fires DURING that warmup, before any +# data has ever arrived, and tears the link down in a reconnect loop that never +# lets video through. So until the first bundle is received we allow this longer +# window; once the stream is flowing, _STALL_TIMEOUT_S governs a real stall. +_COLD_START_TIMEOUT_S = 15.0 +_CONNECT_WAIT_LOG_INTERVAL_S = 5.0 +_DEFAULT_MAX_RECORD_BYTES = 256 * 1024 * 1024 +_DEFAULT_STATS_LOG_INTERVAL_S = 5.0 +_DEFAULT_RECEIVE_FPS_WARNING_INTERVAL_S = 10.0 +_OUTPUT_RGBA8 = "rgba8" +_OUTPUT_NV12 = "nv12" +_NV12_TO_RGBA_SRC = r""" +extern "C" __global__ +void nv12_to_rgba(const unsigned char* nv12, unsigned char* out, int W, int H, + int rotation) { + int x = blockIdx.x * blockDim.x + threadIdx.x; + int y = blockIdx.y * blockDim.y + threadIdx.y; + if (x >= W || y >= H) return; + const unsigned char* Yp = nv12; + const unsigned char* UVp = nv12 + W * H; + float Y = (float)Yp[y * W + x]; + int uv = (y >> 1) * W + (x & ~1); + float U = (float)UVp[uv] - 128.0f; + float V = (float)UVp[uv + 1] - 128.0f; + float Yn = (Y - 16.0f) * 1.16438f; + float R = Yn + 1.79274f * V; + float G = Yn - 0.21325f * U - 0.53291f * V; + float B = Yn + 2.11240f * U; + int ox = rotation == 180 ? W - 1 - x : x; + int oy = rotation == 180 ? H - 1 - y : y; + int o = (oy * W + ox) * 4; + out[o + 0] = (unsigned char)fminf(fmaxf(R, 0.0f), 255.0f); + out[o + 1] = (unsigned char)fminf(fmaxf(G, 0.0f), 255.0f); + out[o + 2] = (unsigned char)fminf(fmaxf(B, 0.0f), 255.0f); + out[o + 3] = 255; +} +""" +_NV12_ROTATE_180_SRC = r""" +extern "C" __global__ +void nv12_rotate_180(const unsigned char* src, unsigned char* dst, int W, int H) { + int x = blockIdx.x * blockDim.x + threadIdx.x; + int y = blockIdx.y * blockDim.y + threadIdx.y; + if (x < W && y < H) { + dst[(H - 1 - y) * W + W - 1 - x] = src[y * W + x]; + } + if (x < W / 2 && y < H / 2) { + int sx = x * 2; + int dx = W - 2 - sx; + int dy = H / 2 - 1 - y; + const unsigned char* src_uv = src + W * H; + unsigned char* dst_uv = dst + W * H; + dst_uv[dy * W + dx] = src_uv[y * W + sx]; + dst_uv[dy * W + dx + 1] = src_uv[y * W + sx + 1]; + } +} +""" + + +def _read_wall_clock_ns() -> tuple[int, str]: + # Use the chrony-disciplined Unix wall clock for cross-host latency. + return int(time.time_ns()), "realtime" + + +def _ns_delta_ms(later_ns: Optional[int], earlier_ns: Optional[int]) -> Optional[float]: + if later_ns is None or earlier_ns is None: + return None + return (int(later_ns) - int(earlier_ns)) / 1.0e6 + + +def _parse_optional_gpu_id(value: object, source: str = "gpu_id") -> Optional[int]: + if value is None: + return None + try: + gpu_id = int(value) + except (TypeError, ValueError) as e: + raise ValueError(f"{source} must be an integer or null, got {value!r}") from e + if gpu_id < 0: + raise ValueError(f"{source} must be >= 0 or null, got {gpu_id}") + return gpu_id + + +def _parse_stats_log_interval_s( + value: object, + source: str = "receiver.stats_log_interval_s", +) -> float: + try: + interval_s = float(value) + except (TypeError, ValueError) as e: + raise ValueError(f"{source} must be a number >= 0, got {value!r}") from e + if not math.isfinite(interval_s) or interval_s < 0: + raise ValueError(f"{source} must be a finite number >= 0, got {value!r}") + return interval_s + + +@dataclass(frozen=True) +class ReceiveFpsWarningConfig: + threshold_fps: float + interval_s: float + + @property + def enabled(self) -> bool: + return self.threshold_fps > 0 + + +def _resolve_receive_fps_warning_config( + receiver_cfg: dict[str, Any], +) -> ReceiveFpsWarningConfig: + threshold_fps = _parse_stats_log_interval_s( + receiver_cfg.get("receive_fps_warning_threshold", 0.0), + "receiver.receive_fps_warning_threshold", + ) + interval_s = _parse_stats_log_interval_s( + receiver_cfg.get( + "receive_fps_warning_interval_s", + _DEFAULT_RECEIVE_FPS_WARNING_INTERVAL_S, + ), + "receiver.receive_fps_warning_interval_s", + ) + if threshold_fps > 0 and interval_s <= 0: + raise ValueError( + "receiver.receive_fps_warning_interval_s must be > 0 when " + "receive_fps_warning_threshold is enabled" + ) + return ReceiveFpsWarningConfig( + threshold_fps=threshold_fps, + interval_s=interval_s, + ) + + +def _resolve_stats_log_interval_s( + receiver_cfg: dict[str, Any], + cli_value: Optional[float], +) -> float: + if cli_value is not None: + return _parse_stats_log_interval_s(cli_value, "--stats-log-interval-s") + if "stats_log_interval_s" in receiver_cfg: + return _parse_stats_log_interval_s(receiver_cfg["stats_log_interval_s"]) + # Backward compatibility for configs created before the interval setting. + if not bool(receiver_cfg.get("log_timing", True)): + return 0.0 + return _DEFAULT_STATS_LOG_INTERVAL_S + + +class PeriodicStats: + def __init__(self, label: str, interval_s: float = 5.0) -> None: + self._label = label + self._interval_s = float(interval_s) + if self._interval_s <= 0: + raise ValueError("PeriodicStats interval_s must be > 0") + self._lock = threading.Lock() + self._reset(time.monotonic()) + + def _reset(self, now: float) -> None: + self._n = 0 + self._sum_ms = 0.0 + self._max_ms = 0.0 + self._sof_delta_n = 0 + self._sof_delta_sum_ns = 0 + self._sof_delta_min_ns: Optional[int] = None + self._sof_delta_max_ns: Optional[int] = None + self._latency_stats: dict[str, list[float]] = {} + self._stream_counts: dict[str, int] = {} + self._window_start = now + + def add( + self, + stream_name: str, + dt_s: float, + sof_delta_ns: Optional[int] = None, + latencies_ms: Optional[dict[str, Optional[float]]] = None, + ) -> None: + dt_ms = dt_s * 1000.0 + log_line = None + with self._lock: + self._n += 1 + self._stream_counts[stream_name] = ( + self._stream_counts.get(stream_name, 0) + 1 + ) + self._sum_ms += dt_ms + self._max_ms = max(self._max_ms, dt_ms) + if sof_delta_ns is not None: + delta_ns = int(sof_delta_ns) + self._sof_delta_n += 1 + self._sof_delta_sum_ns += delta_ns + if self._sof_delta_min_ns is None: + self._sof_delta_min_ns = delta_ns + self._sof_delta_max_ns = delta_ns + else: + self._sof_delta_min_ns = min(self._sof_delta_min_ns, delta_ns) + assert self._sof_delta_max_ns is not None + self._sof_delta_max_ns = max(self._sof_delta_max_ns, delta_ns) + if latencies_ms: + for name, value in latencies_ms.items(): + if value is None: + continue + bucket = self._latency_stats.setdefault(name, [0.0, 0.0, 0.0]) + bucket[0] += 1.0 + bucket[1] += float(value) + bucket[2] = max(bucket[2], float(value)) + now = time.monotonic() + elapsed = now - self._window_start + if elapsed >= self._interval_s and self._n > 0: + mean = self._sum_ms / self._n + fps = self._n / elapsed + stream_parts = [ + f"{name}={count / elapsed:.1f}fps" + for name, count in sorted(self._stream_counts.items()) + ] + log_line = ( + f"{self._label}: {fps:.1f} streams/s, {mean:.2f} ms mean, " + f"{self._max_ms:.2f} ms max (n={self._n} over {elapsed:.1f}s)" + ) + if stream_parts: + log_line += ", " + " ".join(stream_parts) + if self._sof_delta_n > 0: + assert self._sof_delta_min_ns is not None + assert self._sof_delta_max_ns is not None + mean_delta = self._sof_delta_sum_ns / self._sof_delta_n + log_line += ( + f", sof_delta={mean_delta:+.0f} ns mean " + f"[{self._sof_delta_min_ns:+d}," + f"{self._sof_delta_max_ns:+d}]" + ) + else: + log_line += ", sof_delta=n/a" + if self._latency_stats: + parts = [] + for name in sorted(self._latency_stats): + count, total, max_value = self._latency_stats[name] + if count > 0: + parts.append(f"{name}={total / count:.2f}/{max_value:.2f}") + if parts: + log_line += ", latency_ms(avg/max) " + " ".join(parts) + self._reset(now) + if log_line is not None: + logger.info(log_line) + + +class ReceiveFpsMonitor: + """Warn when a configured logical AVS stream arrives below the target FPS.""" + + def __init__( + self, + stream_names: Iterable[str], + config: ReceiveFpsWarningConfig, + clock: Callable[[], float] = time.monotonic, + ) -> None: + if not config.enabled or config.interval_s <= 0: + raise ValueError("ReceiveFpsMonitor requires an enabled config") + names = tuple(sorted(set(stream_names))) + if not names: + raise ValueError("ReceiveFpsMonitor requires at least one stream") + self._stream_names = names + self._threshold_fps = config.threshold_fps + self._interval_s = config.interval_s + self._clock = clock + self._active = False + self._window_start = 0.0 + self._counts = dict.fromkeys(self._stream_names, 0) + + def start(self) -> None: + self._active = True + self._reset(self._clock()) + + def stop(self) -> None: + self._active = False + + def observe(self, stream_names: Iterable[str]) -> None: + if not self._active: + return + for name in stream_names: + if name in self._counts: + self._counts[name] += 1 + self._maybe_warn(self._clock()) + + def poll(self) -> None: + if self._active: + self._maybe_warn(self._clock()) + + def _reset(self, now: float) -> None: + self._window_start = now + self._counts = dict.fromkeys(self._stream_names, 0) + + def _maybe_warn(self, now: float) -> None: + elapsed = now - self._window_start + if elapsed < self._interval_s: + return + rates = {name: self._counts[name] / elapsed for name in self._stream_names} + low_rates = { + name: fps for name, fps in rates.items() if fps < self._threshold_fps + } + if low_rates: + details = " ".join( + f"{name}={fps:.1f}fps" for name, fps in low_rates.items() + ) + logger.warning( + "AVS1 receive FPS below %.1ffps target over %.1fs: %s", + self._threshold_fps, + elapsed, + details, + ) + self._reset(now) + + +@dataclass(frozen=True) +class ReceivedRecord: + payload: bytes + receiver_rx_prefix_done_ns: int + receiver_rx_payload_done_ns: int + receiver_rx_prefix_done_mono_ns: int + receiver_rx_payload_done_mono_ns: int + receiver_clock_domain: str + + +@dataclass(frozen=True) +class StreamConfig: + index: int + name: str + stereo: bool + width: int + height: int + output_format: str + ring_depth: int + ipc_names: tuple[str, ...] + full_range: bool = False + + +class _StreamRuntime: + def __init__(self, config: StreamConfig, gpu_id: int) -> None: + import cupy as cp + + self.config = config + self.gpu_id = int(gpu_id) + self.camera_count = 2 if config.stereo else 1 + self.write_index = 0 + self.slot_generations = [ + [0 for _ in range(config.ring_depth)] for _ in range(self.camera_count) + ] + self.decoders: list[Optional[NvVideoDecoder]] = [None] * self.camera_count + self.executor = ThreadPoolExecutor( + max_workers=1, + thread_name_prefix=f"camera_receiver_stream_{config.name}", + ) + self.inflight = threading.BoundedSemaphore(config.ring_depth) + self.block = (16, 16, 1) + self.grid = ( + (config.width + 15) // 16, + (config.height + 15) // 16, + 1, + ) + with cp.cuda.Device(gpu_id): + if config.output_format == _OUTPUT_RGBA8: + self.output_buffers = [ + [ + cp.empty((config.height, config.width, 4), dtype=cp.uint8) + for _ in range(config.ring_depth) + ] + for _ in range(self.camera_count) + ] + for camera_buffers in self.output_buffers: + for buf in camera_buffers: + buf[..., 3] = 255 + self.raw_buffers = [ + cp.empty(config.width * config.height * 3 // 2, dtype=cp.uint8) + for _ in range(self.camera_count) + ] + self.raw_kernel = cp.RawKernel(_NV12_TO_RGBA_SRC, "nv12_to_rgba") + elif config.output_format == _OUTPUT_NV12: + self.output_buffers = [ + [ + cp.empty((config.height * 3 // 2, config.width), dtype=cp.uint8) + for _ in range(config.ring_depth) + ] + for _ in range(self.camera_count) + ] + self.raw_buffers = [ + cp.empty(config.width * config.height * 3 // 2, dtype=cp.uint8) + for _ in range(self.camera_count) + ] + self.raw_kernel = cp.RawKernel(_NV12_ROTATE_180_SRC, "nv12_rotate_180") + else: + raise ValueError(f"unsupported output format {config.output_format!r}") + + def reset_decoders(self) -> None: + for index, decoder in enumerate(self.decoders): + if decoder is not None: + try: + decoder.reset() + except Exception: + logger.debug( + "decoder reset failed for %s[%d]", + self.config.name, + index, + exc_info=True, + ) + self.decoders[index] = None + + def submit(self, stop: threading.Event, fn, *args) -> bool: + """Bound decode and reset work while allowing shutdown to interrupt waiting.""" + while not stop.is_set(): + if not self.inflight.acquire(timeout=0.1): + continue + if stop.is_set(): + self.inflight.release() + return False + try: + future = self.executor.submit(fn, *args) + except BaseException: + self.inflight.release() + raise + future.add_done_callback(lambda _future: self.inflight.release()) + return True + return False + + def shutdown(self) -> None: + self.executor.shutdown(wait=True) + self.reset_decoders() + + def next_generation(self, camera_index: int, slot: int) -> int: + self.slot_generations[camera_index][slot] += 1 + return self.slot_generations[camera_index][slot] + + +class Avs1TcpReceiver: + def __init__( + self, + host: str, + port: int, + gpu_id: Optional[int], + max_record_bytes: int, + stats_log_interval_s: float, + receive_fps_warning: ReceiveFpsWarningConfig, + ipc_publisher: CudaIpcFramePublisher, + streams: list[StreamConfig], + ) -> None: + try: + import cupy as cp + except ImportError as e: + raise RuntimeError( + "camera_receiver requires CuPy. Install the receiver extra as described in examples/camera_streamer/README.md." + ) from e + if not streams: + raise ValueError("camera_receiver requires at least one AVS stream") + indices = [stream.index for stream in streams] + if len(indices) != len(set(indices)): + raise ValueError("AVS stream indexes must be unique") + if max_record_bytes <= AVS1_HEADER.size: + raise ValueError( + f"max_record_bytes must exceed AVS1 header size, got {max_record_bytes}" + ) + + parsed_gpu_id = _parse_optional_gpu_id(gpu_id) + self._gpu_device_id = ( + int(cp.cuda.runtime.getDevice()) if parsed_gpu_id is None else parsed_gpu_id + ) + self._host = str(host) + self._port = int(port) + self._max_record_bytes = int(max_record_bytes) + self._ipc_publisher = ipc_publisher + self._runtimes = { + stream.index: _StreamRuntime(stream, self._gpu_device_id) + for stream in streams + } + self._sample_pool: Optional[ThreadPoolExecutor] = None + self._header_buf = bytearray(4) + self._record_buf = bytearray() + self._stop = threading.Event() + self._thread: Optional[threading.Thread] = None + self._conn: Optional[socket.socket] = None + self._stats = None + if stats_log_interval_s > 0: + self._stats = PeriodicStats( + "decode[avs1]", + interval_s=stats_log_interval_s, + ) + self._receive_fps_monitor = None + if receive_fps_warning.enabled: + self._receive_fps_monitor = ReceiveFpsMonitor( + (runtime.config.name for runtime in self._runtimes.values()), + receive_fps_warning, + ) + self._session_id: Optional[int] = None + + for runtime in self._runtimes.values(): + cfg = runtime.config + layout = _ipc_layout(cfg.output_format, cfg.width, cfg.height) + for camera_index, ipc_name in enumerate(cfg.ipc_names): + self._ipc_publisher.register_stream( + name=ipc_name, + width=cfg.width, + height=cfg.height, + pixel_format=cfg.output_format, + gpu_id=self._gpu_device_id, + buffers=runtime.output_buffers[camera_index], + layout=layout, + ) + + @property + def host(self) -> str: + return self._host + + @property + def port(self) -> int: + return self._port + + @property + def socket_path(self) -> str: + return self._ipc_publisher.socket_path + + def start(self) -> None: + if self._thread is not None: + return + sample_workers = sum( + runtime.camera_count for runtime in self._runtimes.values() + ) + self._sample_pool = ThreadPoolExecutor( + max_workers=max(1, sample_workers), + thread_name_prefix="camera_receiver_sample", + ) + self._ipc_publisher.start() + self._stop.clear() + self._thread = threading.Thread( + target=self._produce_loop, + name="camera_receiver_avs1", + daemon=False, + ) + self._thread.start() + + def stop(self) -> None: + self._stop.set() + if self._conn is not None: + try: + self._conn.close() + except OSError: + logger.debug("Failed to close receiver socket", exc_info=True) + if self._thread is not None: + self._thread.join(timeout=5.0) + if self._thread.is_alive(): + logger.warning("receiver thread did not exit within 5s") + else: + self._thread = None + for runtime in self._runtimes.values(): + runtime.shutdown() + if self._sample_pool is not None: + self._sample_pool.shutdown(wait=False) + self._sample_pool = None + self._ipc_publisher.stop() + + def wait(self, stop_event: threading.Event) -> None: + while not stop_event.is_set(): + time.sleep(0.2) + + def _produce_loop(self) -> None: + import cupy as cp + + with cp.cuda.Device(self._gpu_device_id): + while not self._stop.is_set(): + conn = self._connect() + if conn is None: + break + self._conn = conn + # Not yet warm: the sender may still be in cold-start. Armed to + # the tight stall timeout only after the first bundle arrives. + self._link_warm = False + if self._receive_fps_monitor is not None: + self._receive_fps_monitor.start() + try: + while not self._stop.is_set(): + received = self._read_record(conn) + if received is None: + break + self._link_warm = True + bundle = parse_avs1_record(received.payload) + self._dispatch_bundle(received, bundle) + except Exception as e: + if not self._stop.is_set(): + logger.warning("receive/parse error (%s); reconnecting", e) + finally: + try: + conn.close() + except OSError: + logger.debug("Failed to close sender connection", exc_info=True) + self._conn = None + if self._receive_fps_monitor is not None: + self._receive_fps_monitor.stop() + self._session_id = None + for runtime in self._runtimes.values(): + runtime.submit(self._stop, runtime.reset_decoders) + + def _connect(self) -> Optional[socket.socket]: + next_wait_log_at = 0.0 + while not self._stop.is_set(): + try: + conn = socket.create_connection( + (self._host, self._port), timeout=_SOCK_TIMEOUT_S + ) + conn.settimeout(_SOCK_TIMEOUT_S) + conn.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + # Enable TCP keepalive with a short probe schedule so the kernel + # also detects a silently-dead peer (the AGX rebooted mid-stream, + # leaving this socket ESTABLISHED but never sending a FIN/RST). + # Belt-and-suspenders with the app-level _STALL_TIMEOUT_S below. + conn.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) + for _opt, _val in ( + ("TCP_KEEPIDLE", 2), + ("TCP_KEEPINTVL", 1), + ("TCP_KEEPCNT", 3), + ): + if hasattr(socket, _opt): + conn.setsockopt(socket.IPPROTO_TCP, getattr(socket, _opt), _val) + logger.info("connected to AVS1 sender %s:%d", self._host, self._port) + return conn + except OSError as e: + now = time.monotonic() + if now >= next_wait_log_at: + logger.info( + "waiting for AVS1 sender %s:%d (%s). Please check that " + "the robot is powered on and wait at least 2 minutes after " + "powering it on.", + self._host, + self._port, + e, + ) + next_wait_log_at = now + _CONNECT_WAIT_LOG_INTERVAL_S + deadline = now + 1.0 + while time.monotonic() < deadline and not self._stop.is_set(): + time.sleep(0.05) + return None + + def _read_record(self, conn: socket.socket) -> Optional[ReceivedRecord]: + header = memoryview(self._header_buf) + if not self._recv_exact(conn, header): + return None + receiver_rx_prefix_done_mono_ns = time.monotonic_ns() + receiver_rx_prefix_done_ns, receiver_clock_domain = _read_wall_clock_ns() + (length,) = struct.unpack("!I", header) + if length == 0: + raise ValueError("zero-length AVS1 record") + if length > self._max_record_bytes: + raise ValueError( + f"AVS1 record too large: {length} bytes > {self._max_record_bytes}" + ) + if len(self._record_buf) < length: + self._record_buf = bytearray(length) + view = memoryview(self._record_buf)[:length] + if not self._recv_exact(conn, view): + return None + receiver_rx_payload_done_mono_ns = time.monotonic_ns() + receiver_rx_payload_done_ns, receiver_payload_clock_domain = ( + _read_wall_clock_ns() + ) + if receiver_payload_clock_domain != receiver_clock_domain: + receiver_clock_domain = receiver_payload_clock_domain + return ReceivedRecord( + payload=bytes(view), + receiver_rx_prefix_done_ns=receiver_rx_prefix_done_ns, + receiver_rx_payload_done_ns=receiver_rx_payload_done_ns, + receiver_rx_prefix_done_mono_ns=receiver_rx_prefix_done_mono_ns, + receiver_rx_payload_done_mono_ns=receiver_rx_payload_done_mono_ns, + receiver_clock_domain=receiver_clock_domain, + ) + + def _dispatch_bundle(self, received: ReceivedRecord, bundle: Avs1Bundle) -> None: + if self._session_id != bundle.session_id: + if self._session_id is not None: + logger.info( + "AVS1 session changed: %s -> %s; resetting decoders", + self._session_id, + bundle.session_id, + ) + for runtime in self._runtimes.values(): + runtime.submit(self._stop, runtime.reset_decoders) + self._session_id = bundle.session_id + + dispatch_items: list[tuple[Avs1Stream, _StreamRuntime]] = [] + for stream in bundle.streams: + runtime = self._runtimes.get(stream.index) + if runtime is None or not stream.present: + continue + dispatch_items.append((stream, runtime)) + if self._receive_fps_monitor is not None: + self._receive_fps_monitor.observe( + runtime.config.name for _, runtime in dispatch_items + ) + for stream, runtime in dispatch_items: + runtime.submit( + self._stop, + self._process_stream_safe, + received, + bundle, + stream, + runtime, + ) + + def _process_stream_safe( + self, + received: ReceivedRecord, + bundle: Avs1Bundle, + stream: Avs1Stream, + runtime: _StreamRuntime, + ) -> None: + try: + self._process_stream(received, bundle, stream, runtime) + except Exception as exc: + logger.warning( + "AVS1 stream %s processing failed: %s", + runtime.config.name, + exc, + exc_info=logger.isEnabledFor(logging.DEBUG), + ) + + def _process_stream( + self, + received: ReceivedRecord, + bundle: Avs1Bundle, + stream: Avs1Stream, + runtime: _StreamRuntime, + ) -> None: + self._validate_stream_descriptor(stream, runtime.config) + slot = runtime.write_index + t0 = time.monotonic() + receiver_process_submit_mono_ns = time.monotonic_ns() + receiver_process_submit_ns, _ = _read_wall_clock_ns() + assert self._sample_pool is not None + futures: list[Future[bool]] = [ + self._sample_pool.submit( + self._process_sample, runtime, stream, sample, slot + ) + for sample in stream.samples + ] + # Keep the slot and decoders owned until both eyes finish, even on failure. + wait(futures) + results = [future.result() for future in futures] + if not all(results): + return + receiver_process_done_mono_ns = time.monotonic_ns() + receiver_process_done_ns, _ = _read_wall_clock_ns() + + timestamp_ns = time.monotonic_ns() + receiver_ipc_publish_mono_ns = timestamp_ns + receiver_ipc_publish_ns, _ = _read_wall_clock_ns() + latencies = self._receiver_latencies( + bundle=bundle, + stream=stream, + received=received, + receiver_process_submit_ns=receiver_process_submit_ns, + receiver_process_done_ns=receiver_process_done_ns, + receiver_ipc_publish_ns=receiver_ipc_publish_ns, + receiver_process_submit_mono_ns=receiver_process_submit_mono_ns, + receiver_process_done_mono_ns=receiver_process_done_mono_ns, + receiver_ipc_publish_mono_ns=receiver_ipc_publish_mono_ns, + ) + + for sample in stream.samples: + generation = runtime.next_generation(sample.camera_index, slot) + metadata = self._frame_metadata( + bundle=bundle, + stream=stream, + sample=sample, + runtime=runtime, + received=received, + receiver_process_submit_ns=receiver_process_submit_ns, + receiver_process_done_ns=receiver_process_done_ns, + receiver_ipc_publish_ns=receiver_ipc_publish_ns, + receiver_process_submit_mono_ns=receiver_process_submit_mono_ns, + receiver_process_done_mono_ns=receiver_process_done_mono_ns, + receiver_ipc_publish_mono_ns=receiver_ipc_publish_mono_ns, + latencies=latencies, + ) + self._ipc_publisher.publish( + runtime.config.ipc_names[sample.camera_index], + slot=slot, + slot_generation=generation, + seq=stream.sequence, + timestamp_ns=timestamp_ns, + metadata=metadata, + ) + + runtime.write_index = (slot + 1) % runtime.config.ring_depth + if self._stats is not None: + self._stats.add( + runtime.config.name, + time.monotonic() - t0, + stream.tsc_delta_ns, + latencies, + ) + + def _process_sample(self, runtime, stream, sample, slot) -> bool: + import cupy as cp + + name = runtime.config.ipc_names[sample.camera_index] + with self._ipc_publisher.write_slot(name, slot, self._stop): + with cp.cuda.Device(self._gpu_device_id): + try: + return self._decode_sample(runtime, stream, sample, slot) + finally: + # CuPy's per-thread default stream belongs to this worker. + # Complete reads/writes even on failure before releasing the slot. + cp.cuda.get_current_stream().synchronize() + + def _decode_sample( + self, + runtime: _StreamRuntime, + stream: Avs1Stream, + sample: Avs1Sample, + slot: int, + ) -> bool: + import cupy as cp + import numpy as np + + cfg = runtime.config + output = runtime.output_buffers[sample.camera_index][slot] + with cp.cuda.Device(self._gpu_device_id): + if stream.codec == CODEC_HEVC: + decoder = runtime.decoders[sample.camera_index] + if decoder is None: + decoder = NvVideoDecoder( + width=stream.width, + height=stream.height, + full_range=cfg.full_range, + gpu_id=self._gpu_device_id, + codec="hevc", + low_latency=True, + ) + runtime.decoders[sample.camera_index] = decoder + if cfg.output_format == _OUTPUT_NV12: + return bool( + decoder.decode_nv12( + sample.payload, output, rotation=stream.rotation + ) + ) + return bool( + decoder.decode(sample.payload, output, rotation=stream.rotation) + ) + + if stream.codec != CODEC_RAW_NV12: + raise ValueError(f"unsupported AVS1 codec {stream.codec}") + + raw_host = np.frombuffer(sample.payload, dtype=np.uint8) + if cfg.output_format == _OUTPUT_NV12: + if stream.rotation == 0: + output.set(raw_host.reshape(output.shape)) + return True + raw = runtime.raw_buffers[sample.camera_index] + raw.set(raw_host) + assert runtime.raw_kernel is not None + runtime.raw_kernel( + runtime.grid, + runtime.block, + (raw, output, stream.width, stream.height), + ) + return True + + raw = runtime.raw_buffers[sample.camera_index] + raw.set(raw_host) + assert runtime.raw_kernel is not None + runtime.raw_kernel( + runtime.grid, + runtime.block, + (raw, output, stream.width, stream.height, stream.rotation), + ) + return True + + def _validate_stream_descriptor( + self, stream: Avs1Stream, cfg: StreamConfig + ) -> None: + if stream.width != cfg.width or stream.height != cfg.height: + raise ValueError( + f"AVS1 stream {stream.index} size {stream.width}x{stream.height} " + f"does not match config {cfg.width}x{cfg.height}" + ) + if stream.stereo != cfg.stereo: + raise ValueError( + f"AVS1 stream {stream.index} stereo={stream.stereo} " + f"does not match config stereo={cfg.stereo}" + ) + if len(stream.samples) != (2 if cfg.stereo else 1): + raise ValueError(f"AVS1 stream {stream.index} sample count mismatch") + + def _receiver_latencies( + self, + bundle: Avs1Bundle, + stream: Avs1Stream, + received: ReceivedRecord, + receiver_process_submit_ns: int, + receiver_process_done_ns: int, + receiver_ipc_publish_ns: int, + receiver_process_submit_mono_ns: int, + receiver_process_done_mono_ns: int, + receiver_ipc_publish_mono_ns: int, + ) -> dict[str, Optional[float]]: + local_rx_to_process_submit_ms = _ns_delta_ms( + receiver_process_submit_mono_ns, received.receiver_rx_payload_done_mono_ns + ) + local_rx_to_process_done_ms = _ns_delta_ms( + receiver_process_done_mono_ns, received.receiver_rx_payload_done_mono_ns + ) + local_rx_to_ipc_publish_ms = _ns_delta_ms( + receiver_ipc_publish_mono_ns, received.receiver_rx_payload_done_mono_ns + ) + latencies: dict[str, Optional[float]] = { + "rx_to_process_submit": local_rx_to_process_submit_ms, + "rx_to_process_done": local_rx_to_process_done_ms, + "rx_to_ipc_publish": local_rx_to_ipc_publish_ms, + } + if bundle.sender_clock_domain == received.receiver_clock_domain: + process_done_ns = max( + ( + int(sample.sender_process_done_ns) + for sample in stream.samples + if sample.sender_process_done_ns is not None + ), + default=None, + ) + capture_ready_ns = _stream_capture_ready_ns(bundle, stream) + sof_ns = _stream_sensor_sof_ns(bundle, stream) + process_submit_ns = min( + ( + int(sample.sender_process_submit_ns) + for sample in stream.samples + if sample.sender_process_submit_ns is not None + ), + default=None, + ) + cross_host_values = { + "sender_capture_ready_to_rx": _ns_delta_ms( + received.receiver_rx_payload_done_ns, capture_ready_ns + ), + "sender_process_submit_to_rx": _ns_delta_ms( + received.receiver_rx_payload_done_ns, process_submit_ns + ), + "sender_process_done_to_rx": _ns_delta_ms( + received.receiver_rx_payload_done_ns, process_done_ns + ), + "sender_tcp_enqueue_to_rx": _ns_delta_ms( + received.receiver_rx_payload_done_ns, bundle.sender_tcp_enqueue_ns + ), + "sender_tcp_enqueue_to_ipc_publish": _ns_delta_ms( + receiver_ipc_publish_ns, bundle.sender_tcp_enqueue_ns + ), + "sender_capture_ready_to_ipc_publish": _ns_delta_ms( + receiver_ipc_publish_ns, capture_ready_ns + ), + "sensor_sof_to_rx": _ns_delta_ms( + received.receiver_rx_payload_done_ns, sof_ns + ), + "sensor_sof_to_ipc_publish": _ns_delta_ms( + receiver_ipc_publish_ns, sof_ns + ), + } + latencies.update(cross_host_values) + return latencies + + def _frame_metadata( + self, + bundle: Avs1Bundle, + stream: Avs1Stream, + sample: Avs1Sample, + runtime: _StreamRuntime, + received: ReceivedRecord, + receiver_process_submit_ns: int, + receiver_process_done_ns: int, + receiver_ipc_publish_ns: int, + receiver_process_submit_mono_ns: int, + receiver_process_done_mono_ns: int, + receiver_ipc_publish_mono_ns: int, + latencies: dict[str, Optional[float]], + ) -> dict[str, object]: + sender_capture_ready_ns = sample.sender_capture_ready_ns + if sender_capture_ready_ns is None: + sender_capture_ready_ns = bundle.sender_capture_ready_ns + sof_lag_ns = sensor_sof_lag_ns(bundle, sample) + sample_sof_ns = sensor_sof_ns(bundle, sample) + metadata: dict[str, object] = { + "source_format": "avs1", + "avs1_version": 1, + "avs1_flags": bundle.flags, + "session_id": bundle.session_id, + "bundle_sequence": bundle.sequence, + "sender_sequence": stream.sequence, + "stream_index": stream.index, + "stream_name": runtime.config.name, + "stream_sequence": stream.sequence, + "stream_stereo": stream.stereo, + "stream_codec": stream.codec_name, + "stream_rotation": stream.rotation, + "stream_fps_numerator": stream.fps_numerator, + "stream_fps_denominator": stream.fps_denominator, + "camera_index": sample.camera_index, + "role": sample.role_name, + "device_index": sample.device_index, + "sensor_mode_index": sample.sensor_mode_index, + "frame_number": sample.frame_number, + "width": stream.width, + "height": stream.height, + "source_pixel_format": "nv12", + "ipc_pixel_format": runtime.config.output_format, + "sender_clock": "tegra_tsc_ns", + "sender_wall_clock": bundle.sender_clock_domain, + "sof_tsc_ns": sample.sof_tsc_ns, + "sof_tsc_valid": sample.sof_tsc_ns is not None, + "frame_timestamp_ns": sample.frame_timestamp_ns, + "sensor_timestamp_ns": sample.frame_timestamp_ns, + "sender_capture_ready_ns": sender_capture_ready_ns, + "sender_capture_ready_tsc_ns": bundle.sender_capture_ready_tsc_ns, + # Sensor start-of-frame in the sender wall domain (``sender_wall_clock``); + # None when the sender TSC anchor is absent or the SOF->ready lag is + # out of range. + "sensor_sof_ns": sample_sof_ns, + # Same value under the key the CUDA-IPC consumer tools and TeleViz + # already read for their "sof_to_consumer_ms" column. + "sof_wall_clock_ns": sample_sof_ns, + "sensor_sof_lag_ms": ( + sof_lag_ns / 1.0e6 if sof_lag_ns is not None else None + ), + "sender_process_submit_ns": sample.sender_process_submit_ns, + "sender_process_done_ns": sample.sender_process_done_ns, + "sender_tcp_enqueue_ns": bundle.sender_tcp_enqueue_ns, + "receiver_clock": received.receiver_clock_domain, + "receiver_rx_prefix_done_ns": received.receiver_rx_prefix_done_ns, + "receiver_rx_payload_done_ns": received.receiver_rx_payload_done_ns, + "receiver_rx_prefix_done_mono_ns": received.receiver_rx_prefix_done_mono_ns, + "receiver_rx_payload_done_mono_ns": received.receiver_rx_payload_done_mono_ns, + "receiver_process_submit_ns": receiver_process_submit_ns, + "receiver_process_done_ns": receiver_process_done_ns, + "receiver_ipc_publish_ns": receiver_ipc_publish_ns, + "receiver_process_submit_mono_ns": receiver_process_submit_mono_ns, + "receiver_process_done_mono_ns": receiver_process_done_mono_ns, + "receiver_ipc_publish_mono_ns": receiver_ipc_publish_mono_ns, + } + if stream.stereo and len(stream.samples) == 2: + metadata["left_sof_tsc_ns"] = stream.samples[0].sof_tsc_ns + metadata["right_sof_tsc_ns"] = stream.samples[1].sof_tsc_ns + metadata["tsc_delta_ns"] = stream.tsc_delta_ns + metadata["left_frame_timestamp_ns"] = stream.samples[0].frame_timestamp_ns + metadata["right_frame_timestamp_ns"] = stream.samples[1].frame_timestamp_ns + metadata["left_sensor_sof_ns"] = sensor_sof_ns(bundle, stream.samples[0]) + metadata["right_sensor_sof_ns"] = sensor_sof_ns(bundle, stream.samples[1]) + for name, value in latencies.items(): + metadata[f"{name}_ms"] = value + return metadata + + def _recv_exact(self, conn: socket.socket, view: memoryview) -> bool: + got = 0 + total = len(view) + last_progress = time.monotonic() + # Tight stall timeout only once the stream is flowing; a fresh, not-yet- + # warm link gets the cold-start grace so the sender's NVENC warmup does + # not look like a dead peer. + stall_timeout = ( + _STALL_TIMEOUT_S + if getattr(self, "_link_warm", False) + else _COLD_START_TIMEOUT_S + ) + while got < total: + if self._stop.is_set(): + return False + try: + n = conn.recv_into(view[got:], total - got) + except socket.timeout: + if self._receive_fps_monitor is not None: + self._receive_fps_monitor.poll() + # Bound the stall: a half-open peer (the AGX rebooted) leaves the + # socket ESTABLISHED and recv just keeps timing out with no data. + # Give up so the produce loop reconnects instead of waiting on a + # dead socket forever; the reconnect loop then re-establishes the + # link as soon as the sender is listening again, in any order. + if time.monotonic() - last_progress > stall_timeout: + logger.warning( + "no data from %s:%d for %.1fs; treating the link as dead " + "and reconnecting", + self._host, + self._port, + stall_timeout, + ) + return False + continue + except OSError: + return False + if n == 0: + return False + got += n + last_progress = time.monotonic() + return True + + +def _stream_capture_ready_ns(bundle: Avs1Bundle, stream: Avs1Stream) -> Optional[int]: + sample_values = [ + int(sample.sender_capture_ready_ns) + for sample in stream.samples + if sample.sender_capture_ready_ns is not None + ] + if sample_values: + return max(sample_values) + return bundle.sender_capture_ready_ns + + +def _stream_sensor_sof_ns(bundle: Avs1Bundle, stream: Avs1Stream) -> Optional[int]: + # Earliest exposure of the stream's samples (stereo pairs differ by ~us). + sample_values = [ + int(sof) + for sof in (sensor_sof_ns(bundle, sample) for sample in stream.samples) + if sof is not None + ] + if sample_values: + return min(sample_values) + return None + + +def _ipc_layout(output_format: str, width: int, height: int) -> dict[str, Any]: + if output_format == _OUTPUT_RGBA8: + return { + "planes": [ + { + "name": "rgba", + "offset": 0, + "width": width, + "height": height, + "pitch": width * 4, + "channels": 4, + } + ] + } + if output_format == _OUTPUT_NV12: + y_bytes = width * height + return { + "planes": [ + { + "name": "y", + "offset": 0, + "width": width, + "height": height, + "pitch": width, + }, + { + "name": "uv", + "offset": y_bytes, + "width": width, + "height": height // 2, + "pitch": width, + }, + ] + } + raise ValueError(f"unsupported output_format {output_format!r}") + + +def _parse_stream_config(raw: dict[str, Any]) -> StreamConfig: + index = int(raw["index"]) + name = str(raw.get("name", f"stream{index}")) + stream_type = str( + raw.get("type", "stereo" if raw.get("stereo", False) else "mono") + ).lower() + stereo = bool(raw.get("stereo", stream_type == "stereo")) + width = int(raw["width"]) + height = int(raw["height"]) + output_format = str(raw.get("output_format", _OUTPUT_NV12)).lower() + if output_format not in (_OUTPUT_NV12, _OUTPUT_RGBA8): + raise ValueError( + f"streams[{index}].output_format must be nv12|rgba8, got {output_format!r}" + ) + ring_depth = int(raw.get("ring_depth", 6)) + if ring_depth < 3: + raise ValueError(f"streams[{index}].ring_depth must be >= 3") + ipc_cfg = raw.get("ipc", {}) + if ipc_cfg is None: + ipc_cfg = {} + if not isinstance(ipc_cfg, dict): + raise ValueError(f"streams[{index}].ipc must be a mapping") + if stereo: + left = str(ipc_cfg.get("left", f"{name}.left")) + right = str(ipc_cfg.get("right", f"{name}.right")) + ipc_names = (left, right) + else: + mono = str(ipc_cfg.get("mono", name)) + ipc_names = (mono,) + return StreamConfig( + index=index, + name=name, + stereo=stereo, + width=width, + height=height, + output_format=output_format, + ring_depth=ring_depth, + ipc_names=ipc_names, + full_range=bool(raw.get("full_range", False)), + ) + + +def _build_receiver(cfg: dict, args: argparse.Namespace) -> Avs1TcpReceiver: + receiver_cfg = cfg.get("receiver", {}) + ipc_cfg = cfg.get("ipc", {}) + streams_raw = cfg.get("streams") + if not isinstance(streams_raw, list) or not streams_raw: + raise ValueError("config must include a non-empty streams list") + streams = [_parse_stream_config(raw) for raw in streams_raw] + ipc_names = [name for stream in streams for name in stream.ipc_names] + if len(ipc_names) != len(set(ipc_names)): + raise ValueError("CUDA IPC stream names must be unique") + + env_host = os.environ.get("CAMERA_RECEIVER_HOST") or None + env_port = os.environ.get("CAMERA_RECEIVER_PORT") or None + host = str(args.host or env_host or receiver_cfg.get("host", "127.0.0.1")) + port = int( + args.port + if args.port is not None + else (env_port or receiver_cfg.get("port", 5000)) + ) + gpu_id = _parse_optional_gpu_id( + args.gpu_id if args.gpu_id is not None else receiver_cfg.get("gpu_id"), + "receiver.gpu_id", + ) + socket_path = str( + args.socket + or ipc_cfg.get("socket_path", "/tmp/camera_receiver_avs1_cuda_ipc.sock") + ) + publisher = CudaIpcFramePublisher( + socket_path=socket_path, + owner_name=str(ipc_cfg.get("owner_name", "camera_receiver_avs1")), + queue_size=int(ipc_cfg.get("queue_size", 1024)), + ) + return Avs1TcpReceiver( + host=host, + port=port, + gpu_id=gpu_id, + max_record_bytes=int( + receiver_cfg.get("max_record_bytes", _DEFAULT_MAX_RECORD_BYTES) + ), + stats_log_interval_s=_resolve_stats_log_interval_s( + receiver_cfg, + args.stats_log_interval_s, + ), + receive_fps_warning=_resolve_receive_fps_warning_config(receiver_cfg), + ipc_publisher=publisher, + streams=streams, + ) + + +def _dry_run_summary( + cfg: dict, + args: argparse.Namespace, +) -> tuple[ + str, + int, + str, + list[StreamConfig], + float, + ReceiveFpsWarningConfig, +]: + receiver_cfg = cfg.get("receiver", {}) + ipc_cfg = cfg.get("ipc", {}) + streams_raw = cfg.get("streams") + if not isinstance(streams_raw, list) or not streams_raw: + raise ValueError("config must include a non-empty streams list") + streams = [_parse_stream_config(raw) for raw in streams_raw] + ipc_names = [name for stream in streams for name in stream.ipc_names] + if len(ipc_names) != len(set(ipc_names)): + raise ValueError("CUDA IPC stream names must be unique") + env_host = os.environ.get("CAMERA_RECEIVER_HOST") or None + env_port = os.environ.get("CAMERA_RECEIVER_PORT") or None + host = str(args.host or env_host or receiver_cfg.get("host", "127.0.0.1")) + port = int( + args.port + if args.port is not None + else (env_port or receiver_cfg.get("port", 5000)) + ) + socket_path = str( + args.socket + or ipc_cfg.get("socket_path", "/tmp/camera_receiver_avs1_cuda_ipc.sock") + ) + stats_log_interval_s = _resolve_stats_log_interval_s( + receiver_cfg, + args.stats_log_interval_s, + ) + receive_fps_warning = _resolve_receive_fps_warning_config(receiver_cfg) + return ( + host, + port, + socket_path, + streams, + stats_log_interval_s, + receive_fps_warning, + ) + + +def _load_config(path: Path) -> dict: + with path.open("r", encoding="utf-8") as f: + cfg = yaml.safe_load(f) + if not isinstance(cfg, dict): + raise ValueError(f"config must be a YAML mapping: {path}") + if "network" in cfg: + # AVS1 stream indices follow sender list order; geometry and port are shared. + receiver = dict(cfg.get("receiver", {})) + if "port" in receiver: + raise ValueError("shared config uses network.port, not receiver.port") + receiver["port"] = cfg["network"].get("port", 5000) + streams = [] + for index, stream in enumerate(cfg.get("streams", [])): + settings = dict(stream.get("receiver", {})) + if set(settings) & {"index", "name", "type", "width", "height"}: + raise ValueError( + "shared stream identity and geometry belong in id/type/output" + ) + output = stream["output"] + if int(output.get("width", 0)) <= 0 or int(output.get("height", 0)) <= 0: + raise ValueError( + "shared config requires explicit positive output width and height" + ) + streams.append( + dict( + settings, + index=index, + name=stream["id"], + type=stream["type"], + width=output["width"], + height=output["height"], + ) + ) + cfg = dict(cfg, receiver=receiver, streams=streams) + return cfg + + +def main(argv: Optional[list[str]] = None) -> int: + default_config = ( + Path(__file__).resolve().parents[1] / "configs" / "s56-stereo-2560x1984-30.yaml" + ) + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--config", default=str(default_config)) + parser.add_argument("--host", help="override receiver.host or CAMERA_RECEIVER_HOST") + parser.add_argument( + "--port", type=int, help="override receiver.port or CAMERA_RECEIVER_PORT" + ) + parser.add_argument("--socket", help="override ipc.socket_path") + parser.add_argument("--gpu-id", type=int, help="override receiver.gpu_id") + parser.add_argument( + "--stats-log-interval-s", + type=float, + help="seconds between periodic stats summaries; 0 disables them", + ) + parser.add_argument( + "--dry-run", action="store_true", help="parse and instantiate only" + ) + parser.add_argument("--log-level", default="INFO") + args = parser.parse_args(argv) + + logging.basicConfig( + level=getattr(logging, str(args.log_level).upper(), logging.INFO), + format="%(asctime)s %(levelname)s %(name)s: %(message)s", + ) + + cfg = _load_config(Path(args.config)) + if args.dry_run: + ( + host, + port, + socket_path, + streams, + stats_log_interval_s, + receive_fps_warning, + ) = _dry_run_summary(cfg, args) + stats_summary = ( + "disabled" + if stats_log_interval_s == 0 + else f"every {stats_log_interval_s:g}s" + ) + fps_warning_summary = ( + "disabled" + if not receive_fps_warning.enabled + else ( + f"below {receive_fps_warning.threshold_fps:g}fps " + f"every {receive_fps_warning.interval_s:g}s" + ) + ) + print( + "camera_receiver: config ok; " + f"sender={host}:{port}, socket={socket_path}, " + f"streams={','.join(stream.name for stream in streams)}, " + f"periodic_stats={stats_summary}, " + f"receive_fps_warning={fps_warning_summary}", + flush=True, + ) + return 0 + + receiver = _build_receiver(cfg, args) + + stop_event = threading.Event() + + def _on_signal(signum, _frame): + logger.info("stopping on signal %s", signum) + stop_event.set() + + signal.signal(signal.SIGINT, _on_signal) + signal.signal(signal.SIGTERM, _on_signal) + + receiver.start() + try: + logger.info( + "camera_receiver running: sender=%s:%d, socket=%s", + receiver.host, + receiver.port, + receiver.socket_path, + ) + receiver.wait(stop_event) + finally: + receiver.stop() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/examples/camera_streamer/camera_receiver/cuda_ipc.py b/examples/camera_streamer/camera_receiver/cuda_ipc.py new file mode 100644 index 0000000000..d75f0e9ed3 --- /dev/null +++ b/examples/camera_streamer/camera_receiver/cuda_ipc.py @@ -0,0 +1,472 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +"""CUDA IPC frame publisher for camera_receiver. + +Protocol v3 uses a newline-delimited JSON control socket: + +1. Publisher -> client: ``hello`` stream catalog, no CUDA handles yet. +2. Client -> publisher: ``subscribe`` with stream names, or omitted/empty for all. +3. Publisher -> client: ``subscribed`` with CUDA IPC handles for selected streams. +4. Publisher -> client: ``frame`` notifications for selected streams only. +5. Client locks the slot's guard file, checks its generation, and completes GPU + reads before unlocking. Stale notifications are discarded. + +Image bytes stay in same-host CUDA IPC memory. The socket only carries control +messages and per-frame metadata. +""" + +from __future__ import annotations + +import base64 +import json +import logging +import os +import queue +import signal +import socket +import threading +import tempfile +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Optional + +from slot_guard import commit_slot, read_slot, write_slot + +logger = logging.getLogger("isaaccapture.camera_streamer." + __name__) + +_CLIENT_SEND_TIMEOUT_S = 2.0 +_CLIENT_SUBSCRIBE_TIMEOUT_S = 5.0 +_MAX_CONTROL_BYTES = 1024 * 1024 +_SEND_FLAGS = getattr(socket, "MSG_NOSIGNAL", 0) + + +@dataclass(frozen=True) +class _SlotInfo: + slot: int + handle_b64: str + nbytes: int + shape: tuple[int, ...] + strides: tuple[int, ...] + guard_path: str + + +@dataclass(frozen=True) +class _StreamInfo: + name: str + width: int + height: int + pixel_format: str + dtype: str + gpu_id: int + layout: dict[str, Any] + slots: tuple[_SlotInfo, ...] + + +@dataclass +class _ClientState: + sock: socket.socket + streams: frozenset[str] + + +class CudaIpcFramePublisher: + """Publish decoded GPU frames to same-host CUDA IPC consumers.""" + + def __init__( + self, + socket_path: str = "/tmp/camera_receiver_cuda_ipc.sock", + owner_name: str = "camera_receiver", + queue_size: int = 512, + ) -> None: + self._socket_path = str(socket_path) + self._owner_name = owner_name + self._queue: "queue.Queue[dict[str, Any]]" = queue.Queue(maxsize=queue_size) + self._streams: dict[str, _StreamInfo] = {} + self._clients: list[_ClientState] = [] + self._lock = threading.Lock() + self._stop = threading.Event() + self._server: Optional[socket.socket] = None + self._accept_thread: Optional[threading.Thread] = None + self._broadcast_thread: Optional[threading.Thread] = None + self._started = False + self._guard_dir = tempfile.TemporaryDirectory(prefix="camera-ipc-slots-") + + @property + def socket_path(self) -> str: + return self._socket_path + + def register_stream( + self, + name: str, + width: int, + height: int, + pixel_format: str, + gpu_id: int, + buffers: list[Any], + layout: Optional[dict[str, Any]] = None, + ) -> None: + """Register a ring of CuPy-like GPU buffers and export IPC handles.""" + try: + import cupy as cp + except ImportError as e: + raise RuntimeError("CudaIpcFramePublisher requires CuPy") from e + + slots: list[_SlotInfo] = [] + with cp.cuda.Device(gpu_id): + for idx, buf in enumerate(buffers): + iface = getattr(buf, "__cuda_array_interface__", None) + if iface is None: + raise TypeError( + f"CUDA IPC stream {name!r} slot {idx} does not expose " + "__cuda_array_interface__" + ) + ptr = int(iface["data"][0]) + handle = cp.cuda.runtime.ipcGetMemHandle(ptr) + strides = iface.get("strides") or getattr(buf, "strides", None) + if strides is None: + raise TypeError( + f"CUDA IPC stream {name!r} slot {idx} has no strides" + ) + guard_path = str( + Path(self._guard_dir.name) / f"{len(self._streams)}-{idx}" + ) + Path(guard_path).write_bytes(bytes(8)) + slots.append( + _SlotInfo( + slot=idx, + guard_path=guard_path, + handle_b64=base64.b64encode(handle).decode("ascii"), + nbytes=int(buf.nbytes), + shape=tuple(int(v) for v in iface["shape"]), + strides=tuple(int(v) for v in strides), + ) + ) + + with self._lock: + if name in self._streams: + raise ValueError(f"CUDA IPC stream {name!r} registered twice") + self._streams[name] = _StreamInfo( + name=name, + width=int(width), + height=int(height), + pixel_format=str(pixel_format), + dtype="uint8", + gpu_id=int(gpu_id), + layout=dict(layout or {}), + slots=tuple(slots), + ) + logger.info( + "CUDA IPC: registered stream '%s' %s with %d slots on GPU %d", + name, + pixel_format, + len(slots), + gpu_id, + ) + + def start(self) -> None: + with self._lock: + if self._started: + return + self._ignore_sigpipe() + self._stop.clear() + try: + self._open_server() + except Exception: + with self._lock: + self._started = False + raise + with self._lock: + self._started = True + self._accept_thread = threading.Thread( + target=self._accept_loop, + name="camera_receiver_cuda_ipc_accept", + daemon=False, + ) + self._broadcast_thread = threading.Thread( + target=self._broadcast_loop, + name="camera_receiver_cuda_ipc_broadcast", + daemon=False, + ) + self._accept_thread.start() + self._broadcast_thread.start() + logger.info("CUDA IPC: listening on %s", self._socket_path) + + def stop(self) -> None: + self._stop.set() + if self._server is not None: + try: + self._server.close() + except OSError: + logger.debug("CUDA IPC: socket close failed", exc_info=True) + self._server = None + with self._lock: + clients = list(self._clients) + self._clients.clear() + self._started = False + for client in clients: + try: + client.sock.close() + except OSError: + logger.debug("CUDA IPC: socket close failed", exc_info=True) + for thread in (self._accept_thread, self._broadcast_thread): + if thread is not None: + thread.join(timeout=2.0) + self._accept_thread = None + self._broadcast_thread = None + try: + if os.path.exists(self._socket_path): + os.unlink(self._socket_path) + except OSError: + logger.debug( + "CUDA IPC: failed to unlink %s", self._socket_path, exc_info=True + ) + + def write_slot(self, stream_name: str, slot: int, stop: threading.Event): + return write_slot(self._streams[stream_name].slots[slot].guard_path, stop) + + def _current_frame(self, msg): + return read_slot( + self._streams[msg["stream"]].slots[msg["slot"]].guard_path, + msg["slot_generation"], + ) + + def publish( + self, + stream_name: str, + slot: int, + slot_generation: int, + seq: int, + timestamp_ns: int, + metadata: Optional[dict[str, Any]] = None, + ) -> None: + if not self._started: + return + commit_slot(self._streams[stream_name].slots[slot].guard_path, slot_generation) + msg = { + "type": "frame", + "stream": stream_name, + "slot": int(slot), + "slot_generation": int(slot_generation), + "seq": int(seq), + "timestamp_ns": int(timestamp_ns), + } + if metadata is not None: + msg["metadata"] = metadata + try: + self._queue.put_nowait(msg) + except queue.Full: + try: + self._queue.get_nowait() + except queue.Empty: + # The broadcaster already drained the notification queue. + pass + try: + self._queue.put_nowait(msg) + except queue.Full: + logger.debug( + "CUDA IPC: dropping frame notification for %s", stream_name + ) + + def _open_server(self) -> None: + path = Path(self._socket_path) + path.parent.mkdir(parents=True, exist_ok=True) + if path.exists(): + probe = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + try: + probe.settimeout(0.1) + probe.connect(str(path)) + raise RuntimeError( + f"CUDA IPC socket {self._socket_path} is already in use" + ) + except OSError: + path.unlink() + finally: + probe.close() + + srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + srv.bind(self._socket_path) + os.chmod(self._socket_path, 0o600) + srv.listen(8) + srv.settimeout(0.5) + self._server = srv + + def _stream_catalog(self, stream: _StreamInfo) -> dict[str, Any]: + return { + "name": stream.name, + "width": stream.width, + "height": stream.height, + "pixel_format": stream.pixel_format, + "dtype": stream.dtype, + "gpu_id": stream.gpu_id, + "slot_count": len(stream.slots), + "layout": stream.layout, + } + + def _stream_with_slots(self, stream: _StreamInfo) -> dict[str, Any]: + msg = self._stream_catalog(stream) + msg["slots"] = [ + { + "slot": slot.slot, + "guard_path": slot.guard_path, + "handle": slot.handle_b64, + "nbytes": slot.nbytes, + "shape": list(slot.shape), + "strides": list(slot.strides), + } + for slot in stream.slots + ] + return msg + + def _hello_message(self) -> dict[str, Any]: + with self._lock: + streams = list(self._streams.values()) + return { + "type": "hello", + "version": 3, + "features": [ + "subscriptions", + "slot_read_locks", + "frame_metadata", + "avs1_metadata", + "latency_metadata", + "slot_generation", + "nv12", + ], + "owner": self._owner_name, + "socket_path": self._socket_path, + "streams": [self._stream_catalog(s) for s in streams], + } + + def _subscribed_message( + self, requested: list[str] + ) -> tuple[dict[str, Any], frozenset[str]]: + with self._lock: + all_streams = dict(self._streams) + selected_names = requested or list(all_streams) + unknown = [name for name in selected_names if name not in all_streams] + if unknown: + raise ValueError(f"unknown CUDA IPC stream(s): {', '.join(unknown)}") + selected = frozenset(selected_names) + return ( + { + "type": "subscribed", + "version": 3, + "streams": [ + self._stream_with_slots(all_streams[name]) + for name in selected_names + ], + }, + selected, + ) + + def _accept_loop(self) -> None: + assert self._server is not None + while not self._stop.is_set(): + try: + client, _ = self._server.accept() + except socket.timeout: + continue + except OSError: + break + try: + client.settimeout(_CLIENT_SEND_TIMEOUT_S) + self._send_json(client, self._hello_message()) + client.settimeout(_CLIENT_SUBSCRIBE_TIMEOUT_S) + subscribe = self._recv_json(client) + if subscribe.get("type") != "subscribe": + raise ValueError(f"expected subscribe message, got {subscribe!r}") + if subscribe.get("version") != 3: + raise ValueError( + "CUDA IPC requires protocol 3 slot read locks; update the consumer" + ) + requested_raw = subscribe.get("streams") or [] + if not isinstance(requested_raw, list): + raise ValueError("subscribe.streams must be a list") + requested = [str(name) for name in requested_raw] + msg, selected = self._subscribed_message(requested) + client.settimeout(_CLIENT_SEND_TIMEOUT_S) + self._send_json(client, msg) + except Exception as exc: + logger.info("CUDA IPC: rejecting consumer (%s)", exc) + try: + self._send_json(client, {"type": "error", "error": str(exc)}) + except OSError: + logger.debug( + "CUDA IPC: cannot notify rejected client", exc_info=True + ) + client.close() + continue + with self._lock: + self._clients.append(_ClientState(client, selected)) + logger.info( + "CUDA IPC: consumer connected (%s)", + ", ".join(sorted(selected)) if selected else "no streams", + ) + + def _broadcast_loop(self) -> None: + while not self._stop.is_set(): + try: + msg = self._queue.get(timeout=0.5) + except queue.Empty: + continue + with self._current_frame(msg) as current: + if not current: + continue + stream_name = str(msg.get("stream", "")) + with self._lock: + clients = list(self._clients) + if not clients: + continue + dead: list[_ClientState] = [] + for client in clients: + if stream_name not in client.streams: + continue + try: + self._send_json(client.sock, msg) + except socket.timeout: + logger.warning( + "CUDA IPC: dropping slow consumer after send timeout" + ) + dead.append(client) + except OSError as e: + logger.info("CUDA IPC: consumer disconnected (%s)", e) + dead.append(client) + if dead: + with self._lock: + self._clients = [c for c in self._clients if c not in dead] + for client in dead: + try: + client.sock.close() + except OSError: + logger.debug("CUDA IPC: client close failed", exc_info=True) + + @staticmethod + def _send_json(sock: socket.socket, msg: dict[str, Any]) -> None: + payload = json.dumps(msg, separators=(",", ":")).encode("utf-8") + b"\n" + sock.sendall(payload, _SEND_FLAGS) + + @staticmethod + def _recv_json(sock: socket.socket) -> dict[str, Any]: + data = bytearray() + while b"\n" not in data: + chunk = sock.recv(4096) + if not chunk: + raise EOFError("consumer disconnected before subscribe") + data.extend(chunk) + if len(data) > _MAX_CONTROL_BYTES: + raise ValueError("control message too large") + line = bytes(data[: data.index(ord("\n"))]) + msg = json.loads(line.decode("utf-8")) + if not isinstance(msg, dict): + raise ValueError("control message must be a JSON object") + return msg + + @staticmethod + def _ignore_sigpipe() -> None: + sigpipe = getattr(signal, "SIGPIPE", None) + if sigpipe is None: + return + try: + signal.signal(sigpipe, signal.SIG_IGN) + except (ValueError, OSError): + # Worker threads cannot change signal handlers; sends use MSG_NOSIGNAL. + pass diff --git a/examples/camera_streamer/camera_receiver/requirements.txt b/examples/camera_streamer/camera_receiver/requirements.txt new file mode 100644 index 0000000000..09ae935803 --- /dev/null +++ b/examples/camera_streamer/camera_receiver/requirements.txt @@ -0,0 +1,8 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +PyYAML>=6.0 +numpy>=1.26,<2 +cupy-cuda12x>=13.0,<14 +pybind11>=2.12 +ninja>=1.11 diff --git a/examples/camera_streamer/camera_receiver/slot_guard.py b/examples/camera_streamer/camera_receiver/slot_guard.py new file mode 100644 index 0000000000..97ff93295c --- /dev/null +++ b/examples/camera_streamer/camera_receiver/slot_guard.py @@ -0,0 +1,52 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Cross-process slot ownership; locks cover GPU work, not just submission.""" + +import fcntl +import os +import struct +from contextlib import contextmanager + +_GENERATION = struct.Struct("!Q") + + +@contextmanager +def write_slot(path, stop): + """Invalidate old notifications before writing; wait interruptibly for readers.""" + with open(path, "r+b", buffering=0) as guard: + while True: + if stop.is_set(): + raise InterruptedError("CUDA IPC slot write cancelled") + try: + fcntl.flock(guard, fcntl.LOCK_EX | fcntl.LOCK_NB) + break + except BlockingIOError: + stop.wait(0.01) + guard.write(_GENERATION.pack(0)) + yield + + +def commit_slot(path, generation): + if generation <= 0: + raise ValueError("slot generation must be positive") + with open(path, "r+b", buffering=0) as guard: + fcntl.flock(guard, fcntl.LOCK_EX) + guard.write(_GENERATION.pack(generation)) + + +@contextmanager +def read_slot(path, generation): + """Yield False for stale/busy slots; hold a shared lock through a valid read.""" + try: + guard = open(path, "rb", buffering=0) + except FileNotFoundError: + yield False + return + with guard: + try: + fcntl.flock(guard, fcntl.LOCK_SH | fcntl.LOCK_NB) + except BlockingIOError: + yield False + return + current = os.pread(guard.fileno(), _GENERATION.size, 0) + yield generation > 0 and current == _GENERATION.pack(generation) diff --git a/examples/camera_streamer/tests/python/test_backpressure.py b/examples/camera_streamer/tests/python/test_backpressure.py new file mode 100644 index 0000000000..01cd8ed45d --- /dev/null +++ b/examples/camera_streamer/tests/python/test_backpressure.py @@ -0,0 +1,131 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +import threading +from concurrent.futures import ThreadPoolExecutor + +import pytest + +from camera_receiver import _StreamRuntime +from cuda_ipc import CudaIpcFramePublisher + + +@pytest.fixture +def runtime(): + runtime = _StreamRuntime.__new__(_StreamRuntime) + runtime.inflight = threading.BoundedSemaphore(1) + runtime.executor = ThreadPoolExecutor(max_workers=1) + yield runtime + runtime.executor.shutdown(wait=True) + + +def test_full_queue_blocks_and_preserves_order(runtime): + stop = threading.Event() + release = threading.Event() + submitted = threading.Event() + order = [] + + def first(): + release.wait(5) + order.append(1) + + assert runtime.submit(stop, first) + + def submit_second(): + runtime.submit(stop, lambda: order.append(2)) + submitted.set() + + thread = threading.Thread(target=submit_second) + thread.start() + try: + assert not submitted.wait(0.1) + finally: + release.set() + thread.join(2) + assert submitted.is_set() + runtime.executor.shutdown(wait=True) + assert order == [1, 2] + + +def test_stop_interrupts_full_queue(runtime): + stop = threading.Event() + runtime.inflight.acquire() + result = [] + thread = threading.Thread( + target=lambda: result.append(runtime.submit(stop, lambda: None)) + ) + thread.start() + stop.set() + thread.join(2) + runtime.inflight.release() + assert not thread.is_alive() + assert result == [False] + + +def test_submit_failure_returns_capacity(runtime): + runtime.executor.shutdown(wait=True) + with pytest.raises(RuntimeError): + runtime.submit(threading.Event(), lambda: None) + assert runtime.inflight.acquire(blocking=False) + + +def test_worker_failure_returns_capacity(runtime): + def fail(): + raise ValueError("decode failed") + + assert runtime.submit(threading.Event(), fail) + assert runtime.inflight.acquire(timeout=2) + + +def test_ipc_socket_is_owner_only(tmp_path): + path = tmp_path / "frames.sock" + publisher = CudaIpcFramePublisher(str(path)) + try: + publisher._open_server() + assert path.stat().st_mode & 0o777 == 0o600 + finally: + publisher.stop() + + +@pytest.mark.parametrize("raises", [False, True]) +def test_stream_waits_for_other_eye_after_failure(monkeypatch, raises): + import sys + from concurrent.futures import Future, wait + from types import SimpleNamespace + + import camera_receiver as module + + first, second = Future(), Future() + if raises: + first.set_exception(RuntimeError("left decode failed")) + else: + first.set_result(False) + futures = iter([first, second]) + receiver = module.Avs1TcpReceiver.__new__(module.Avs1TcpReceiver) + receiver._sample_pool = SimpleNamespace(submit=lambda *args: next(futures)) + receiver._validate_stream_descriptor = lambda *args: None + runtime = SimpleNamespace(write_index=0, config=SimpleNamespace(name="head")) + waiting, finished = threading.Event(), threading.Event() + + def observed_wait(tasks): + waiting.set() + return wait(tasks) + + def process(): + try: + receiver._process_stream_safe( + None, None, SimpleNamespace(samples=[object(), object()]), runtime + ) + finally: + finished.set() + + monkeypatch.setitem(sys.modules, "cupy", SimpleNamespace()) + monkeypatch.setattr(module, "wait", observed_wait) + thread = threading.Thread(target=process) + thread.start() + try: + assert waiting.wait(2) + assert not finished.is_set() + finally: + second.set_result(False) + thread.join(2) + assert finished.is_set() diff --git a/examples/camera_streamer/tests/python/test_reconnect.py b/examples/camera_streamer/tests/python/test_reconnect.py new file mode 100755 index 0000000000..2be07e8ff4 --- /dev/null +++ b/examples/camera_streamer/tests/python/test_reconnect.py @@ -0,0 +1,143 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Regression tests for the receiver's half-open-link detection. + +If the AGX reboots mid-stream the PC socket stays ESTABLISHED but never delivers +another byte. `_recv_exact` must give up after `_STALL_TIMEOUT_S` so the produce +loop reconnects, instead of retrying `recv` on the dead socket forever (which +left the camera black until someone killed the receiver by hand).""" + +from __future__ import annotations + +import socket +import sys +import threading +import unittest +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "camera_receiver")) + +import camera_receiver # noqa: E402 +from camera_receiver import Avs1TcpReceiver # noqa: E402 + + +def _bare_receiver() -> Avs1TcpReceiver: + """An instance with only the attributes `_recv_exact` touches, skipping the + real __init__ (which needs a config, decoders and a GPU).""" + recv = Avs1TcpReceiver.__new__(Avs1TcpReceiver) + recv._stop = threading.Event() + recv._receive_fps_monitor = None + recv._host = "6.6.7.100" + recv._port = 5000 + recv._link_warm = True # warmed stream by default; cold-start tests flip this + return recv + + +class _AlwaysTimeoutSocket: + def recv_into(self, *_args): + raise socket.timeout() + + +class _DataThenBlockSocket: + """Delivers `chunks` (bytes) on successive recv_into calls, then times out.""" + + def __init__(self, chunks): + self._chunks = list(chunks) + + def recv_into(self, view, _nbytes=None): + if not self._chunks: + raise socket.timeout() + chunk = self._chunks.pop(0) + view[: len(chunk)] = chunk + return len(chunk) + + +class RecvExactHalfOpenTest(unittest.TestCase): + def test_gives_up_after_stall_timeout(self): + recv = _bare_receiver() + # monotonic starts at 0 (last_progress), then jumps past the deadline on + # the first in-loop check so the stall is detected without a real wait. + with patch.object( + camera_receiver.time, + "monotonic", + side_effect=[0.0, camera_receiver._STALL_TIMEOUT_S + 1.0], + ): + ok = recv._recv_exact(_AlwaysTimeoutSocket(), memoryview(bytearray(16))) + self.assertFalse( + ok, + "a half-open (never-delivering) socket must return False to force a reconnect", + ) + + def test_returns_true_when_data_arrives(self): + recv = _bare_receiver() + sock = _DataThenBlockSocket([b"abcd", b"efgh"]) + ok = recv._recv_exact(sock, memoryview(bytearray(8))) + self.assertTrue(ok) + + def test_tolerates_a_brief_gap_then_data(self): + recv = _bare_receiver() + + # one timeout (within the deadline), then the record's bytes arrive. + class _GapThenData: + def __init__(self): + self._sent = False + + def recv_into(self, view, _n=None): + if not self._sent: + self._sent = True + raise socket.timeout() + view[:4] = b"wxyz" + return 4 + + with patch.object( + camera_receiver.time, "monotonic", side_effect=[0.0, 1.0, 1.0] + ): + ok = recv._recv_exact(_GapThenData(), memoryview(bytearray(4))) + self.assertTrue(ok, "a brief sub-deadline gap must not trigger a reconnect") + + +class RecvExactColdStartTest(unittest.TestCase): + def test_cold_start_does_not_give_up_before_first_bundle(self): + # Fresh, not-yet-warm link: the sender's NVENC warmup can take a few + # seconds before the first byte. The 3s stall must NOT fire yet. + recv = _bare_receiver() + recv._link_warm = False # no bundle received yet + # times out a few times (~4s elapsed, past _STALL_TIMEOUT_S=3 but under + # _COLD_START_TIMEOUT_S=15), then the first bytes arrive. + _DataThenBlockSocket([]) # will raise timeout first + + class _WarmupSock: + def __init__(self): + self.calls = 0 + + def recv_into(self, view, _n=None): + self.calls += 1 + if self.calls < 3: + raise socket.timeout() + view[:4] = b"cold" + return 4 + + with patch.object( + camera_receiver.time, "monotonic", side_effect=[0.0, 2.0, 4.0, 4.0] + ): + ok = recv._recv_exact(_WarmupSock(), memoryview(bytearray(4))) + self.assertTrue( + ok, "cold start must wait past _STALL_TIMEOUT_S for the first bundle" + ) + + def test_warm_link_still_gives_up_at_stall_timeout(self): + recv = _bare_receiver() # warm by default + with patch.object( + camera_receiver.time, + "monotonic", + side_effect=[0.0, camera_receiver._STALL_TIMEOUT_S + 1.0], + ): + ok = recv._recv_exact(_AlwaysTimeoutSocket(), memoryview(bytearray(16))) + self.assertFalse(ok) + + +if __name__ == "__main__": + unittest.main() diff --git a/examples/camera_streamer/tests/python/test_shared_config.py b/examples/camera_streamer/tests/python/test_shared_config.py new file mode 100644 index 0000000000..b527a4308f --- /dev/null +++ b/examples/camera_streamer/tests/python/test_shared_config.py @@ -0,0 +1,88 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +from pathlib import Path + +import pytest +import yaml + +from camera_receiver import _load_config, _parse_stream_config + + +PROFILES = Path(__file__).resolve().parents[2] / "configs" + + +@pytest.mark.parametrize( + "path", sorted(PROFILES.glob("s56-*.yaml")), ids=lambda p: p.stem +) +def test_shared_profiles(path): + source = yaml.safe_load(path.read_text()) + receiver = _load_config(path) + assert receiver["receiver"]["port"] == source["network"]["port"] + for index, (sent, received) in enumerate( + zip(source["streams"], receiver["streams"], strict=True) + ): + stream = _parse_stream_config(received) + assert (stream.index, stream.name) == (index, sent["id"]) + assert (stream.width, stream.height) == ( + sent["output"]["width"], + sent["output"]["height"], + ) + assert stream.stereo == (sent["type"] == "stereo") + assert stream.ipc_names == ( + ("head.left", "head.right") if stream.stereo else ("head",) + ) + assert len(sent["cameras"]) == (2 if stream.stereo else 1) + + +def test_multiple_streams_share_order_geometry_and_port(tmp_path): + cfg = yaml.safe_load((PROFILES / "s56-stereo-1080p60.yaml").read_text()) + cfg["network"]["port"] = 54321 + cfg["streams"].append( + dict( + id="aux", + type="mono", + cameras=[dict(device=2)], + output=dict(width=640, height=480, fps=30, codec="hevc"), + ) + ) + path = tmp_path / "shared.yaml" + path.write_text(yaml.safe_dump(cfg)) + received = _load_config(path) + assert received["receiver"]["port"] == 54321 + aux = _parse_stream_config(received["streams"][1]) + assert (aux.index, aux.width, aux.height, aux.ipc_names) == (1, 640, 480, ("aux",)) + + +@pytest.mark.parametrize("conflict", ["port", "width", "missing_dimensions"]) +def test_shared_config_rejects_ambiguous_geometry_or_port(tmp_path, conflict): + cfg = yaml.safe_load((PROFILES / "s56-stereo-1080p60.yaml").read_text()) + if conflict == "port": + cfg["receiver"]["port"] = 6000 + elif conflict == "width": + cfg["streams"][0]["receiver"]["width"] = 640 + else: + del cfg["streams"][0]["output"]["width"] + path = tmp_path / "invalid.yaml" + path.write_text(yaml.safe_dump(cfg)) + with pytest.raises(ValueError): + _load_config(path) + + +def test_legacy_receiver_profile(tmp_path): + cfg = dict( + receiver=dict(port=6000), + streams=[dict(index=0, name="old", type="mono", width=640, height=480)], + ) + path = tmp_path / "legacy.yaml" + path.write_text(yaml.safe_dump(cfg)) + assert _load_config(path) == cfg + + +def test_default_profile_is_independent_of_working_directory( + tmp_path, monkeypatch, capsys +): + from camera_receiver import main + + monkeypatch.chdir(tmp_path) + assert main(["--dry-run", "--host", "127.0.0.1"]) == 0 + assert "config ok" in capsys.readouterr().out diff --git a/examples/camera_streamer/tests/python/test_slot_guard.py b/examples/camera_streamer/tests/python/test_slot_guard.py new file mode 100644 index 0000000000..1fcfb17b82 --- /dev/null +++ b/examples/camera_streamer/tests/python/test_slot_guard.py @@ -0,0 +1,94 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +import threading +from types import SimpleNamespace + +from cuda_ipc import CudaIpcFramePublisher +from slot_guard import commit_slot, read_slot, write_slot + + +def test_delayed_broadcast_discards_frames_after_multiple_ring_wraps(tmp_path): + publisher = CudaIpcFramePublisher(str(tmp_path / "ipc.sock")) + paths = [tmp_path / str(i) for i in range(3)] + for path in paths: + path.write_bytes(bytes(8)) + publisher._streams["eye"] = SimpleNamespace( + slots=[SimpleNamespace(guard_path=str(p)) for p in paths] + ) + publisher._started = True + pixels = [None] * 3 + for seq in range(1, 8): + slot = (seq - 1) % 3 + with publisher.write_slot("eye", slot, publisher._stop): + pixels[slot] = seq + publisher.publish("eye", slot, (seq - 1) // 3 + 1, seq, seq) + sent = [] + publisher._clients = [SimpleNamespace(streams={"eye"}, sock=None)] + + def send(sock, msg): + with publisher._current_frame(msg) as valid: + assert valid + assert pixels[msg["slot"]] == msg["seq"] + sent.append(msg["seq"]) + if msg["seq"] == 7: + publisher._stop.set() + + publisher._send_json = send + publisher._broadcast_loop() + assert sent == [5, 6, 7] + # A message already in the socket must also be checked at consumption time. + with read_slot(paths[0], 1) as valid: + assert not valid + publisher._guard_dir.cleanup() + + +def test_reader_holds_pixels_until_read_finishes(tmp_path): + path = tmp_path / "slot" + path.write_bytes(bytes(8)) + commit_slot(path, 1) + waiting, finished = threading.Event(), threading.Event() + stop = threading.Event() + original_wait = stop.wait + + def observed_wait(timeout): + waiting.set() + return original_wait(timeout) + + stop.wait = observed_wait + + def overwrite(): + with write_slot(path, stop): + finished.set() + + with read_slot(path, 1) as valid: + assert valid + worker = threading.Thread(target=overwrite) + worker.start() + assert waiting.wait(2) + assert not finished.is_set() + worker.join(2) + assert finished.is_set() + with read_slot(path, 1) as valid: + assert not valid + + +def test_stop_interrupts_writer_waiting_on_reader(tmp_path): + path = tmp_path / "slot" + path.write_bytes(bytes(8)) + commit_slot(path, 1) + stop, cancelled = threading.Event(), threading.Event() + + def write(): + try: + with write_slot(path, stop): + raise AssertionError("must not overwrite a reader") + except InterruptedError: + cancelled.set() + + with read_slot(path, 1): + worker = threading.Thread(target=write) + worker.start() + stop.set() + worker.join(2) + assert cancelled.is_set() diff --git a/examples/camera_streamer/tests/python/test_wire_format.py b/examples/camera_streamer/tests/python/test_wire_format.py new file mode 100755 index 0000000000..4d3f3e047b --- /dev/null +++ b/examples/camera_streamer/tests/python/test_wire_format.py @@ -0,0 +1,445 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import dataclasses +import sys +import unittest +from pathlib import Path +from typing import Optional +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "camera_receiver")) + +from avs_wire import ( # noqa: E402 + BUNDLE_SENDER_CLOCK_VALID, + BUNDLE_SENDER_TSC_VALID, + CODEC_HEVC, + CODEC_RAW_NV12, + HEADER, + SAMPLE, + SENSOR_SOF_MAX_LAG_NS, + STREAM, + Avs1Bundle, + parse_avs1_record, + sensor_sof_lag_ns, + sensor_sof_ns, +) +from camera_receiver import ( # noqa: E402 + ReceiveFpsMonitor, + ReceiveFpsWarningConfig, + _OUTPUT_NV12, + _ipc_layout, + _parse_optional_gpu_id, + _parse_stats_log_interval_s, + _parse_stream_config, + _resolve_receive_fps_warning_config, + _resolve_stats_log_interval_s, +) + + +_CAPTURE_READY_NS = 10_000 +_SOF_TSC_NS = (1000, 1210, 3000) # head.left, head.right, wrist + + +def _avs1_record( + *, + flags: Optional[int] = None, + bundle_sequence: int = 3, + session_id: int = 99, + stream1_present: bool = True, + sender_capture_ready_tsc_ns: Optional[int] = None, +) -> bytes: + # flags=None derives CLOCK_VALID plus SENDER_TSC_VALID when a TSC is given; + # pass flags explicitly to build deliberately inconsistent records. + if flags is None: + flags = BUNDLE_SENDER_CLOCK_VALID + if sender_capture_ready_tsc_ns is not None: + flags |= BUNDLE_SENDER_TSC_VALID + head_left = b"\x00\x00\x00\x01head-left" + head_right = b"\x00\x00\x00\x01head-right" + wrist = bytes(640 * 480 * 3 // 2) if stream1_present else b"" + payloads = [head_left, head_right, wrist] + payload_offsets = [] + offset = 0 + for payload in payloads: + payload_offsets.append(offset) + offset += len(payload) + payload = b"".join(payloads) + + stream0_flags = 0x0001 | 0x0002 + stream1_flags = 0x0001 if stream1_present else 0 + streams = [ + STREAM.pack( + 0, + stream0_flags, + CODEC_HEVC, + 1, + 180, + 2560, + 1984, + 60, + 1, + 42, + 0, + 2, + 0, + ), + STREAM.pack( + 1, + stream1_flags, + CODEC_RAW_NV12, + 1, + 0, + 640, + 480, + 60, + 1, + 7 if stream1_present else 0, + 2, + 1, + 0, + ), + ] + samples = [ + SAMPLE.pack( + 0, + 0, + 1, + 0x000F, + 0, + 0, + 0, + 100, + _SOF_TSC_NS[0], + 2000, + 10_000, + 10_100, + 10_500, + payload_offsets[0], + len(head_left), + ), + SAMPLE.pack( + 0, + 1, + 2, + 0x000F, + 0, + 1, + 0, + 101, + _SOF_TSC_NS[1], + 2210, + 10_000, + 10_100, + 10_550, + payload_offsets[1], + len(head_right), + ), + SAMPLE.pack( + 1, + 0, + 0, + 0x000F if stream1_present else 0, + 0, + 4, + 0, + 5 if stream1_present else 0, + _SOF_TSC_NS[2] if stream1_present else 0, + 4000 if stream1_present else 0, + 11_000 if stream1_present else 0, + 11_050 if stream1_present else 0, + 11_100 if stream1_present else 0, + payload_offsets[2], + len(wrist), + ), + ] + descriptor_bytes = len(streams) * STREAM.size + len(samples) * SAMPLE.size + header = HEADER.pack( + b"AVS1", + 1, + flags, + HEADER.size, + len(streams), + len(samples), + bundle_sequence, + session_id, + _CAPTURE_READY_NS, + 10_800, + descriptor_bytes, + len(payload), + sender_capture_ready_tsc_ns or 0, + ) + return header + b"".join(streams) + b"".join(samples) + payload + + +class Avs1ParserTests(unittest.TestCase): + def test_optional_gpu_id_semantics(self) -> None: + self.assertIsNone(_parse_optional_gpu_id(None)) + self.assertEqual(_parse_optional_gpu_id(0), 0) + self.assertEqual(_parse_optional_gpu_id("2"), 2) + with self.assertRaisesRegex(ValueError, "must be >= 0"): + _parse_optional_gpu_id(-1) + + def test_stats_log_interval_semantics(self) -> None: + self.assertEqual(_parse_stats_log_interval_s(0), 0.0) + self.assertEqual(_parse_stats_log_interval_s("2.5"), 2.5) + with self.assertRaisesRegex(ValueError, "finite number >= 0"): + _parse_stats_log_interval_s(-1) + with self.assertRaisesRegex(ValueError, "finite number >= 0"): + _parse_stats_log_interval_s(float("inf")) + + def test_stats_log_interval_resolution(self) -> None: + self.assertEqual(_resolve_stats_log_interval_s({}, None), 5.0) + self.assertEqual( + _resolve_stats_log_interval_s({"stats_log_interval_s": 0}, None), + 0.0, + ) + self.assertEqual( + _resolve_stats_log_interval_s({"stats_log_interval_s": 1}, 3.0), + 3.0, + ) + self.assertEqual( + _resolve_stats_log_interval_s({"log_timing": False}, None), + 0.0, + ) + + def test_receive_fps_warning_config(self) -> None: + disabled = _resolve_receive_fps_warning_config({}) + self.assertFalse(disabled.enabled) + + enabled = _resolve_receive_fps_warning_config( + { + "receive_fps_warning_threshold": 55, + "receive_fps_warning_interval_s": 10, + } + ) + self.assertTrue(enabled.enabled) + self.assertEqual(enabled.threshold_fps, 55.0) + self.assertEqual(enabled.interval_s, 10.0) + + with self.assertRaisesRegex(ValueError, "must be > 0"): + _resolve_receive_fps_warning_config( + { + "receive_fps_warning_threshold": 55, + "receive_fps_warning_interval_s": 0, + } + ) + + def test_receive_fps_monitor_warns_only_for_low_streams(self) -> None: + now = [0.0] + monitor = ReceiveFpsMonitor( + ("head", "wrist1"), + ReceiveFpsWarningConfig(threshold_fps=55.0, interval_s=2.0), + clock=lambda: now[0], + ) + monitor.start() + for _ in range(120): + monitor.observe(("head",)) + for _ in range(100): + monitor.observe(("wrist1",)) + now[0] = 2.0 + + with patch("camera_receiver.logger.warning") as warning: + monitor.poll() + + warning.assert_called_once() + message_args = warning.call_args.args + self.assertIn("wrist1=50.0fps", message_args[-1]) + self.assertNotIn("head", message_args[-1]) + + def test_receive_fps_monitor_reports_zero_during_connected_stall(self) -> None: + now = [0.0] + monitor = ReceiveFpsMonitor( + ("head",), + ReceiveFpsWarningConfig(threshold_fps=55.0, interval_s=1.0), + clock=lambda: now[0], + ) + monitor.start() + now[0] = 1.0 + + with patch("camera_receiver.logger.warning") as warning: + monitor.poll() + + warning.assert_called_once() + self.assertIn("head=0.0fps", warning.call_args.args[-1]) + + def test_parses_mixed_stream_bundle(self) -> None: + bundle = parse_avs1_record(_avs1_record()) + + self.assertIsInstance(bundle, Avs1Bundle) + self.assertEqual(bundle.sequence, 3) + self.assertEqual(bundle.session_id, 99) + self.assertEqual(bundle.sender_clock_domain, "realtime") + self.assertEqual(len(bundle.streams), 2) + + head = bundle.streams[0] + self.assertTrue(head.present) + self.assertTrue(head.stereo) + self.assertEqual(head.codec, CODEC_HEVC) + self.assertEqual(head.rotation, 180) + self.assertEqual(head.tsc_delta_ns, 210) + self.assertEqual(bytes(head.samples[0].payload), b"\x00\x00\x00\x01head-left") + self.assertEqual(bytes(head.samples[1].payload), b"\x00\x00\x00\x01head-right") + + wrist = bundle.streams[1] + self.assertTrue(wrist.present) + self.assertFalse(wrist.stereo) + self.assertEqual(wrist.codec, CODEC_RAW_NV12) + self.assertEqual(wrist.samples[0].payload_length, 640 * 480 * 3 // 2) + + def test_accepts_absent_lower_rate_stream(self) -> None: + bundle = parse_avs1_record(_avs1_record(stream1_present=False)) + + self.assertTrue(bundle.streams[0].present) + self.assertFalse(bundle.streams[1].present) + self.assertEqual(bundle.streams[1].sequence, 0) + self.assertEqual(bundle.streams[1].samples[0].payload_length, 0) + + def test_rejects_tai_without_clock_valid(self) -> None: + with self.assertRaisesRegex(ValueError, "CLOCK_TAI"): + parse_avs1_record(_avs1_record(flags=0x0002)) + + def test_legacy_bundle_has_no_sender_tsc(self) -> None: + bundle = parse_avs1_record(_avs1_record()) + + self.assertEqual(bundle.flags, 0x0001) + self.assertIsNone(bundle.sender_capture_ready_tsc_ns) + for stream in bundle.streams: + for sample in stream.samples: + self.assertIsNone(sensor_sof_lag_ns(bundle, sample)) + self.assertIsNone(sensor_sof_ns(bundle, sample)) + + def test_parses_sender_tsc_when_flagged(self) -> None: + bundle = parse_avs1_record(_avs1_record(sender_capture_ready_tsc_ns=3_500)) + + self.assertEqual(bundle.flags, 0x0001 | 0x0004) + self.assertEqual(bundle.sender_capture_ready_ns, 10_000) + self.assertEqual(bundle.sender_capture_ready_tsc_ns, 3_500) + + def test_rejects_sender_tsc_flag_without_field(self) -> None: + with self.assertRaisesRegex(ValueError, "zero sender_capture_ready_tsc_ns"): + parse_avs1_record(_avs1_record(flags=0x0001 | 0x0004)) + + def test_rejects_sender_tsc_field_without_flag(self) -> None: + with self.assertRaisesRegex(ValueError, "without SENDER_TSC_VALID"): + parse_avs1_record( + _avs1_record(flags=0x0001, sender_capture_ready_tsc_ns=3_500) + ) + + def test_rejects_sender_tsc_without_clock_valid(self) -> None: + with self.assertRaisesRegex(ValueError, "SENDER_TSC_VALID flag requires"): + parse_avs1_record( + _avs1_record(flags=0x0004, sender_capture_ready_tsc_ns=3_500) + ) + + def test_sensor_sof_math_per_sample(self) -> None: + # capture_ready wall 10_000 ns, capture_ready TSC 3_500 ns. + bundle = parse_avs1_record(_avs1_record(sender_capture_ready_tsc_ns=3_500)) + left, right = bundle.streams[0].samples + (wrist,) = bundle.streams[1].samples + + self.assertEqual(sensor_sof_lag_ns(bundle, left), 2_500) + self.assertEqual(sensor_sof_lag_ns(bundle, right), 2_290) + self.assertEqual(sensor_sof_lag_ns(bundle, wrist), 500) + self.assertEqual(sensor_sof_ns(bundle, left), 10_000 - 2_500) + # Right eye uses its own SOF: 210 ns later than the left eye. + self.assertEqual(sensor_sof_ns(bundle, right), 10_000 - 2_290) + self.assertEqual( + sensor_sof_ns(bundle, right) - sensor_sof_ns(bundle, left), + bundle.streams[0].tsc_delta_ns, + ) + self.assertEqual(sensor_sof_ns(bundle, wrist), 10_000 - 500) + + def test_sensor_sof_lag_window(self) -> None: + (wrist,) = ( + parse_avs1_record(_avs1_record(sender_capture_ready_tsc_ns=3_500)) + .streams[1] + .samples + ) + + # Zero lag (SOF == capture-ready instant) is accepted. + at_sof = parse_avs1_record(_avs1_record(sender_capture_ready_tsc_ns=3_000)) + self.assertEqual(sensor_sof_lag_ns(at_sof, wrist), 0) + self.assertEqual(sensor_sof_ns(at_sof, wrist), 10_000) + + # Exactly the maximum lag is accepted. + at_max = parse_avs1_record( + _avs1_record(sender_capture_ready_tsc_ns=3_000 + SENSOR_SOF_MAX_LAG_NS) + ) + self.assertEqual(sensor_sof_lag_ns(at_max, wrist), SENSOR_SOF_MAX_LAG_NS) + self.assertEqual(sensor_sof_ns(at_max, wrist), 10_000 - SENSOR_SOF_MAX_LAG_NS) + + # One ns past the window: lag is still reported, SOF is not. + too_late = parse_avs1_record( + _avs1_record(sender_capture_ready_tsc_ns=3_000 + SENSOR_SOF_MAX_LAG_NS + 1) + ) + self.assertEqual(sensor_sof_lag_ns(too_late, wrist), SENSOR_SOF_MAX_LAG_NS + 1) + self.assertIsNone(sensor_sof_ns(too_late, wrist)) + + # SOF after the capture-ready instant (negative lag) is rejected. + before_sof = parse_avs1_record(_avs1_record(sender_capture_ready_tsc_ns=2_999)) + self.assertEqual(sensor_sof_lag_ns(before_sof, wrist), -1) + self.assertIsNone(sensor_sof_ns(before_sof, wrist)) + + def test_sensor_sof_requires_every_input(self) -> None: + bundle = parse_avs1_record(_avs1_record(sender_capture_ready_tsc_ns=3_500)) + (wrist,) = bundle.streams[1].samples + + no_wall = dataclasses.replace(bundle, sender_capture_ready_ns=None) + self.assertEqual(sensor_sof_lag_ns(no_wall, wrist), 500) + self.assertIsNone(sensor_sof_ns(no_wall, wrist)) + + no_tsc = dataclasses.replace(bundle, sender_capture_ready_tsc_ns=None) + self.assertIsNone(sensor_sof_lag_ns(no_tsc, wrist)) + self.assertIsNone(sensor_sof_ns(no_tsc, wrist)) + + no_sof = dataclasses.replace(wrist, sof_tsc_ns=None) + self.assertIsNone(sensor_sof_lag_ns(bundle, no_sof)) + self.assertIsNone(sensor_sof_ns(bundle, no_sof)) + + def test_rejects_length_mismatch(self) -> None: + record = bytearray(_avs1_record()) + fields = list(HEADER.unpack_from(record)) + fields[11] += 1 + HEADER.pack_into(record, 0, *fields) + with self.assertRaisesRegex(ValueError, "byte count mismatch"): + parse_avs1_record(record) + + def test_rejects_raw_payload_size_mismatch(self) -> None: + record = bytearray(_avs1_record()) + sample2_offset = HEADER.size + 2 * STREAM.size + 2 * SAMPLE.size + fields = list(SAMPLE.unpack_from(record, sample2_offset)) + fields[-1] -= 1 + SAMPLE.pack_into(record, sample2_offset, *fields) + with self.assertRaisesRegex(ValueError, "raw AVS1 payload"): + parse_avs1_record(record) + + def test_stream_config_defaults(self) -> None: + cfg = _parse_stream_config( + { + "index": 0, + "name": "head", + "type": "stereo", + "width": 2560, + "height": 1984, + } + ) + + self.assertTrue(cfg.stereo) + self.assertEqual(cfg.output_format, _OUTPUT_NV12) + self.assertEqual(cfg.ring_depth, 6) + self.assertEqual(cfg.ipc_names, ("head.left", "head.right")) + + def test_nv12_ipc_layout(self) -> None: + layout = _ipc_layout("nv12", 640, 480) + + self.assertEqual(layout["planes"][0]["name"], "y") + self.assertEqual(layout["planes"][1]["name"], "uv") + self.assertEqual(layout["planes"][1]["offset"], 640 * 480) + + +if __name__ == "__main__": + unittest.main() diff --git a/examples/camera_streamer/tests/python/test_worker_stream.py b/examples/camera_streamer/tests/python/test_worker_stream.py new file mode 100644 index 0000000000..43c3dfbead --- /dev/null +++ b/examples/camera_streamer/tests/python/test_worker_stream.py @@ -0,0 +1,70 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +import contextlib +import sys +import threading +from concurrent.futures import ThreadPoolExecutor +from types import SimpleNamespace + +import pytest + +from camera_receiver import Avs1TcpReceiver, CODEC_RAW_NV12 + + +@pytest.mark.parametrize( + "pixel_format,rotation", [("nv12", 0), ("nv12", 180), ("rgba8", 0), ("rgba8", 180)] +) +@pytest.mark.parametrize("fails", [False, True]) +def test_raw_work_completes_on_sample_thread_before_slot_release( + monkeypatch, pixel_format, rotation, fails +): + pending = threading.local() + coordinator = threading.get_ident() + sync_threads = [] + + def submit(*args): + pending.busy = True + if fails: + raise RuntimeError("submitted work failed") + + def synchronize(): + sync_threads.append(threading.get_ident()) + pending.busy = False + + @contextlib.contextmanager + def slot(*args): + try: + yield + finally: + assert not pending.busy + + cp = SimpleNamespace( + cuda=SimpleNamespace( + Device=lambda _: contextlib.nullcontext(), + get_current_stream=lambda: SimpleNamespace(synchronize=synchronize), + ) + ) + monkeypatch.setitem(sys.modules, "cupy", cp) + output = SimpleNamespace(shape=(3, 4), set=submit) + runtime = SimpleNamespace( + config=SimpleNamespace(ipc_names=["eye"], output_format=pixel_format), + output_buffers=[[output]], + raw_buffers=[SimpleNamespace(set=submit)], + raw_kernel=submit, + grid=None, + block=None, + ) + receiver = Avs1TcpReceiver.__new__(Avs1TcpReceiver) + receiver._gpu_device_id = 0 + receiver._stop = threading.Event() + receiver._ipc_publisher = SimpleNamespace(write_slot=slot) + sample = SimpleNamespace(camera_index=0, payload=bytes(12)) + stream = SimpleNamespace(codec=CODEC_RAW_NV12, width=4, height=2, rotation=rotation) + with ThreadPoolExecutor(max_workers=1) as pool: + future = pool.submit(receiver._process_sample, runtime, stream, sample, 0) + if fails: + with pytest.raises(RuntimeError, match="submitted"): + future.result() + else: + assert future.result() + assert len(sync_threads) == 1 and sync_threads[0] != coordinator