diff --git a/examples/camera_streamer/AGENTS.md b/examples/camera_streamer/AGENTS.md index 819fcdac49..62bff46128 100644 --- a/examples/camera_streamer/AGENTS.md +++ b/examples/camera_streamer/AGENTS.md @@ -5,10 +5,43 @@ SPDX-License-Identifier: Apache-2.0 # Standalone GMSL/Argus streamer -- Keep the wire/configuration tests runnable without Jetson or CUDA dependencies. +- Keep this example scoped to GMSL/Argus capture, AVS1/TCP, NVDEC and CUDA IPC. + Do not restore unrelated camera wrappers, USB senders, RTP/recording paths or + rig-specific deployment assets without an explicit scope change. - Preserve upstream notices when adding SPDX headers; missing SPDX does not mean a file lacks a license. Keep third-party exceptions separate from Apache-2.0 code. -- Keep each review independently buildable: add tests and documentation with their dependencies. -- Normalize generated build fragments to one final newline before staging. +- Keep the native sender and standalone receiver independent of robot control + software and the Isaac Capture runtime. Use the isaaccapture logger tree without importing it. +- Document Install, Preview, Send, Receive and Viewer stages with explicit + commands, without a launcher wrapper. Keep operator instructions in this README + rather than duplicating them across subdirectories. Local Argus preview is a separate application + and must release the sensor before network capture starts. +- Use HOST_IP/ROBOT_IP discovered for the installation, never developer addresses. + Explain source transfer before referencing robot deployment directories. +- Select sensor modes from --list-modes and match PWM to capture rate. Record + hardware validation evidence in the review; keep operator mode tables focused on configuration. +- Deploy directory renames with documentation and rebuild stale CMake + caches. Do not transfer virtual environments, builds or caches between machines. +- Receiver tests must establish the standalone import path. Check runtime imports + and native/receiver tests after trimming dependencies, not just formatting. +- Preserve encoded frame dependencies across disconnects. CUDA IPC handles stay + local to the receiving machine; they are not network transport handles. +- Preview loops must handle window-close events before rendering another frame. +- CuPy pixel diagnostics need matching NVRTC; exclude first-use JIT from timing. +- Normalize final newlines and imported executable permissions before staging. +- Describe camera support generically; separate upstream device testing from local validation. +- Use one shared profile per device/resolution; define wire geometry and port once. +- Keep tests under the example-level tests/ directory, split by language and hardware needs. +- Public setup instructions target main; describe pre-merge branch overrides separately. +- Do not infer Argus mode indices from vendor specifications or another model; keep unknown modes explicit in profiles. +- Keep camera YAMLs directly under configs/ with model-prefixed filenames so profile selection needs no subdirectory. +- Use matching columns for all camera profile tables so resolution, frame rate and capture constraints are directly comparable. + Keep profile alternatives in those tables; show PROFILE assignments in fenced code blocks with an inline selection hint. +- Preserve HEVC ordering through TCP enqueue and bound receiver work queues; backpressure must remain interruptible during shutdown. +- Verify unresolved review findings against the current PR head and regression tests; + an open thread can describe code already fixed in an earlier commit. +- Concurrency regression tests must synchronize inside the operation under test; + a pre-call flag and a fixed sleep do not establish lock contention. +- Keep portable tests buildable without Jetson dependencies and each review independently buildable. - Keep argus_sender, camera_receiver and cuda_ipc_consumer directly under camera_streamer; do not group standalone transport components under the TeleViz visualization name. diff --git a/examples/camera_streamer/README.md b/examples/camera_streamer/README.md index 6a1edb9fae..113fc39652 100644 --- a/examples/camera_streamer/README.md +++ b/examples/camera_streamer/README.md @@ -5,78 +5,431 @@ SPDX-License-Identifier: Apache-2.0 # Isaac Camera Streamer — GMSL / Argus -This component streams Argus camera data using AVS1/TCP. The protocol and -configuration tests run without camera hardware. +Stream Argus-compatible GMSL stereo video from a Jetson Orin robot computer to an NVIDIA +workstation: **Argus → hardware HEVC → AVS1/TCP → NVDEC → CUDA IPC viewer**. +This standalone example runs independently of robot control software. -## Protocol and configuration tests +| Camera | Robot Install | Robot Preview | Robot Send | Host Receive | Host Viewer | +|---|---|---|---|---|---| +| Sensing SHW5G / S56 (GMSL) | Required: driver + native build | Optional: external Argus app | Jetson HEVC/TCP | NVDEC receiver | Stereo desktop view | + +Complete common setup once, then follow the five GMSL sections below. +Network streaming uses three terminals: Robot Send, Host Receive, Host Viewer. + +## Common setup — once per computer + +### Identify the two network addresses + +On **both computers**, run: ```bash -cmake -S examples/camera_streamer/argus_sender -B /tmp/camera-protocol-build -DARGUS_SENDER_CPU_TESTS_ONLY=ON -cmake --build /tmp/camera-protocol-build --parallel 2 -ctest --test-dir /tmp/camera-protocol-build --output-on-failure +ip -brief -4 address ``` -See [the wire specification](argus_sender/WIRE_FORMAT.md). Shared camera -profiles are in `configs/`; select sensor modes using the installed camera driver. +Choose the IPv4 address of the Ethernet, Wi-Fi, or VPN interface that connects +these two sites. Use the address without its `/prefix` suffix. Do not select +loopback (`127.0.0.1`) or a Docker-only interface. For a shared LAN, choose the +addresses on that LAN; for routed sites, choose mutually reachable addresses. -## Jetson sender +Replace the placeholders below and export both values in each terminal where +you run the network commands: -Install the camera vendor driver and Jetson Multimedia API on the Orin. Stop any -preview using the sensors and configure the vendor PWM for the selected frame rate. -From `examples/camera_streamer` (or a copy of that directory on the robot): +```bash +export HOST_IP="" +export ROBOT_IP="" +``` + +`HOST_IP` always means the workstation and `ROBOT_IP` always means the Orin, +regardless of which computer runs the command. Exports do not carry into a new +terminal or SSH session; repeat them there. You can check routing from each side: ```bash -cmake -S argus_sender -B argus_sender/build -DCMAKE_BUILD_TYPE=Release -cmake --build argus_sender/build --parallel 2 -ctest --test-dir argus_sender/build --output-on-failure -argus_sender/build/argus_sender --list-modes -PROFILE=s56-stereo-2560x1984-30 # Choose your resolution -argus_sender/build/argus_sender --config "configs/${PROFILE}.yaml" --check-config -argus_sender/build/argus_sender --config "configs/${PROFILE}.yaml" +# On the Distributed Cloud workstation: +ip route get "${ROBOT_IP:?Set ROBOT_IP first}" +# On the Robot Site: +ip route get "${HOST_IP:?Set HOST_IP first}" +``` + +The receiver connects to the sender at `ROBOT_IP:5000`. The sender listens on +all local interfaces by default, so it does not need `HOST_IP` as an argument. + +### Put the source on both computers + +`~/isaac-camera-streamer` is a deployment directory. The copy commands below +create it explicitly; the robot does not need a separate repository there. +Choose **one** of the following approaches. + +**Recommended for a disk-constrained robot: clone on the host, copy the example.** +The robot does not need a full Isaac Capture clone. Start with a host +checkout containing `examples/camera_streamer` (Git and Git LFS required). +The commands use `main` for the merged example. Before merge, reviewers must +substitute the pull request's head branch for `main`. + +```bash +# On the Distributed Cloud workstation; skip cloning if you have this checkout. +export CAMERA_REPO_URL="https://github.com/NVIDIA/IsaacCapture.git" +export CAMERA_BRANCH="main" +GIT_LFS_SKIP_SMUDGE=1 git clone --single-branch --branch "$CAMERA_BRANCH" \ + "$CAMERA_REPO_URL" "$HOME/IsaacCapture" +cd ~/IsaacCapture +git lfs pull --include='examples/camera_streamer/**' --exclude='' ``` -The sender listens for a TCP receiver on port 5000 by default. +With SSH access configured and `rsync` installed on both computers, run this on +the **host**. Replace the username with the account you use on the robot: + +```bash +export ROBOT_USER="" +ssh "${ROBOT_USER}@${ROBOT_IP:?Set ROBOT_IP first}" \ + 'mkdir -p "$HOME/isaac-camera-streamer/camera_streamer"' +rsync -av --exclude='.venv*/' --exclude='build/' \ + --exclude='__pycache__/' --exclude='.pytest_cache/' \ + --exclude='*.egg-info/' --exclude='*.so' \ + "$HOME/IsaacCapture/examples/camera_streamer/" \ + "${ROBOT_USER}@${ROBOT_IP}:isaac-camera-streamer/camera_streamer/" +``` -## Workstation NVDEC decoder +The trailing source slash copies the example's contents into the destination. +Run this transfer again to deploy source updates; it overwrites matching files, +so keep robot-specific profiles outside that directory or copy edits back first. +Virtual environments and native builds stay on their own machine: workstation +binaries cannot be reused on the ARM64 Orin. -On a Linux workstation with an NVIDIA GPU, CUDA toolkit, CMake and `uv`, run from -`examples/camera_streamer`: +**Alternative: clone directly on both computers.** On the robot, run the same +clone and LFS commands above using the same branch/revision as the host. No SSH +file transfer or `isaac-camera-streamer` directory is then needed. Use +`~/IsaacCapture/examples/camera_streamer` instead of the deployment path in the +Robot Site commands below. This approach uses more disk space for Git history +and the rest of the checkout. + +### Test a clean robot deployment + +Stop the sender and any local preview, and save custom profiles or captures outside +`~/isaac-camera-streamer` before deleting that directory. Repeating the source +transfer and Robot Install steps then builds the sender from scratch. This resets +the example's source, build products and any environments inside the directory; +it does not uninstall JetPack, camera drivers, system packages or PWM setup. + +For a fuller check, use fresh terminals, enumerate cameras with `--list-modes`, +validate the shared profile with `--check-config`, run the native tests, then +start sender, receiver and viewer. Check both eyes, stop/restart the receiver to +exercise reconnect, and switch profiles after stopping all three processes. +A reboot followed by the documented vendor driver/PWM setup checks cold-start +behavior. A clean OS installation is a separate test of system prerequisites. + +### Install the shared host receiver and viewer + +Do this once, for any supported GMSL mode. The host needs Linux, an NVIDIA NVDEC-capable +GPU, CUDA development tools, CMake, a C++ compiler and `uv`. Argus and the camera vendor drivers are needed only on the robot. The native build downloads a pinned NVIDIA Video +Codec SDK archive from the source-configured NVIDIA artifact server. ```bash +cd ~/IsaacCapture/examples/camera_streamer uv venv .venv --python 3.12 -uv pip install --python .venv/bin/python pybind11 numpy +uv pip install --python .venv/bin/python -e '.[receiver,test]' cmake -S camera_receiver/codec -B camera_receiver/codec/build \ -DCMAKE_BUILD_TYPE=Release -DPython3_EXECUTABLE="$PWD/.venv/bin/python" cmake --build camera_receiver/codec/build --parallel 2 ``` -The build downloads the pinned NVIDIA Video Codec SDK. CPU-side Python tests: +Create the environment only once. Reinstall dependencies/rebuild after relevant +source updates. Keep host and robot environments separate even when both are +named `.venv`. + +### How to select and switch modes + +All profiles live directly in `configs/`, with model-prefixed filenames. +Each device/resolution has one shared YAML. Set `PROFILE` to the filename without +`.yaml`, for example: + +```bash +PROFILE=s56-stereo-2560x1984-30 # Choose your resolution +``` + +Then `configs/${PROFILE}.yaml` selects the file. Copy the same file to both computers and select the **same PROFILE on both computers**. +`network.port` and `streams[].id/type/output` define the common wire format. +`cameras`, `capture`, `runtime` and `alignment` control capture; `receiver`, `ipc` +and `streams[].receiver` control decoding and local IPC. Stream order determines +the AVS1 index. The receiver derives dimensions from `output` without a second copy. +Set `PROFILE` again in each new terminal; shell variables are not shared. +No Python copy/modify script is needed for the supplied profiles. + +Before switching mode, stop the viewer, receiver and sender with +Ctrl+C. Then choose a new profile and restart those three processes. Stop local +previews before sending: they compete for camera ownership. All profiles use +TCP port 5000 and host IPC socket `/tmp/isaac-camera-streamer.sock`, so run one +pipeline at a time. TCP is unauthenticated; use a trusted LAN or VPN. + +A sensor's maximum rate is not a guarantee of the complete capture/encode/network/display +rate. Link bandwidth, lighting, driver modes and hardware sync can limit it. + +## GMSL cameras: Sensing + +Stereo GMSL2 → Argus → Jetson hardware HEVC → AVS1/TCP. Dimensions are per eye. +The sender uses Argus device IDs and sensor modes for Sensing cameras, including +SHW5G and S56. Device models are not hard-coded in the sender. + +### S56 stereo profiles + +The supplied stereo profiles use sensor indices 0/1 on the S56 adapter. +Resolution and frame rate below apply to each eye. + +| PROFILE | Output resolution | Frame rate | Capture mode / notes | +|---|---|---|---| +| `s56-stereo-2560x1984-30` | 2560×1984 | 30 fps | Mode 1, native dimensions; highest resolution exposed by this driver | +| `s56-stereo-1080p60` | 1920×1080 | 60 fps | Mode 0, native dimensions; highest frame rate exposed by this driver | + +### S56 mono profiles + +Use one S56 sensor to exercise the mono capture, HEVC, network and viewer path. +Here, mono means one color camera stream, not grayscale. The supplied profiles +select device 0; select device 1 instead to use the other eye. + +| PROFILE | Output resolution | Frame rate | Capture mode / notes | +|---|---|---|---| +| `s56-mono-2560x1984-30` | 2560×1984 | 30 fps | Mode 1, native dimensions | +| `s56-mono-1080p30` | 1920×1080 | 30 fps | Mode 0, native dimensions | +| `s56-mono-1080p60` | 1920×1080 | 60 fps | Mode 0, native dimensions | +| `s56-mono-720p30` | 1280×720 | 30 fps | Mode 0, scaled from 1920×1080 | +| `s56-mono-720p60` | 1280×720 | 60 fps | Mode 0, scaled from 1920×1080 | +| `s56-mono-2400x1440-30` | 2400×1440 | 30 fps | Mode 1, center-cropped/scaled from 2560×1984 | + +Set `PROFILE` to the same mono profile in the Robot Send and Host Receive commands +below. The viewer automatically displays the single `head` stream. No runtime +code or separate receiver configuration is needed. All mono profiles use 30 Mbps. + +### SHW5G mono profiles + +SENSING lists SHW5G as a monocular camera with 2560×1984 at 30 fps and +2400×1440 at 60 fps ([SHW5G specifications](https://www.sgwrd.com/products/3mp-global-shutter-monocular-camera-shw5g)). +The [S56 specifications](https://wiki.sensing-world.com/docs/6_1_Camera/Binocular_Camera/S56) +list 2560×1984 at 30 fps per eye. Argus modes depend on the installed driver; +vendor specifications do not define their numeric indices. + +`configs/` contains one shared YAML for each published SHW5G mode: + +| PROFILE | Output resolution | Frame rate | Capture mode / notes | +|---|---|---|---| +| `shw5g-mono-2560x1984-30` | 2560×1984 | 30 fps | Set mode from `--list-modes`; highest published resolution | +| `shw5g-mono-2400x1440-60` | 2400×1440 | 60 fps | Set mode from `--list-modes`; highest published frame rate | + +Edit the desired profile in `configs/`: set `cameras[].device` for your rig, +and replace `mode: SET_FROM_LIST_MODES` with the matching Argus mode index. +Then copy the edited YAML to both computers and use its filename without `.yaml` as `PROFILE` +(for example, `shw5g-mono-2560x1984-30`). +The placeholder intentionally prevents starting a sender with a guessed mode. + +The supplied S56 driver's 60 fps mode is 1920×1080, not 2400×1440. Its +2400×1440 mono profile uses the full-resolution sensor mode at 30 fps; it +exercises that output geometry without claiming the SHW5G 60 fps capture mode. +Smaller scaled outputs are not additional native sensor modes. Use +`--list-modes` as the source of truth for a particular camera/driver combination. + +Stereo profiles use 30 Mbps per eye. At 60 fps, a hardware-triggered rig must also have its +vendor PWM/sync configured for 60 Hz; changing YAML alone does not change PWM. + +### Robot Install + +Complete the common source transfer first. Install the matching JetPack +Multimedia API/CUDA development headers, CMake and a C++ compiler. Install the +Sensing adapter's vendor driver and configure its PWM/sync for the selected rate. +These are rig-specific steps; the sender does not install drivers or change PWM. ```bash -uv run --project tests --extra dev python -m pytest tests/python -q +cd ~/isaac-camera-streamer/camera_streamer +cmake -S argus_sender -B argus_sender/build \ + -DCMAKE_BUILD_TYPE=Release +cmake --build argus_sender/build --parallel 2 +argus_sender/build/argus_sender --list-modes ``` -## Workstation receiver +Verify both eye devices and mode indices against the chosen shared YAML. The +build downloads pinned yaml-cpp. Build initially and after source changes; after +moving the source directory, use a fresh build directory rather than an old +CMake cache. Do not assume supplied rig calibration maps fit your camera. -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: +### Robot Preview + +Optional, with the sender stopped, in a terminal on the **Orin desktop**: ```bash -export ROBOT_IP="" -uv pip install --python .venv/bin/python -r camera_receiver/requirements.txt +argus_camera -d 0 +``` + +Exit it, then use `argus_camera -d 1` to inspect the other eye. This is NVIDIA's +separate Argus sample app; install/build it from the Jetson Multimedia API Argus +samples if absent. It shows one sensor locally and does not use the sender YAML +or transmit to the host. Its GUI controls select preview settings independently. +Exit the preview before sending. No host process is needed for this check. + +### Robot Send + +```bash +cd ~/isaac-camera-streamer/camera_streamer +PROFILE=s56-stereo-2560x1984-30 # Choose your resolution +argus_sender/build/argus_sender \ + --config "configs/${PROFILE}.yaml" --check-config +argus_sender/build/argus_sender \ + --config "configs/${PROFILE}.yaml" +``` + +The sender opens no preview and waits for the receiver before capturing. +`--check-config` validates configuration; it is not an end-to-end hardware test. + +### Host Receive + +After the shared host installation, run in the first host terminal: + +```bash +cd ~/IsaacCapture/examples/camera_streamer 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}" + --config "configs/${PROFILE}.yaml" \ + --host "${ROBOT_IP:?Set ROBOT_IP first}" +``` + +This decodes the configured mono or stereo stream using NVDEC and publishes GPU frames through local CUDA +IPC. It does not open a display window. + +### Host Viewer + +In a second host terminal: + +```bash +cd ~/IsaacCapture/examples/camera_streamer +.venv/bin/python cuda_ipc_consumer/consumer.py \ + --socket /tmp/isaac-camera-streamer.sock ``` -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. +Expect `head.left` and `head.right` for stereo, or `head` for mono. This command is not a +CloudXR launcher; headset display needs a compatible AVS1 CUDA IPC source in +camera_viz. +Close the window, press q/Esc with the window focused, or Ctrl+C in its terminal. + +#### Display size and stream selection + +`--scale` controls each image's display dimensions. The default is `1.0`, which +preserves the decoded resolution before window display. A 2560×1984 stream +produces a 2560×1984 image at `1.0`, or 1280×992 at `0.5`. This affects only the +local preview, not capture, encoding, transmission or the decoded CUDA IPC frames. +The desktop may fit the window to your screen; full-resolution stereo can exceed +its available space. Restart only the viewer to change these options. + +`--streams` selects a comma-separated list of IPC stream names. Omit it to display +all streams. For the supplied profiles, stereo names are `head.left` and +`head.right`; mono uses `head`. -## Licensing +```bash +# Show both eyes at half size to fit a smaller screen. +.venv/bin/python cuda_ipc_consumer/consumer.py \ + --socket /tmp/isaac-camera-streamer.sock \ + --scale 0.5 --streams head.left,head.right + +# Inspect one eye at full resolution (1.0 is also the default). +.venv/bin/python cuda_ipc_consumer/consumer.py \ + --socket /tmp/isaac-camera-streamer.sock \ + --scale 1.0 --streams head.left + +# Display the single stream from a mono profile. +.venv/bin/python cuda_ipc_consumer/consumer.py \ + --socket /tmp/isaac-camera-streamer.sock --streams head +``` + +## Tuning and diagnostics + +To add a camera mode, copy one shared YAML under a new device/resolution profile name. +Change `streams[].output` and choose the matching `cameras[].mode` from +`--list-modes`. Dimensions are shared; adjust the receiver FPS warning threshold and +configure hardware PWM for the capture rate. Save changes on the host and +transfer profiles before restarting the sender, receiver and viewer. + +`bitrate_mbps` is Mbps per eye; `gop` is in frames. Sensor timestamps and timing +statistics are not by themselves a photon-to-photon latency measurement. + +On a CUDA 13-only host, the optional CuPy pixel-mean diagnostic needs NVRTC 12: + +```bash +cd ~/IsaacCapture/examples/camera_streamer +uv pip install --python .venv/bin/python nvidia-cuda-nvrtc-cu12 +export LD_LIBRARY_PATH="$PWD/.venv/lib/python3.12/site-packages/nvidia/cuda_nvrtc/lib${LD_LIBRARY_PATH:+:$LD_LIBRARY_PATH}" +.venv/bin/python cuda_ipc_consumer/consumer.py \ + --socket /tmp/isaac-camera-streamer.sock --no-display --frames 120 --sample-mean +``` + +First-use kernel compilation can pause this diagnostic. Omit `--sample-mean` +when measuring steady frame delivery. + +To record frame timing without opening a window, use the same consumer: + +```bash +.venv/bin/python cuda_ipc_consumer/consumer.py \ + --socket /tmp/isaac-camera-streamer.sock --no-display \ + --frames 120 --latency-csv /tmp/camera_latency.csv +``` + +## Validation + +Portable protocol/configuration tests can run on a workstation without Jetson libraries: + +```bash +cd ~/IsaacCapture/examples/camera_streamer +cmake -S argus_sender -B /tmp/camera-protocol-build -DARGUS_SENDER_CPU_TESTS_ONLY=ON +cmake --build /tmp/camera-protocol-build --parallel 2 +ctest --test-dir /tmp/camera-protocol-build --output-on-failure +``` + +```bash +cd ~/IsaacCapture/examples/camera_streamer +uv pip install --python .venv/bin/python -e ".[test]" +.venv/bin/python -m pytest tests +``` + +On the robot, run the native sender tests where the sender was built. The default +suite checks configuration, scheduling, alignment and AVS1/TCP behavior without +opening a camera: + +```bash +cd ~/isaac-camera-streamer/camera_streamer +ctest --test-dir argus_sender/build --output-on-failure +``` + +For an opt-in stereo hardware test, stop the sender and local preview first. +Check the devices, sensor modes and PWM rate against the smoke-test YAML before +running it; pass a different YAML path if your camera setup differs. Install +PyYAML for the interpreter selected by CMake (`Python3_EXECUTABLE`) before this test: + +```bash +cd ~/isaac-camera-streamer/camera_streamer/argus_sender +cmake -S . -B build \ + -DARGUS_SENDER_HARDWARE_TESTS=ON \ + -DARGUS_SENDER_HARDWARE_CONFIG="$PWD/configs/hardware_stereo_smoke.yaml" +cmake --build build --parallel 2 +ctest --test-dir build -R '^avs1_hardware_smoke$' --output-on-failure -V +``` + +This receives 120 stereo bundles and checks sequence continuity, nonempty HEVC +payloads, increasing sensor timestamps when available, initial HEVC parameter +sets/keyframes, and clean sender exit. The receiver reads `network.port` from +the same YAML as the sender (default 5000); change the YAML to select another port. +Configuration checks alone do not validate a camera mode on hardware. + +## Source layout and provenance + +| Directory | Purpose | +|---|---| +| `argus_sender` | Native Argus capture, hardware HEVC and AVS1/TCP | +| `camera_receiver` | AVS1 parsing, NVDEC extension, GPU rings and CUDA IPC | +| `cuda_ipc_consumer` | Standalone viewer and timing diagnostics | +| `tests` | C++ unit tests, Python receiver/config tests and opt-in hardware tests | +| `configs` | Shared device/resolution profiles for sender and receiver | + +The GMSL/Argus pipeline was adapted from NVIDIA camera sources at commit +`1fdb06d91ddb02ad527563cbacfc894a76d09126`. The original extraction and hardware-validation history are retained in +[PR #1358](https://github.com/NVIDIA/IsaacCapture/pull/1358). This example is Apache-2.0 licensed. Existing upstream notices and external SDK licenses remain in effect. The Argus-derived `EGLGlobal.h`, `EGLGlobal.cpp` and -`Error.h` retain their BSD-3-Clause license and notices. +`Error.h` retain their BSD-3-Clause license and notices. This example is not included in Isaac Capture wheels. ## Rebuilding after a source-directory move @@ -84,11 +437,6 @@ CMake caches absolute source paths. After updating an existing deployment, confi and build the native sender and decoder again at the paths shown above. Do not move or reuse an old build directory from a different source location. -For opt-in hardware smoke tests, install PyYAML for the Python interpreter selected -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 diff --git a/examples/camera_streamer/__init__.py b/examples/camera_streamer/__init__.py new file mode 100644 index 0000000000..544e2d4224 --- /dev/null +++ b/examples/camera_streamer/__init__.py @@ -0,0 +1,4 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Standalone Sensing/Argus camera streaming example.""" diff --git a/examples/camera_streamer/cuda_ipc_consumer/.gitignore b/examples/camera_streamer/cuda_ipc_consumer/.gitignore new file mode 100644 index 0000000000..13c4cad133 --- /dev/null +++ b/examples/camera_streamer/cuda_ipc_consumer/.gitignore @@ -0,0 +1,18 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +# Python environment and caches +.venv/ +__pycache__/ +*.py[cod] +*$py.class +.pytest_cache/ +.mypy_cache/ +.ruff_cache/ +.coverage +htmlcov/ + +# Local logs and runtime artifacts +*.log +*.pid +*.sock diff --git a/examples/camera_streamer/cuda_ipc_consumer/consumer.py b/examples/camera_streamer/cuda_ipc_consumer/consumer.py new file mode 100755 index 0000000000..b5c6a4645f --- /dev/null +++ b/examples/camera_streamer/cuda_ipc_consumer/consumer.py @@ -0,0 +1,737 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Example CUDA IPC consumer for camera_receiver AVS1 streams.""" + +from __future__ import annotations + +import argparse +import base64 +import csv +import json +import select +import signal +import socket +import sys +import time +from contextlib import contextmanager +from dataclasses import dataclass +from pathlib import Path +from typing import Optional + + +# Support direct execution without installing the example as a package. +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "camera_receiver")) +from slot_guard import read_slot # noqa: E402 + + +@dataclass +class IpcSlot: + stream: str + slot: int + array: object + owner: "IpcMemory" + guard_path: str + gpu_id: int + + +class IpcMemory: + def __init__(self, ptr: int) -> None: + self.ptr = ptr + self.closed = False + + def close(self) -> None: + if self.closed: + return + import cupy as cp + + cp.cuda.runtime.ipcCloseMemHandle(self.ptr) + self.closed = True + + def __del__(self) -> None: + try: + self.close() + except Exception: + # CUDA/Python modules may already be torn down during finalization. + pass + + +class CudaIpcClient: + def __init__(self, socket_path: str, streams: Optional[list[str]] = None) -> None: + self.socket_path = socket_path + self.requested_streams = list(streams or []) + self.sock: Optional[socket.socket] = None + self.recv_buf = bytearray() + self.slots: dict[tuple[str, int], IpcSlot] = {} + self.stream_order: list[str] = [] + self.stream_info: dict[str, dict] = {} + + def connect(self) -> dict: + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self.sock = sock + try: + sock.connect(self.socket_path) + hello = self._read_json_blocking() + if hello is None or hello.get("type") != "hello": + raise RuntimeError(f"expected hello message, got {hello!r}") + if hello.get("version") != 3: + raise RuntimeError( + "CUDA IPC requires protocol 3 slot read locks; update the receiver" + ) + self._send_json( + {"type": "subscribe", "version": 3, "streams": self.requested_streams} + ) + subscribed = self._read_json_blocking() + if subscribed is None: + raise RuntimeError("publisher disconnected before subscription") + if subscribed.get("type") == "error": + raise RuntimeError(str(subscribed.get("error", "subscription failed"))) + if subscribed.get("type") != "subscribed": + raise RuntimeError(f"expected subscribed message, got {subscribed!r}") + self._open_streams(subscribed) + sock.setblocking(False) + return {"hello": hello, "subscribed": subscribed} + except Exception: + self.close() + raise + + def close(self) -> None: + if self.sock is not None: + try: + self.sock.close() + except OSError: + # A socket close failure must not prevent releasing CUDA mappings. + pass + self.sock = None + for slot in self.slots.values(): + slot.owner.close() + self.slots.clear() + self.recv_buf.clear() + self.stream_order.clear() + self.stream_info.clear() + + @contextmanager + def read_frame(self, msg): + """Hold ownership until all GPU reads finish; yield None for stale frames.""" + import cupy as cp + + slot = self.slots.get((str(msg["stream"]), int(msg["slot"]))) + if slot is None: + yield None + return + with read_slot(slot.guard_path, int(msg["slot_generation"])) as valid: + if not valid: + yield None + return + with cp.cuda.Device(slot.gpu_id): + try: + yield slot.array + finally: + cp.cuda.get_current_stream().synchronize() + + def read_messages(self, timeout_s: float = 0.05) -> Optional[list[dict]]: + """Return all currently available messages, or None on socket EOF.""" + if self.sock is None: + return None + + messages = self._pop_complete_messages() + if messages: + return messages + + readable, _, _ = select.select([self.sock], [], [], timeout_s) + if not readable: + return [] + + while True: + try: + chunk = self.sock.recv(65536) + except BlockingIOError: + break + if not chunk: + return None + self.recv_buf.extend(chunk) + return self._pop_complete_messages() + + def _send_json(self, msg: dict) -> None: + assert self.sock is not None + self.sock.sendall( + json.dumps(msg, separators=(",", ":")).encode("utf-8") + b"\n" + ) + + def _read_json_blocking(self) -> Optional[dict]: + assert self.sock is not None + while b"\n" not in self.recv_buf: + chunk = self.sock.recv(65536) + if not chunk: + return None + self.recv_buf.extend(chunk) + line = self._pop_line() + return json.loads(line.decode("utf-8")) + + def _pop_complete_messages(self) -> list[dict]: + messages: list[dict] = [] + while b"\n" in self.recv_buf: + line = self._pop_line() + messages.append(json.loads(line.decode("utf-8"))) + return messages + + def _pop_line(self) -> bytes: + idx = self.recv_buf.index(ord("\n")) + line = bytes(self.recv_buf[:idx]) + del self.recv_buf[: idx + 1] + return line + + def _open_streams(self, subscribed: dict) -> None: + import cupy as cp + + for stream in subscribed.get("streams", []): + name = str(stream["name"]) + gpu_id = int(stream.get("gpu_id", 0)) + if name not in self.stream_order: + self.stream_order.append(name) + self.stream_info[name] = dict(stream) + with cp.cuda.Device(gpu_id): + for slot in stream["slots"]: + handle = base64.b64decode(slot["handle"]) + ptr = cp.cuda.runtime.ipcOpenMemHandle( + handle, cp.cuda.runtime.cudaIpcMemLazyEnablePeerAccess + ) + owner = IpcMemory(ptr) + try: + mem = cp.cuda.UnownedMemory(ptr, int(slot["nbytes"]), owner) + memptr = cp.cuda.MemoryPointer(mem, 0) + arr = cp.ndarray( + tuple(slot["shape"]), + dtype=cp.uint8, + memptr=memptr, + strides=tuple(slot["strides"]), + ) + self.slots[(name, int(slot["slot"]))] = IpcSlot( + stream=name, + slot=int(slot["slot"]), + array=arr, + owner=owner, + guard_path=str(slot["guard_path"]), + gpu_id=gpu_id, + ) + except Exception: + # A mapping is owned before its array can be registered in slots. + owner.close() + raise + + +def _to_bgr_cpu(array, stream_info: dict): + import cupy as cp + import cv2 + + pixel_format = str(stream_info.get("pixel_format", "rgba8")).lower() + img = cp.asnumpy(array) + if pixel_format == "nv12": + return cv2.cvtColor(img, cv2.COLOR_YUV2BGR_NV12) + if pixel_format == "rgba8": + return cv2.cvtColor(img, cv2.COLOR_RGBA2BGR) + raise ValueError(f"unsupported IPC pixel_format {pixel_format!r}") + + +def _resize_for_display(img, scale: float): + if scale == 1.0: + return img + import cv2 + + return cv2.resize(img, None, fx=scale, fy=scale, interpolation=cv2.INTER_AREA) + + +def _compose_canvas(images: list[object]): + if not images: + return None + if len(images) == 1: + return images[0] + import cv2 + + max_h = max(int(img.shape[0]) for img in images) + padded = [] + for img in images: + h = int(img.shape[0]) + if h < max_h: + img = cv2.copyMakeBorder( + img, + 0, + max_h - h, + 0, + 0, + cv2.BORDER_CONSTANT, + value=(0, 0, 0), + ) + padded.append(img) + return cv2.hconcat(padded) + + +def _format_ns(value: object, signed: bool = False) -> str: + if value is None: + return "n/a" + try: + ns = int(value) + except (TypeError, ValueError): + return "n/a" + return f"{ns:+d} ns" if signed else f"{ns} ns" + + +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 _ms_delta(later_ns: object, earlier_ns: object) -> Optional[float]: + if later_ns is None or earlier_ns is None: + return None + try: + return (int(later_ns) - int(earlier_ns)) / 1.0e6 + except (TypeError, ValueError): + return None + + +def _format_ms(value: object) -> str: + if value is None: + return "n/a" + try: + return f"{float(value):.2f} ms" + except (TypeError, ValueError): + return "n/a" + + +def _capture_latency_status(metadata: dict) -> str: + if metadata.get("sender_capture_ready_to_consumer_ms") is not None: + return _format_ms(metadata.get("sender_capture_ready_to_consumer_ms")) + source_format = metadata.get("source_format") + sender_clock = metadata.get("sender_wall_clock") + consumer_clock = metadata.get("consumer_clock") + if source_format == "avs1" and metadata.get("sender_capture_ready_ns") is None: + return "n/a(no_sender_capture_ts)" + if sender_clock != consumer_clock: + return f"n/a(clock:{sender_clock}->{consumer_clock})" + return "n/a" + + +def _annotate_consumer_receive(msg: dict) -> dict: + metadata = msg.get("metadata") + if not isinstance(metadata, dict): + metadata = {} + else: + metadata = dict(metadata) + consumer_recv_mono_ns = time.monotonic_ns() + consumer_recv_ns, consumer_clock = _read_wall_clock_ns() + metadata["consumer_clock"] = consumer_clock + metadata["consumer_ipc_msg_recv_ns"] = consumer_recv_ns + metadata["consumer_ipc_msg_recv_mono_ns"] = consumer_recv_mono_ns + + if metadata.get("sender_wall_clock") == consumer_clock: + for source_key, output_key in ( + ("sender_capture_ready_ns", "sender_capture_ready_to_consumer_ms"), + ("sender_process_submit_ns", "sender_process_submit_to_consumer_ms"), + ("sender_process_done_ns", "sender_process_done_to_consumer_ms"), + ("sender_tcp_enqueue_ns", "sender_tcp_enqueue_to_consumer_ms"), + ("sof_wall_clock_ns", "sof_to_consumer_ms"), + ): + metadata[output_key] = _ms_delta(consumer_recv_ns, metadata.get(source_key)) + + if metadata.get("receiver_clock") == consumer_clock: + metadata["receiver_rx_to_consumer_ms"] = _ms_delta( + consumer_recv_ns, metadata.get("receiver_rx_payload_done_ns") + ) + metadata["receiver_ipc_publish_to_consumer_ms"] = _ms_delta( + consumer_recv_ns, metadata.get("receiver_ipc_publish_ns") + ) + + if metadata.get("receiver_rx_payload_done_mono_ns") is not None: + metadata["receiver_rx_to_consumer_mono_ms"] = _ms_delta( + consumer_recv_mono_ns, metadata.get("receiver_rx_payload_done_mono_ns") + ) + if metadata.get("receiver_ipc_publish_mono_ns") is not None: + metadata["receiver_ipc_publish_to_consumer_mono_ms"] = _ms_delta( + consumer_recv_mono_ns, metadata.get("receiver_ipc_publish_mono_ns") + ) + msg["metadata"] = metadata + return msg + + +def _latest_metadata( + latest_metadata_by_stream: dict[str, dict], stream_order: list[str] +) -> Optional[dict]: + for stream in reversed(stream_order): + metadata = latest_metadata_by_stream.get(stream) + if metadata: + return metadata + return None + + +def _format_summary_metadata(metadata: Optional[dict]) -> str: + if not metadata: + return "" + fields = [] + stream_name = metadata.get("stream_name") + if stream_name is not None: + fields.append(f"stream={stream_name}") + sender_sequence = metadata.get("sender_sequence") + if sender_sequence is not None: + fields.append(f"sender_seq={sender_sequence}") + if "tsc_delta_ns" in metadata: + fields.append( + f"sof_delta={_format_ns(metadata.get('tsc_delta_ns'), signed=True)}" + ) + if sender_sequence is not None or metadata.get("source_format") == "avs1": + fields.append(f"capture_to_consumer={_capture_latency_status(metadata)}") + latency_fields = ( + ("sender_tcp_enqueue_to_consumer_ms", "txq_to_consumer"), + ("receiver_rx_to_consumer_mono_ms", "rx_to_consumer"), + ("receiver_ipc_publish_to_consumer_mono_ms", "ipc_to_consumer"), + ("sof_to_consumer_ms", "sof_to_consumer"), + ) + for key, label in latency_fields: + if metadata.get(key) is not None: + fields.append(f"{label}={_format_ms(metadata.get(key))}") + return ", ".join(fields) + + +def _format_stream_metadata(metadata: Optional[dict]) -> str: + if not metadata: + return "" + fields = [] + sender_sequence = metadata.get("sender_sequence") + if sender_sequence is not None: + fields.append(f"sender_seq={sender_sequence}") + role = metadata.get("role") + if role is not None: + fields.append(f"role={role}") + fields.append(f"sof_tsc={_format_ns(metadata.get('sof_tsc_ns'))}") + if "tsc_delta_ns" in metadata: + fields.append( + f"sof_delta={_format_ns(metadata.get('tsc_delta_ns'), signed=True)}" + ) + if sender_sequence is not None or metadata.get("source_format") == "avs1": + fields.append(f"capture_to_consumer={_capture_latency_status(metadata)}") + if metadata.get("sender_tcp_enqueue_to_consumer_ms") is not None: + fields.append( + "txq_to_consumer=" + f"{_format_ms(metadata.get('sender_tcp_enqueue_to_consumer_ms'))}" + ) + if metadata.get("receiver_rx_to_consumer_mono_ms") is not None: + fields.append( + "rx_to_consumer=" + f"{_format_ms(metadata.get('receiver_rx_to_consumer_mono_ms'))}" + ) + return " " + " ".join(fields) + + +_LATENCY_CSV_FIELDS = [ + "consumer_ipc_msg_recv_ns", + "consumer_clock", + "consumer_ipc_msg_recv_mono_ns", + "stream", + "slot", + "slot_generation", + "ipc_seq", + "source_format", + "avs1_version", + "avs1_flags", + "session_id", + "bundle_sequence", + "stream_index", + "stream_name", + "stream_sequence", + "stream_stereo", + "stream_codec", + "stream_rotation", + "camera_index", + "role", + "device_index", + "sensor_mode_index", + "frame_number", + "width", + "height", + "source_pixel_format", + "ipc_pixel_format", + "sender_wall_clock", + "sof_tsc_ns", + "tsc_delta_ns", + "frame_timestamp_ns", + "left_sof_tsc_ns", + "right_sof_tsc_ns", + "left_frame_timestamp_ns", + "right_frame_timestamp_ns", + "sender_capture_ready_ns", + "sender_process_submit_ns", + "sender_process_done_ns", + "sender_tcp_enqueue_ns", + "receiver_clock", + "receiver_rx_prefix_done_ns", + "receiver_rx_payload_done_ns", + "receiver_process_submit_ns", + "receiver_process_done_ns", + "receiver_ipc_publish_ns", + "receiver_rx_prefix_done_mono_ns", + "receiver_rx_payload_done_mono_ns", + "receiver_process_submit_mono_ns", + "receiver_process_done_mono_ns", + "receiver_ipc_publish_mono_ns", + "rx_to_process_submit_ms", + "rx_to_process_done_ms", + "rx_to_ipc_publish_ms", + "sender_capture_ready_to_rx_ms", + "sender_process_submit_to_rx_ms", + "sender_process_done_to_rx_ms", + "sender_tcp_enqueue_to_rx_ms", + "sender_tcp_enqueue_to_ipc_publish_ms", + "sender_capture_ready_to_ipc_publish_ms", + "sender_capture_ready_to_consumer_ms", + "sender_process_submit_to_consumer_ms", + "sender_process_done_to_consumer_ms", + "sender_tcp_enqueue_to_consumer_ms", + "receiver_rx_to_consumer_ms", + "receiver_ipc_publish_to_consumer_ms", + "receiver_rx_to_consumer_mono_ms", + "receiver_ipc_publish_to_consumer_mono_ms", + "sof_to_consumer_ms", +] + + +def _csv_row(msg: dict) -> dict[str, object]: + metadata = msg.get("metadata") if isinstance(msg.get("metadata"), dict) else {} + row = {field: metadata.get(field, "") for field in _LATENCY_CSV_FIELDS} + row["stream"] = msg.get("stream", "") + row["slot"] = msg.get("slot", "") + row["slot_generation"] = msg.get("slot_generation", "") + row["ipc_seq"] = msg.get("seq", "") + return row + + +def _parse_stream_filter(value: Optional[str]) -> list[str]: + if not value: + return [] + return [part.strip() for part in value.split(",") if part.strip()] + + +def run(args: argparse.Namespace) -> int: + import cupy as cp + + requested_streams = _parse_stream_filter(args.streams) + client = CudaIpcClient(args.socket, requested_streams) + stop = False + + def on_sigint(_signum, _frame): + nonlocal stop + stop = True + + signal.signal(signal.SIGINT, on_sigint) + + handshake = client.connect() + subscribed = handshake["subscribed"] + print( + f"connected to {args.socket}: " + + ", ".join( + f"{s['name']} {s['width']}x{s['height']} " + f"{s.get('pixel_format', 'unknown')} slots={len(s['slots'])}" + for s in subscribed.get("streams", []) + ), + flush=True, + ) + + cv2 = None + if not args.no_display: + try: + import cv2 as _cv2 + + cv2 = _cv2 + except ImportError: + print( + "OpenCV not available; falling back to --no-display stats", flush=True + ) + + latest_cpu: dict[str, object] = {} + latest_metadata_by_stream: dict[str, dict] = {} + latest_generation_by_slot: dict[tuple[str, int], int] = {} + counts: dict[str, int] = dict.fromkeys(client.stream_order, 0) + stats_interval = max(1, args.stats_every) + last_report = time.monotonic() + last_counts = dict(counts) + frames_seen = 0 + overwritten = 0 + csv_file = None + csv_writer = None + if args.latency_csv: + csv_file = open(args.latency_csv, "w", newline="", encoding="utf-8") + csv_writer = csv.DictWriter(csv_file, fieldnames=_LATENCY_CSV_FIELDS) + csv_writer.writeheader() + + try: + if cv2 is not None: + cv2.namedWindow(args.window_name, cv2.WINDOW_NORMAL) + print( + "Close a preview window, press 'q' or Esc in the preview, or Ctrl+C in this terminal to quit.", + flush=True, + ) + while not stop: + if cv2 is not None: + # Process window-close events before displaying another frame. + if cv2.waitKey(1) & 0xFF in (ord("q"), 27): + break + try: + if ( + cv2.getWindowProperty(args.window_name, cv2.WND_PROP_VISIBLE) + < 1 + ): + break + except cv2.error: + break + messages = client.read_messages() + if messages is None: + print("publisher disconnected", flush=True) + break + latest_msg_by_stream: dict[str, dict] = {} + sampled_messages: list[dict] = [] + for msg in messages: + if msg.get("type") != "frame": + continue + msg = _annotate_consumer_receive(msg) + stream = str(msg["stream"]) + slot_id = int(msg["slot"]) + generation = int(msg.get("slot_generation", 0)) + key = (stream, slot_id) + previous = latest_generation_by_slot.get(key) + if previous is not None and generation > previous + 1: + overwritten += generation - previous - 1 + latest_generation_by_slot[key] = generation + counts[stream] = counts.get(stream, 0) + 1 + frames_seen += 1 + latest_msg_by_stream[stream] = msg + if cv2 is None and counts[stream] % stats_interval == 0: + sampled_messages.append(msg) + metadata = msg.get("metadata") + if isinstance(metadata, dict): + latest_metadata_by_stream[stream] = metadata + if csv_writer is not None: + csv_writer.writerow(_csv_row(msg)) + + updated_display = False + for stream in client.stream_order: + msg = latest_msg_by_stream.get(stream) + if msg is None: + continue + slot_id = int(msg["slot"]) + slot = client.slots.get((stream, slot_id)) + if slot is None: + print(f"unknown frame slot: {stream}[{slot_id}]", flush=True) + continue + if cv2 is not None: + with client.read_frame(msg) as array: + if array is None: + overwritten += 1 + continue + latest_cpu[stream] = _to_bgr_cpu( + array, client.stream_info.get(stream, {}) + ) + updated_display = True + + if cv2 is None: + for msg in sampled_messages: + stream = str(msg["stream"]) + slot_id = int(msg["slot"]) + slot = client.slots.get((stream, slot_id)) + if slot is None: + print(f"unknown frame slot: {stream}[{slot_id}]", flush=True) + continue + metadata_text = _format_stream_metadata(msg.get("metadata")) + mean_text = "" + if args.sample_mean: + with client.read_frame(msg) as array: + if array is None: + overwritten += 1 + continue + mean = float(cp.mean(array).get()) + mean_text = f" mean={mean:.2f}" + print( + f"{stream}[{slot_id}] gen={msg.get('slot_generation')} " + f"seq={msg.get('seq')}{mean_text}{metadata_text}", + flush=True, + ) + + if cv2 is not None: + if updated_display: + images = [ + _resize_for_display(latest_cpu[name], args.scale) + for name in client.stream_order + if name in latest_cpu + ] + if images: + canvas = _compose_canvas(images) + if canvas is not None: + cv2.imshow(args.window_name, canvas) + + now = time.monotonic() + if now - last_report >= 1.0: + rates = [] + for name in client.stream_order: + current = counts.get(name, 0) + prev = last_counts.get(name, 0) + rates.append(f"{name}: {current - prev} fps") + last_counts[name] = current + if overwritten: + rates.append(f"overwritten_slots={overwritten}") + overwritten = 0 + metadata_text = _format_summary_metadata( + _latest_metadata(latest_metadata_by_stream, client.stream_order) + ) + if metadata_text: + rates.append(metadata_text) + print(", ".join(rates), flush=True) + last_report = now + if args.frames > 0 and frames_seen >= args.frames: + break + finally: + if csv_file is not None: + csv_file.close() + if cv2 is not None: + cv2.destroyAllWindows() + client.close() + return 0 + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument( + "--socket", + default="/tmp/camera_receiver_avs1_cuda_ipc.sock", + help="Unix socket path exposed by camera_receiver ipc.socket_path", + ) + ap.add_argument( + "--streams", + help="comma-separated IPC stream names to subscribe to; default subscribes to all", + ) + ap.add_argument( + "--scale", + type=float, + default=1.0, + help="OpenCV display scale (default: 1.0, full resolution)", + ) + ap.add_argument("--window-name", default="camera_receiver CUDA IPC") + ap.add_argument("--no-display", action="store_true", help="print stats only") + ap.add_argument( + "--stats-every", + type=int, + default=60, + help="print one detailed sample per stream every N frames (default: 60)", + ) + ap.add_argument( + "--sample-mean", + action="store_true", + help="compute a GPU mean for occasional no-display frame samples", + ) + ap.add_argument("--frames", type=int, default=0, help="exit after this many frames") + ap.add_argument( + "--latency-csv", + help="write one CSV row per received IPC frame notification", + ) + return run(ap.parse_args()) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/examples/camera_streamer/cuda_ipc_consumer/requirements.txt b/examples/camera_streamer/cuda_ipc_consumer/requirements.txt new file mode 100644 index 0000000000..814bee05b9 --- /dev/null +++ b/examples/camera_streamer/cuda_ipc_consumer/requirements.txt @@ -0,0 +1,6 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +numpy>=1.26,<2 +cupy-cuda12x>=13.0,<14 +opencv-python>=4.9,<4.12 diff --git a/examples/camera_streamer/pyproject.toml b/examples/camera_streamer/pyproject.toml new file mode 100644 index 0000000000..4639ce6b23 --- /dev/null +++ b/examples/camera_streamer/pyproject.toml @@ -0,0 +1,20 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "isaac-camera-streamer-example" +version = "0.1.0" +requires-python = ">=3.10" +dependencies = ["numpy>=1.24,<2", "PyYAML>=6", "opencv-python>=4.8"] + +[project.optional-dependencies] +receiver = ["cupy-cuda12x>=13,<14", "pybind11>=2.12", "ninja>=1.11"] +test = ["pytest>=8"] + +[tool.setuptools] +packages = ["camera_streamer"] +package-dir = {camera_streamer = "."} diff --git a/examples/camera_streamer/tests/python/test_consumer_cleanup.py b/examples/camera_streamer/tests/python/test_consumer_cleanup.py new file mode 100644 index 0000000000..4f74b2492e --- /dev/null +++ b/examples/camera_streamer/tests/python/test_consumer_cleanup.py @@ -0,0 +1,114 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +import base64 +import contextlib +import sys +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "cuda_ipc_consumer")) +from consumer import CudaIpcClient # noqa: E402 + + +@pytest.mark.parametrize("failure", ["second_handle", "array"]) +def test_connect_failure_closes_socket_and_partial_mappings(monkeypatch, failure): + import consumer + + runtime = SimpleNamespace( + cudaIpcMemLazyEnablePeerAccess=1, + ipcOpenMemHandle=Mock(side_effect=[123, RuntimeError("mapping failed")]), + ipcCloseMemHandle=Mock(), + ) + cp = SimpleNamespace( + cuda=SimpleNamespace( + runtime=runtime, + Device=lambda _: contextlib.nullcontext(), + UnownedMemory=Mock(), + MemoryPointer=Mock(), + ), + uint8=object(), + ndarray=Mock( + side_effect=RuntimeError("array failed") if failure == "array" else None + ), + ) + sock = Mock() + monkeypatch.setitem(sys.modules, "cupy", cp) + monkeypatch.setattr(consumer.socket, "socket", Mock(return_value=sock)) + slots = [ + dict( + slot=i, + guard_path="/unused", + handle=base64.b64encode(b"handle").decode(), + nbytes=4, + shape=[2, 2], + strides=[2, 1], + ) + for i in range(2) + ] + client = CudaIpcClient("/unused.sock") + client._read_json_blocking = Mock( + side_effect=[ + {"type": "hello", "version": 3}, + {"type": "subscribed", "streams": [{"name": "head.left", "slots": slots}]}, + ] + ) + with pytest.raises(RuntimeError, match="failed"): + client.connect() + runtime.ipcCloseMemHandle.assert_called_once_with(123) + sock.close.assert_called_once() + assert client.sock is None + assert not client.slots and not client.stream_order and not client.stream_info + client.close() + runtime.ipcCloseMemHandle.assert_called_once_with(123) + + +@pytest.mark.parametrize("fails", [False, True]) +def test_read_frame_holds_lock_through_gpu_completion(monkeypatch, tmp_path, fails): + import fcntl + import consumer + from slot_guard import commit_slot + + path = tmp_path / "guard" + path.write_bytes(bytes(8)) + commit_slot(path, 7) + synced = [] + + def sync(): + with path.open("rb") as probe: + with pytest.raises(BlockingIOError): + fcntl.flock(probe, fcntl.LOCK_EX | fcntl.LOCK_NB) + synced.append(True) + + monkeypatch.setitem( + sys.modules, + "cupy", + SimpleNamespace( + cuda=SimpleNamespace( + Device=lambda _: contextlib.nullcontext(), + get_current_stream=lambda: SimpleNamespace(synchronize=sync), + ) + ), + ) + client = consumer.CudaIpcClient("unused") + array = object() + client.slots[("eye", 0)] = SimpleNamespace( + array=array, guard_path=str(path), gpu_id=0 + ) + msg = dict(stream="eye", slot=0, slot_generation=7) + try: + with client.read_frame(msg) as actual: + assert actual is array + if fails: + raise RuntimeError("conversion failed") + except RuntimeError: + assert fails + assert synced == [True] + with path.open("rb") as probe: + fcntl.flock(probe, fcntl.LOCK_EX | fcntl.LOCK_NB) + msg["slot_generation"] = 6 + with client.read_frame(msg) as actual: + assert actual is None + assert synced == [True]