examples/camera_streamer: carry a qualified sensor capture time to LiveKit

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)

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

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

examples/camera_livekit_bridge/bridge/main.cpp

@@ -7,6 +7,7 @@
 
 #include <livekit_publisher/livekit_session.hpp>
 #include <nvenc/cuda_context.hpp>
+#include <sender_clock/sender_clock_client.hpp>
 
 #include <atomic>
 #include <chrono>
@@ -67,6 +68,10 @@ int main(int argc, char** argv)
     std::vector<StreamSpec> 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<int64_t>(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<uint16_t>(std::atoi(endpoint.substr(colon + 1).c_str()));
         }
+        else if (arg == "--clock-uncertainty-budget-ms")
+            defaults.clock_budget_ns = static_cast<int64_t>(std::atof(next().c_str()) * 1e6);
         else if (arg == "--gpu")
             gpu = std::atoi(next().c_str());
         else if (arg == "--stream")
@@ -142,13 +154,32 @@ 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);
 
     try
     {
         CudaContext context(gpu);
         auto session = LiveKitSession::connect(url, read_token(token_file));
+        std::unique_ptr<SenderClockClient> sender_clock;
+        if (!sender_host.empty())
+        {
+            SenderClockConfig clock_config;
+            clock_config.host = sender_host;
+            clock_config.port = sender_port;
+            sender_clock = std::make_unique<SenderClockClient>(clock_config);
+            sender_clock->start();
+            defaults.sender_clock = sender_clock.get();
+        }
         std::vector<std::unique_ptr<StreamWorker>> workers;
         for (const auto& spec : streams)
         {
@@ -206,6 +237,18 @@ int main(int argc, char** argv)
                                   << static_cast<double>(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<CaptureVerdict>(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;
                 }

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<FrameNotice> 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<bool>();
+                input.capture_sm_ns = static_cast<int64_t>(unsigned_field(notice.metadata, "sensor_capture_sm_ns"));
+                input.upstream_uncertainty_ns =
+                    static_cast<int64_t>(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<std::size_t>(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<uint64_t>(result.capture_bridge_ns) / 1000 :
+                                              0;
+            }
             if (sensor_sof_ns != 0)
             {
                 const int64_t age_ns = ready_realtime_ns - static_cast<int64_t>(sensor_sof_ns);

examples/camera_livekit_bridge/bridge/stream_worker.hpp

@@ -3,8 +3,10 @@
 
 #pragma once
 
+#include <clock_exchange/capture_qualifier.hpp>
 #include <livekit_publisher/livekit_session.hpp>
 #include <nvenc/cuda_context.hpp>
+#include <sender_clock/sender_clock_client.hpp>
 #include <video_types/timestamp_basis.hpp>
 
 #include <atomic>
@@ -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<uint64_t> capture_to_ready_us_total{ 0 };
     std::atomic<uint64_t> capture_to_ready_samples{ 0 };
     std::atomic<uint32_t> 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<uint64_t> verdicts[8] = {};
+    std::atomic<int64_t> last_uncertainty_ns{ 0 };
+    std::atomic<uint64_t> last_epoch{ 0 };
     /// The latest target the transport reported (0 until it reports one).
     std::atomic<uint32_t> target_bps{ 0 };
 };

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
+        $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/inc>
+)
+
+add_library(camera_livekit_bridge::clock_exchange ALIAS clock_exchange)

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 <clock_exchange/clock_estimator.hpp>
+
+#include <cstdint>
+#include <optional>
+
+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<ClockMapping>& 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