diff --git a/examples/camera_livekit_bridge/CMakeLists.txt b/examples/camera_livekit_bridge/CMakeLists.txt index 0c8bb954fa..b74d14ac0c 100644 --- a/examples/camera_livekit_bridge/CMakeLists.txt +++ b/examples/camera_livekit_bridge/CMakeLists.txt @@ -60,6 +60,8 @@ add_subdirectory(video_types) add_subdirectory(ipc_client) add_subdirectory(gpu_ops) add_subdirectory(nvenc) +add_subdirectory(clock_exchange) +add_subdirectory(sender_clock) add_subdirectory(livekit_publisher) add_subdirectory(probe) add_subdirectory(bridge) diff --git a/examples/camera_livekit_bridge/README.md b/examples/camera_livekit_bridge/README.md index f14ffae051..6a52b9c0b1 100644 --- a/examples/camera_livekit_bridge/README.md +++ b/examples/camera_livekit_bridge/README.md @@ -64,12 +64,42 @@ the same basis: | Basis | Value | | --- | --- | | `capture_unix_us` (default) | The sensor start-of-frame as Unix time on the sender's clock. | +| `capture_monotonic_us` | The sensor capture instant from the sender's own timing record (AVS1 timing extension, on the sender's monotonic clock), moved onto this host's monotonic clock by a measured mapping. Needs `--sender-clock`; see below. | | `monotonic_us` | This host's monotonic clock at the instant the picture is ready to encode (before encoding): frame-ready time, later than the sensor capture. | +### Capture time through a measured sender clock + +With `--timestamp-basis capture_monotonic_us --sender-clock IPV4:PORT`, the bridge measures the sender's monotonic +clock against its own over the sender's UDP clock responder (a 13-byte request `Q`, u32 sequence, i64 t1 and a +29-byte reply `R`, sequence, t1, t2, t3, little-endian, five times a second) and applies the mapping to each +frame's capture instant. The responder is stateless and unauthenticated; the bridge accepts only a datagram from the +sender's address, of exactly that size and tag, that answers the outstanding request and was sent since the last +reset. A lost or late reply is a missed sample. A new capture session (a changed `session_id`) resets the mapping and +voids the outstanding request, so a reply in flight across the change cannot seed the new epoch. + +A frame gets a capture `user_timestamp` only when all of these hold; otherwise it carries none, and frame-ready +time is never offered in its place: + +- the sender marked the frame's timing record valid; +- there is a current mapping (a reply within 5 s) with at least three accepted exchanges in its epoch, so a new + capture session or a detected step of the sender's clock waits for fresh exchanges; +- the exchange uncertainty (half the round trip plus a 200 ppm drift allowance) plus the uncertainty the sender + states for its sensor mapping, both legs together, is within the budget (`--clock-uncertainty-budget-ms`, + default 5, or `EGOCAP_CLOCK_UNCERTAINTY_BUDGET_MS`). The budget covers the sensor-to-host chain only; egocap's own + publisher-to-recorder uncertainty adds to it; +- the mapped capture time is not after the frame was ready or more than 1 s before it. + +A step of the sender's clock shows at the next exchange (a discontinuity, a new epoch, warm-up); frames mapped in +between, at most one exchange interval, carry the old offset. A real monotonic clock does not step, so this +exposure is the cost of not trusting a learned age level, which cannot tell a step from a stall. + +The `[stats]` line counts qualified frames and each reason frames were withheld, with the last combined +uncertainty and the mapping epoch. + `capture_unix_us` is used only when the receiver reports a realtime sender clock; a frame with no usable sensor start-of-frame (missing, in the future, older than 5 s, or on another clock domain) carries no `user_timestamp`, so -a consumer that needs a capture time says so instead of using a guess. `capture_monotonic_us` is refused: it would -need a measured sender-to-host clock mapping with a stated uncertainty, which does not exist yet. +a consumer that needs a capture time says so instead of using a guess. `capture_monotonic_us` is refused without +`--sender-clock`. `--bitrate-mbps` is the maximum. The encoder starts at 2 Mbps and follows the target bitrate that LiveKit's congestion control reports for each track (with 10% headroom, never above the maximum or below 0.5 Mbps), so a diff --git a/examples/camera_livekit_bridge/bridge/CMakeLists.txt b/examples/camera_livekit_bridge/bridge/CMakeLists.txt index c647bca7ef..ecdbb67684 100644 --- a/examples/camera_livekit_bridge/bridge/CMakeLists.txt +++ b/examples/camera_livekit_bridge/bridge/CMakeLists.txt @@ -12,6 +12,8 @@ target_link_libraries(camera_livekit_bridge camera_livekit_bridge::gpu_ops camera_livekit_bridge::nvenc camera_livekit_bridge::livekit_publisher + camera_livekit_bridge::sender_clock + camera_livekit_bridge::clock_exchange camera_livekit_bridge::video_types CUDA::cudart Threads::Threads diff --git a/examples/camera_livekit_bridge/bridge/main.cpp b/examples/camera_livekit_bridge/bridge/main.cpp index 45af62f354..e9fbef9f5f 100644 --- a/examples/camera_livekit_bridge/bridge/main.cpp +++ b/examples/camera_livekit_bridge/bridge/main.cpp @@ -7,6 +7,7 @@ #include #include +#include #include #include @@ -67,6 +68,10 @@ int main(int argc, char** argv) std::vector streams; WorkerConfig defaults; int gpu = 0; + std::string sender_host; + uint16_t sender_port = 0; + if (const char* budget = std::getenv("EGOCAP_CLOCK_UNCERTAINTY_BUDGET_MS")) + defaults.clock_budget_ns = static_cast(std::atof(budget) * 1e6); for (int i = 1; i < argc; ++i) { const std::string arg = argv[i]; @@ -99,15 +104,22 @@ int main(int argc, char** argv) std::cerr << "--timestamp-basis must be monotonic_us, capture_monotonic_us or capture_unix_us\n"; return 2; } - if (*basis == TimestampBasis::CaptureMonotonicUs) + defaults.timestamp_basis = *basis; + } + else if (arg == "--sender-clock") + { + const std::string endpoint = next(); + const auto colon = endpoint.rfind(':'); + if (colon == std::string::npos || colon == 0 || colon + 1 == endpoint.size()) { - std::cerr - << "--timestamp-basis capture_monotonic_us is not available: it needs a measured sender-to-host clock " - "mapping, which is not implemented. Use monotonic_us (frame-ready time) or capture_unix_us.\n"; + std::cerr << "--sender-clock needs IPV4:PORT, got '" << endpoint << "'\n"; return 2; } - defaults.timestamp_basis = *basis; + sender_host = endpoint.substr(0, colon); + sender_port = static_cast(std::atoi(endpoint.substr(colon + 1).c_str())); } + else if (arg == "--clock-uncertainty-budget-ms") + defaults.clock_budget_ns = static_cast(std::atof(next().c_str()) * 1e6); else if (arg == "--gpu") gpu = std::atoi(next().c_str()); else if (arg == "--stream") @@ -142,6 +154,15 @@ int main(int argc, char** argv) return 2; } + if (defaults.timestamp_basis == TimestampBasis::CaptureMonotonicUs && sender_host.empty()) + { + std::cerr + << "--timestamp-basis capture_monotonic_us needs --sender-clock IPV4:PORT: the " + "sender's capture time is only used through a measured mapping of its clock. Use monotonic_us for the " + "frame-ready time or capture_unix_us for a Unix-time sensor stamp.\n"; + return 2; + } + std::signal(SIGINT, on_signal); std::signal(SIGTERM, on_signal); @@ -149,6 +170,16 @@ int main(int argc, char** argv) { CudaContext context(gpu); auto session = LiveKitSession::connect(url, read_token(token_file)); + std::unique_ptr sender_clock; + if (!sender_host.empty()) + { + SenderClockConfig clock_config; + clock_config.host = sender_host; + clock_config.port = sender_port; + sender_clock = std::make_unique(clock_config); + sender_clock->start(); + defaults.sender_clock = sender_clock.get(); + } std::vector> workers; for (const auto& spec : streams) { @@ -206,6 +237,18 @@ int main(int argc, char** argv) << static_cast(stats.capture_to_ready_us_total.load()) / ready_samples / 1000.0 << " ms mean " << stats.capture_to_ready_max_us.load() / 1000.0 << " ms max"; } + if (defaults.timestamp_basis == TimestampBasis::CaptureMonotonicUs) + { + std::cerr << ", capture claims: " << stats.verdicts[0].load() << " qualified"; + for (std::size_t v = 1; v < 7; ++v) + { + if (stats.verdicts[v].load() > 0) + std::cerr << ", " << to_string(static_cast(v)) << " " + << stats.verdicts[v].load(); + } + std::cerr << ", combined uncertainty " << stats.last_uncertainty_ns.load() / 1000.0 + << " us, mapping epoch " << stats.last_epoch.load(); + } std::cerr << "\n"; last_frames[i] = frames; } diff --git a/examples/camera_livekit_bridge/bridge/stream_worker.cpp b/examples/camera_livekit_bridge/bridge/stream_worker.cpp index 44ebba2a33..62a542cc9d 100644 --- a/examples/camera_livekit_bridge/bridge/stream_worker.cpp +++ b/examples/camera_livekit_bridge/bridge/stream_worker.cpp @@ -134,6 +134,7 @@ struct StreamWorker::Impl bool need_idr = true; uint32_t own_frame_id = 0; uint64_t last_session = 0; + CaptureQualifier qualifier(config.clock_budget_ns); FrameNotice notice; std::optional pending; // a newer notice seen while checking the last slot; used next while (running) @@ -221,6 +222,9 @@ struct StreamWorker::Impl // A restarted source: start the track's picture sequence over at an IDR. need_idr = true; last_session = session_id; + // Nothing measured before the restart applies to the frames after it. + if (config.sender_clock != nullptr) + config.sender_clock->begin_session(session_id); } const auto feedback = sink->take_feedback(); const bool force_idr = need_idr || feedback.keyframe_requested; @@ -272,6 +276,27 @@ struct StreamWorker::Impl const uint64_t sensor_sof_ns = wall_is_realtime ? unsigned_field(notice.metadata, "sensor_sof_ns") : 0; frame.user_timestamp_us = user_timestamp_us(config.timestamp_basis, sensor_sof_ns, ready_realtime_ns, ready_monotonic_ns); + if (config.timestamp_basis == TimestampBasis::CaptureMonotonicUs) + { + // The capture time comes from the sender's own timing record, on its monotonic clock, moved onto this + // host's by the measured mapping. Without a qualified mapping no capture claim is made; the frame-ready + // time is never offered in its place. + const auto valid = notice.metadata.find("sensor_capture_valid"); + CaptureInput input; + input.sender_valid = valid != notice.metadata.end() && valid->is_boolean() && valid->get(); + input.capture_sm_ns = static_cast(unsigned_field(notice.metadata, "sensor_capture_sm_ns")); + input.upstream_uncertainty_ns = + static_cast(unsigned_field(notice.metadata, "sensor_capture_uncertainty_ns")); + input.ready_monotonic_ns = ready_monotonic_ns; + const SenderClockSnapshot clock = config.sender_clock->snapshot(); + const CaptureResult result = qualifier.qualify(input, clock.mapping, clock.accepted_in_epoch); + ++stats.verdicts[static_cast(result.verdict)]; + stats.last_uncertainty_ns = result.combined_uncertainty_ns; + stats.last_epoch = result.epoch; + frame.user_timestamp_us = result.verdict == CaptureVerdict::Qualified ? + static_cast(result.capture_bridge_ns) / 1000 : + 0; + } if (sensor_sof_ns != 0) { const int64_t age_ns = ready_realtime_ns - static_cast(sensor_sof_ns); diff --git a/examples/camera_livekit_bridge/bridge/stream_worker.hpp b/examples/camera_livekit_bridge/bridge/stream_worker.hpp index f1879cbb04..361fb58df5 100644 --- a/examples/camera_livekit_bridge/bridge/stream_worker.hpp +++ b/examples/camera_livekit_bridge/bridge/stream_worker.hpp @@ -3,8 +3,10 @@ #pragma once +#include #include #include +#include #include #include @@ -30,6 +32,11 @@ struct WorkerConfig uint32_t scale_div = 1; /// What each frame's user_timestamp means; the consumer's streams file must name the same basis. TimestampBasis timestamp_basis = TimestampBasis::CaptureUnixUs; + /// The sender's clock, measured over its management channel; required for capture_monotonic_us. Shared by every + /// stream (one sender, one clock) and owned by the caller. + SenderClockClient* sender_clock = nullptr; + /// The largest combined clock uncertainty a capture time may carry. + int64_t clock_budget_ns = 5'000'000; }; struct WorkerStats @@ -49,6 +56,11 @@ struct WorkerStats std::atomic capture_to_ready_us_total{ 0 }; std::atomic capture_to_ready_samples{ 0 }; std::atomic capture_to_ready_max_us{ 0 }; + /// capture_monotonic_us: how many frames got each verdict, and the last qualified frame's combined uncertainty and + /// the mapping epoch it used. + std::atomic verdicts[8] = {}; + std::atomic last_uncertainty_ns{ 0 }; + std::atomic last_epoch{ 0 }; /// The latest target the transport reported (0 until it reports one). std::atomic target_bps{ 0 }; }; diff --git a/examples/camera_livekit_bridge/clock_exchange/CMakeLists.txt b/examples/camera_livekit_bridge/clock_exchange/CMakeLists.txt new file mode 100644 index 0000000000..f007a2eaf5 --- /dev/null +++ b/examples/camera_livekit_bridge/clock_exchange/CMakeLists.txt @@ -0,0 +1,12 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +# The four-timestamp clock estimator, header only. +add_library(clock_exchange INTERFACE) + +target_include_directories(clock_exchange + INTERFACE + $ +) + +add_library(camera_livekit_bridge::clock_exchange ALIAS clock_exchange) diff --git a/examples/camera_livekit_bridge/clock_exchange/inc/clock_exchange/capture_qualifier.hpp b/examples/camera_livekit_bridge/clock_exchange/inc/clock_exchange/capture_qualifier.hpp new file mode 100644 index 0000000000..df5c8a9d44 --- /dev/null +++ b/examples/camera_livekit_bridge/clock_exchange/inc/clock_exchange/capture_qualifier.hpp @@ -0,0 +1,136 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#pragma once + +#include + +#include +#include + +namespace camera_bridge +{ + +enum class CaptureVerdict +{ + /// The capture time is mapped onto the host's monotonic clock with a stated uncertainty inside the budget. + Qualified, + /// The sender marked the frame's timing record not valid, or sent none. + SenderInvalid, + /// No current mapping of the sender clock (none yet, or the replies stopped). + NoMapping, + /// A new mapping epoch has too few exchanges behind it (after a session restart or a clock step). + Warmup, + /// Exchange uncertainty plus the sender's stated sensor uncertainty is over the budget. + OverBudget, + /// The mapped capture time is after the frame was ready, or implausibly long before it. + ImplausibleAge, +}; + +const char* to_string(CaptureVerdict verdict); + +struct CaptureInput +{ + bool sender_valid = false; + /// The sensor capture instant on the sender's monotonic clock, as the sender transported it. + int64_t capture_sm_ns = 0; + /// The sender's stated bound on its sensor-to-sender-clock mapping. + int64_t upstream_uncertainty_ns = 0; + /// The host's CLOCK_MONOTONIC when the frame was ready to encode. + int64_t ready_monotonic_ns = 0; +}; + +struct CaptureResult +{ + CaptureVerdict verdict = CaptureVerdict::SenderInvalid; + /// The capture instant on the host's monotonic clock; meaningful only when qualified. + int64_t capture_bridge_ns = 0; + /// Exchange uncertainty plus the sender's upstream uncertainty. + int64_t combined_uncertainty_ns = 0; + uint64_t epoch = 0; +}; + +/** + * @brief Decides, per frame, whether a capture time may be claimed, and converts it to the host's monotonic clock. + * + * It never substitutes another time for a capture time it cannot vouch for: the answer is a verdict, and only a + * qualified one carries a time. The checks are validity, a current mapping with enough exchanges behind it, + * upstream plus mapping uncertainty inside the budget, and a physically possible capture-to-ready age. A learned + * age level is not a check: it cannot tell a clock step from a stall. A step of the sender clock is seen at the + * next exchange (a discontinuity starts a new epoch and warm-up); frames in between are exposed for at most one + * exchange interval. Not thread-safe. + */ +class CaptureQualifier +{ +public: + CaptureQualifier(int64_t budget_ns = 5'000'000, std::size_t warmup_exchanges = 3, int64_t max_age_ns = 1'000'000'000) + : m_budget_ns(budget_ns), m_warmup(warmup_exchanges), m_max_age_ns(max_age_ns) + { + } + + CaptureResult qualify(const CaptureInput& input, + const std::optional& mapping, + std::size_t accepted_in_epoch) + { + CaptureResult result; + if (!input.sender_valid) + { + result.verdict = CaptureVerdict::SenderInvalid; + return result; + } + if (!mapping) + { + result.verdict = CaptureVerdict::NoMapping; + return result; + } + result.epoch = mapping->epoch; + result.combined_uncertainty_ns = mapping->uncertainty_ns + input.upstream_uncertainty_ns; + if (accepted_in_epoch < m_warmup) + { + result.verdict = CaptureVerdict::Warmup; + return result; + } + if (result.combined_uncertainty_ns > m_budget_ns) + { + result.verdict = CaptureVerdict::OverBudget; + return result; + } + const int64_t capture = mapping->to_local_ns(input.capture_sm_ns); + const int64_t age = input.ready_monotonic_ns - capture; + if (age < 0 || age > m_max_age_ns) + { + result.verdict = CaptureVerdict::ImplausibleAge; + return result; + } + result.verdict = CaptureVerdict::Qualified; + result.capture_bridge_ns = capture; + return result; + } + +private: + int64_t m_budget_ns; + std::size_t m_warmup; + int64_t m_max_age_ns; +}; + +inline const char* to_string(CaptureVerdict verdict) +{ + switch (verdict) + { + case CaptureVerdict::Qualified: + return "qualified"; + case CaptureVerdict::SenderInvalid: + return "sender_invalid"; + case CaptureVerdict::NoMapping: + return "no_mapping"; + case CaptureVerdict::Warmup: + return "warmup"; + case CaptureVerdict::OverBudget: + return "over_budget"; + case CaptureVerdict::ImplausibleAge: + return "implausible_age"; + } + return "unknown"; +} + +} // namespace camera_bridge diff --git a/examples/camera_livekit_bridge/clock_exchange/inc/clock_exchange/clock_estimator.hpp b/examples/camera_livekit_bridge/clock_exchange/inc/clock_exchange/clock_estimator.hpp new file mode 100644 index 0000000000..22a078afd0 --- /dev/null +++ b/examples/camera_livekit_bridge/clock_exchange/inc/clock_exchange/clock_estimator.hpp @@ -0,0 +1,216 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#pragma once + +#include +#include +#include +#include +#include + +namespace camera_bridge +{ + +/// Round-trip budget, staleness limit, window and relative clock-rate allowance of the estimator; the same values +/// and the same decisions as the four-timestamp estimator egocap applies to the clock of a camera publisher. +inline constexpr int64_t kDefaultMaxRttNs = 250'000'000; +inline constexpr int64_t kDefaultStaleAfterNs = 5'000'000'000; +inline constexpr std::size_t kDefaultWindow = 8; +inline constexpr double kDefaultDriftAllowance = 200e-6; + +/// Floor division, so a negative offset rounds the same way in C++ as in the integer arithmetic of the reference. +inline int64_t floor_div(int64_t numerator, int64_t denominator) +{ + int64_t quotient = numerator / denominator; + if ((numerator % denominator != 0) && ((numerator < 0) != (denominator < 0))) + --quotient; + return quotient; +} + +/// One completed exchange: t1 and t4 on the local clock, t2 and t3 on the remote clock. +struct Exchange +{ + int64_t t1 = 0; + int64_t t2 = 0; + int64_t t3 = 0; + int64_t t4 = 0; + + /// Round trip with the remote's own processing time removed. + int64_t rtt_ns() const + { + return (t4 - t1) - (t3 - t2); + } + /// Remote minus local, assuming the two directions took equally long. + int64_t offset_ns() const + { + return floor_div((t2 - t1) + (t3 - t4), 2); + } + /// Time moved forward on both clocks and the round trip is not negative. + bool well_formed() const + { + return t4 >= t1 && t3 >= t2 && rtt_ns() >= 0; + } +}; + +/// A mapping of the remote clock onto the local one. The uncertainty is half the round trip plus the drift the age +/// allows; it is an error bound for asymmetric paths and clock rates, not an accuracy claim about anything else. +struct ClockMapping +{ + uint64_t epoch = 0; + int64_t offset_ns = 0; + int64_t rtt_ns = 0; + int64_t uncertainty_ns = 0; + /// Local time at which the exchange behind this mapping completed. + int64_t accepted_at_ns = 0; + + int64_t to_local_ns(int64_t remote_ns) const + { + return remote_ns - offset_ns; + } + bool operator==(const ClockMapping& other) const + { + return epoch == other.epoch && offset_ns == other.offset_ns && rtt_ns == other.rtt_ns && + uncertainty_ns == other.uncertainty_ns && accepted_at_ns == other.accepted_at_ns; + } +}; + +/** + * @brief Keeps the best recent exchange as the current mapping and notices when the clocks jump apart. + * + * Not thread-safe: one owner calls add(), reset() and mapping(). + */ +class ClockEstimator +{ +public: + explicit ClockEstimator(int64_t max_rtt_ns = kDefaultMaxRttNs, + int64_t stale_after_ns = kDefaultStaleAfterNs, + std::size_t window = kDefaultWindow, + double drift_allowance = kDefaultDriftAllowance) + : m_max_rtt_ns(max_rtt_ns), m_stale_after_ns(stale_after_ns), m_window(window), m_drift_allowance(drift_allowance) + { + } + + /// Start over, for a new source session or a restarted remote: the old mapping no longer applies. + void reset() + { + m_recent.clear(); + m_current.reset(); + m_newest_ns.reset(); + m_discontinuity = false; + m_accepted_in_epoch = 0; + ++m_epoch; + } + + /// Takes one exchange into account; returns why it was rejected, or nothing when it was accepted. + std::optional add(const Exchange& exchange) + { + if (!exchange.well_formed()) + return "malformed exchange"; + if (exchange.rtt_ns() > m_max_rtt_ns) + return "round trip too long"; + const std::optional previous = m_current; + m_newest_ns = exchange.t4; + const ClockMapping newest = mapping_of(exchange, exchange.t4); + ClockMapping candidate; + if (previous && disagrees(*previous, newest)) + { + // Not a refinement: the clocks moved. Judged on the newest exchange alone, so a jump is seen at once. + m_discontinuity = true; + ++m_epoch; + m_accepted_in_epoch = 0; + m_recent.clear(); + m_recent.push_back(exchange); + candidate = mapping_of(exchange, exchange.t4); + } + else + { + m_recent.push_back(exchange); + while (m_recent.size() > m_window) + m_recent.pop_front(); + const Exchange* best = &m_recent.front(); + for (const Exchange& e : m_recent) + { + const int64_t ue = uncertainty_ns(e, exchange.t4); + const int64_t ub = uncertainty_ns(*best, exchange.t4); + if (ue < ub || (ue == ub && e.t4 > best->t4)) + best = &e; + } + candidate = mapping_of(*best, exchange.t4); + } + ++m_accepted_in_epoch; + m_current = candidate; + return std::nullopt; + } + + /// The current mapping with its uncertainty aged to `now_ns`, or nothing when there is none yet or no exchange + /// was accepted within the staleness limit. + std::optional mapping(int64_t now_ns) const + { + if (!m_current || !m_newest_ns || now_ns - *m_newest_ns > m_stale_after_ns) + return std::nullopt; + ClockMapping aged = *m_current; + const int64_t age = std::max(0, now_ns - aged.accepted_at_ns); + const int64_t wanted = aged.rtt_ns / 2 + static_cast(m_drift_allowance * static_cast(age)); + if (wanted > aged.uncertainty_ns) + aged.uncertainty_ns = wanted; + return aged; + } + + /// Set when a new best mapping disagreed with the last by more than their uncertainties and the allowed drift + /// explain, and left set until reset(). + bool discontinuity() const + { + return m_discontinuity; + } + uint64_t epoch() const + { + return m_epoch; + } + /// Exchanges accepted since the epoch began (a new epoch starts at reset() and at a discontinuity). + std::size_t accepted_in_epoch() const + { + return m_accepted_in_epoch; + } + +private: + int64_t uncertainty_ns(const Exchange& e, int64_t now_ns) const + { + return e.rtt_ns() / 2 + + static_cast(m_drift_allowance * static_cast(std::max(0, now_ns - e.t4))); + } + + ClockMapping mapping_of(const Exchange& e, int64_t now_ns) const + { + ClockMapping m; + m.epoch = m_epoch; + m.offset_ns = e.offset_ns(); + m.rtt_ns = e.rtt_ns(); + m.uncertainty_ns = uncertainty_ns(e, now_ns); + m.accepted_at_ns = e.t4; + return m; + } + + bool disagrees(const ClockMapping& previous, const ClockMapping& candidate) const + { + const int64_t elapsed = std::max(0, candidate.accepted_at_ns - previous.accepted_at_ns); + const double allowed = static_cast(previous.uncertainty_ns) + + static_cast(candidate.uncertainty_ns) + + m_drift_allowance * static_cast(elapsed); + const int64_t delta = candidate.offset_ns - previous.offset_ns; + return static_cast(delta < 0 ? -delta : delta) > allowed; + } + + int64_t m_max_rtt_ns; + int64_t m_stale_after_ns; + std::size_t m_window; + double m_drift_allowance; + std::deque m_recent; + uint64_t m_epoch = 1; + std::optional m_current; + std::optional m_newest_ns; + bool m_discontinuity = false; + std::size_t m_accepted_in_epoch = 0; +}; + +} // namespace camera_bridge diff --git a/examples/camera_livekit_bridge/sender_clock/CMakeLists.txt b/examples/camera_livekit_bridge/sender_clock/CMakeLists.txt new file mode 100644 index 0000000000..06cee09b1b --- /dev/null +++ b/examples/camera_livekit_bridge/sender_clock/CMakeLists.txt @@ -0,0 +1,22 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +# Measures the sender's clock over its UDP clock responder. +add_library(sender_clock STATIC sender_clock_client.cpp) + +target_include_directories(sender_clock + INTERFACE + $ +) + +target_link_libraries(sender_clock + PUBLIC + camera_livekit_bridge::clock_exchange + PRIVATE + camera_livekit_bridge::video_types + Threads::Threads +) + +target_compile_options(sender_clock PRIVATE -Wall -Wextra) + +add_library(camera_livekit_bridge::sender_clock ALIAS sender_clock) diff --git a/examples/camera_livekit_bridge/sender_clock/inc/sender_clock/sender_clock_client.hpp b/examples/camera_livekit_bridge/sender_clock/inc/sender_clock/sender_clock_client.hpp new file mode 100644 index 0000000000..389bf67875 --- /dev/null +++ b/examples/camera_livekit_bridge/sender_clock/inc/sender_clock/sender_clock_client.hpp @@ -0,0 +1,98 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#pragma once + +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace camera_bridge +{ + +struct SenderClockConfig +{ + /// IPv4 address of the sender (a literal address, not a name) and its clock responder's UDP port. + std::string host; + std::uint16_t port = 0; + /// Time between requests. A step of the sender's clock is seen at the next reply, so this bounds how long a + /// stale mapping can be used. + std::chrono::milliseconds interval{ 200 }; + /// How long a reply is waited for. Later replies are discarded: a measurement needs t4 near the receipt. + std::chrono::milliseconds reply_timeout{ 150 }; +}; + +struct SenderClockSnapshot +{ + std::optional mapping; + bool discontinuity = false; + uint64_t epoch = 0; + std::size_t accepted_in_epoch = 0; + uint64_t exchanges = 0; + uint64_t rejected = 0; + /// Replies that were not a measurement: wrong size or tag, wrong peer, not matching the outstanding request, or + /// from before a reset. + uint64_t ignored = 0; + uint64_t timeouts = 0; + /// Bumped by every reset; an exchange only counts if its request was sent in the current generation. + uint64_t generation = 0; + /// Why the last request failed, empty while healthy. + std::string last_error; +}; + +/** + * @brief Measures the sender's clock against this host's CLOCK_MONOTONIC over the sender's UDP clock responder. + * + * A thread sends the 13-byte request ('Q', u32 seq, i64 t1, little-endian) to the sender at the configured interval + * and feeds the 29-byte reply ('R', seq, t1, t2, t3: the sender's clock at receipt and at send) to a + * ClockEstimator. Only a reply from the sender's address, of exactly that size and tag, matching the one + * outstanding request, and sent in the current generation counts. A loss is a missed sample; there is no + * connection to lose, so a reset (new session) is the only thing that invalidates earlier measurements. + */ +class SenderClockClient +{ +public: + explicit SenderClockClient(SenderClockConfig config); + ~SenderClockClient(); + + SenderClockClient(const SenderClockClient&) = delete; + SenderClockClient& operator=(const SenderClockClient&) = delete; + + void start(); + void stop(); + + /// Start a new mapping epoch: nothing measured before applies, and the outstanding request is void. + void reset(); + + /// Tell the client which capture session the frames now belong to. The first call and every change of session id + /// reset it (once, however many streams report it). + void begin_session(uint64_t session_id); + + /// A consistent copy of the estimator's state, with the mapping's uncertainty aged to now. + SenderClockSnapshot snapshot() const; + +private: + void run(); + + SenderClockConfig m_config; + mutable std::mutex m_mutex; + ClockEstimator m_estimator; + uint64_t m_generation = 0; + uint64_t m_exchanges = 0; + uint64_t m_rejected = 0; + uint64_t m_ignored = 0; + uint64_t m_timeouts = 0; + std::string m_last_error; + std::optional m_session_id; + std::atomic m_running{ false }; + std::thread m_thread; +}; + +} // namespace camera_bridge diff --git a/examples/camera_livekit_bridge/sender_clock/sender_clock_client.cpp b/examples/camera_livekit_bridge/sender_clock/sender_clock_client.cpp new file mode 100644 index 0000000000..a4719ca1cb --- /dev/null +++ b/examples/camera_livekit_bridge/sender_clock/sender_clock_client.cpp @@ -0,0 +1,191 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#include "inc/sender_clock/sender_clock_client.hpp" + +#include +#include +#include + +#include +#include +#include +#include +#include + +namespace camera_bridge +{ + +namespace +{ + +constexpr std::size_t kRequestSize = 13; +constexpr std::size_t kReplySize = 29; + +void put_le(std::uint8_t* out, std::uint64_t value, std::size_t bytes) +{ + for (std::size_t i = 0; i < bytes; ++i) + out[i] = static_cast((value >> (8 * i)) & 0xFF); +} + +std::uint64_t get_le(const std::uint8_t* in, std::size_t bytes) +{ + std::uint64_t value = 0; + for (std::size_t i = 0; i < bytes; ++i) + value |= static_cast(in[i]) << (8 * i); + return value; +} + +} // namespace + +SenderClockClient::SenderClockClient(SenderClockConfig config) : m_config(std::move(config)) +{ +} + +SenderClockClient::~SenderClockClient() +{ + stop(); +} + +void SenderClockClient::start() +{ + if (m_running.exchange(true)) + return; + m_thread = std::thread([this] { run(); }); +} + +void SenderClockClient::stop() +{ + m_running = false; + if (m_thread.joinable()) + m_thread.join(); +} + +void SenderClockClient::reset() +{ + std::lock_guard lock(m_mutex); + ++m_generation; + m_estimator.reset(); +} + +void SenderClockClient::begin_session(uint64_t session_id) +{ + std::lock_guard lock(m_mutex); + if (m_session_id && *m_session_id == session_id) + return; + // A change of session: every exchange so far was made before it. The first session also starts clean, so a + // mapping measured before any frame arrived is not trusted for frames of a session that began later. + m_session_id = session_id; + ++m_generation; + m_estimator.reset(); +} + +SenderClockSnapshot SenderClockClient::snapshot() const +{ + std::lock_guard lock(m_mutex); + SenderClockSnapshot out; + out.mapping = m_estimator.mapping(monotonic_ns()); + out.discontinuity = m_estimator.discontinuity(); + out.epoch = m_estimator.epoch(); + out.accepted_in_epoch = m_estimator.accepted_in_epoch(); + out.exchanges = m_exchanges; + out.rejected = m_rejected; + out.ignored = m_ignored; + out.timeouts = m_timeouts; + out.generation = m_generation; + out.last_error = m_last_error; + return out; +} + +void SenderClockClient::run() +{ + sockaddr_in peer{}; + peer.sin_family = AF_INET; + peer.sin_port = htons(m_config.port); + const int fd = ::socket(AF_INET, SOCK_DGRAM, 0); + if (fd < 0 || inet_pton(AF_INET, m_config.host.c_str(), &peer.sin_addr) != 1) + { + if (fd >= 0) + ::close(fd); + std::lock_guard lock(m_mutex); + m_last_error = "sender clock host must be an IPv4 address: " + m_config.host; + m_running = false; + return; + } + + std::uint32_t sequence = 0; + while (m_running) + { + ++sequence; + std::uint8_t request[kRequestSize]; + request[0] = 'Q'; + put_le(request + 1, sequence, 4); + uint64_t generation; + { + std::lock_guard lock(m_mutex); + generation = m_generation; + } + // t1 is read as late as possible, immediately before the send. + const int64_t t1 = monotonic_ns(); + put_le(request + 5, static_cast(t1), 8); + const ssize_t sent = ::sendto(fd, request, sizeof(request), 0, reinterpret_cast(&peer), sizeof(peer)); + + bool answered = false; + if (sent == static_cast(sizeof(request))) + { + const auto deadline = std::chrono::steady_clock::now() + m_config.reply_timeout; + while (!answered && m_running) + { + const auto remaining = + std::chrono::duration_cast(deadline - std::chrono::steady_clock::now()); + if (remaining.count() <= 0) + break; + pollfd wait{ fd, POLLIN, 0 }; + const int ready = ::poll(&wait, 1, static_cast(std::min(remaining.count(), 50))); + if (ready <= 0) + continue; + std::uint8_t reply[64]; + sockaddr_in from{}; + socklen_t from_len = sizeof(from); + const ssize_t got = + ::recvfrom(fd, reply, sizeof(reply), 0, reinterpret_cast(&from), &from_len); + const int64_t t4 = monotonic_ns(); // first, so it is as close to the receipt as software allows + if (got < 0) + continue; + std::lock_guard lock(m_mutex); + // A datagram that is not the answer to this request is not a measurement; neither is one sent + // before the last reset. + if (from.sin_addr.s_addr != peer.sin_addr.s_addr || from.sin_port != peer.sin_port || + got != static_cast(kReplySize) || reply[0] != 'R' || get_le(reply + 1, 4) != sequence || + static_cast(get_le(reply + 5, 8)) != t1 || generation != m_generation) + { + ++m_ignored; + continue; + } + Exchange exchange; + exchange.t1 = t1; + exchange.t2 = static_cast(get_le(reply + 13, 8)); + exchange.t3 = static_cast(get_le(reply + 21, 8)); + exchange.t4 = t4; + ++m_exchanges; + if (m_estimator.add(exchange)) + ++m_rejected; + m_last_error.clear(); + answered = true; + } + } + if (!answered) + { + std::lock_guard lock(m_mutex); + ++m_timeouts; + m_last_error = sent == static_cast(sizeof(request)) ? "no clock reply" : std::strerror(errno); + } + // Sleep in small steps so a stop request is honoured promptly. + const auto wake = std::chrono::steady_clock::now() + m_config.interval; + while (m_running && std::chrono::steady_clock::now() < wake) + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + ::close(fd); +} + +} // namespace camera_bridge diff --git a/examples/camera_livekit_bridge/tests/capture_qualifier/CMakeLists.txt b/examples/camera_livekit_bridge/tests/capture_qualifier/CMakeLists.txt new file mode 100644 index 0000000000..06ea53f2ac --- /dev/null +++ b/examples/camera_livekit_bridge/tests/capture_qualifier/CMakeLists.txt @@ -0,0 +1,13 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +add_executable(test_capture_qualifier test_capture_qualifier.cpp) + +target_link_libraries(test_capture_qualifier + PRIVATE + camera_livekit_bridge::clock_exchange +) + +target_compile_options(test_capture_qualifier PRIVATE -Wall -Wextra) + +add_test(NAME capture_qualifier COMMAND test_capture_qualifier) diff --git a/examples/camera_livekit_bridge/tests/capture_qualifier/test_capture_qualifier.cpp b/examples/camera_livekit_bridge/tests/capture_qualifier/test_capture_qualifier.cpp new file mode 100644 index 0000000000..bdc0da8e43 --- /dev/null +++ b/examples/camera_livekit_bridge/tests/capture_qualifier/test_capture_qualifier.cpp @@ -0,0 +1,258 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#include +#include + +#include +#include +#include +#include + +namespace +{ + +using camera_bridge::CaptureInput; +using camera_bridge::CaptureQualifier; +using camera_bridge::CaptureResult; +using camera_bridge::CaptureVerdict; +using camera_bridge::ClockEstimator; +using camera_bridge::ClockMapping; +using camera_bridge::Exchange; + +int g_failures = 0; + +void check(bool condition, const std::string& what) +{ + if (!condition) + { + std::cerr << "FAILED: " << what << "\n"; + ++g_failures; + } +} + +constexpr int64_t kMs = 1'000'000; +constexpr int64_t kOffset = 5'000'000'000; +constexpr int64_t kReady = 2'000'000'000'000; + +ClockMapping make_mapping(uint64_t epoch = 1, int64_t uncertainty_ns = 100'000, int64_t offset_ns = kOffset) +{ + ClockMapping m; + m.epoch = epoch; + m.offset_ns = offset_ns; + m.rtt_ns = 200'000; + m.uncertainty_ns = uncertainty_ns; + m.accepted_at_ns = kReady - 100 * kMs; + return m; +} + +/// A frame ready at `ready` whose capture was `age_ns` earlier, stamped on a sender clock `offset_ns` ahead. +CaptureInput make_input(int64_t age_ns, int64_t ready = kReady, int64_t offset_ns = kOffset, int64_t upstream = 0) +{ + CaptureInput in; + in.sender_valid = true; + in.capture_sm_ns = ready - age_ns + offset_ns; + in.upstream_uncertainty_ns = upstream; + in.ready_monotonic_ns = ready; + return in; +} + +/// Checks the verdict and that nothing but a qualified frame carries a capture time. +CaptureResult expect(CaptureQualifier& q, + const CaptureInput& in, + const std::optional& m, + std::size_t accepted, + CaptureVerdict verdict, + const std::string& what) +{ + const CaptureResult r = q.qualify(in, m, accepted); + check(r.verdict == verdict, + what + ": expected " + camera_bridge::to_string(verdict) + ", got " + camera_bridge::to_string(r.verdict)); + if (r.verdict != CaptureVerdict::Qualified) + check(r.capture_bridge_ns == 0, what + ": a non-qualified frame carries no capture time"); + return r; +} + +/// A qualifier that has learned a steady 20 ms age from 8 qualified frames. + +void test_verdict_table() +{ + const ClockMapping m = make_mapping(); + { + CaptureQualifier q; + CaptureInput invalid = make_input(20 * kMs); + invalid.sender_valid = false; + expect(q, invalid, m, 10, CaptureVerdict::SenderInvalid, "invalid sender record"); + expect(q, invalid, std::nullopt, 0, CaptureVerdict::SenderInvalid, "invalid record wins over no mapping"); + expect(q, make_input(20 * kMs), std::nullopt, 0, CaptureVerdict::NoMapping, "no mapping"); + } + { + CaptureQualifier q; + expect(q, make_input(20 * kMs), m, 2, CaptureVerdict::Warmup, "two accepted exchanges"); + expect(q, make_input(20 * kMs), m, 3, CaptureVerdict::Qualified, "three accepted exchanges"); + } + { + // Budget is 5 ms on mapping + upstream uncertainty; exactly at it qualifies. + CaptureQualifier q; + const ClockMapping at_budget = make_mapping(1, 3 * kMs); + const CaptureResult ok = expect( + q, make_input(20 * kMs, kReady, kOffset, 2 * kMs), at_budget, 10, CaptureVerdict::Qualified, "at budget"); + check(ok.combined_uncertainty_ns == 5 * kMs, "combined uncertainty at the budget"); + const CaptureResult over = expect(q, make_input(20 * kMs, kReady, kOffset, 2 * kMs + 1), at_budget, 10, + CaptureVerdict::OverBudget, "one ns over budget"); + check(over.combined_uncertainty_ns == 5 * kMs + 1, "over-budget result reports the combined uncertainty"); + expect(q, make_input(20 * kMs), make_mapping(1, 5 * kMs + 1), 10, CaptureVerdict::OverBudget, + "mapping alone over budget"); + } + { + // Age plausibility: not after ready, not more than 1 s before it. + const struct + { + int64_t age; + CaptureVerdict verdict; + const char* what; + } cases[] = { + { -1, CaptureVerdict::ImplausibleAge, "capture 1 ns after ready" }, + { 0, CaptureVerdict::Qualified, "capture at ready" }, + { 1'000'000'000, CaptureVerdict::Qualified, "capture exactly max_age before ready" }, + { 1'000'000'001, CaptureVerdict::ImplausibleAge, "capture over max_age before ready" }, + }; + for (const auto& c : cases) + { + CaptureQualifier q; + expect(q, make_input(c.age), m, 10, c.verdict, c.what); + } + } +} + +void test_qualified_result() +{ + CaptureQualifier q; + const ClockMapping m = make_mapping(7, 150'000); + CaptureInput in = make_input(20 * kMs, kReady, kOffset, 250'000); + const CaptureResult r = q.qualify(in, m, 10); + check(r.verdict == CaptureVerdict::Qualified, "qualified"); + check(r.capture_bridge_ns == in.capture_sm_ns - kOffset, "capture_bridge_ns = capture_sm_ns - offset_ns"); + check(r.capture_bridge_ns == kReady - 20 * kMs, "capture_bridge_ns is the capture instant on the host clock"); + check(r.combined_uncertainty_ns == 400'000, "combined uncertainty is mapping plus upstream"); + check(r.epoch == 7, "epoch is the mapping's"); + + // Negative and large offsets map the same way. + const ClockMapping negative = make_mapping(1, 100'000, -3'000'000'000); + CaptureQualifier q2; + const CaptureResult r2 = q2.qualify(make_input(20 * kMs, kReady, -3'000'000'000), negative, 10); + check(r2.verdict == CaptureVerdict::Qualified && r2.capture_bridge_ns == kReady - 20 * kMs, + "negative offset maps the capture time"); +} + +void test_no_learned_age_level() +{ + // A steady history decides nothing: the age checks are the physical ones (not negative, not over the maximum). + const ClockMapping m = make_mapping(); + CaptureQualifier q; + for (int i = 0; i < 20; ++i) + expect(q, make_input(20 * kMs), m, 10, CaptureVerdict::Qualified, "steady age"); + expect(q, make_input(400 * kMs), m, 10, CaptureVerdict::Qualified, "a longer but possible age is not withheld"); + expect(q, make_input(-1), m, 10, CaptureVerdict::ImplausibleAge, "a capture after ready is withheld"); +} + +void test_epoch_change() +{ + const ClockMapping m = make_mapping(); + CaptureQualifier q; + expect(q, make_input(20 * kMs), m, 10, CaptureVerdict::Qualified, "first epoch"); + const ClockMapping next = make_mapping(2, 100'000, kOffset + 80 * kMs); + expect(q, make_input(20 * kMs, kReady, kOffset + 80 * kMs), next, 1, CaptureVerdict::Warmup, + "a new epoch waits for exchanges"); + const CaptureResult r = expect(q, make_input(20 * kMs, kReady, kOffset + 80 * kMs), next, 3, + CaptureVerdict::Qualified, "a warm new epoch qualifies"); + check(r.epoch == 2, "the result carries the new epoch"); +} + +/// Simulates a sender whose monotonic clock steps by `step_ns` while frames keep being stamped on it, with the +/// estimator and the qualifier wired together as the bridge does. +void run_clock_step(int64_t step_ns) +{ + const std::string label = "step " + std::to_string(step_ns / kMs) + " ms: "; + constexpr int64_t kRtt = 200'000; + constexpr int64_t kProcessing = 50'000; + constexpr int64_t kAge = 20 * kMs; + constexpr int64_t kStepAt = 1'100 * kMs; // between two exchanges + constexpr int64_t kEnd = 4'000 * kMs; + constexpr int64_t kExchange = 200 * kMs; + const auto sender_offset = [&](int64_t t) { return kOffset + (t >= kStepAt ? step_ns : 0); }; + + ClockEstimator estimator; + CaptureQualifier qualifier; + int qualified_before = 0; + int withheld_after_step = 0; + int qualified_at_end = 0; + int wrong_in_exposure = 0; + for (int64_t t = 0; t <= kEnd; t += kMs) + { + const int64_t now = kReady + t; + if (t % (200 * kMs) == 0) + { + const int64_t t1 = now - kRtt - kProcessing; + const int64_t t2 = t1 + sender_offset(t) + kRtt / 2; + estimator.add(Exchange{ t1, t2, t2 + kProcessing, now }); + } + if (t % (33 * kMs) != 0) + continue; + const CaptureInput in = make_input(kAge, now, sender_offset(t), 500'000); + const auto mapping = estimator.mapping(now); + const CaptureResult r = qualifier.qualify(in, mapping, estimator.accepted_in_epoch()); + if (r.verdict == CaptureVerdict::Qualified) + { + const int64_t truth = now - kAge; + const int64_t error = r.capture_bridge_ns > truth ? r.capture_bridge_ns - truth : truth - r.capture_bridge_ns; + // The one exposure: frames between the step and the next exchange are mapped with the old offset. + const bool exposed = t >= kStepAt && t < kStepAt + kExchange; + if (!exposed) + check(error <= r.combined_uncertainty_ns, + label + "a qualified frame at t=" + std::to_string(t / kMs) + " ms has a wrong capture time"); + if (exposed) + ++wrong_in_exposure; + if (t < kStepAt) + ++qualified_before; + if (t >= kEnd - 200 * kMs) + ++qualified_at_end; + } + else + { + check(r.capture_bridge_ns == 0, label + "a non-qualified frame carries no capture time"); + if (t >= kStepAt && t < kStepAt + kExchange) + ++withheld_after_step; + } + } + check(qualified_before > 0, label + "frames qualify before the step"); + (void)wrong_in_exposure; // bounded by construction: only frames before the next exchange are in the window + (void)withheld_after_step; + check(estimator.discontinuity() && estimator.epoch() == 2, label + "the estimator sees the step"); + check(qualified_at_end > 0, label + "frames qualify again once the new epoch is warm"); +} + +void test_clock_steps() +{ + for (const int64_t step : { 50 * kMs, -50 * kMs, 250 * kMs, -250 * kMs, 2'000 * kMs, -2'000 * kMs }) + run_clock_step(step); +} + +} // namespace + +int main() +{ + test_verdict_table(); + test_qualified_result(); + test_no_learned_age_level(); + test_epoch_change(); + test_clock_steps(); + + if (g_failures != 0) + { + std::cerr << g_failures << " check(s) failed\n"; + return 1; + } + std::cout << "all capture_qualifier checks passed\n"; + return 0; +} diff --git a/examples/camera_livekit_bridge/tests/clock_estimator/CMakeLists.txt b/examples/camera_livekit_bridge/tests/clock_estimator/CMakeLists.txt new file mode 100644 index 0000000000..0ce69a05b2 --- /dev/null +++ b/examples/camera_livekit_bridge/tests/clock_estimator/CMakeLists.txt @@ -0,0 +1,13 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +add_executable(test_clock_estimator test_clock_estimator.cpp) + +target_link_libraries(test_clock_estimator + PRIVATE + camera_livekit_bridge::clock_exchange +) + +target_compile_options(test_clock_estimator PRIVATE -Wall -Wextra) + +add_test(NAME clock_estimator COMMAND test_clock_estimator) diff --git a/examples/camera_livekit_bridge/tests/clock_estimator/test_clock_estimator.cpp b/examples/camera_livekit_bridge/tests/clock_estimator/test_clock_estimator.cpp new file mode 100644 index 0000000000..48c57e27dd --- /dev/null +++ b/examples/camera_livekit_bridge/tests/clock_estimator/test_clock_estimator.cpp @@ -0,0 +1,280 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +// Golden numbers were produced by running egocap's Python ClockEstimator +// (examples/egocap/backend/egocap_web/clock_exchange.py) over the sequences built by make_exchange() below; +// the test does not run Python. + +#include + +#include +#include +#include +#include +#include +#include + +namespace +{ + +using camera_bridge::ClockEstimator; +using camera_bridge::ClockMapping; +using camera_bridge::Exchange; + +int g_failures = 0; + +void check(bool condition, const char* what) +{ + if (!condition) + { + std::cerr << "FAILED: " << what << "\n"; + ++g_failures; + } +} + +constexpr int64_t kT0 = 1'000'000'000'000; +constexpr int64_t kSpacing = 200'000'000; +constexpr int64_t kProcessing = 50'000; +constexpr int64_t kOffset = 5'000'000'000; + +/// Exchange i of a sequence: t1 local, t2 = t1 + offset + rtt/2 (symmetric path), t3 = t2 + processing time, +/// t4 = t1 + rtt + processing time. +Exchange make_exchange(int i, int64_t rtt_ns, int64_t offset_ns) +{ + const int64_t t1 = kT0 + i * kSpacing; + const int64_t t2 = t1 + offset_ns + rtt_ns / 2; + return Exchange{ t1, t2, t2 + kProcessing, t1 + rtt_ns + kProcessing }; +} + +struct Golden +{ + uint64_t epoch; + int64_t offset_ns; + int64_t rtt_ns; + int64_t uncertainty_ns; + int64_t accepted_at_ns; +}; + +bool matches(const std::optional& m, const Golden& g) +{ + return m && m->epoch == g.epoch && m->offset_ns == g.offset_ns && m->rtt_ns == g.rtt_ns && + m->uncertainty_ns == g.uncertainty_ns && m->accepted_at_ns == g.accepted_at_ns; +} + +// Sequence A: 12 exchanges, rtt 200 us, offset 5e9 ns, 200 ms apart; mapping(t4) after each. +const std::array kGoldenA = { { + { 1, 5000000000, 200000, 100000, 1000000250000 }, + { 1, 5000000000, 200000, 100000, 1000200250000 }, + { 1, 5000000000, 200000, 100000, 1000400250000 }, + { 1, 5000000000, 200000, 100000, 1000600250000 }, + { 1, 5000000000, 200000, 100000, 1000800250000 }, + { 1, 5000000000, 200000, 100000, 1001000250000 }, + { 1, 5000000000, 200000, 100000, 1001200250000 }, + { 1, 5000000000, 200000, 100000, 1001400250000 }, + { 1, 5000000000, 200000, 100000, 1001600250000 }, + { 1, 5000000000, 200000, 100000, 1001800250000 }, + { 1, 5000000000, 200000, 100000, 1002000250000 }, + { 1, 5000000000, 200000, 100000, 1002200250000 }, +} }; + +// Sequence B: as A, but exchange 2 has rtt 40 us, so best-of-window keeps it (the mapping keeps its +// accepted_at_ns) until the drift allowance over its age makes a fresh 200 us exchange cheaper. +const std::array kRttsB = { 200000, 200000, 40000, 200000, 200000, 200000, + 200000, 200000, 200000, 200000, 200000, 200000 }; +const std::array kGoldenB = { { + { 1, 5000000000, 200000, 100000, 1000000250000 }, + { 1, 5000000000, 200000, 100000, 1000200250000 }, + { 1, 5000000000, 40000, 20000, 1000400090000 }, + { 1, 5000000000, 40000, 60032, 1000400090000 }, + { 1, 5000000000, 200000, 100000, 1000800250000 }, + { 1, 5000000000, 200000, 100000, 1001000250000 }, + { 1, 5000000000, 200000, 100000, 1001200250000 }, + { 1, 5000000000, 200000, 100000, 1001400250000 }, + { 1, 5000000000, 200000, 100000, 1001600250000 }, + { 1, 5000000000, 200000, 100000, 1001800250000 }, + { 1, 5000000000, 200000, 100000, 1002000250000 }, + { 1, 5000000000, 200000, 100000, 1002200250000 }, +} }; + +// Sequence C: six exchanges at offset 5e9, then three at 5.3e9 (a +300 ms jump). +const std::array kGoldenC = { { + { 1, 5000000000, 200000, 100000, 1000000250000 }, + { 1, 5000000000, 200000, 100000, 1000200250000 }, + { 1, 5000000000, 200000, 100000, 1000400250000 }, + { 1, 5000000000, 200000, 100000, 1000600250000 }, + { 1, 5000000000, 200000, 100000, 1000800250000 }, + { 1, 5000000000, 200000, 100000, 1001000250000 }, + { 2, 5300000000, 200000, 100000, 1001200250000 }, + { 2, 5300000000, 200000, 100000, 1001400250000 }, + { 2, 5300000000, 200000, 100000, 1001600250000 }, +} }; + +void test_clean_sequence() +{ + ClockEstimator est; + check(!est.mapping(kT0).has_value(), "no mapping before the first exchange"); + for (int i = 0; i < 12; ++i) + { + const Exchange x = make_exchange(i, 200'000, kOffset); + check(!est.add(x).has_value(), "clean exchange accepted"); + check(matches(est.mapping(x.t4), kGoldenA[i]), "sequence A mapping equals the Python estimator's"); + check(est.accepted_in_epoch() == static_cast(i + 1), "accepted_in_epoch counts exchanges"); + check(!est.discontinuity(), "clean sequence has no discontinuity"); + check(est.epoch() == 1, "clean sequence stays in epoch 1"); + } +} + +void test_best_of_window() +{ + ClockEstimator est; + for (int i = 0; i < 12; ++i) + { + const Exchange x = make_exchange(i, kRttsB[i], kOffset); + check(!est.add(x).has_value(), "exchange B accepted"); + check(matches(est.mapping(x.t4), kGoldenB[i]), "sequence B mapping equals the Python estimator's"); + } +} + +void test_rejections() +{ + ClockEstimator est; + const auto slow = est.add(make_exchange(0, 300'000'000, kOffset)); + check(slow && *slow == "round trip too long", "slow exchange rejected as too long"); + check(!est.mapping(kT0 + 400'000'000).has_value() && est.accepted_in_epoch() == 0, + "a rejected exchange leaves no mapping"); + + const Exchange good = make_exchange(0, 200'000, kOffset); + const auto t4_before_t1 = est.add(Exchange{ good.t1, good.t2, good.t3, good.t1 - 1 }); + check(t4_before_t1 && *t4_before_t1 == "malformed exchange", "t4 < t1 is malformed"); + const auto t3_before_t2 = est.add(Exchange{ good.t1, good.t2, good.t2 - 1, good.t4 }); + check(t3_before_t2 && *t3_before_t2 == "malformed exchange", "t3 < t2 is malformed"); + const auto negative_rtt = est.add(Exchange{ 0, 0, 500, 100 }); + check(negative_rtt && *negative_rtt == "malformed exchange", "negative round trip is malformed"); + check(est.accepted_in_epoch() == 0 && !est.mapping(good.t4).has_value(), "malformed exchanges leave no mapping"); + + check(!est.add(make_exchange(0, 250'000'000, kOffset)).has_value(), "rtt exactly at the limit is accepted"); + const auto over = est.add(make_exchange(1, 250'000'001, kOffset)); + check(over && *over == "round trip too long", "rtt one ns over the limit is rejected"); +} + +void test_offset_jump() +{ + ClockEstimator est; + for (int i = 0; i < 9; ++i) + { + const Exchange x = make_exchange(i, 200'000, i < 6 ? kOffset : kOffset + 300'000'000); + check(!est.add(x).has_value(), "exchange C accepted"); + check(matches(est.mapping(x.t4), kGoldenC[i]), "sequence C mapping equals the Python estimator's"); + check(est.discontinuity() == (i >= 6), "discontinuity set from the jump on and left set"); + check(est.epoch() == (i >= 6 ? 2U : 1U), "epoch increments once at the jump"); + const std::size_t expected = i < 6 ? static_cast(i + 1) : static_cast(i - 5); + check(est.accepted_in_epoch() == expected, "accepted_in_epoch restarts at the jump"); + } + + // A step inside the uncertainties (100 us) is a refinement, not a discontinuity (Python: epoch 1, offset + // 5000100000). + ClockEstimator small; + for (int i = 0; i < 3; ++i) + small.add(make_exchange(i, 200'000, kOffset)); + small.add(make_exchange(3, 200'000, kOffset + 100'000)); + check(!small.discontinuity() && small.epoch() == 1, "a step within the uncertainty is not a discontinuity"); + const auto m = small.mapping(make_exchange(3, 200'000, 0).t4); + check(m && m->offset_ns == 5'000'100'000, "the refined offset is adopted"); +} + +void test_staleness_and_aging() +{ + ClockEstimator est; + for (int i = 0; i < 12; ++i) + est.add(make_exchange(i, 200'000, kOffset)); + const int64_t last = make_exchange(11, 200'000, 0).t4; + + // Golden uncertainties from the Python estimator at last + dt: 200 ppm of the age on top of rtt/2. + struct Aged + { + int64_t dt; + int64_t uncertainty; + }; + for (const Aged a : { Aged{ 0, 100000 }, Aged{ 1'000'000'000, 300000 }, Aged{ 2'500'000'000, 600000 }, + Aged{ 5'000'000'000, 1100000 } }) + { + const auto m = est.mapping(last + a.dt); + check(m && m->uncertainty_ns == a.uncertainty && m->accepted_at_ns == last && m->offset_ns == kOffset, + "uncertainty grows with age by 200 ppm"); + } + check(!est.mapping(last + 5'000'000'001).has_value(), "no mapping more than 5 s after the last exchange"); + const auto before = est.mapping(last - 1'000'000); + check(before && before->uncertainty_ns == 100'000, "a query before the exchange does not shrink the uncertainty"); + + // A rejected exchange does not refresh staleness. + est.add(make_exchange(30, 300'000'000, kOffset)); + check(!est.mapping(last + 5'000'000'001).has_value(), "a rejected exchange does not keep the mapping alive"); +} + +void test_reset() +{ + ClockEstimator est; + for (int i = 0; i < 5; ++i) + est.add(make_exchange(i, 200'000, kOffset)); + est.add(make_exchange(5, 200'000, kOffset + 300'000'000)); + check(est.discontinuity() && est.epoch() == 2, "setup: discontinuity in epoch 2"); + + est.reset(); + check(!est.mapping(make_exchange(5, 0, 0).t4).has_value(), "reset clears the mapping"); + check(est.epoch() == 3, "reset bumps the epoch"); + check(est.accepted_in_epoch() == 0, "reset clears accepted_in_epoch"); + // This C++ reset also clears the discontinuity flag; egocap's Python reset leaves it set. + check(!est.discontinuity(), "reset clears the discontinuity flag (C++ behaviour)"); + + const Exchange next = make_exchange(6, 200'000, kOffset); + check(!est.add(next).has_value(), "exchange after reset accepted"); + const auto m = est.mapping(next.t4); + check(m && m->epoch == 3 && m->offset_ns == kOffset && est.accepted_in_epoch() == 1, + "the new epoch starts from the first exchange after reset"); +} + +void test_floor_division() +{ + using camera_bridge::floor_div; + // Python: -1 // 2 == -1, -3 // 2 == -2, -5 // 2 == -3, -7 // 2 == -4, -1000000001 // 2 == -500000001, + // -5 // 3 == -2, -1000000001 // 3 == -333333334, 7 // 2 == 3, 5 // -2 == -3, -4 // 2 == -2, 0 // 2 == 0. + check(floor_div(-1, 2) == -1, "floor_div(-1, 2)"); + check(floor_div(-3, 2) == -2, "floor_div(-3, 2)"); + check(floor_div(-5, 2) == -3, "floor_div(-5, 2)"); + check(floor_div(-7, 2) == -4, "floor_div(-7, 2)"); + check(floor_div(-1'000'000'001, 2) == -500'000'001, "floor_div(-1000000001, 2)"); + check(floor_div(-5, 3) == -2, "floor_div(-5, 3)"); + check(floor_div(-1'000'000'001, 3) == -333'333'334, "floor_div(-1000000001, 3)"); + check(floor_div(7, 2) == 3, "floor_div(7, 2)"); + check(floor_div(5, -2) == -3, "floor_div(5, -2)"); + check(floor_div(-4, 2) == -2, "floor_div(-4, 2)"); + check(floor_div(0, 2) == 0, "floor_div(0, 2)"); + + // Python: Exchange(0,-4,-3,2).offset_ns == -5, .rtt_ns == 1; Exchange(0,5,6,3).offset_ns == 4. + check(Exchange{ 0, -4, -3, 2 }.offset_ns() == -5, "negative odd offset rounds down"); + check(Exchange{ 0, -4, -3, 2 }.rtt_ns() == 1, "rtt of the negative-offset exchange"); + check(Exchange{ 0, 5, 6, 3 }.offset_ns() == 4, "positive odd offset"); + check(Exchange{ 0, -4, -3, 0 }.offset_ns() == -4, "offset of a malformed exchange still floors"); + check(!Exchange{ 0, -4, -3, 0 }.well_formed(), "negative round trip is not well formed"); +} + +} // namespace + +int main() +{ + test_clean_sequence(); + test_best_of_window(); + test_rejections(); + test_offset_jump(); + test_staleness_and_aging(); + test_reset(); + test_floor_division(); + + if (g_failures != 0) + { + std::cerr << g_failures << " check(s) failed\n"; + return 1; + } + std::cout << "all clock_estimator checks passed\n"; + return 0; +} diff --git a/examples/camera_livekit_bridge/tests/python/test_capture_time.py b/examples/camera_livekit_bridge/tests/python/test_capture_time.py new file mode 100644 index 0000000000..8123f67dc9 --- /dev/null +++ b/examples/camera_livekit_bridge/tests/python/test_capture_time.py @@ -0,0 +1,687 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Capture time through a measured sender clock, end to end. + +mock sender (own clock, ground truth, UDP clock responder) -> camera_receiver -> +camera_livekit_bridge (`--timestamp-basis capture_monotonic_us --sender-clock +127.0.0.1:PORT`) -> LiveKit -> a +Python subscriber. Each published frame is joined to the mock's ground truth on +(stream, stream_sequence, session_id); the published `user_timestamp` is compared +with the true capture instant on this host's CLOCK_MONOTONIC. The bridge must +publish an accurate capture time or none at all: a wrong claim anywhere fails. + +Run: LIVEKIT_PYTHON=/path/to/python python3 test_capture_time.py -v +(about 2.5 minutes; needs a CUDA device, ffmpeg and the receiver venv). + +Environment overrides: LIVEKIT_SERVER_BIN, CAMERA_LIVEKIT_BRIDGE_BIN, +LIVEKIT_PYTHON (interpreter with the `livekit` SDK and numpy, used for the +subscriber). Skipped, never faked, when a prerequisite is missing. +""" + +from __future__ import annotations + +import json +import re +import socket +import statistics +import subprocess +import sys +import tempfile +import time +import unittest +import uuid +from pathlib import Path + +from test_bridge_pipeline import ( + BRIDGE_BIN, + MOCK_DIR, + RECEIVER_DIR, + RECEIVER_PY, + SERVER_BIN, + SOURCE_FPS, + SUBSCRIBER_PY, + _Proc, + _skip_reason, + free_port, + make_token, +) + +# name, source width, source height, track name, bridge divisor +STREAMS = [ + ("head", 640, 496, "head_track", 2), + ("wrist", 320, 240, "wrist_track", 1), +] +TRACK_STREAM = {track: name for name, _, _, track, _ in STREAMS} + +# Accuracy bounds, on a live run of the accuracy scenario the error was +# median +0.3 us, p95 22 us, max 37 us. The bounds sit well above that and well +# below anything a consumer would notice. +MEDIAN_BOUND_NS = 100_000 +P95_BOUND_NS = 500_000 +MAX_BOUND_NS = 2_000_000 + +# Runs under LIVEKIT_PYTHON. Waits for the first frame, then records every frame +# (track, arrival on CLOCK_MONOTONIC, frame_id, user_timestamp) for run_s. +SUBSCRIBER_SCRIPT = r""" +import asyncio, json, sys, time +from livekit import rtc + +url, token_file, run_s, out_file = sys.argv[1], sys.argv[2], float(sys.argv[3]), sys.argv[4] +token = open(token_file).read().strip() + + +async def main(): + room = rtc.Room() + rows = [] + state = {"end": None} + + async def read(name, track): + async for event in rtc.VideoStream(track, capacity=8): + now = time.monotonic_ns() + if state["end"] is None: + state["end"] = time.monotonic() + run_s + m = event.metadata + rows.append([ + name, now, + m.frame_id if m is not None and m.HasField("frame_id") else None, + m.user_timestamp if m is not None and m.HasField("user_timestamp") else None, + ]) + + @room.on("track_subscribed") + def on_track(track, publication, participant): + if track.kind == rtc.TrackKind.KIND_VIDEO: + asyncio.create_task(read(publication.name, track)) + + await room.connect(url, token) + hard_end = time.monotonic() + 120 + run_s + while time.monotonic() < hard_end: + await asyncio.sleep(0.25) + if state["end"] is not None and time.monotonic() > state["end"]: + break + await room.disconnect() + with open(out_file, "w") as f: + json.dump(rows, f) + print("RESULT ok", len(rows)) + + +asyncio.run(main()) +""" + + +def _pctl(sorted_values: list[float], fraction: float) -> float: + return sorted_values[ + min(len(sorted_values) - 1, int(len(sorted_values) * fraction)) + ] + + +class Run: + """One scenario's frames joined to ground truth, plus the bridge log.""" + + def __init__(self, rows, truth_lines, bridge_log: str): + self.bridge_log = bridge_log + # session_id -> start (first ready instant, the mock's connection start) + self.sessions: dict[int, float] = {} + truth = {} + by_key: dict[tuple, list[tuple[int, dict]]] = {} + for line in truth_lines: + r = json.loads(line) + sid = r["session_id"] + start = self.sessions.get(sid) + if start is None or r["ready_pc_mono_ns"] < start: + self.sessions[sid] = r["ready_pc_mono_ns"] + truth[(r["stream"], r["stream_sequence"], sid)] = r + by_key.setdefault((r["stream"], r["stream_sequence"]), []).append((sid, r)) + self.frames: list[dict] = [] + self.unjoined = 0 + for track, arrival, frame_id, ts in rows: + stream = TRACK_STREAM.get(track) + if stream is None or frame_id is None: + self.unjoined += 1 + continue + # The sequence repeats across sessions; the session whose frame was + # ready most recently before this arrival is the one delivered. + cands = [ + (arrival - r["ready_pc_mono_ns"], sid, r) + for sid, r in by_key.get((stream, frame_id), []) + if arrival >= r["ready_pc_mono_ns"] + ] + if not cands: + self.unjoined += 1 + continue + _, sid, r = min(cands, key=lambda c: c[0]) + self.frames.append( + { + "stream": stream, + "session": sid, + "id": frame_id, + "ts": ts, + "arrival": arrival, + "t_rel": (r["ready_pc_mono_ns"] - self.sessions[sid]) / 1e9, + "err_ns": None + if ts is None + else ts * 1000 - r["true_capture_pc_mono_ns"], + "age_ms": (arrival - r["true_capture_pc_mono_ns"]) / 1e6, + } + ) + self.frames.sort(key=lambda f: f["arrival"]) + + def session_order(self) -> list[int]: + return sorted(self.sessions, key=self.sessions.get) + + def select(self, session=None, lo=None, hi=None, stream=None) -> list[dict]: + return [ + f + for f in self.frames + if (session is None or f["session"] == session) + and (stream is None or f["stream"] == stream) + and (lo is None or f["t_rel"] >= lo) + and (hi is None or f["t_rel"] < hi) + ] + + def errors(self, frames: list[dict]) -> list[float]: + return sorted(f["err_ns"] for f in frames if f["err_ns"] is not None) + + def summary(self, frames: list[dict] | None = None) -> str: + frames = self.frames if frames is None else frames + e = self.errors(frames) + spans = [f["t_rel"] for f in frames if f["ts"] is None] + withheld = len(spans) + if spans: + withheld = f"{withheld} (t_rel {min(spans):.2f}..{max(spans):.2f} s)" + if not e: + return f"frames={len(frames)} withheld={withheld} no claims" + return ( + f"frames={len(frames)} withheld={withheld} claims={len(e)} error_us " + f"median={statistics.median(e) / 1e3:.1f} " + f"p95_abs={_pctl(sorted(abs(x) for x in e), 0.95) / 1e3:.1f} " + f"max_abs={max(abs(x) for x in e) / 1e3:.1f}" + ) + + +def _stat_counts(log: str, reason: str) -> int: + """Largest cumulative count the bridge's stats lines reported for a withhold reason.""" + best = 0 + for line in log.splitlines(): + marker = f", {reason} " + if "[stats]" in line and marker in line: + tail = line.split(marker, 1)[1].split(",", 1)[0] + if tail.strip().isdigit(): + best = max(best, int(tail.strip())) + return best + + +def _qualified_count(log: str) -> int: + """Largest cumulative qualified-claim count in the bridge's stats lines.""" + found = [int(m) for m in re.findall(r"capture claims: (\d+) qualified", log)] + return max(found, default=0) + + +@unittest.skipIf(_skip_reason() is not None, _skip_reason() or "") +class CaptureTimeTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls._tmp = Path(tempfile.mkdtemp(prefix="capture_time_")) + cls._procs: list[_Proc] = [] + lk_port, lk_tcp, lk_udp = ( + free_port(socket.SOCK_STREAM), + free_port(socket.SOCK_STREAM), + free_port(socket.SOCK_DGRAM), + ) + cls.url = f"ws://127.0.0.1:{lk_port}" + try: + server = cls._spawn( + "livekit", + [ + str(SERVER_BIN), "--dev", "--bind", "127.0.0.1", "--node-ip", "127.0.0.1", + "--port", str(lk_port), "--rtc.tcp_port", str(lk_tcp), "--udp-port", str(lk_udp), + ], + ) # fmt: skip + server.wait_for("starting LiveKit server", 20) + deadline = time.monotonic() + 20 + while time.monotonic() < deadline: + try: + socket.create_connection(("127.0.0.1", lk_port), timeout=1).close() + break + except OSError: + time.sleep(0.2) + else: + raise AssertionError("livekit-server never accepted connections") + except BaseException: + cls._stop_all() + raise + + @classmethod + def tearDownClass(cls): + cls._stop_all() + + @classmethod + def _stop_all(cls): + for proc in reversed(cls._procs): + proc.stop() + cls._procs.clear() + + @classmethod + def _spawn(cls, name, argv, cwd=None) -> _Proc: + proc = _Proc(name, argv, cls._tmp, cwd) + cls._procs.append(proc) + return proc + + @classmethod + def _scenario(cls, name: str, mock_args: list[str], run_s: float) -> Run: + """One mock, receiver, bridge and subscriber against the shared server.""" + tmp = cls._tmp / name + tmp.mkdir() + n_before = len(cls._procs) + room = f"capture-time-{name}-{uuid.uuid4().hex[:10]}" + mock_port = free_port(socket.SOCK_STREAM) + clock_port = free_port(socket.SOCK_DGRAM) + sock = tmp / "receiver.sock" + truth_file = tmp / "ground_truth.jsonl" + + def spawn(label, argv, cwd=None): + return cls._spawn(f"{name}_{label}", argv, cwd) + + try: + cfg = tmp / "receiver.yaml" + lines = [ + "version: 1", "receiver:", " host: 127.0.0.1", f" port: {mock_port}", " gpu_id: 0", + " stats_log_interval_s: 5", "ipc:", f" socket_path: {sock}", + f" owner_name: capture_time_{name}_{uuid.uuid4().hex[:6]}", " queue_size: 1024", "streams:", + ] # fmt: skip + for i, (sname, w, h, _, _) in enumerate(STREAMS): + lines += [ + f" - index: {i}", f" name: {sname}", " type: mono", f" width: {w}", + f" height: {h}", " output_format: nv12", " ring_depth: 6", " ipc:", + f" mono: {sname}", + ] # fmt: skip + cfg.write_text("\n".join(lines) + "\n") + + # The receiver and bridge wait first and the mock starts last, so the + # mock's clock (session start, --clock-mute) begins as the receiver connects. + receiver = spawn( + "receiver", + [ + str(RECEIVER_PY), str(RECEIVER_DIR / "camera_receiver.py"), + "--config", str(cfg), "--port", str(mock_port), "--socket", str(sock), + ], + cwd=RECEIVER_DIR, + ) # fmt: skip + receiver.wait_for("camera_receiver running", 60) + + pub_token = tmp / "pub.token" + pub_token.write_text(make_token(room, f"capture-time-bridge-{name}", True)) + bridge_cmd = [ + str(BRIDGE_BIN), "--livekit-url", cls.url, "--livekit-token-file", str(pub_token), + "--socket", str(sock), "--fps", str(SOURCE_FPS), + "--timestamp-basis", "capture_monotonic_us", + "--sender-clock", f"127.0.0.1:{clock_port}", + ] # fmt: skip + for sname, _, _, track, div in STREAMS: + bridge_cmd += ["--stream", f"{sname}={track}:{div}"] + bridge = spawn("bridge", bridge_cmd) + + sub_token = tmp / "sub.token" + sub_token.write_text(make_token(room, f"capture-time-sub-{name}", False)) + script = tmp / "subscriber.py" + script.write_text(SUBSCRIBER_SCRIPT) + out_file = tmp / "frames.json" + sub_log = open(tmp / "subscriber.log", "wb") + sub = subprocess.Popen( + [str(SUBSCRIBER_PY), str(script), cls.url, str(sub_token), str(run_s), str(out_file)], + stdout=sub_log, + stderr=subprocess.STDOUT, + start_new_session=True, + ) # fmt: skip + try: + mock_cmd = [ + sys.executable, str(MOCK_DIR / "mock_sender.py"), + "--port", str(mock_port), "--duration", str(run_s + 60), + "--cache-dir", str(Path(tempfile.gettempdir()) / "camera_livekit_bridge_test_cache"), + "--timing-ext", "--ground-truth", str(truth_file), + "--clock-port", str(clock_port), + ] # fmt: skip + for sname, w, h, _, _ in STREAMS: + mock_cmd += ["--stream", f"{sname}:{w}x{h}@{SOURCE_FPS}"] + mock = spawn("mock", mock_cmd + mock_args) + mock.wait_for("mock sender listening", 120) + receiver.wait_for("connected to AVS1 sender", 60) + bridge.wait_for("publishing", 60) + try: + sub.wait(timeout=run_s + 150) + except subprocess.TimeoutExpired as exc: + raise AssertionError("subscriber timed out") from exc + finally: + if sub.poll() is None: + sub.terminate() + try: + sub.wait(timeout=10) + except subprocess.TimeoutExpired: + sub.kill() + sub_log.close() + if not out_file.is_file(): + raise AssertionError( + f"subscriber wrote no frames:\n{(tmp / 'subscriber.log').read_text()[-2000:]}\n" + f"bridge:\n{bridge.text()[-2000:]}" + ) + rows = json.loads(out_file.read_text()) + truth_lines = ( + truth_file.read_text().splitlines() if truth_file.is_file() else [] + ) + run = Run(rows, truth_lines, bridge.text()) + print( + f"\n[{name}] {run.summary()} unjoined={run.unjoined}", file=sys.stderr + ) + return run + finally: + for proc in reversed(cls._procs[n_before:]): + proc.stop() + del cls._procs[n_before:] + + # ---- shared checks ------------------------------------------------------ + + def _no_wrong_claim(self, run: Run, label: str): + wrong = [ + f + for f in run.frames + if f["err_ns"] is not None and abs(f["err_ns"]) > MAX_BOUND_NS + ] + self.assertFalse( + wrong, + f"{label}: {len(wrong)} frames carry a wrong capture time; first: " + f"{[(f['stream'], f['id'], round(f['t_rel'], 2), round(f['err_ns'] / 1e3)) for f in wrong[:8]]} us", + ) + + def _frame_ids_continuous(self, run: Run, label: str): + """Frames keep their source stream_sequence (they join to ground truth) without gaps from renumbering.""" + self.assertLess( + run.unjoined, + 0.05 * max(len(run.frames), 1) + 5, + f"{label}: {run.unjoined} frames did not join to ground truth", + ) + for sid in run.session_order(): + for stream in TRACK_STREAM.values(): + frames = run.select(session=sid, stream=stream) + if len(frames) < 30: + continue + ids = [f["id"] for f in frames] + steps = [b - a for a, b in zip(ids, ids[1:])] + self.assertTrue( + all(s > 0 for s in steps), + f"{label}: {stream} ids not increasing: {steps[:40]}", + ) + self.assertGreaterEqual( + sum(1 for s in steps if s == 1) / len(steps), 0.8, + f"{label}: {stream} ids skip: {steps[:60]}", + ) # fmt: skip + + def _claims_are_never_ready_time(self, run: Run, label: str): + """A claim is the mapped capture time, never the frame-ready time standing in for it.""" + for f in run.frames: + if f["ts"] is None: + continue + # Ready time is the sof lag (20 ms) after capture: a substituted claim would show that much error. + self.assertLess( + abs(f["err_ns"]), + MAX_BOUND_NS, + f"{label}: frame {f['id']} claim off by {f['err_ns']} ns", + ) + + def _assert_all_withheld_without_stamp(self, frames: list[dict], label: str): + stamped = [f for f in frames if f["ts"] is not None] + self.assertFalse( + stamped, + f"{label}: {len(stamped)} of {len(frames)} frames in the window carry a timestamp: " + f"{[(f['stream'], f['id'], round(f['t_rel'], 2)) for f in stamped[:8]]}", + ) + + # ---- scenarios ---------------------------------------------------------- + + def test_1_accuracy(self): + run = self._scenario( + "accuracy", ["--sm-offset-ms", "1234", "--sm-drift-ppm", "50"], 20.0 + ) + self._frame_ids_continuous(run, "accuracy") + self.assertGreaterEqual(len(run.frames), 100, run.summary()) + claims = run.errors(run.frames) + self.assertGreaterEqual( + len(claims), 100, f"too few claimed frames: {run.summary()}" + ) + # Warm-up frames at the start have no claim; after it nearly every frame has one. + steady = run.select(lo=3.0) + stamped = [f for f in steady if f["ts"] is not None] + self.assertGreaterEqual(len(stamped), 0.9 * len(steady), run.summary(steady)) + self.assertLess(abs(statistics.median(claims)), MEDIAN_BOUND_NS, run.summary()) + self.assertLess( + _pctl(sorted(abs(x) for x in claims), 0.95), P95_BOUND_NS, run.summary() + ) + self.assertLess(max(abs(x) for x in claims), MAX_BOUND_NS, run.summary()) + self._claims_are_never_ready_time(run, "accuracy") + + def test_2_step_recovery(self): + step_s = 8.0 + # After a sender-clock step, frames mapped before the next exchange may carry the old + # offset: a bounded exposure of one exchange interval (~200 ms) plus an RTT. + exposure_s = 0.4 + run = self._scenario( + "step", ["--sm-offset-ms", "1234", "--sm-step", f"{step_s:g}:250"], 24.0 + ) + self._frame_ids_continuous(run, "step") + sid = run.session_order()[0] + per_stream = {} + for stream in TRACK_STREAM.values(): + frames = run.select(session=sid, stream=stream) + wrong = [ + f + for f in frames + if f["err_ns"] is not None and abs(f["err_ns"]) > MAX_BOUND_NS + ] + outside = [ + f for f in wrong if not step_s <= f["t_rel"] <= step_s + exposure_s + ] + self.assertFalse( + outside, + f"{stream}: wrong claims outside the exposure window: " + f"{[(f['id'], round(f['t_rel'], 2), round(f['err_ns'] / 1e3)) for f in outside[:8]]} us", + ) + self.assertLess( + len(wrong), + 15, + f"{stream}: {len(wrong)} wrong claims inside the exposure window", + ) + near = run.select(session=sid, stream=stream, lo=step_s, hi=step_s + 3.0) + withheld = [f for f in near if f["ts"] is None] + per_stream[stream] = (len(wrong), len(withheld)) + self.assertGreater( + len(withheld), + 0, + f"{stream}: nothing withheld after the step; {run.summary(near)}", + ) + self.assertLess( + len(withheld), + 3 * SOURCE_FPS, + f"{stream}: {len(withheld)} withheld, more than 3 s worth", + ) + before = run.select(session=sid, stream=stream, lo=3.0, hi=step_s - 0.5) + self.assertTrue( + before and all(f["ts"] is not None for f in before), + f"{stream}: claims withheld well before the step; {run.summary(before)}", + ) + after = run.select(session=sid, stream=stream, lo=step_s + 3.0) + self.assertGreater( + len(after), 60, f"{stream}: too few frames after the step" + ) + self.assertTrue( + all(f["ts"] is not None for f in after), + f"{stream}: no recovery; {run.summary(after)}", + ) + self.assertLess(max(abs(f["err_ns"]) for f in after), MAX_BOUND_NS) + print( + f"[step] {stream}: wrong in window {len(wrong)}, withheld {len(withheld)}, " + f"{run.summary(after)}", + file=sys.stderr, + ) + + def test_3_restart(self): + run = self._scenario( + "restart", ["--sm-offset-ms", "1234", "--disconnect-at", "6"], 22.0 + ) + self._frame_ids_continuous(run, "restart") + self._no_wrong_claim(run, "restart") + order = run.session_order() + self.assertGreaterEqual( + len(order), 2, f"the receiver did not reconnect with a new session: {order}" + ) + for sid in order[1:]: + for stream in TRACK_STREAM.values(): + warm = run.select(session=sid, stream=stream, lo=0.0, hi=2.0) + if not warm: + continue + self.assertTrue( + any(f["ts"] is None for f in warm), + f"{stream}: session {sid} claimed from its first frame, no warm-up; {run.summary(warm)}", + ) + gap = [ + f for f in run.select(session=sid, stream=stream) if f["ts"] is None + ] + self.assertLess( + len(gap), 4 * SOURCE_FPS, f"{stream}: warm-up gap too long" + ) + later = run.select(session=sid, stream=stream, lo=3.0) + if len(later) >= 20: + stamped = [f for f in later if f["ts"] is not None] + self.assertGreaterEqual( + len(stamped), + 0.9 * len(later), + f"{stream}: no recovery; {run.summary(later)}", + ) + resumed = [ + f + for sid in order[1:] + for f in run.select(session=sid, lo=3.0) + if f["ts"] is not None + ] + self.assertGreater(len(resumed), 20, "no claims resumed after a restart") + print( + f"[restart] sessions={len(order)} " + run.summary(resumed), file=sys.stderr + ) + + def test_4_stale_mapping(self): + mute_start, mute_s = 5.0, 9.0 + run = self._scenario( + "stale", + ["--sm-offset-ms", "1234", "--clock-mute", f"{mute_start:g}+{mute_s:g}"], + 25.0, + ) + self._frame_ids_continuous(run, "stale") + self._no_wrong_claim(run, "stale") + sid = run.session_order()[0] + # The mapping goes stale 5 s after the last reply (about mute_start) and replies return at mute_start + mute_s. + # The mute is timed from the mock's start, so it begins up to about a second before the session's t_rel 0 reference; + # the window is inset to stay inside the stale interval for any such gap. + stale_lo, stale_hi = mute_start + 5.0 + 0.5, mute_start + mute_s - 1.5 + window = run.select(session=sid, lo=stale_lo, hi=stale_hi) + self.assertGreater(len(window), 30, run.summary(window)) + self._assert_all_withheld_without_stamp(window, "stale window") + after = run.select(session=sid, lo=mute_start + mute_s + 4.0) + self.assertGreater(len(after), 30, run.summary(after)) + stamped = [f for f in after if f["ts"] is not None] + self.assertGreaterEqual( + len(stamped), 0.9 * len(after), f"no recovery; {run.summary(after)}" + ) + self.assertGreater( + _stat_counts(run.bridge_log, "no_mapping"), + 0, + f"no no_mapping in:\n{run.bridge_log[-3000:]}", + ) + print( + f"[stale] window {run.summary(window)}; after {run.summary(after)}", + file=sys.stderr, + ) + + def test_5_loss_and_reorder(self): + run = self._scenario( + "lossy", + [ + "--sm-offset-ms", "1234", "--sm-drift-ppm", "50", + "--clock-loss", "0.3", "--clock-reorder", "0.2", + ], + 20.0, + ) # fmt: skip + self._frame_ids_continuous(run, "lossy") + claims = run.errors(run.frames) + self.assertGreaterEqual( + len(claims), 100, f"too few claimed frames: {run.summary()}" + ) + steady = run.select(lo=5.0) + stamped = [f for f in steady if f["ts"] is not None] + self.assertGreaterEqual(len(stamped), 0.9 * len(steady), run.summary(steady)) + self.assertLess(abs(statistics.median(claims)), MEDIAN_BOUND_NS, run.summary()) + self.assertLess( + _pctl(sorted(abs(x) for x in claims), 0.95), P95_BOUND_NS, run.summary() + ) + self.assertLess(max(abs(x) for x in claims), 3_000_000, run.summary()) + self.assertGreater( + _qualified_count(run.bridge_log), + 100, + f"no qualified claims in:\n{run.bridge_log[-3000:]}", + ) + + def test_6_invalid_records(self): + start_s, dur_s = 5.0, 3.0 + run = self._scenario( + "invalid", + ["--sm-offset-ms", "1234", "--timing-invalid", f"{start_s:g}+{dur_s:g}"], + 16.0, + ) + self._frame_ids_continuous(run, "invalid") + self._no_wrong_claim(run, "invalid") + sid = run.session_order()[0] + # Inset 0.1 s: a frame ready exactly at a window edge may fall on either side. + inside = run.select(session=sid, lo=start_s + 0.1, hi=start_s + dur_s - 0.1) + self.assertGreater(len(inside), 30, run.summary(inside)) + self._assert_all_withheld_without_stamp(inside, "invalid window") + before = run.select(session=sid, lo=3.0, hi=start_s - 0.1) + self.assertGreater(len(before), 30, run.summary(before)) + claimed = [f for f in before if f["ts"] is not None] + self.assertGreaterEqual( + len(claimed), + 0.9 * len(before), + f"claims missing before the window; {run.summary(before)}", + ) + after = run.select(session=sid, lo=start_s + dur_s + 2.0) + self.assertGreater(len(after), 30, run.summary(after)) + stamped = [f for f in after if f["ts"] is not None] + self.assertGreaterEqual( + len(stamped), 0.9 * len(after), f"no recovery; {run.summary(after)}" + ) + self.assertGreater( + _stat_counts(run.bridge_log, "sender_invalid"), + 0, + f"no sender_invalid in:\n{run.bridge_log[-3000:]}", + ) + print( + f"[invalid] inside {run.summary(inside)}; after {run.summary(after)}", + file=sys.stderr, + ) + + def test_7_refusal_without_sender_clock(self): + token = self._tmp / "refuse.token" + token.write_text(make_token("unused", "capture-time-refuse", True)) + cmd = [ + str(BRIDGE_BIN), "--livekit-url", "ws://127.0.0.1:1", "--livekit-token-file", str(token), + "--socket", str(self._tmp / "none.sock"), "--stream", "head=head_track", + "--timestamp-basis", "capture_monotonic_us", + ] # fmt: skip + started = time.monotonic() + out = subprocess.run( + cmd, capture_output=True, text=True, check=False, timeout=30 + ) + self.assertLess(time.monotonic() - started, 15) + self.assertEqual(out.returncode, 2, out.stdout + out.stderr) + self.assertIn("--sender-clock", out.stderr) + self.assertIn("measured mapping", out.stderr) + + +if __name__ == "__main__": + unittest.main() diff --git a/examples/camera_livekit_bridge/tests/sender_clock/CMakeLists.txt b/examples/camera_livekit_bridge/tests/sender_clock/CMakeLists.txt new file mode 100644 index 0000000000..10a624aa21 --- /dev/null +++ b/examples/camera_livekit_bridge/tests/sender_clock/CMakeLists.txt @@ -0,0 +1,15 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +add_executable(test_sender_clock test_sender_clock.cpp) + +target_link_libraries(test_sender_clock + PRIVATE + camera_livekit_bridge::sender_clock + camera_livekit_bridge::video_types + Threads::Threads +) + +target_compile_options(test_sender_clock PRIVATE -Wall -Wextra) + +add_test(NAME sender_clock COMMAND test_sender_clock) diff --git a/examples/camera_livekit_bridge/tests/sender_clock/test_sender_clock.cpp b/examples/camera_livekit_bridge/tests/sender_clock/test_sender_clock.cpp new file mode 100644 index 0000000000..3ee32c5fe1 --- /dev/null +++ b/examples/camera_livekit_bridge/tests/sender_clock/test_sender_clock.cpp @@ -0,0 +1,269 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +// SenderClockClient against a fake UDP responder that misbehaves on demand. + +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace +{ + +using camera_bridge::monotonic_ns; +using camera_bridge::SenderClockClient; +using camera_bridge::SenderClockConfig; + +int g_failures = 0; + +void check(bool condition, const std::string& what) +{ + if (!condition) + { + std::cerr << "FAILED: " << what << "\n"; + ++g_failures; + } +} + +enum class Mode +{ + Good, + Mute, + WrongSequence, + WrongT1, + WrongTag, + Short, + Long, + Delayed, // answers every request only after the next one arrives, so every reply is stale +}; + +void put_le(std::uint8_t* out, std::uint64_t value, std::size_t bytes) +{ + for (std::size_t i = 0; i < bytes; ++i) + out[i] = static_cast((value >> (8 * i)) & 0xFF); +} + +std::uint64_t get_le(const std::uint8_t* in, std::size_t bytes) +{ + std::uint64_t value = 0; + for (std::size_t i = 0; i < bytes; ++i) + value |= static_cast(in[i]) << (8 * i); + return value; +} + +/// A responder on the sender's clock = host clock + offset. +class FakeSender +{ +public: + explicit FakeSender(int64_t offset_ns) : m_offset(offset_ns) + { + m_fd = ::socket(AF_INET, SOCK_DGRAM, 0); + sockaddr_in addr{}; + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + ::bind(m_fd, reinterpret_cast(&addr), sizeof(addr)); + socklen_t len = sizeof(addr); + ::getsockname(m_fd, reinterpret_cast(&addr), &len); + m_port = ntohs(addr.sin_port); + timeval tv{ 0, 50'000 }; + ::setsockopt(m_fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); + m_thread = std::thread([this] { serve(); }); + } + ~FakeSender() + { + m_stop = true; + m_thread.join(); + ::close(m_fd); + } + std::uint16_t port() const + { + return m_port; + } + void set_mode(Mode mode) + { + m_mode = mode; + } + +private: + void serve() + { + std::vector held; + sockaddr_in held_peer{}; + while (!m_stop) + { + std::uint8_t in[64]; + sockaddr_in peer{}; + socklen_t len = sizeof(peer); + const ssize_t got = ::recvfrom(m_fd, in, sizeof(in), 0, reinterpret_cast(&peer), &len); + if (got != 13) + continue; + const int64_t t2 = monotonic_ns() + m_offset; + const Mode mode = m_mode; + if (mode == Mode::Mute) + continue; + std::uint8_t out[40] = {}; + out[0] = mode == Mode::WrongTag ? 'X' : 'R'; + std::uint32_t seq = static_cast(get_le(in + 1, 4)); + if (mode == Mode::WrongSequence) + seq += 1; + put_le(out + 1, seq, 4); + std::int64_t t1 = static_cast(get_le(in + 5, 8)); + if (mode == Mode::WrongT1) + t1 += 1; + put_le(out + 5, static_cast(t1), 8); + put_le(out + 13, static_cast(t2), 8); + put_le(out + 21, static_cast(monotonic_ns() + m_offset), 8); + std::size_t size = 29; + if (mode == Mode::Short) + size = 28; + if (mode == Mode::Long) + size = 30; + if (mode == Mode::Delayed) + { + if (!held.empty()) + ::sendto( + m_fd, held.data(), held.size(), 0, reinterpret_cast(&held_peer), sizeof(held_peer)); + held.assign(out, out + size); + held_peer = peer; + continue; + } + ::sendto(m_fd, out, size, 0, reinterpret_cast(&peer), len); + } + } + + int m_fd = -1; + std::uint16_t m_port = 0; + int64_t m_offset; + std::atomic m_mode{ Mode::Good }; + std::atomic m_stop{ false }; + std::thread m_thread; +}; + +void wait_ms(int ms) +{ + std::this_thread::sleep_for(std::chrono::milliseconds(ms)); +} + +SenderClockConfig fast_config(std::uint16_t port) +{ + SenderClockConfig config; + config.host = "127.0.0.1"; + config.port = port; + config.interval = std::chrono::milliseconds(20); + config.reply_timeout = std::chrono::milliseconds(60); + return config; +} + +void test_measures_the_offset() +{ + const int64_t offset = 7'000'000'000; + FakeSender sender(offset); + SenderClockClient client(fast_config(sender.port())); + client.start(); + wait_ms(400); + const auto snap = client.snapshot(); + client.stop(); + check(snap.mapping.has_value(), "a mapping exists after replies"); + check(snap.accepted_in_epoch >= 3, "several exchanges accepted"); + if (snap.mapping) + { + const int64_t error = snap.mapping->offset_ns - offset; + check(error > -2'000'000 && error < 2'000'000, "offset within 2 ms of the fake sender's"); + } +} + +void test_unmatched_replies_are_not_measurements() +{ + for (Mode mode : { Mode::WrongSequence, Mode::WrongT1, Mode::WrongTag, Mode::Short, Mode::Long, Mode::Delayed }) + { + FakeSender sender(1'000'000'000); + sender.set_mode(mode); + SenderClockClient client(fast_config(sender.port())); + client.start(); + wait_ms(300); + const auto snap = client.snapshot(); + client.stop(); + check(!snap.mapping && snap.exchanges == 0, "no measurement from a reply that does not answer the request (mode " + + std::to_string(static_cast(mode)) + ")"); + check(snap.ignored > 0 || mode == Mode::Delayed, "the bad replies are counted as ignored"); + } +} + +void test_loss_and_recovery() +{ + FakeSender sender(2'000'000'000); + SenderClockClient client(fast_config(sender.port())); + client.start(); + wait_ms(300); + check(client.snapshot().mapping.has_value(), "mapping before the loss"); + sender.set_mode(Mode::Mute); + wait_ms(200); + const auto muted = client.snapshot(); + check(muted.timeouts > 0, "silence is counted as timeouts"); + sender.set_mode(Mode::Good); + wait_ms(300); + const auto after = client.snapshot(); + client.stop(); + check(after.exchanges > muted.exchanges, "measurements resume after the loss"); +} + +void test_reset_voids_the_outstanding_request() +{ + FakeSender sender(3'000'000'000); + SenderClockClient client(fast_config(sender.port())); + client.start(); + wait_ms(300); + const auto before = client.snapshot(); + check(before.mapping.has_value() && before.accepted_in_epoch >= 3, "warm before the reset"); + client.begin_session(1); + const auto reset = client.snapshot(); + check(!reset.mapping && reset.accepted_in_epoch == 0, "a new session discards every earlier measurement"); + check(reset.generation > before.generation, "the generation advanced"); + client.begin_session(1); + check(client.snapshot().generation == reset.generation, "the same session does not reset again"); + wait_ms(300); + check(client.snapshot().mapping.has_value(), "measurements resume in the new generation"); + client.stop(); +} + +void test_wrong_peer_is_ignored() +{ + FakeSender sender(1'000'000'000); + FakeSender other(5'000'000'000); + // The client asks `sender`; `other` cannot be heard because replies must come from the address asked. + SenderClockClient client(fast_config(sender.port())); + client.start(); + wait_ms(200); + const auto snap = client.snapshot(); + client.stop(); + check(snap.mapping && snap.mapping->offset_ns < 2'000'000'000, "only the configured sender's clock is used"); +} + +} // namespace + +int main() +{ + test_measures_the_offset(); + test_unmatched_replies_are_not_measurements(); + test_loss_and_recovery(); + test_reset_voids_the_outstanding_request(); + test_wrong_peer_is_ignored(); + if (g_failures != 0) + { + std::cerr << g_failures << " check(s) failed\n"; + return 1; + } + std::cout << "all sender_clock checks passed\n"; + return 0; +} diff --git a/examples/camera_streamer/camera_receiver/avs_wire.py b/examples/camera_streamer/camera_receiver/avs_wire.py index edca77d1c6..a469913df1 100644 --- a/examples/camera_streamer/camera_receiver/avs_wire.py +++ b/examples/camera_streamer/camera_receiver/avs_wire.py @@ -21,12 +21,24 @@ HEADER = struct.Struct("!4sHHIHHQQQQIIQ") STREAM = struct.Struct("!HHBBHIIIIQHHI") SAMPLE = struct.Struct("!HBBHHIIQQQQQQII") +# Timing extension: one record per sample after the sample descriptors, present when BUNDLE_TIMING_EXT is set. +# capture_sm_ns: the qualified sensor capture instant on the sender CLOCK_MONOTONIC; uncertainty_ns: the sender's +# bound on the sensor-to-sender-clock mapping; anchor_age_ns: age of the sensor anchor when the frame was stamped; +# flags bit0 VALID (otherwise every other field is zero and the frame carries no capture claim); clock_domain 1 = +# CLOCK_MONOTONIC. +TIMING = struct.Struct("!QIIHHI") +TIMING_VALID = 1 << 0 +TIMING_DOMAIN_MONOTONIC = 1 BUNDLE_SENDER_CLOCK_VALID = 1 << 0 BUNDLE_SENDER_CLOCK_TAI = 1 << 1 BUNDLE_SENDER_TSC_VALID = 1 << 2 +BUNDLE_TIMING_EXT = 1 << 3 BUNDLE_KNOWN_FLAGS = ( - BUNDLE_SENDER_CLOCK_VALID | BUNDLE_SENDER_CLOCK_TAI | BUNDLE_SENDER_TSC_VALID + BUNDLE_SENDER_CLOCK_VALID + | BUNDLE_SENDER_CLOCK_TAI + | BUNDLE_SENDER_TSC_VALID + | BUNDLE_TIMING_EXT ) # Sensor SOF -> sender capture-ready lag accepted by ``sensor_sof_ns``. @@ -72,6 +84,11 @@ class Avs1Sample: payload_offset: int payload_length: int payload: memoryview + # From the timing extension; None/False when the bundle has none or the record is not valid. + sensor_capture_sm_ns: Optional[int] = None + capture_uncertainty_ns: Optional[int] = None + capture_anchor_age_ns: Optional[int] = None + capture_valid: bool = False @property def present(self) -> bool: @@ -184,6 +201,8 @@ def parse_avs1_record(record: bytes | bytearray | memoryview) -> Avs1Bundle: raise ValueError("invalid AVS1 bundle identity/count") expected_descriptors = stream_count * STREAM.size + sample_count * SAMPLE.size + if flags & BUNDLE_TIMING_EXT: + expected_descriptors += sample_count * TIMING.size if descriptor_bytes != expected_descriptors: raise ValueError("AVS1 descriptor byte count mismatch") expected_length = header_bytes + descriptor_bytes + payload_bytes @@ -244,6 +263,11 @@ def parse_avs1_record(record: bytes | bytearray | memoryview) -> Avs1Bundle: if fields[4] != 0: raise ValueError("AVS1 sample reserved field is nonzero") raw_samples.append(fields) + raw_timing = [] + if flags & BUNDLE_TIMING_EXT: + for _ in range(sample_count): + raw_timing.append(TIMING.unpack_from(packet, offset)) + offset += TIMING.size payload_start = header_bytes + descriptor_bytes streams: list[Avs1Stream] = [] @@ -321,6 +345,31 @@ def parse_avs1_record(record: bytes | bytearray | memoryview) -> Avs1Bundle: (payload_offset, payload_offset + payload_length) ) sender_timing_valid = bool(sample_flags & SAMPLE_SENDER_TIMING_VALID) + capture_sm_ns = capture_uncertainty_ns = capture_anchor_age_ns = None + capture_valid = False + if raw_timing: + (t_sm, t_unc, t_age, t_flags, t_domain, t_reserved) = raw_timing[ + first_sample + camera_index + ] + if t_reserved != 0 or t_flags & ~TIMING_VALID: + raise ValueError("AVS1 timing record has reserved bits set") + if t_flags & TIMING_VALID: + if not sample_present: + raise ValueError( + "AVS1 timing record is valid for an absent sample" + ) + if t_domain != TIMING_DOMAIN_MONOTONIC or t_sm == 0: + raise ValueError( + "AVS1 valid timing record has no monotonic capture time" + ) + capture_sm_ns, capture_uncertainty_ns, capture_anchor_age_ns = ( + int(t_sm), + int(t_unc), + int(t_age), + ) + capture_valid = True + elif t_sm or t_unc or t_age or t_domain: + raise ValueError("AVS1 invalid timing record carries values") samples.append( Avs1Sample( stream_index=stream_index, @@ -350,6 +399,10 @@ def parse_avs1_record(record: bytes | bytearray | memoryview) -> Avs1Bundle: payload_offset=int(payload_offset), payload_length=int(payload_length), payload=packet[begin:end], + sensor_capture_sm_ns=capture_sm_ns, + capture_uncertainty_ns=capture_uncertainty_ns, + capture_anchor_age_ns=capture_anchor_age_ns, + capture_valid=capture_valid, ) ) streams.append( diff --git a/examples/camera_streamer/camera_receiver/camera_receiver.py b/examples/camera_streamer/camera_receiver/camera_receiver.py index c1f5672745..937d217b1e 100755 --- a/examples/camera_streamer/camera_receiver/camera_receiver.py +++ b/examples/camera_streamer/camera_receiver/camera_receiver.py @@ -1042,6 +1042,13 @@ def _frame_metadata( # Same value under the key the CUDA IPC consumer tools and other viewers # already read for their "sof_to_consumer_ms" column. "sof_wall_clock_ns": sample_sof_ns, + # Qualified sensor capture instant on the sender CLOCK_MONOTONIC (timing extension), with the sender's + # stated uncertainty and the age of its sensor anchor; None when the sender sent none or marked it invalid. + "sensor_capture_sm_ns": sample.sensor_capture_sm_ns, + "sensor_capture_uncertainty_ns": sample.capture_uncertainty_ns, + "sensor_anchor_age_ns": sample.capture_anchor_age_ns, + "sensor_capture_valid": sample.capture_valid, + "sender_capture_clock": ("monotonic" if sample.capture_valid else None), "sensor_sof_lag_ms": ( sof_lag_ns / 1.0e6 if sof_lag_ns is not None else None ), diff --git a/examples/camera_streamer/mock_sender/README.md b/examples/camera_streamer/mock_sender/README.md index 754a3dd0f1..325a303aa6 100644 --- a/examples/camera_streamer/mock_sender/README.md +++ b/examples/camera_streamer/mock_sender/README.md @@ -30,6 +30,15 @@ tick, as in the real sender. | `--disconnect-at SECONDS` | Close the receiver connection after that long. | | `--clock-offset-ms`, `--clock-drift-ppm` | Offset (default 0, as with chrony-synced clocks) and drift of the synthetic sender clock. | | `--sof-lag-ms` | Synthetic sensor start-of-frame to capture-ready lag (default 20, range 0-500). | +| `--timing-ext` | Emit the AVS1 timing extension: per frame a qualified sensor capture instant on the sender's monotonic clock, the sender's stated uncertainty and anchor age. The receiver in this tree parses it; an unmodified receiver rejects the bundle flag. | +| `--sm-offset-ms`, `--sm-drift-ppm` | Offset and drift of the sender's monotonic clock from this host's. The UDP clock responder and the timing extension are both on this clock. | +| `--sm-step AT_S:DELTA_MS` | Step the sender's monotonic clock this long after the receiver connects; repeatable. | +| `--sensor-uncertainty-us` | The bound the extension states for the sensor-to-sender-clock mapping (default 500). | +| `--timing-invalid START_S+DURATION_S` | Mark the timing records not valid (no capture claim) in that window; repeatable. | +| `--ground-truth FILE` | Append one JSON line per published frame: the true capture instant and the stamps, on this host's monotonic clock, for checking a consumer's mapped capture time. | +| `--clock-port PORT` | Serve the sender clock responder on this UDP port (0, the default, disables it). `--clock-bind` (127.0.0.1). Alive for the whole process, independent of media connections. | +| `--clock-mute START_S+DURATION_S` | Stop answering clock requests in that window (seconds after the mock starts); repeatable. | +| `--clock-loss F`, `--clock-reorder F` | Leave a fraction of requests unanswered, or send a fraction of replies after the next one. | Timing fields come from the synthetic clock only. Every frame carries a synthetic sensor start-of-frame TSC, in a TSC domain of the mock's own, so the @@ -38,6 +47,14 @@ Argus. Argus frame timestamps are never marked valid, so `sof_delta` stays `n/a`. Nothing here measures real sensor timing. Frame numbers count published frames per stream. +## Clock responder + +`--clock-port` serves one UDP port, stateless and unauthenticated, as the real sender does. A datagram of exactly +13 bytes, `Q`, u32 sequence, i64 t1 (little-endian), is answered to its source with 29 bytes: `R`, sequence, t1, t2, +t3, with t2 taken at receipt and t3 just before the send, in nanoseconds on the sender's monotonic clock +(`--sm-offset-ms`, `--sm-drift-ppm`, `--sm-step`), the clock `sensor_capture_sm_ns` is on. Anything else is +ignored silently. + Tests: `python3 -m unittest discover -s mock_sender/tests` and `python3 -m unittest tests.test_mock_sender_receiver_wire`, both from this directory's parent. The full mock-to-CUDA-IPC pipeline (needs a GPU): diff --git a/examples/camera_streamer/mock_sender/clock_responder.py b/examples/camera_streamer/mock_sender/clock_responder.py new file mode 100644 index 0000000000..72a247e3da --- /dev/null +++ b/examples/camera_streamer/mock_sender/clock_responder.py @@ -0,0 +1,94 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Stateless UDP clock responder of the sender: the PC measures the sender's CLOCK_MONOTONIC with it. + +Request, 13 bytes little-endian: 'Q', u32 seq, i64 t1. Reply, 29 bytes: 'R', seq, t1, t2, t3, with t2 taken +at receipt and t3 just before the send, both on the sender's monotonic clock (the one sensor_capture_sm_ns is +on). Anything else is ignored. The responder lives for the whole process, independent of media connections. +""" + +from __future__ import annotations + +import logging +import random +import socket +import struct +import threading +import time +from collections.abc import Callable + +logger = logging.getLogger(__name__) + +REQUEST = struct.Struct(" None: + self._now_ns = now_ns + self._muted = muted or (lambda: False) + self._loss = loss + self._reorder = reorder + self._rng = random.Random(seed) + self._held: tuple[bytes, tuple] | None = None + self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + self._sock.bind((bind, port)) + self._sock.settimeout(0.1) + self.port = self._sock.getsockname()[1] + self._stop = threading.Event() + self._thread = threading.Thread( + target=self._serve, name="clock-responder", daemon=True + ) + self._thread.start() + + def close(self) -> None: + self._stop.set() + self._thread.join(timeout=1.0) + self._sock.close() + + def _serve(self) -> None: + while not self._stop.is_set(): + try: + data, peer = self._sock.recvfrom(2048) + except TimeoutError: + continue + except OSError: + return + t2 = self._now_ns() + if len(data) != REQUEST.size or data[:1] != b"Q" or self._muted(): + continue + if self._rng.random() < self._loss: + continue + _, sequence, t1 = REQUEST.unpack(data) + reply = REPLY.pack(b"R", sequence, t1, t2, self._now_ns()) + if self._held is not None: + # A held reply goes out after the newer one: reordered, as a lossy network may do. + self._sock.sendto(reply, peer) + self._sock.sendto(*self._held) + self._held = None + elif self._rng.random() < self._reorder: + self._held = (reply, peer) + else: + self._sock.sendto(reply, peer) + + +def parse_window(text: str) -> tuple[float, float]: + start, _, duration = text.partition("+") + return float(start), float(duration) + + +def mute_predicate(windows: list[tuple[float, float]]) -> Callable[[], bool]: + started = time.monotonic() + return lambda: any(a <= time.monotonic() - started < a + d for a, d in windows) diff --git a/examples/camera_streamer/mock_sender/mock_sender.py b/examples/camera_streamer/mock_sender/mock_sender.py index 0c87e8f9db..7e6979a23d 100755 --- a/examples/camera_streamer/mock_sender/mock_sender.py +++ b/examples/camera_streamer/mock_sender/mock_sender.py @@ -16,6 +16,7 @@ from __future__ import annotations import argparse +import json import logging import random import re @@ -29,6 +30,8 @@ from dataclasses import dataclass, field from pathlib import Path +import clock_responder + logger = logging.getLogger("isaaccapture.camera_streamer.mock_sender") HEADER = struct.Struct("!4sHHIHHQQQQIIQ") # 64 bytes @@ -43,6 +46,14 @@ BUNDLE_SENDER_CLOCK_VALID = 0x0001 # realtime domain; the TAI flag stays clear BUNDLE_SENDER_TSC_VALID = 0x0004 TSC_BASE_NS = 7_000_000_000_000 # arbitrary origin of the synthetic TSC domain +BUNDLE_TIMING_EXT = ( + 0x0008 # one TIMING record per sample follows the sample descriptors +) +TIMING_VALID = 0x0001 +CLOCK_DOMAIN_MONOTONIC = 1 +TIMING = struct.Struct( + "!QIIHHI" +) # 24 bytes: capture_sm_ns, uncertainty_ns, anchor_age_ns, flags, domain, reserved LOOP_SECONDS = 12 GOP_SECONDS = 2 @@ -152,20 +163,79 @@ def split_access_units(data: bytes) -> list[tuple[bytes, bool]]: class MockClock: - """Synthetic sender clock: this host's wall clock plus an offset and a drift.""" + """The mock sender's clocks, all derived from this host's CLOCK_MONOTONIC reading so that ground truth is exact. + + Wall: a synthetic Unix clock (the sender's wall clock) with an offset and a drift. + SM: the sender's CLOCK_MONOTONIC, a different machine's boot-relative clock, with its own offset and drift and + deterministic steps. The clock exchange is answered on SM and the transported capture timestamps are on SM. + TSC: the sensor counter domain, unrelated to either. + """ - def __init__(self, offset_ns: int, drift_ppm: float) -> None: + def __init__( + self, + offset_ns: int, + drift_ppm: float, + sm_offset_ns: int = 0, + sm_drift_ppm: float = 0.0, + ) -> None: self._t0 = time.monotonic_ns() + self._wall_epoch = time.time_ns() - self._t0 self._offset = offset_ns self._drift = drift_ppm * 1e-6 + self._sm_offset = sm_offset_ns + self._sm_drift = sm_drift_ppm * 1e-6 + self._step_ns = 0 + self._lock = threading.Lock() + + def wall_at(self, mono_ns: int) -> int: + return ( + mono_ns + + self._wall_epoch + + self._offset + + int((mono_ns - self._t0) * self._drift) + ) def now_ns(self) -> int: - elapsed = time.monotonic_ns() - self._t0 - return time.time_ns() + self._offset + int(elapsed * self._drift) + return self.wall_at(time.monotonic_ns()) + + def sm_at(self, mono_ns: int) -> int: + with self._lock: + step = self._step_ns + return ( + mono_ns + + self._sm_offset + + int((mono_ns - self._t0) * self._sm_drift) + + step + ) + + def sm_ns(self) -> int: + return self.sm_at(time.monotonic_ns()) + + def step_sm(self, delta_ns: int) -> None: + """Step the sender's monotonic clock, as a suspend/resume or a clock fix would.""" + with self._lock: + self._step_ns += delta_ns + + def tsc_at(self, mono_ns: int) -> int: + return TSC_BASE_NS + mono_ns def tsc_ns(self) -> int: """The synthetic sensor TSC: monotonic, unrelated to the wall clock's value.""" - return TSC_BASE_NS + time.monotonic_ns() + return self.tsc_at(time.monotonic_ns()) + + +@dataclass +class TimingOptions: + """What the mock adds for the capture-time contract (see README: timing extension).""" + + ext: bool = False + #: The bound the sender states for its sensor-to-SM mapping. + uncertainty_ns: int = 500_000 + #: (start_s, duration_s) windows, since the receiver connected, in which records are not valid. + invalid_windows: list[tuple[float, float]] = field(default_factory=list) + #: (at_s, delta_ns) steps of the SM clock, since the receiver connected. + sm_steps: list[tuple[float, int]] = field(default_factory=list) + ground_truth: Path | None = None @dataclass @@ -199,6 +269,7 @@ def build_record( enqueue_ns: int, ready_tsc_ns: int, sof_lag_ns: int, + timing_records: list[bytes] | None = None, ) -> bytes: """One length-prefixed AVS1 record: every stream has a descriptor, present or not.""" streams, samples, body = [], [], b"" @@ -226,8 +297,12 @@ def build_record( SAMPLE.pack(index, 0, 0, 0, 0, index, 0, 0, 0, 0, 0, 0, 0, 0, 0) ) descriptors = b"".join(streams) + b"".join(samples) + bundle_flags = BUNDLE_SENDER_CLOCK_VALID | BUNDLE_SENDER_TSC_VALID + if timing_records is not None: + descriptors += b"".join(timing_records) + bundle_flags |= BUNDLE_TIMING_EXT header = HEADER.pack( - b"AVS1", 1, BUNDLE_SENDER_CLOCK_VALID | BUNDLE_SENDER_TSC_VALID, HEADER.size, + b"AVS1", 1, bundle_flags, HEADER.size, len(streams), len(samples), bundle_sequence, session_id, ready_ns, enqueue_ns, len(descriptors), len(body), ready_tsc_ns, ) # fmt: skip @@ -247,11 +322,13 @@ def __init__( pauses: list[Pause], disconnects_s: list[float], sof_lag_ns: int = 20_000_000, + timing: TimingOptions | None = None, ) -> None: self.specs = specs self.clock = clock self.sof_lag_ns = sof_lag_ns self.pauses = pauses + self.timing = timing or TimingOptions() self.disconnects_s = sorted(disconnects_s) self._units = { spec: split_access_units(sources[spec].read_bytes()) for spec in specs @@ -288,13 +365,41 @@ def _stream_to(self, conn: socket.socket) -> None: disconnects = list(self.disconnects_s) start = time.monotonic() tick = 0 + steps = sorted(self.timing.sm_steps) + truth = ( + self.timing.ground_truth.open("a", encoding="utf-8") + if self.timing.ground_truth is not None + else None + ) + try: + self._stream_loop( + conn, cursors, session_id, disconnects, start, tick, steps, truth + ) + finally: + if truth is not None: + truth.close() + + def _stream_loop( + self, conn, cursors, session_id, disconnects, start, tick, steps, truth + ) -> None: while not self.stop_event.is_set(): elapsed = time.monotonic() - start if disconnects and elapsed >= disconnects[0]: logger.info("scripted disconnect at %.1fs", disconnects.pop(0)) return - ready_ns = self.clock.now_ns() - ready_tsc_ns = self.clock.tsc_ns() + while steps and elapsed >= steps[0][0]: + at_s, delta_ns = steps.pop(0) + logger.info( + "stepping the sender monotonic clock by %.3f ms at %.1fs", + delta_ns / 1e6, + at_s, + ) + self.clock.step_sm(delta_ns) + # One instant stamps every clock, so the ground truth is exact. + t_mono = time.monotonic_ns() + ready_ns = self.clock.wall_at(t_mono) + ready_tsc_ns = self.clock.tsc_at(t_mono) + ready_sm_ns = self.clock.sm_at(t_mono) payloads: list[bytes | None] = [] for cursor in cursors: paused = any( @@ -319,15 +424,52 @@ def _stream_to(self, conn: socket.socket) -> None: cursor.frame_number += 1 payloads.append(payload) tick += 1 + timing_records = None + if self.timing.ext: + invalid = any( + a <= elapsed < a + d for a, d in self.timing.invalid_windows + ) + capture_sm_ns = ready_sm_ns - self.sof_lag_ns + # The sensor anchor is refreshed once a second since the connection started. + anchor_age_ns = int((elapsed * 1e9) % 1_000_000_000) + timing_records = [ + TIMING.pack( + capture_sm_ns, self.timing.uncertainty_ns, anchor_age_ns, + TIMING_VALID, CLOCK_DOMAIN_MONOTONIC, 0, + ) + if payload is not None and not invalid + else TIMING.pack(0, 0, 0, 0, 0, 0) + for payload in payloads + ] # fmt: skip + if truth is not None: + for cursor, payload in zip(cursors, payloads): + if payload is None: + continue + truth.write( + json.dumps( + { + "session_id": session_id, + "stream": cursor.spec.name, + "stream_sequence": cursor.stream_sequence, + "true_capture_pc_mono_ns": t_mono - self.sof_lag_ns, + "ready_pc_mono_ns": t_mono, + "capture_sm_ns": ready_sm_ns - self.sof_lag_ns, + "sm_minus_pc_mono_ns": ready_sm_ns - t_mono, + } + ) + + "\n" + ) + truth.flush() record = build_record( cursors, payloads, tick, session_id, ready_ns, - self.clock.now_ns(), + self.clock.wall_at(time.monotonic_ns()), ready_tsc_ns, self.sof_lag_ns, + timing_records, ) conn.sendall(record) self.bundles_sent += 1 @@ -380,6 +522,54 @@ def main(argv: list[str] | None = None) -> int: "--disconnect-at", type=float, action="append", default=[], metavar="SECONDS", help="close the receiver connection this long after it connects", ) # fmt: skip + parser.add_argument( + "--timing-ext", action="store_true", + help="emit the AVS1 timing extension (a qualified sensor capture time on the sender monotonic clock)", + ) # fmt: skip + parser.add_argument( + "--sm-offset-ms", + type=float, + default=0.0, + help="sender monotonic clock offset from this host's", + ) + parser.add_argument("--sm-drift-ppm", type=float, default=0.0) + parser.add_argument( + "--sm-step", action="append", default=[], metavar="AT_S:DELTA_MS", + help="step the sender monotonic clock by DELTA_MS this many seconds after the receiver connects; repeatable", + ) # fmt: skip + parser.add_argument( + "--sensor-uncertainty-us", type=float, default=500.0, + help="the bound the timing extension states for the sensor-to-sender-clock mapping", + ) # fmt: skip + parser.add_argument( + "--timing-invalid", action="append", default=[], metavar="START_S+DURATION_S", + help="mark the timing records not valid (no capture claim) in this window; repeatable", + ) # fmt: skip + parser.add_argument( + "--ground-truth", type=Path, + help="append one JSON line per published frame with the true capture instant on this host's monotonic clock", + ) # fmt: skip + parser.add_argument( + "--clock-port", type=int, default=0, + help="UDP port of the sender clock responder (0 disables); the bridge's --sender-clock points here", + ) # fmt: skip + parser.add_argument("--clock-bind", default="127.0.0.1") + parser.add_argument( + "--clock-mute", action="append", default=[], metavar="START_S+DURATION_S", + help="stop answering clock requests in this window, seconds after the mock starts; repeatable", + ) # fmt: skip + parser.add_argument( + "--clock-loss", + type=float, + default=0.0, + help="fraction of requests left unanswered", + ) + parser.add_argument( + "--clock-reorder", + type=float, + default=0.0, + help="fraction of replies sent after the next one", + ) parser.add_argument("--duration", type=float, default=0.0) parser.add_argument("--log-level", default="INFO") args = parser.parse_args(argv) @@ -395,18 +585,56 @@ def main(argv: list[str] | None = None) -> int: names = [s.name for s in specs] if len(set(names)) != len(names): parser.error("stream names must be unique") - for pause in args.pause: - if pause.stream not in names: - parser.error(f"--pause names unknown stream {pause.stream!r}") + for item in args.pause: + if item.stream not in names: + parser.error(f"unknown stream {item.stream!r}") if not 0.0 <= args.sof_lag_ms <= 500.0: parser.error("--sof-lag-ms must be in [0, 500]") + clock = MockClock( + int(args.clock_offset_ms * 1e6), + args.clock_drift_ppm, + int(args.sm_offset_ms * 1e6), + args.sm_drift_ppm, + ) + timing = TimingOptions( + ext=args.timing_ext, + uncertainty_ns=int(args.sensor_uncertainty_us * 1000), + ground_truth=args.ground_truth, + ) + for text in args.sm_step: + match = re.fullmatch(r"([0-9.]+):(-?[0-9.]+)", text) + if not match: + parser.error(f"--sm-step expects AT_S:DELTA_MS, got {text!r}") + timing.sm_steps.append((float(match[1]), int(float(match[2]) * 1e6))) + for text in args.timing_invalid: + match = re.fullmatch(r"([0-9.]+)\+([0-9.]+)", text) + if not match: + parser.error(f"--timing-invalid expects START_S+DURATION_S, got {text!r}") + timing.invalid_windows.append((float(match[1]), float(match[2]))) + responder = None + if args.clock_port: + mute_windows = [] + for text in args.clock_mute: + if not re.fullmatch(r"[0-9.]+\+[0-9.]+", text): + parser.error(f"--clock-mute expects START_S+DURATION_S, got {text!r}") + mute_windows.append(clock_responder.parse_window(text)) + responder = clock_responder.ClockResponder( + args.clock_bind, + args.clock_port, + clock.sm_ns, + clock_responder.mute_predicate(mute_windows), + args.clock_loss, + args.clock_reorder, + ) + logger.info("clock responder on udp %s:%d", args.clock_bind, responder.port) sender = MockSender( specs, ensure_sources(specs, args.cache_dir), - MockClock(int(args.clock_offset_ms * 1e6), args.clock_drift_ppm), + clock, args.pause, args.disconnect_at, int(args.sof_lag_ms * 1e6), + timing, ) listener = socket.create_server((args.bind, args.port), reuse_port=False) logger.info( @@ -418,6 +646,8 @@ def main(argv: list[str] | None = None) -> int: pass finally: listener.close() + if responder: + responder.close() return 0 diff --git a/examples/camera_streamer/mock_sender/tests/test_clock_responder.py b/examples/camera_streamer/mock_sender/tests/test_clock_responder.py new file mode 100644 index 0000000000..76252ee937 --- /dev/null +++ b/examples/camera_streamer/mock_sender/tests/test_clock_responder.py @@ -0,0 +1,64 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +import socket +import struct +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +import clock_responder # noqa: E402 + + +class ClockResponderTest(unittest.TestCase): + def setUp(self): + self.now = 1_000 + self.muted = False + + def tick(): + self.now += 10 + return self.now + + self.responder = clock_responder.ClockResponder( + "127.0.0.1", 0, tick, lambda: self.muted + ) + self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + self.sock.settimeout(0.3) + self.addr = ("127.0.0.1", self.responder.port) + + def tearDown(self): + self.sock.close() + self.responder.close() + + def test_reply_layout(self): + self.sock.sendto(struct.pack("