diff --git a/CMakeLists.txt b/CMakeLists.txt index 4a90a6ce9e..96c996c4c9 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -71,7 +71,7 @@ option(BUILD_PLUGIN_NOITOM_MOCAP "Build Noitom mocap plugin (downloads MocapApi option(BUILD_PLUGIN_OGLO "Build OGLO tactile glove plugin (BLE, Linux only; fetches SimpleBLE + nlohmann/json)" OFF) option(BUILD_PLUGIN_WUJI_GLOVE "Build Wuji glove plugin (requires the wuji_sdk C SDK)" OFF) option(BUILD_PLUGIN_SHARPA_AVATAR "Build Sharpa AvatarExo Glove plugin (Linux; skips if the Avatar SDK is missing)" OFF) -option(BUILD_PLUGIN_SENSING "Build SENSING GMSL camera plugin (Jetson only; requires SIPL, NVENC and the LiveKit SDK)" OFF) +option(BUILD_PLUGIN_SENSING "Build SENSING GMSL camera plugin (Jetson only; requires SIPL and NVENC)" OFF) option(BUILD_VIDEO_STAGE "Build the video stage: the Orin-to-PC video transport and the PC relay" OFF) option(BUILD_EXAMPLES "Build examples" ON) option(BUILD_EXAMPLE_TELEOP_ROS2 "Build only the teleop_ros2 ROS 2 reference integration (e.g. for Docker)" OFF) diff --git a/docs/source/device/camera/sensing.rst b/docs/source/device/camera/sensing.rst index 24a375482b..adec05d4d1 100644 --- a/docs/source/device/camera/sensing.rst +++ b/docs/source/device/camera/sensing.rst @@ -156,17 +156,18 @@ on the side that fails for a read-only runtime report: - Check the Jetson power mode. Maximum performance can be selected with ``sudo nvpmodel -m 0 && sudo jetson_clocks``. -Build and run the LiveKit publisher ------------------------------------ +Build and run the camera producer +--------------------------------- ``camera_plugin_sensing`` runs on the Orin. It captures the GMSL cameras with -SIPL, encodes H.264 on the hardware encoder and publishes one LiveKit video -track per camera. CaptureServer, the egocap backend, recording and the browser -run on the PC. +SIPL, encodes H.264 on the hardware encoder and sends each camera's video over a +direct TCP link to the PC video stage (``video_stage_worker``), which publishes it +to LiveKit. The producer has no LiveKit client of its own. CaptureServer, the +egocap backend, recording and the browser run on the PC. Prerequisites: an AGX Orin on JetPack 7.2.1 / L4T R39.2.1 with the SENSING -drivers installed and the cameras working, CUDA, CMake, and network access to -the LiveKit server. The build also needs two header trees; neither is fetched +drivers installed and the cameras working, CUDA, CMake, and a network path to the +PC running ``video_stage_worker``. The build also needs two header trees; neither is fetched by ``setup.sh``: - ``JETSON_MMAPI_DIR``: the Jetson Multimedia API tree, containing @@ -179,9 +180,7 @@ by ``setup.sh``: .. code-block:: bash - # Build. The LiveKit SDK v1.12.4 prebuilt is downloaded and hash-checked; use - # -DLIVEKIT_SDK_DIR=... for a local copy (it needs LICENSE and NOTICE files in that - # directory, or -DLIVEKIT_SDK_LICENSE_FILE / -DLIVEKIT_SDK_NOTICE_FILE). + # Build. cmake -B build -DBUILD_PLUGIN_SENSING=ON -DBUILD_VIZ=OFF -DBUILD_EXAMPLES=OFF \ -DBUILD_TESTING=OFF -DBUILD_PYTHON_BINDINGS=OFF \ -DSIPL_API_DIR=... -DJETSON_MMAPI_DIR=... @@ -190,16 +189,16 @@ by ``setup.sh``: # Check the cameras are visible. build/src/plugins/sensing/app/camera_plugin_sensing --list-sensors - # Publish. The token file is a LiveKit token for the room and publisher identity - # named in the PC's streams file. + # Send. Each stream's direct= is the PC address and port its video_stage_worker + # --stream=PORT:TRACK listens on; start the worker first. build/src/plugins/sensing/app/camera_plugin_sensing \ - --livekit-url=ws://LIVEKIT_HOST:7880 --livekit-token-file=join.token \ - --add-stream=sensor=0,livekit=sensing-left --add-stream=sensor=1,livekit=sensing-right + --add-stream=sensor=0,name=sensing-left,direct=PC_HOST:5600 \ + --add-stream=sensor=1,name=sensing-right,direct=PC_HOST:5601 -The ``livekit=`` names must equal the ``track`` names in the PC's streams file. -Stop with Ctrl-C. ``--help`` lists every option. The example uses plain ``ws://`` -for a LiveKit server on a trusted LAN; the join token is sent in the clear, so -use a ``wss://`` endpoint on any other network. +The ``direct=`` ports must match the ``--stream`` ports of the PC's +``video_stage_worker``, whose ``TRACK`` names must equal the ``track`` names in +the PC's streams file. Stop with Ctrl-C. ``--help`` lists every option. The direct +link is plain TCP: use a wired link or a trusted network. ``SENSING_TSC_PROBE=1`` logs the SIPL capture counter beside the host clocks twice a second, as separate uncalibrated samples. The counter's units are unconfirmed, so diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 1ea2f91e2b..e7edca168b 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -25,17 +25,15 @@ if(BUILD_VIZ) add_subdirectory(viz) endif() -# The LiveKit SDK and publisher library: the PC relay publishes to LiveKit; the camera producer still does until it sends -# direct to the relay. -if(BUILD_VIDEO_STAGE OR BUILD_PLUGIN_SENSING) - add_subdirectory(livekit_publisher) -endif() - +# The LiveKit SDK and publisher library: the PC relay publishes to LiveKit. The camera producer on the Orin does not use +# LiveKit; it sends over its direct links. if(BUILD_VIDEO_STAGE) + add_subdirectory(livekit_publisher) add_subdirectory(video_stage) elseif(BUILD_PLUGIN_SENSING) - # The LiveKit library uses the record codec and clock-exchange helpers. + # The camera producer sends AVS1 records to the video stage and needs only its record codec and sender. add_subdirectory(video_stage/wire) + add_subdirectory(video_stage/sender) endif() if(BUILD_PLUGINS) diff --git a/src/plugins/sensing/CMakeLists.txt b/src/plugins/sensing/CMakeLists.txt index 2e47a2e159..2f944156fb 100644 --- a/src/plugins/sensing/CMakeLists.txt +++ b/src/plugins/sensing/CMakeLists.txt @@ -2,9 +2,9 @@ # SPDX-License-Identifier: Apache-2.0 # ============================================================================== -# SENSING Camera Plugin: Jetson SIPL capture, hardware H.264 encode, LiveKit publication +# SENSING Camera Plugin: Jetson SIPL capture, hardware H.264 encode, direct streaming to the PC video stage # ============================================================================== -# Two leaf targets, one per directory (cmake/cmake-structure.md rule 1); the LiveKit wrapper is src/livekit_publisher: +# Two leaf targets, one per directory (cmake/cmake-structure.md rule 1): # core/ sensing_core -- SIPL capture, encoder, conversion, track output # app/ camera_plugin_sensing # diff --git a/src/plugins/sensing/README.md b/src/plugins/sensing/README.md index 1906b555ab..0f3057586c 100644 --- a/src/plugins/sensing/README.md +++ b/src/plugins/sensing/README.md @@ -39,11 +39,11 @@ before replacing it; do not hand-edit the board wiring. Its CSI ports, I2C buses, deserializer addresses, GPIOs, and link modes must match the installed SIPL drivers. -## Orin: LiveKit publisher +## Orin: camera producer -`camera_plugin_sensing` (built from [`app/`](app/), [`core/`](core/) and -[`livekit/`](livekit/)) captures the cameras with SIPL, encodes H.264 on the -hardware encoder and publishes one LiveKit video track per camera. The +`camera_plugin_sensing` (built from [`app/`](app/) and [`core/`](core/)) captures the cameras with SIPL, encodes H.264 +on the hardware encoder and sends each camera's video over a direct TCP link to the PC video stage (`src/video_stage`), +which publishes it to LiveKit. The producer has no LiveKit client of its own. The [SENSING user guide](https://nvidia.github.io/IsaacCapture/main/device/camera/sensing.html) has the build and run steps. @@ -53,4 +53,4 @@ guessing). SIPL is built by default (`-DSENSING_WITH_SIPL=OFF` for an Argus-only (`-DSENSING_WITH_ARGUS=ON`, Argus device index with `--add-stream=sensor=N,...`, `--argus-mode=K`) captures NV12 into DMA-BUFs that the hardware encoder reads without a copy. **The Argus code has never been built or run: there is no Argus SDK or Argus camera available to the author.** Argus DMA-BUF input cannot be combined with -`--livekit-scale-div` above 1. An Argus-only build (`-DSENSING_WITH_SIPL=OFF -DSENSING_WITH_ARGUS=ON`) also builds on JetPack 6.2 / L4T R36.4.3. +`--scale-div` above 1. An Argus-only build (`-DSENSING_WITH_SIPL=OFF -DSENSING_WITH_ARGUS=ON`) also builds on JetPack 6.2 / L4T R36.4.3. diff --git a/src/plugins/sensing/app/CMakeLists.txt b/src/plugins/sensing/app/CMakeLists.txt index 613d396602..a414ba79ff 100644 --- a/src/plugins/sensing/app/CMakeLists.txt +++ b/src/plugins/sensing/app/CMakeLists.txt @@ -24,16 +24,6 @@ install(DIRECTORY "${SENSING_CONFIGS_DIR}/" DESTINATION plugins/sensing_camera/configs ) -# The plugin links the LiveKit runtime, whose RUNPATH points into the build tree; ship it beside the -# binary, where INSTALL_RPATH=$ORIGIN (and liblivekit.so's own $ORIGIN) finds it. -install(FILES - "${SENSING_LIVEKIT_SDK_DIR}/lib/liblivekit.so" - "${SENSING_LIVEKIT_SDK_DIR}/lib/liblivekit_ffi.so" - DESTINATION plugins/sensing_camera -) -install(FILES "${SENSING_LIVEKIT_LICENSE_FILE}" "${SENSING_LIVEKIT_NOTICE_FILE}" - DESTINATION plugins/sensing_camera/licenses/livekit-sdk -) # The vendored encoder classes are compiled into the binary, so their license text ships with it. install(FILES "${SENSING_MMAPI_VENDORED_DIR}/LICENSE" DESTINATION plugins/sensing_camera/licenses/jetson-multimedia-api diff --git a/src/plugins/sensing/app/main.cpp b/src/plugins/sensing/app/main.cpp index 530c438dd4..3e32a37a61 100644 --- a/src/plugins/sensing/app/main.cpp +++ b/src/plugins/sensing/app/main.cpp @@ -46,7 +46,7 @@ void signal_handler(int signal) // Argument parsers // ============================================================================= -/// "sensor=N,livekit=NAME" -> one sensor and the track that carries it. +/// "sensor=N,name=NAME[,direct=HOST:PORT]" -> one sensor and the stream that carries it. static StreamConfig parse_stream_arg(const std::string& arg) { StreamConfig cfg{}; @@ -68,9 +68,19 @@ static StreamConfig parse_stream_arg(const std::string& arg) cfg.sensor_id = static_cast(std::stoul(val)); has_sensor = true; } - else if (key == "livekit") + else if (key == "name") { - cfg.livekit_track = val; + // What the PC, the management channel and the logs call this stream. + cfg.name = val; + } + else if (key == "direct") + { + // HOST:PORT of this stream's PC video-stage port; the producer connects and reconnects. + const auto colon = val.rfind(':'); + if (colon == std::string::npos || colon == 0 || colon + 1 == val.size()) + throw std::runtime_error("--add-stream direct= must be HOST:PORT"); + cfg.direct_host = val.substr(0, colon); + cfg.direct_port = static_cast(std::stoul(val.substr(colon + 1))); } else { @@ -78,8 +88,8 @@ static StreamConfig parse_stream_arg(const std::string& arg) } } - if (!has_sensor || cfg.livekit_track.empty()) - throw std::runtime_error("--add-stream requires sensor=,livekit="); + if (!has_sensor || cfg.name.empty()) + throw std::runtime_error("--add-stream requires sensor=,name="); return cfg; } @@ -125,13 +135,15 @@ static std::string default_platform_config() static void print_usage(const char* program_name) { std::cout - << "Usage: " << program_name << " --livekit-url=URL --livekit-token-file=PATH --add-stream=... [options]\n" + << "Usage: " << program_name << " --add-stream=... [options]\n" << "\n" - << "Publishes each SENSING sensor as a LiveKit camera track: SIPL or Argus capture, hardware H.264,\n" - << "sent as it is.\n" + << "Captures each SENSING sensor (SIPL or Argus), encodes it as H.264 on the hardware encoder and sends\n" + << "it over TCP to the PC video stage, which publishes it onwards.\n" << "\n" << "Streams (repeat for each sensor):\n" - << " --add-stream=sensor=N,livekit=NAME camera N (SIPL or Argus device), published as track NAME\n" + << " --add-stream=sensor=N,name=NAME[,direct=HOST:PORT]\n" + << " camera N (SIPL or Argus device), called NAME;\n" + << " direct= is where the PC video stage listens for it\n" << " --list-sensors print the cameras, then exit\n" << " --camera-backend=auto|sipl|argus auto picks the one backend that matches (default);\n" << " (a backend must also be built in)\n" @@ -144,18 +156,18 @@ static void print_usage(const char* program_name) << " --link-masks='M M' one mask per deserializer (default: '0x0000 0x1100')\n" << " --nito-dir=DIR ISP tuning directory (default: /var/nvidia/nvcam/settings/sipl)\n" << "\n" - << "LiveKit:\n" - << " --livekit-url=URL server, e.g. wss://livekit.example.org\n" - << " --livekit-token-file=PATH file holding the join token; it names the room and the\n" - << " participant identity\n" - << " --livekit-fps=N publication rate cap (default: 30)\n" - << " --livekit-scale-div=N reduce every track by this integer divisor (default: 1)\n" + << "Encoding:\n" + << " --fps=N frame rate cap of each stream (default: 30)\n" + << " --scale-div=N reduce every stream by this integer divisor (default: 1)\n" << " --timestamp-basis=B what each frame's timestamp is: monotonic_us (default) is this host's\n" << " monotonic clock when the frame was ready to encode; capture_monotonic_us\n" << " (UNVERIFIED on hardware) is the sensor's capture event on that clock and fails\n" - << " the stream if it cannot be placed; processing_unix_us is Unix processing time.\n" + << " the stream if it cannot be placed. The direct link carries this clock only.\n" << " The streams file on the PC must declare the same basis.\n" - << " --livekit-bitrate=N target bitrate of each track's encoder in bps (default: 8000000)\n" + << " --frame-correlation=C how an encoded unit is matched to its capture: stamp (default) by the stamp the\n" + << " encoder returns, sending a unit that matches none without a capture time; fifo\n" + << " assumes one unit per input in order (approximate; not for recording)\n" + << " --bitrate=N target bitrate of each stream's encoder in bps (default: 8000000)\n" << " --gop=N IDR period in frames (default: fps*5)\n" << "\n" << "Other:\n" @@ -165,9 +177,8 @@ static void print_usage(const char* program_name) << " --swap-uv swap the chroma planes\n" << "\n" << "Example:\n" - << " " << program_name << " --livekit-url=wss://livekit.example.org \\\n" - << " --livekit-token-file=join.token --add-stream=sensor=0,livekit=sensing-left \\\n" - << " --add-stream=sensor=1,livekit=sensing-right\n"; + << " " << program_name << " --add-stream=sensor=0,name=sensing-left,direct=192.168.1.205:5600 \\\n" + << " --add-stream=sensor=1,name=sensing-right,direct=192.168.1.205:5601\n"; } // ============================================================================= @@ -179,7 +190,6 @@ try { SensingConfig camera_config; std::map stream_map; - std::string livekit_token_file; bool list_sensors = false; CameraBackend requested_backend = CameraBackend::Auto; @@ -232,41 +242,45 @@ try } requested_backend = *parsed; } - else if (arg.find("--livekit-url=") == 0) - { - camera_config.livekit.url = arg.substr(14); - } - else if (arg.find("--livekit-token-file=") == 0) + else if (arg.find("--fps=") == 0) { - livekit_token_file = arg.substr(21); - } - else if (arg.find("--livekit-fps=") == 0) - { - camera_config.livekit.fps = static_cast(std::stoul(arg.substr(14))); + camera_config.encode.fps = static_cast(std::stoul(arg.substr(6))); } else if (arg.find("--timestamp-basis=") == 0) { const std::string basis = arg.substr(18); if (basis == "monotonic_us") - camera_config.livekit.timestamp_basis = TimestampBasis::FrameReadyMonotonic; + camera_config.encode.timestamp_basis = TimestampBasis::FrameReadyMonotonic; else if (basis == "capture_monotonic_us") - camera_config.livekit.timestamp_basis = TimestampBasis::SensorCaptureMonotonic; - else if (basis == "processing_unix_us") - camera_config.livekit.timestamp_basis = TimestampBasis::ProcessingUnix; + camera_config.encode.timestamp_basis = TimestampBasis::SensorCaptureMonotonic; + else + { + std::cerr + << "Error: --timestamp-basis must be monotonic_us or capture_monotonic_us: the direct link carries the sender's monotonic clock only." + << std::endl; + return 1; + } + } + else if (arg.find("--frame-correlation=") == 0) + { + const std::string how = arg.substr(20); + if (how == "stamp") + camera_config.encode.correlation = FrameCorrelation::Stamp; + else if (how == "fifo") + camera_config.encode.correlation = FrameCorrelation::Fifo; else { - std::cerr << "Error: --timestamp-basis must be monotonic_us, capture_monotonic_us or processing_unix_us." - << std::endl; + std::cerr << "Error: --frame-correlation must be stamp or fifo." << std::endl; return 1; } } - else if (arg.find("--livekit-bitrate=") == 0) + else if (arg.find("--bitrate=") == 0) { - camera_config.livekit.bitrate_bps = static_cast(std::stoul(arg.substr(18))); + camera_config.encode.bitrate_bps = static_cast(std::stoul(arg.substr(10))); } - else if (arg.find("--livekit-scale-div=") == 0) + else if (arg.find("--scale-div=") == 0) { - camera_config.livekit.scale_div = static_cast(std::stoul(arg.substr(20))); + camera_config.encode.scale_div = static_cast(std::stoul(arg.substr(12))); } else if (arg.find("--gop=") == 0) { @@ -405,22 +419,18 @@ try } // Check the settings that need no hardware or network before anything is opened. - if (camera_config.livekit.url.empty() || livekit_token_file.empty()) - throw std::runtime_error("--livekit-url and --livekit-token-file are required"); - if (camera_config.livekit.fps == 0) - throw std::runtime_error("--livekit-fps must be at least 1"); - if (camera_config.livekit.bitrate_bps < 100'000) - throw std::runtime_error("--livekit-bitrate must be at least 100000"); - - // The token is read from a file so it never appears in a process listing, and is not printed. - std::ifstream token_file(livekit_token_file); - if (!token_file) - throw std::runtime_error("cannot read --livekit-token-file " + livekit_token_file); - std::getline(token_file, camera_config.livekit.token); - const auto last = camera_config.livekit.token.find_last_not_of(" \t\r\n"); - camera_config.livekit.token.erase(last == std::string::npos ? 0 : last + 1); - if (camera_config.livekit.token.empty()) - throw std::runtime_error("--livekit-token-file " + livekit_token_file + " is empty"); + if (camera_config.encode.fps == 0) + throw std::runtime_error("--fps must be at least 1"); + if (camera_config.encode.bitrate_bps < 100'000) + throw std::runtime_error("--bitrate must be at least 100000"); + { + const bool any_direct = std::any_of( + stream_configs.begin(), stream_configs.end(), [](const StreamConfig& s) { return !s.direct_host.empty(); }); + if (!any_direct) + std::cerr << "Warning: no stream has a direct=HOST:PORT destination: nothing will receive the video. " + "Give --add-stream a direct=HOST:PORT." + << std::endl; + } std::signal(SIGINT, signal_handler); std::signal(SIGTERM, signal_handler); diff --git a/src/plugins/sensing/core/CMakeLists.txt b/src/plugins/sensing/core/CMakeLists.txt index e36c7447f9..572e93fe56 100644 --- a/src/plugins/sensing/core/CMakeLists.txt +++ b/src/plugins/sensing/core/CMakeLists.txt @@ -1,14 +1,14 @@ # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -# SIPL capture, colour conversion, the hardware encoder and the LiveKit track output. +# SIPL capture, colour conversion, the hardware encoder and the output that sends encoded video to the PC. add_library(sensing_core STATIC sensing_camera.cpp camera_backend.cpp sensor_clock.cpp capture_factory.cpp jetson_encoder.cpp - livekit_video_output.cpp + video_output.cpp rgba_to_nv12.cu rgba_box_downscale.cu inc/sensing_core/argus_capture.hpp @@ -20,6 +20,7 @@ add_library(sensing_core STATIC inc/sensing_core/sensor_clock.hpp inc/sensing_core/sensing_types.hpp inc/sensing_core/sipl_camera.hpp + inc/sensing_core/video_output.hpp ${JETSON_MMAPI_SOURCES} ) add_library(sensing::core ALIAS sensing_core) @@ -66,7 +67,7 @@ target_compile_options(sensing_core PRIVATE target_link_libraries(sensing_core PUBLIC - sensing::livekit + video_stage::sender PRIVATE CUDA::cudart CUDA::cuda_driver diff --git a/src/plugins/sensing/core/inc/sensing_core/camera_capture.hpp b/src/plugins/sensing/core/inc/sensing_core/camera_capture.hpp index d16dfba181..1aab427e7d 100644 --- a/src/plugins/sensing/core/inc/sensing_core/camera_capture.hpp +++ b/src/plugins/sensing/core/inc/sensing_core/camera_capture.hpp @@ -124,7 +124,7 @@ struct CaptureFrame uint64_t sensor_time = 0; /// What `sensor_time` counts; the claim is checked at run time by SensorClockMap, not trusted. SensorTimeDomain sensor_domain = SensorTimeDomain::None; - /// CLOCK_MONOTONIC ns of the sensor's capture event, set by the LiveKit output when it publishes capture + /// CLOCK_MONOTONIC ns of the sensor's capture event, set by the video output when it stamps capture /// time; zero otherwise. uint64_t capture_monotonic_ns = 0; std::variant pixels; diff --git a/src/plugins/sensing/core/inc/sensing_core/sensing_camera.hpp b/src/plugins/sensing/core/inc/sensing_core/sensing_camera.hpp index d022e09036..8e9a761988 100644 --- a/src/plugins/sensing/core/inc/sensing_core/sensing_camera.hpp +++ b/src/plugins/sensing/core/inc/sensing_core/sensing_camera.hpp @@ -5,9 +5,11 @@ #include "camera_capture.hpp" #include "sensing_types.hpp" +#include "video_output.hpp" #include #include +#include #include namespace plugins @@ -15,15 +17,12 @@ namespace plugins namespace sensing { -class LiveKitSession; -class LiveKitVideoOutput; - /** - * @brief Multi-sensor SENSING camera publisher. + * @brief Multi-sensor SENSING camera producer. * * One CameraCapture owns every camera of the rig (SIPL's camera object is process-wide, so there is no * per-camera capture object to hold). Each update() polls every sensor's latest frame and hands a new frame - * to that sensor's LiveKit track. Geometry is per sensor and comes from the capture backend. + * to that sensor's encoder and direct link to the PC. Geometry is per sensor and comes from the capture backend. */ class SensingCamera { @@ -39,10 +38,10 @@ class SensingCamera SensingCamera(SensingCamera&&) = delete; SensingCamera& operator=(SensingCamera&&) = delete; - /** @brief Poll every sensor and submit newly captured frames. Throws once the room is lost. */ + /** @brief Poll every sensor and submit newly captured frames. Throws once a stream can no longer work. */ void update(); - /** @brief Flush each track's queued frames. */ + /** @brief Flush each stream's encoder. */ void flush(); /** @brief Print per-sensor frame counts to stdout. */ @@ -52,7 +51,8 @@ class SensingCamera struct Stream { uint32_t sensor_id = 0; - std::unique_ptr livekit; + std::string track; + std::unique_ptr output; /// SIPL publish counter of the frame already handled; skips re-reads. uint64_t last_sequence = 0; bool have_last_sequence = false; @@ -63,8 +63,6 @@ class SensingCamera SensingConfig m_config; std::unique_ptr m_camera; - /// Declared before the streams so it is destroyed after their tracks have been unpublished. - std::unique_ptr m_livekit; std::vector m_streams; }; diff --git a/src/plugins/sensing/core/inc/sensing_core/sensing_types.hpp b/src/plugins/sensing/core/inc/sensing_core/sensing_types.hpp index d8c68ec693..032d5db09f 100644 --- a/src/plugins/sensing/core/inc/sensing_core/sensing_types.hpp +++ b/src/plugins/sensing/core/inc/sensing_core/sensing_types.hpp @@ -15,40 +15,51 @@ namespace plugins namespace sensing { -/// One captured sensor and the LiveKit track that carries it. +/// One captured sensor and the stream that carries it to the PC. struct StreamConfig { + /// Send every encoded unit of this stream to the PC video stage at host:port (AVS1 over TCP); empty means no + /// destination yet (one can be set over the management channel). Each stream needs its own port on the PC. + std::string direct_host; + uint16_t direct_port = 0; /// The selected backend's own camera index (SIPL pipeline or Argus device), validated at startup. uint32_t sensor_id = 0; - /// Name of the LiveKit camera track carrying this sensor. - std::string livekit_track; + /// The stream's name: what the PC, the management channel and the logs call this sensor. + std::string name; }; -/// Where and how the tracks are published. -/// The clock and event a published frame's application timestamp is read from. +/// The clock and event a frame's application timestamp is read from. enum class TimestampBasis { /// CLOCK_MONOTONIC when the frame was ready to encode (`monotonic_us`); recordable. FrameReadyMonotonic, /// The sensor's capture event converted to CLOCK_MONOTONIC (`capture_monotonic_us`); unverified on hardware. SensorCaptureMonotonic, - /// CLOCK_REALTIME when the frame finished conversion (`processing_unix_us`); not recordable. - ProcessingUnix, }; -struct LiveKitConfig +/// How an encoded unit is matched to the capture it came from. +enum class FrameCorrelation { - std::string url; - /// The join token itself, not a path. It names the room and the participant. Never logged. - std::string token; - /// Publication rate cap, independent of the capture rate. + /// By the stamp the encoder hands back with the unit (the stamp it was given with the input). A unit whose stamp + /// matches no submitted frame carries no capture timestamp, and inputs the encoder never returned are counted. + Stamp, + /// By order: the encoder is assumed to return one unit per input, in order. Not verifiable; an input the encoder + /// skips shifts every later timestamp. For when the encoder does not return the stamp and approximate timing is + /// acceptable (a live preview), never for recording. + Fifo, +}; + +struct EncodeConfig +{ + /// Rate cap of the encoded stream, independent of the capture rate. uint32_t fps = 30; - /// Output-only uniform reduction of every track; 1 publishes the sensor's native size. + /// Output-only uniform reduction of every stream; 1 sends the sensor's native size. uint32_t scale_div = 1; - /// Target bitrate of each track's hardware encoder; the VBR ceiling is half as much again. + /// Target bitrate of each stream's hardware encoder; the VBR ceiling is half as much again. uint32_t bitrate_bps = 8'000'000; /// What each frame's application timestamp is; the receiving side's streams file must declare the same basis. TimestampBasis timestamp_basis = TimestampBasis::FrameReadyMonotonic; + FrameCorrelation correlation = FrameCorrelation::Stamp; }; struct SensingConfig @@ -74,7 +85,7 @@ struct SensingConfig /// Argus only: the sensor mode index applied to every camera. uint32_t argus_sensor_mode = 0; - LiveKitConfig livekit; + EncodeConfig encode; }; } // namespace sensing diff --git a/src/plugins/sensing/core/livekit_video_output.hpp b/src/plugins/sensing/core/inc/sensing_core/video_output.hpp similarity index 56% rename from src/plugins/sensing/core/livekit_video_output.hpp rename to src/plugins/sensing/core/inc/sensing_core/video_output.hpp index a0d624beec..42b0c4eed8 100644 --- a/src/plugins/sensing/core/livekit_video_output.hpp +++ b/src/plugins/sensing/core/inc/sensing_core/video_output.hpp @@ -3,10 +3,8 @@ #pragma once -#include "inc/sensing_core/camera_capture.hpp" -#include "inc/sensing_core/sensing_types.hpp" - -#include +#include "camera_capture.hpp" +#include "sensing_types.hpp" #include #include @@ -17,7 +15,7 @@ namespace plugins namespace sensing { -struct LiveKitOutputConfig +struct VideoOutputConfig { uint32_t source_width = 0; uint32_t source_height = 0; @@ -28,70 +26,82 @@ struct LiveKitOutputConfig /// Publication rate cap, independent of the capture rate. uint32_t fps = 30; int gpu_id = 0; - /// Hardware encoder settings for the published pictures; the peak is also the ceiling for the - /// bitrate that congestion control may ask for. + /// Hardware encoder settings: the target bitrate and the VBR ceiling. uint32_t bitrate_bps = 8'000'000; uint32_t peak_bitrate_bps = 0; /// IDR period in frames; 0 defers to the encoder's default. uint32_t gop = 0; bool full_range = false; + /// Also send every encoded unit to the PC video stage at this address (host:port); empty disables it. + std::string direct_host; + uint16_t direct_port = 0; /// What each frame's application timestamp is. SensorCaptureMonotonic fails the output on a frame whose sensor /// time cannot be placed on CLOCK_MONOTONIC. TimestampBasis timestamp_basis = TimestampBasis::FrameReadyMonotonic; + FrameCorrelation correlation = FrameCorrelation::Stamp; }; -struct LiveKitOutputStats +/// Where this stream's encoded video also goes (the PC video stage), and whether that link is up. +struct DirectLinkStatus +{ + std::string host; + uint16_t port = 0; + bool connected = false; + std::string error; +}; + +struct VideoOutputStats { uint64_t offered = 0; ///< Captures handed to submit(). uint64_t paced_out = 0; ///< Skipped to hold the publication rate. uint64_t replaced = 0; ///< Snapshots overwritten by a newer capture before the worker took them. uint64_t snapshots = 0; ///< GPU snapshots taken. - uint64_t published = 0; ///< Frames the SDK accepted. Not proof that a viewer received them. + uint64_t published = 0; ///< Encoded units handed to the direct link (not proof the PC received them). uint64_t encoder_busy = 0; ///< Snapshots the encoder had no input buffer for, dropped before encoding. - uint64_t rejected = 0; ///< Encoded frames the SDK refused; a keyframe follows. - uint64_t keyframes_requested = 0; ///< Keyframes the receiving side asked for. - uint64_t encoder_dropped = 0; ///< Accepted frames the encoder never returned; their metadata is retired. - uint64_t unmatched_units = 0; ///< Encoded units with no capture to pair with, skipped. + uint64_t keyframes_requested = 0; ///< Keyframes the PC asked for. + uint64_t matched = 0; ///< Units matched to their capture by the stamp the encoder returned. + /// Units whose returned stamp matched no submitted frame (sent without a capture time), and units with no frame + /// in flight (skipped). + uint64_t unmatched = 0; + uint64_t encoder_dropped = 0; ///< Submitted frames the encoder never returned a unit for. }; /** - * @brief Encodes one sensor's frames on hardware and feeds them to a pre-encoded video track. + * @brief Encodes one sensor's frames on hardware and hands them to the direct link to the PC. * - * submit() runs on the caller's polling loop. A pixel-buffer frame (SIPL) is copied into a snapshot in GPU - * memory before returning, because the camera may overwrite its slot at the second following poll. A DMA-BUF - * frame (Argus) is not copied: it goes to the encoder as it is, and its lease returns the buffer to the camera - * when the encoder is done with it. Everything else -- the optional reduction, the hardware encode and the - * SDK call -- happens on this output's own worker thread, so a slow network cannot hold up capture - * or the other sensors. + * submit() runs on the caller's polling loop. A pixel-buffer frame (SIPL) is copied into a snapshot in GPU memory + * before returning, because the camera may overwrite its slot at the second following poll. A DMA-BUF frame (Argus) + * is not copied: it goes to the encoder as it is, and its lease returns the buffer to the camera when the encoder is + * done with it. Everything else -- the optional reduction, the hardware encode and the send queue -- happens on this + * output's own worker thread, so a slow network cannot hold up capture or the other sensors. * - * Two snapshot slots bound the memory: one the worker is encoding, one waiting. A newer capture - * replaces the waiting one; the worker never waits for it. The encoder is the JetsonEncoder class: - * the Jetson V4L2 hardware engine. There is no software encoder: if the hardware encoder cannot - * start, construction fails. + * Two snapshot slots bound the memory: one the worker is encoding, one waiting. A newer capture replaces the waiting + * one; the worker never waits for it. The encoder is the JetsonEncoder class: the Jetson V4L2 hardware engine. There + * is no software encoder: if the hardware encoder cannot start, construction fails. * - * The receiving side's keyframe requests and bitrate targets are applied before the next frame is - * encoded. A frame the SDK refuses breaks the reference chain, so the output asks for a keyframe and - * publishes nothing else until one arrives. + * The PC's keyframe requests (and a dropped or reconnected link) are applied before the next frame is encoded. */ -class LiveKitVideoOutput +class VideoOutput { public: /// @throws std::runtime_error if the geometry is unusable (see validate_geometry), the hardware /// encoder cannot start, or CUDA fails. - LiveKitVideoOutput(std::unique_ptr sink, const LiveKitOutputConfig& config); + explicit VideoOutput(const VideoOutputConfig& config); /// Stops the worker; a snapshot not yet published is dropped. - ~LiveKitVideoOutput(); + ~VideoOutput(); - LiveKitVideoOutput(const LiveKitVideoOutput&) = delete; - LiveKitVideoOutput& operator=(const LiveKitVideoOutput&) = delete; + VideoOutput(const VideoOutput&) = delete; + VideoOutput& operator=(const VideoOutput&) = delete; /// Never blocks on the network. A frame sooner than the publication rate allows is skipped. Borrowed CUDA /// RGBA pixels are copied out before this returns; a leased DMA-BUF frame's lease is moved into the output and /// released once the encoder is finished with it (or at once if the frame is dropped). void submit(CaptureFrame& frame); - LiveKitOutputStats stats() const; + VideoOutputStats stats() const; + + DirectLinkStatus direct_status() const; bool failed() const; std::string failure() const; diff --git a/src/plugins/sensing/core/jetson_encoder.cpp b/src/plugins/sensing/core/jetson_encoder.cpp index 4546e5ee1f..898b99f8c5 100644 --- a/src/plugins/sensing/core/jetson_encoder.cpp +++ b/src/plugins/sensing/core/jetson_encoder.cpp @@ -43,6 +43,26 @@ void check_cuda(cudaError_t err, const char* what) throw std::runtime_error(std::string("JetsonEncoder: ") + what + ": " + cudaGetErrorString(err)); } +// Asks the driver to return each input buffer's timestamp on its encoded unit. Must precede plane setup. +bool enable_copy_timestamp(NvVideoEncoder& enc) +{ +#ifdef V4L2_CID_MPEG_VIDEOENC_COPY_TIMESTAMP + struct v4l2_ext_control control; + struct v4l2_ext_controls controls; + std::memset(&control, 0, sizeof(control)); + std::memset(&controls, 0, sizeof(controls)); + control.id = V4L2_CID_MPEG_VIDEOENC_COPY_TIMESTAMP; + control.value = 1; + controls.ctrl_class = V4L2_CTRL_CLASS_MPEG; + controls.count = 1; + controls.controls = &control; + return enc.setExtControls(controls) >= 0; +#else + (void)enc; + return false; +#endif +} + } // namespace struct JetsonEncoder::Impl @@ -227,7 +247,17 @@ JetsonEncoder::JetsonEncoder(const EncoderConfig& config) : m_impl(std::make_uni // join mid-stream, and the max-performance clock preset. check(enc->setNumBFrames(0), "setNumBFrames"); check(enc->setInsertSpsPpsAtIdrEnabled(true), "setInsertSpsPpsAtIdrEnabled"); + // Without a VUI the SPS carries no reorder/buffering limits, so a receiver's decoder (NVDEC) holds back about a + // dozen pictures. Non-fatal: a stream without it is valid, only slower to display. + if (enc->setInsertVuiEnabled(true) < 0) + isaaccapture::Logger::get("isaaccapture.plugins.sensing.JetsonEncoder") + ->warn("setInsertVuiEnabled failed; the SPS will have no VUI"); check(enc->setMaxPerfMode(1), "setMaxPerfMode"); + // Without this the driver numbers units from its own frame clock, so a unit cannot be tied to its capture by + // stamp. Non-fatal: the producer then reports unmatched units and sends them without a capture time. + if (!enable_copy_timestamp(*enc)) + isaaccapture::Logger::get("isaaccapture.plugins.sensing.JetsonEncoder") + ->warn("the encoder will not copy input timestamps; encoded units cannot be matched to captures"); // A separate control from the max-performance clocks: it trades quality for encoder throughput. // FAST suits live streams; four native-size images at 30 fps may need ULTRAFAST, to be measured. check(enc->setHWPresetType(V4L2_ENC_HW_PRESET_FAST), "setHWPresetType"); diff --git a/src/plugins/sensing/core/sensing_camera.cpp b/src/plugins/sensing/core/sensing_camera.cpp index 141e713c22..54e6072c3b 100644 --- a/src/plugins/sensing/core/sensing_camera.cpp +++ b/src/plugins/sensing/core/sensing_camera.cpp @@ -3,9 +3,7 @@ #include "inc/sensing_core/sensing_camera.hpp" -#include "livekit_video_output.hpp" - -#include +#include "inc/sensing_core/video_output.hpp" #include #include @@ -21,20 +19,23 @@ namespace sensing namespace { -LiveKitOutputConfig make_output_config(const SensingConfig& config, const CameraInfo& sensor) +VideoOutputConfig make_output_config(const SensingConfig& config, const StreamConfig& stream, const CameraInfo& sensor) { - LiveKitOutputConfig output; + VideoOutputConfig output; output.source_width = sensor.width; output.source_height = sensor.height; - output.scale_div = config.livekit.scale_div; - output.fps = config.livekit.fps; + output.scale_div = config.encode.scale_div; + output.fps = config.encode.fps; output.gpu_id = config.gpu_id; - output.bitrate_bps = config.livekit.bitrate_bps; - output.peak_bitrate_bps = config.livekit.bitrate_bps + config.livekit.bitrate_bps / 2; + output.bitrate_bps = config.encode.bitrate_bps; + output.peak_bitrate_bps = config.encode.bitrate_bps + config.encode.bitrate_bps / 2; output.input = sensor.pixels; output.gop = config.gop; output.full_range = config.full_range; - output.timestamp_basis = config.livekit.timestamp_basis; + output.timestamp_basis = config.encode.timestamp_basis; + output.correlation = config.encode.correlation; + output.direct_host = stream.direct_host; + output.direct_port = stream.direct_port; return output; } @@ -53,13 +54,12 @@ SensingCamera::SensingCamera(const SensingConfig& config, throw std::runtime_error("SensingCamera: no camera backend"); const auto& sensors = m_camera->cameras(); - // Everything that needs no network is checked first, so a bad name or divisor fails before any - // connection is made. + // Everything that needs no hardware is checked first, so a bad name or divisor fails before the cameras start. std::set tracks; for (const auto& stream_config : streams) { - if (!tracks.insert(stream_config.livekit_track).second) - throw std::runtime_error("LiveKit track name '" + stream_config.livekit_track + "' is used twice"); + if (!tracks.insert(stream_config.name).second) + throw std::runtime_error("stream name '" + stream_config.name + "' is used twice"); const auto found = std::find_if( sensors.begin(), sensors.end(), [&](const CameraInfo& s) { return s.id == stream_config.sensor_id; }); if (found == sensors.end()) @@ -72,34 +72,34 @@ SensingCamera::SensingCamera(const SensingConfig& config, oss << ". Note this is the backend's own camera index (a SIPL pipeline index is not the GMSL link index)."; throw std::runtime_error(oss.str()); } - LiveKitVideoOutput::validate_geometry(found->width, found->height, config.livekit.scale_div); - if (found->pixels == PixelKind::DmaBufNv12 && config.livekit.scale_div != 1) + VideoOutput::validate_geometry(found->width, found->height, config.encode.scale_div); + if (found->pixels == PixelKind::DmaBufNv12 && config.encode.scale_div != 1) throw std::runtime_error("sensor " + std::to_string(found->id) + - " delivers DMA-BUF frames, which cannot be reduced: use --livekit-scale-div=1"); + " delivers DMA-BUF frames, which cannot be reduced: use --scale-div=1"); } - m_livekit = LiveKitSession::connect(config.livekit.url, config.livekit.token); m_streams.reserve(streams.size()); for (const auto& stream_config : streams) { const auto found = std::find_if( sensors.begin(), sensors.end(), [&](const CameraInfo& s) { return s.id == stream_config.sensor_id; }); - const LiveKitOutputConfig output = make_output_config(config, *found); - - PublishSpec spec; - spec.name = stream_config.livekit_track; - spec.width = found->width / output.scale_div; - spec.height = found->height / output.scale_div; - spec.max_bitrate_bps = output.peak_bitrate_bps; - spec.max_fps = std::min(output.fps, found->fps); + const VideoOutputConfig output = make_output_config(config, stream_config, *found); + const uint32_t width = found->width / output.scale_div; + const uint32_t height = found->height / output.scale_div; + const double max_fps = std::min(output.fps, found->fps); Stream stream; stream.sensor_id = found->id; - stream.livekit = std::make_unique(m_livekit->publish_video(spec), output); + stream.track = stream_config.name; + stream.output = std::make_unique(output); m_streams.push_back(std::move(stream)); - std::cout << "Sensor " << found->id << " (" << found->name << ") -> LiveKit track '" << spec.name - << "': " << spec.width << "x" << spec.height << " @ up to " << spec.max_fps << " fps" << std::endl; + std::cout << "Sensor " << found->id << " (" << found->name << ") -> stream '" << stream_config.name + << "': " << width << "x" << height << " @ up to " << max_fps << " fps, direct link " + << (stream_config.direct_host.empty() ? + std::string("not set yet") : + stream_config.direct_host + ":" + std::to_string(stream_config.direct_port)) + << std::endl; } m_camera->start(); @@ -113,10 +113,6 @@ SensingCamera::~SensingCamera() void SensingCamera::update() { - // The room is gone for good: stop rather than keep capturing for nobody. - if (m_livekit->failed()) - throw std::runtime_error(m_livekit->failure()); - for (auto& stream : m_streams) { auto view = m_camera->acquire_latest(stream.sensor_id); @@ -129,41 +125,47 @@ void SensingCamera::update() ++stream.frame_count; // A GPU snapshot only; the conversion and the network are on the output's own thread. - stream.livekit->submit(*view); + stream.output->submit(*view); } - if (stream.livekit->failed()) - throw std::runtime_error("sensor " + std::to_string(stream.sensor_id) + ": " + stream.livekit->failure()); + if (stream.output->failed()) + throw std::runtime_error("sensor " + std::to_string(stream.sensor_id) + ": " + stream.output->failure()); } } void SensingCamera::flush() { for (auto& stream : m_streams) - stream.livekit->flush(); + stream.output->flush(); } void SensingCamera::print_stats() const { for (const auto& stream : m_streams) { - const auto s = stream.livekit->stats(); + const auto s = stream.output->stats(); std::cout << " sensor " << stream.sensor_id << ": " << stream.frame_count << " frames"; if (stream.missed_captures) std::cout << " | " << stream.missed_captures << " captures missed"; - std::cout << " | livekit " << s.published << " published, " << s.paced_out << " paced out, " << s.replaced - << " replaced"; + std::cout << " | encoded " << s.published << ", " << s.paced_out << " paced out, " << s.replaced << " replaced"; if (s.encoder_busy) std::cout << ", " << s.encoder_busy << " encoder busy"; - if (s.rejected) - std::cout << ", " << s.rejected << " rejected"; - if (s.encoder_dropped) - std::cout << ", " << s.encoder_dropped << " dropped in encoder"; - if (s.unmatched_units) - std::cout << ", " << s.unmatched_units << " unmatched units"; if (s.keyframes_requested) std::cout << ", " << s.keyframes_requested << " keyframes requested"; - const std::string encoder = stream.livekit->describe(); + if (m_config.encode.correlation == FrameCorrelation::Stamp) + std::cout << " | stamps matched " << s.matched << ", unmatched " << s.unmatched << ", encoder skipped " + << s.encoder_dropped; + else if (s.unmatched || s.encoder_dropped) + std::cout << " | unmatched " << s.unmatched << ", encoder skipped " << s.encoder_dropped; + const DirectLinkStatus direct = stream.output->direct_status(); + if (direct.host.empty()) + std::cout << " | direct: no destination"; + else if (direct.connected) + std::cout << " | direct: " << direct.host << ":" << direct.port << " connected"; + else + std::cout << " | direct: " << direct.host << ":" << direct.port << " NOT connected" + << (direct.error.empty() ? "" : " (" + direct.error + ")"); + const std::string encoder = stream.output->describe(); if (!encoder.empty()) std::cout << " (" << encoder << ")"; std::cout << std::endl; diff --git a/src/plugins/sensing/core/livekit_video_output.cpp b/src/plugins/sensing/core/video_output.cpp similarity index 70% rename from src/plugins/sensing/core/livekit_video_output.cpp rename to src/plugins/sensing/core/video_output.cpp index e4af684843..ccb7d16d2a 100644 --- a/src/plugins/sensing/core/livekit_video_output.cpp +++ b/src/plugins/sensing/core/video_output.cpp @@ -1,12 +1,16 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 -#include "livekit_video_output.hpp" +#include "inc/sensing_core/video_output.hpp" #include "inc/sensing_core/jetson_encoder.hpp" #include "inc/sensing_core/sensor_clock.hpp" #include "rgba_box_downscale.cuh" +#include +#include +#include + #include #include #include @@ -32,51 +36,64 @@ namespace void check_cuda(cudaError_t err, const char* what) { if (err != cudaSuccess) - throw std::runtime_error(std::string("LiveKit output: ") + what + ": " + cudaGetErrorString(err)); + throw std::runtime_error(std::string("Video output: ") + what + ": " + cudaGetErrorString(err)); } // Congestion control never drives the encoder below this. constexpr uint32_t kMinBitrateBps = 100'000; +/// Forced keyframes (the PC asking, or a dropped link recovering) are at least this far apart. +constexpr std::chrono::milliseconds kMinIdrInterval{ 1000 }; +/// How often the link is judged, how far the bitrate falls back, and how long it must stay calm before it rises. +constexpr std::chrono::milliseconds kAdaptInterval{ 1000 }; +constexpr uint32_t kFloorBitrateBps = 1'500'000; +constexpr uint32_t kCalmBeforeRaise = 5; // A target is applied only when it differs from the current one by this fraction, and this long after // the last change. -constexpr double kMinBitrateChange = 0.25; -constexpr std::chrono::seconds kMinBitrateInterval{ 3 }; } // namespace -struct LiveKitVideoOutput::Impl +struct VideoOutput::Impl { struct Slot { uint8_t* ptr = nullptr; size_t pitch = 0; uint64_t timestamp_ns = 0; - uint64_t unix_ns = 0; uint64_t capture_monotonic_ns = 0; uint64_t capture_tsc = 0; uint64_t sequence = 0; }; - Impl(std::unique_ptr track, const LiveKitOutputConfig& cfg) : sink(std::move(track)), config(cfg) + explicit Impl(const VideoOutputConfig& cfg) + : config(cfg), + correlator(cfg.correlation == FrameCorrelation::Stamp ? video_stage::FrameCorrelator::Mode::Stamp : + video_stage::FrameCorrelator::Mode::Fifo) { validate_geometry(config.source_width, config.source_height, config.scale_div); if (config.fps == 0) - throw std::runtime_error("LiveKit output: the publication rate must be at least 1 fps"); + throw std::runtime_error("Video output: the publication rate must be at least 1 fps"); out_width = config.source_width / config.scale_div; out_height = config.source_height / config.scale_div; - if (sink->width() != out_width || sink->height() != out_height) - throw std::runtime_error("LiveKit output: the track size does not match the output size"); if (config.input == PixelKind::DmaBufNv12 && config.scale_div != 1) - throw std::runtime_error("LiveKit output: DMA-BUF input cannot be reduced (scale_div must be 1)"); + throw std::runtime_error("Video output: DMA-BUF input cannot be reduced (scale_div must be 1)"); period_ns = 1'000'000'000ull / config.fps; + // Always present: with no host the sender idles, and a management client can give it one at runtime. + { + DirectSenderConfig direct_config; + direct_config.host = config.direct_host; + direct_config.port = config.direct_port; + direct_config.width = out_width; + direct_config.height = out_height; + direct_config.fps = config.fps; + direct = std::make_unique(direct_config); + } if (config.bitrate_bps < kMinBitrateBps) { - throw std::runtime_error("LiveKit output: the bitrate must be at least " + std::to_string(kMinBitrateBps) + + throw std::runtime_error("Video output: the bitrate must be at least " + std::to_string(kMinBitrateBps) + " bps"); } // Congestion control moves the target between the floor and the ceiling; the configured target // is never below the floor, so the range is never inverted. - max_bitrate = std::max(config.bitrate_bps, config.peak_bitrate_bps); // A constructor that throws never runs the destructor, so free what was allocated here. try @@ -88,13 +105,13 @@ struct LiveKitVideoOutput::Impl encoder_config.width = out_width; encoder_config.height = out_height; encoder_config.bitrate_bps = config.bitrate_bps; + current_bitrate = config.bitrate_bps; encoder_config.peak_bitrate_bps = config.peak_bitrate_bps; encoder_config.fps = config.fps; encoder_config.gop = config.gop; encoder_config.full_range = config.full_range; encoder_config.input = config.input; encoder = std::make_unique(encoder_config); - applied_bitrate = config.bitrate_bps; for (auto& slot : slots) { @@ -162,7 +179,7 @@ struct LiveKitVideoOutput::Impl const auto* rgba = std::get_if(&frame.pixels); if (dma ? !std::holds_alternative(frame.pixels) : rgba == nullptr) { - fail("LiveKit output: the frame's pixel type does not match the encoder input"); + fail("Video output: the frame's pixel type does not match the encoder input"); return; } ++offered; @@ -247,7 +264,6 @@ struct LiveKitVideoOutput::Impl check_cuda(cudaStreamSynchronize(snapshot_stream), "snapshot sync"); slot.timestamp_ns = frame.monotonic_ns; - slot.unix_ns = frame.unix_ns; slot.capture_monotonic_ns = frame.capture_monotonic_ns; slot.capture_tsc = frame.sensor_time; slot.sequence = frame.sequence; @@ -279,7 +295,7 @@ struct LiveKitVideoOutput::Impl std::unique_lock lock(mutex); // Frames in the encoder come out a little later: poll for them soon, and otherwise // wake now and then to pick up the receiving side's requests. - const auto wait = in_encoder.empty() ? std::chrono::milliseconds(100) : std::chrono::milliseconds(2); + const auto wait = correlator.empty() ? std::chrono::milliseconds(100) : std::chrono::milliseconds(2); wake.wait_for(lock, wait, [&] { return stopping || pending >= 0 || pending_dma || flush_requested; }); if (stopping) return; @@ -290,8 +306,6 @@ struct LiveKitVideoOutput::Impl pending_dma.reset(); flush_now = flush_requested; } - - apply_feedback(); if (dma) { encode_dma(*dma); @@ -308,8 +322,7 @@ struct LiveKitVideoOutput::Impl for (auto& unit : encoder->drain()) publish_unit(unit); // Anything still unreturned once the encoder is drained is not coming back. - encoder_dropped += in_encoder.size(); - in_encoder.clear(); + encoder_dropped += correlator.in_flight(); std::lock_guard lock(mutex); flush_requested = false; flushed.notify_all(); @@ -329,29 +342,48 @@ struct LiveKitVideoOutput::Impl } } - // Worker thread: what the receiving side asked for since the last frame. - void apply_feedback() + // A keyframe is several times the size of an ordinary frame. A link that is struggling makes the PC ask for + // keyframes over and over, and each one adds to the load that caused the request, so forced keyframes are spaced. + bool idr_allowed() { - const EncoderFeedback feedback = sink->take_feedback(); - if (feedback.keyframe_requested) + const auto now = std::chrono::steady_clock::now(); + if (now - last_forced_idr < kMinIdrInterval) + return false; + last_forced_idr = now; + return true; + } + + // Worker thread: follow the link. Frames the direct sender had to drop mean the network could not carry the + // stream, so lower the bitrate; after a calm stretch raise it again, never above the configured target. + void adapt_to_link() + { + const auto now = std::chrono::steady_clock::now(); + if (now - last_adapt < kAdaptInterval) + return; + last_adapt = now; + const uint64_t dropped = direct->dropped(); + // Drops while nothing is connected say the destination is absent, not that the network is slow. + const bool congested = dropped != last_dropped && direct->connected(); + last_dropped = dropped; + + uint32_t target = current_bitrate; + if (congested) { - ++keyframes_requested; - keyframe_wanted = true; + calm_intervals = 0; + target = std::max(kFloorBitrateBps, static_cast(current_bitrate * 0.7)); } - if (feedback.target_bitrate_bps) + else if (++calm_intervals >= kCalmBeforeRaise) { - const uint32_t target = std::clamp(*feedback.target_bitrate_bps, kMinBitrateBps, max_bitrate); - // Congestion control re-estimates all the time. Follow it only for a real change, and not - // more than every few seconds: a change reopens the encoder session. - const auto now = std::chrono::steady_clock::now(); - const double change = std::abs(static_cast(target) - applied_bitrate) / applied_bitrate; - if (change >= kMinBitrateChange && now - last_bitrate_change >= kMinBitrateInterval && - encoder->set_bitrate(target)) - { - applied_bitrate = target; - last_bitrate_change = now; - } + calm_intervals = 0; + target = std::min(config.bitrate_bps, static_cast(current_bitrate * 1.15)); } + target = std::min(target, config.bitrate_bps); + if (target == current_bitrate || !encoder->set_bitrate(target)) + return; + isaaccapture::Logger::get("isaaccapture.plugins.sensing.VideoOutput") + ->info("link {}: video bitrate {} -> {} kbit/s", congested ? "congested" : "recovering", + current_bitrate / 1000, target / 1000); + current_bitrate = target; } // Worker thread. @@ -371,7 +403,13 @@ struct LiveKitVideoOutput::Impl pitch = scaled_pitch; } - if (keyframe_wanted) + if (direct && direct->take_keyframe_request()) + { + ++keyframes_requested; + keyframe_wanted = true; + } + adapt_to_link(); + if (keyframe_wanted && idr_allowed()) { encoder->request_keyframe(); keyframe_wanted = false; @@ -385,13 +423,19 @@ struct LiveKitVideoOutput::Impl return; } track_in_flight(static_cast(slot.sequence), slot.capture_tsc, - application_stamp_us(slot.timestamp_ns, slot.capture_monotonic_ns, slot.unix_ns)); + application_stamp_us(slot.timestamp_ns, slot.capture_monotonic_ns), stamp_ns); } // Worker thread: a leased DMA-BUF frame goes to the encoder as it is, and its lease stays with the encoder. void encode_dma(CaptureFrame& frame) { - if (keyframe_wanted) + if (direct && direct->take_keyframe_request()) + { + ++keyframes_requested; + keyframe_wanted = true; + } + adapt_to_link(); + if (keyframe_wanted && idr_allowed()) { encoder->request_keyframe(); keyframe_wanted = false; @@ -403,7 +447,8 @@ struct LiveKitVideoOutput::Impl return; } track_in_flight(static_cast(frame.sequence), frame.sensor_time, - application_stamp_us(frame.monotonic_ns, frame.capture_monotonic_ns, frame.unix_ns)); + application_stamp_us(frame.monotonic_ns, frame.capture_monotonic_ns), + static_cast(frame.monotonic_ns)); } /// Polling thread: the frame's capture event on CLOCK_MONOTONIC. The sensor clock's anchor is built from the @@ -411,7 +456,7 @@ struct LiveKitVideoOutput::Impl uint64_t capture_time_ns(const CaptureFrame& frame) { if (frame.sensor_domain == SensorTimeDomain::None || frame.sensor_time == 0) - throw std::runtime_error("LiveKit output: the camera backend reports no sensor timestamp to publish"); + throw std::runtime_error("Video output: the camera backend reports no sensor timestamp to publish"); if (!clock_map) { SensorClockMap::Config map; @@ -432,7 +477,7 @@ struct LiveKitVideoOutput::Impl } /// The application timestamp published with a frame, in microseconds. - uint64_t application_stamp_us(uint64_t ready_monotonic_ns, uint64_t capture_monotonic_ns, uint64_t unix_ns) const + uint64_t application_stamp_us(uint64_t ready_monotonic_ns, uint64_t capture_monotonic_ns) const { switch (config.timestamp_basis) { @@ -440,100 +485,69 @@ struct LiveKitVideoOutput::Impl return ready_monotonic_ns / 1000; case TimestampBasis::SensorCaptureMonotonic: return capture_monotonic_ns / 1000; - case TimestampBasis::ProcessingUnix: - return unix_ns / 1000; } throw std::logic_error("unknown timestamp basis"); } // Worker thread: remember the capture a submitted frame came from, for the unit that comes back. - void track_in_flight(uint32_t sequence, uint64_t capture_tsc, uint64_t unix_us) + void track_in_flight(uint32_t sequence, uint64_t capture_tsc, uint64_t stamp_us, int64_t submitted_ns) { - in_encoder.push_back({ sequence, capture_tsc, monotonic_us, unix_us }); + correlator.submitted({ sequence, capture_tsc, stamp_us, submitted_ns / 1000 }); } // Worker thread: one encoded unit becomes one published frame, with the sequence and capture time // of the capture it came from. void publish_unit(const EncodedUnit& unit) { - // The Jetson encoder does not return the stamp it was given: it numbers its units from its own frame - // rate (0, 1/fps, ...). Units pair with captures by order, and a jump in that numbering is the only - // evidence of frames the encoder consumed without returning a unit. Metadata is retired only by a - // matched unit or by such a jump; a jump that leaves no frame for this unit, or a numbering that goes - // backwards, is not trusted and just resets the baseline. - const int64_t index = std::llround(static_cast(unit.timestamp_ns) * config.fps / 1e9); - if (have_unit_index) + const int64_t stamp_us = unit.timestamp_ns / 1000; + // Which capture is this unit? See FrameCorrelator: by the stamp the encoder hands back, so a frame it never + // returned is detected, and a unit that matches nothing is sent without a capture time instead of an assumed + // one. Fifo is an explicit choice for approximate timing only. + const auto result = correlator.returned(stamp_us); + if (!result) { - const int64_t gap = index - last_unit_index - 1; - if (gap > 0 && static_cast(gap) < in_encoder.size()) - { - in_encoder.erase(in_encoder.begin(), in_encoder.begin() + gap); - encoder_dropped += static_cast(gap); - } - } - last_unit_index = index; - have_unit_index = true; - - // A unit with no frame to pair with is skipped rather than published with someone else's metadata. - if (in_encoder.empty()) - { - ++unmatched_units; + // A unit with no capture to pair with is skipped, not published with someone else's metadata. + ++unmatched; return; } - const FrameMeta meta = in_encoder.front(); - in_encoder.pop_front(); - - // After a refusal the stream is only usable from its next keyframe. - if (awaiting_keyframe && !unit.keyframe) - return; + const bool identified = result->identified; + const video_stage::InFlightFrame& meta = result->frame; + encoder_dropped += result->skipped; + if (config.correlation == FrameCorrelation::Stamp) + ++(identified ? matched : unmatched); EncodedVideoFrame frame; frame.data = unit.data.data(); frame.size = unit.data.size(); frame.keyframe = unit.keyframe; frame.timestamp_us = static_cast(stamp_us); - frame.frame_id = meta.sequence; - frame.capture_tsc = meta.capture_tsc; - frame.user_timestamp_us = meta.unix_us; - if (sink->submit(frame)) - { - awaiting_keyframe = false; - ++published; - } - else - { - ++rejected; - awaiting_keyframe = true; - keyframe_wanted = true; - } + // Zero identity and time when the unit could not be tied to a capture: the receiver treats that as unknown. + frame.frame_id = identified ? meta.sequence : 0; + frame.capture_tsc = identified ? meta.capture_tsc : 0; + frame.user_timestamp_us = identified ? meta.stamp_us : 0; + // The direct link keeps its own queue and keyframe bookkeeping and never blocks this thread. + direct->submit(frame); + ++published; } - std::unique_ptr sink; - LiveKitOutputConfig config; + std::unique_ptr direct; ///< optional second destination: the PC video stage + VideoOutputConfig config; uint32_t out_width = 0; uint32_t out_height = 0; uint64_t period_ns = 0; - struct FrameMeta - { - uint32_t sequence = 0; - uint64_t capture_tsc = 0; - /// The frame's application timestamp: capture time when publishing it, else processing-time Unix. - uint64_t unix_us = 0; - }; std::unique_ptr encoder; - uint32_t max_bitrate = 0; // Worker thread only. std::optional clock_map; ///< Polling thread only. - std::deque in_encoder; ///< Submitted to the encoder, no unit back yet. - int64_t last_unit_index = 0; ///< The encoder's own number of the last unit returned. - bool have_unit_index = false; + video_stage::FrameCorrelator correlator; ///< Submitted to the encoder, no unit back yet. bool keyframe_wanted = false; - bool awaiting_keyframe = false; - uint32_t applied_bitrate = 0; - std::chrono::steady_clock::time_point last_bitrate_change = std::chrono::steady_clock::now(); + std::chrono::steady_clock::time_point last_forced_idr{}; + std::chrono::steady_clock::time_point last_adapt = std::chrono::steady_clock::now(); + uint64_t last_dropped = 0; + uint32_t calm_intervals = 0; + uint32_t current_bitrate = 0; Slot slots[2]; /// DMA-BUF input only: the newest accepted frame, waiting for the worker. Guarded by `mutex`. @@ -562,84 +576,85 @@ struct LiveKitVideoOutput::Impl std::atomic replaced{ 0 }; std::atomic snapshots{ 0 }; std::atomic published{ 0 }; + std::atomic matched{ 0 }; + std::atomic unmatched{ 0 }; + std::atomic encoder_dropped{ 0 }; std::atomic encoder_busy{ 0 }; - std::atomic rejected{ 0 }; std::atomic keyframes_requested{ 0 }; - std::atomic encoder_dropped{ 0 }; - std::atomic unmatched_units{ 0 }; std::mutex failure_mutex; std::string failure_message; std::atomic failed_flag{ false }; }; -void LiveKitVideoOutput::validate_geometry(uint32_t width, uint32_t height, uint32_t divisor) +void VideoOutput::validate_geometry(uint32_t width, uint32_t height, uint32_t divisor) { if (divisor == 0) - throw std::runtime_error("LiveKit output: the scale divisor must be at least 1"); + throw std::runtime_error("Video output: the scale divisor must be at least 1"); if (width == 0 || height == 0) - throw std::runtime_error("LiveKit output: the source size is empty"); + throw std::runtime_error("Video output: the source size is empty"); if (width % divisor != 0 || height % divisor != 0) { - throw std::runtime_error("LiveKit output: divisor " + std::to_string(divisor) + " does not divide " + + throw std::runtime_error("Video output: divisor " + std::to_string(divisor) + " does not divide " + std::to_string(width) + "x" + std::to_string(height) + " exactly"); } if ((width / divisor) % 2 != 0 || (height / divisor) % 2 != 0) { - throw std::runtime_error("LiveKit output: " + std::to_string(width / divisor) + "x" + + throw std::runtime_error("Video output: " + std::to_string(width / divisor) + "x" + std::to_string(height / divisor) + " is not even; NV12 needs even dimensions"); } } -LiveKitVideoOutput::LiveKitVideoOutput(std::unique_ptr sink, const LiveKitOutputConfig& config) - : m_impl(std::make_unique(std::move(sink), config)) +VideoOutput::VideoOutput(const VideoOutputConfig& config) : m_impl(std::make_unique(config)) { } -LiveKitVideoOutput::~LiveKitVideoOutput() = default; +VideoOutput::~VideoOutput() = default; -void LiveKitVideoOutput::submit(CaptureFrame& frame) +void VideoOutput::submit(CaptureFrame& frame) { m_impl->submit(frame); } -LiveKitOutputStats LiveKitVideoOutput::stats() const +VideoOutputStats VideoOutput::stats() const { - LiveKitOutputStats s; + VideoOutputStats s; s.offered = m_impl->offered.load(); s.paced_out = m_impl->paced_out.load(); s.replaced = m_impl->replaced.load(); s.snapshots = m_impl->snapshots.load(); s.published = m_impl->published.load(); + s.matched = m_impl->matched.load(); + s.unmatched = m_impl->unmatched.load(); + s.encoder_dropped = m_impl->encoder_dropped.load(); s.encoder_busy = m_impl->encoder_busy.load(); - s.rejected = m_impl->rejected.load(); s.keyframes_requested = m_impl->keyframes_requested.load(); - s.encoder_dropped = m_impl->encoder_dropped.load(); - s.unmatched_units = m_impl->unmatched_units.load(); return s; } -bool LiveKitVideoOutput::failed() const +bool VideoOutput::failed() const { return m_impl->failed_flag.load(); } -std::string LiveKitVideoOutput::failure() const +std::string VideoOutput::failure() const { std::lock_guard lock(m_impl->failure_mutex); return m_impl->failure_message; } -std::string LiveKitVideoOutput::describe() +DirectLinkStatus VideoOutput::direct_status() const +{ + return { m_impl->direct->host(), m_impl->direct->port(), m_impl->direct->connected(), m_impl->direct->last_error() }; +} + +std::string VideoOutput::describe() { - std::string text = std::string(JetsonEncoder::backend_name()) + ", pre-encoded passthrough"; - const std::string transport = m_impl->sink->describe(); - if (!transport.empty()) - text += "; transport: " + transport; + std::string text = std::string(JetsonEncoder::backend_name()) + ", H.264 to the PC video stage"; return text; } -void LiveKitVideoOutput::flush() +void VideoOutput::flush() { std::unique_lock lock(m_impl->mutex); m_impl->flush_requested = true; diff --git a/src/plugins/sensing/plugin.yaml b/src/plugins/sensing/plugin.yaml index 2f7cb80087..c1df9fea39 100644 --- a/src/plugins/sensing/plugin.yaml +++ b/src/plugins/sensing/plugin.yaml @@ -2,7 +2,7 @@ # SPDX-License-Identifier: Apache-2.0 name: sensing_camera -description: "SENSING GMSL camera plugin (SIPL capture, hardware H.264, LiveKit publication)" +description: "SENSING GMSL camera plugin (SIPL capture, hardware H.264, direct streaming to the PC video stage)" command: "./camera_plugin_sensing" version: "1.0.0" devices: