feat(livestream): embed per-frame SEI timestamps for latency debugging

Author: jkaczman

camera_hub/src/fmp4.rs

@@ -106,7 +106,7 @@ impl TrakTracker {
             buf.put_u64(self.fragment_start_time);
         });
 
-        let mut data_offset_pos: Option<usize> = None;
+        let data_offset_pos: Option<usize>;
 
         const TRUN_DATA_OFFSET: u32 = 0x000001;
         const TRUN_FIRST_SAMPLE_FL: u32 = 0x000004;
@@ -265,7 +265,7 @@ impl<W: AsyncWrite + Unpin, V: CodecParameters, A: CodecParameters> Fmp4Writer<W
     }
 
     fn write_video_fragment(&self, buf: &mut BytesMut) -> Result<Option<usize>, Error> {
-        let mut trun_off: Option<usize> = None;
+        let trun_off: Option<usize>;
         write_box!(buf, b"traf", {
             write_box!(buf, b"tfhd", {
                 buf.put_u32(0x020000); // default-base-is-moof
@@ -278,7 +278,7 @@ impl<W: AsyncWrite + Unpin, V: CodecParameters, A: CodecParameters> Fmp4Writer<W
     }
 
     fn write_audio_fragment(&self, buf: &mut BytesMut) -> Result<Option<usize>, Error> {
-        let mut trun_off: Option<usize> = None;
+        let trun_off: Option<usize>;
         write_box!(buf, b"traf", {
             write_box!(buf, b"tfhd", {
                 buf.put_u32(0x020000); // default-base-is-moof

camera_hub/src/livestream.rs

@@ -12,6 +12,7 @@ use std::pin::Pin;
 use std::sync::mpsc;
 use std::sync::mpsc::Sender;
 use std::task::{Context, Poll};
+use std::time::{SystemTime, UNIX_EPOCH};
 use tokio::io::AsyncWrite;
 
 /// Used to determine when to end livestream
@@ -104,12 +105,31 @@ pub fn livestream(
         // to prevent a malicious server from reordering the chunks.
         let mut data: Vec<u8> = chunk_number.to_be_bytes().to_vec();
         data.extend(rx.recv().unwrap());
+
+        let start = SystemTime::now();
+        let since_the_epoch = start
+        .duration_since(UNIX_EPOCH)
+        .expect("time should go forward");
+        debug!("Livestream: Received data for chunk {} at {}", chunk_number, since_the_epoch.as_millis());
+
         let enc_data = mls_client.encrypt(&data)?;
 
+        let end = SystemTime::now();
+        let end_since_epoch = end
+        .duration_since(UNIX_EPOCH)
+        .expect("time should go forward");
+        let diff = end_since_epoch.as_millis() - since_the_epoch.as_millis();
+        debug!("Livestream: Took {}ms for chunk {} for encryption", diff, chunk_number);
+
         let num_pending_files =
             http_client.livestream_upload(&group_name, enc_data, chunk_number)?;
         chunk_number += 1;
 
+        let end2 = SystemTime::now();
+        let end_since_epoch2 = end2.duration_since(UNIX_EPOCH).expect("time should go forward");
+        let diff2 = end_since_epoch2.as_millis() - end_since_epoch.as_millis();
+        debug!("Livestream: Took {}ms for chunk {} for uploading (curr time = {})", diff2, chunk_number - 1, end_since_epoch2.as_millis());
+
         // The server returns 0 when the app has explicitly ended livestream
         if num_pending_files == 0 || num_pending_files > MAX_NUM_PENDING_LIVESTREAM_CHUNKS {
             info!("Ending livestream.");

camera_hub/src/raspberry_pi/rpi_camera.rs

@@ -2,7 +2,7 @@
 //!
 //! SPDX-License-Identifier: GPL-3.0-or-later
 
-use std::time::Instant;
+use std::time::{Instant, SystemTime, UNIX_EPOCH};
 use std::{
     collections::VecDeque,
     io,
@@ -22,7 +22,7 @@ use crate::{
     traits::{Camera, CodecParameters},
     write_box,
 };
-use anyhow::{Context, Error};
+use anyhow::{Error};
 use bytes::{BufMut, BytesMut};
 use crossbeam_channel::unbounded;
 use image::RgbImage;
@@ -51,15 +51,15 @@ pub enum VideoFrameKind {
 pub struct VideoFrame {
     pub data: Vec<u8>,
     pub kind: VideoFrameKind,
-    pub timestamp: Instant,
+    pub timestamp: SystemTime,
 }
 
 impl VideoFrame {
     pub fn new(data: Vec<u8>, kind: VideoFrameKind) -> Self {
         Self {
             data,
             kind,
-            timestamp: Instant::now(),
+            timestamp: SystemTime::now(),
         }
     }
 }
@@ -223,11 +223,19 @@ impl RaspberryPiCamera {
                 let ts = frame_count * ticks_per_frame;
 
                 let avcc = Self::annexb_to_avcc_frame(&frame.data, /*strip_aud*/ true, /*strip_ps*/ true);
+
+                // Prepend per-frame SEI carrying capture time
+                let sei = Self::make_sei_unreg_avcc(frame.timestamp.duration_since(UNIX_EPOCH).unwrap().as_millis() as u64);
+                let mut sample = Vec::with_capacity(sei.len() + avcc.len());
+                sample.extend_from_slice(&sei);
+                sample.extend_from_slice(&avcc);
+
                 mp4.video(
-                    &avcc,
+                    &sample,
                     ts,
                     frame.kind == VideoFrameKind::IFrame,
                 ).await?;
+
                 frame_count += 1;
                 samples_in_fragment += 1;
             }
@@ -285,6 +293,63 @@ impl RaspberryPiCamera {
         Ok(())
     }
 
+
+    fn make_sei_unreg_avcc(epoch_ms: u64) -> Vec<u8> {
+        const NAL_TYPE_SEI: u8 = 6;
+        const PAYLOAD_TYPE_UNREG: u8 = 5;
+
+        // 16-byte UUID
+        const UUID: &[u8; 16] = b"SECLUSO_LATENCY_";
+
+        // Build raw RBSP (before EPB insertion)
+        // payload = UUID(16) + timestamp(8, big-endian)
+        let mut payload = Vec::with_capacity(24);
+        payload.extend_from_slice(UUID);
+        payload.extend_from_slice(&epoch_ms.to_be_bytes()); // 8B BE
+
+        // payloadType (5) using 255-extension coding
+        // payloadSize (24) using 255-extension coding
+        let mut rbsp = Vec::with_capacity(1 + 1 + payload.len() + 1);
+        rbsp.push(PAYLOAD_TYPE_UNREG);
+
+        let mut size = payload.len() as u32; // must be 24
+        while size >= 255 {
+            rbsp.push(255);
+            size -= 255;
+        }
+        rbsp.push(size as u8);
+
+        rbsp.extend_from_slice(&payload);
+
+        // rbsp_trailing_bits: 1 bit '1' then pad with zeros to byte boundary
+        rbsp.push(0x80);
+
+        // Emulation-prevention bytes
+        // Scan rbsp and after any 0x00 0x00 {00..03} pattern, insert 0x03 before that
+        let mut rbsp_epb = Vec::with_capacity(rbsp.len() + 8);
+        let mut zero_run = 0usize;
+        for &b in &rbsp {
+            if zero_run >= 2 && b <= 0x03 {
+                rbsp_epb.push(0x03);
+                zero_run = 0; // reset after insertion
+            }
+            rbsp_epb.push(b);
+            if b == 0x00 { zero_run += 1 } else { zero_run = 0 }
+        }
+
+        // Assemble the NAL (header + rbsp_epb)
+        // forbidden_zero_bit=0, nal_ref_idc=0, nal_unit_type=6 (SEI)
+        let mut nal = Vec::with_capacity(1 + rbsp_epb.len());
+        nal.push(0x00 | NAL_TYPE_SEI);
+        nal.extend_from_slice(&rbsp_epb);
+
+        // AVCC length-prefixed output
+        let mut out = Vec::with_capacity(4 + nal.len());
+        out.extend_from_slice(&(nal.len() as u32).to_be_bytes());
+        out.extend_from_slice(&nal);
+        out
+    }
+
     /// Streams fmp4 video.
     async fn write_fmp4(
         livestream_writer: LivestreamWriter,

camera_hub/src/raspberry_pi/rpi_dual_stream.rs

@@ -7,7 +7,7 @@ use std::collections::VecDeque;
 use std::os::unix::net::UnixStream;
 use std::sync::{Arc, Mutex};
 use std::thread::sleep;
-use std::time::Instant;
+use std::time::{SystemTime};
 use std::{
     io::{BufReader, Read, Write},
     process::{Command, Stdio},
@@ -77,7 +77,7 @@ pub fn start(
                             Ok(h264_frame2) => {
                                 if let Some(mut frame) = h264_frame2 {
                                     // Update the frame timestamp on extraction.
-                                    frame.timestamp = Instant::now();
+                                    frame.timestamp = SystemTime::now();
 
                                     if !sps_sent && frame.kind == VideoFrameKind::Sps {
                                         let _ = ps_tx.send(frame.clone());
@@ -166,7 +166,7 @@ fn add_frame_and_drop_old(frame_queue: Arc<Mutex<VecDeque<VideoFrame>>>, frame:
 
     // Remove frames older than the time window.
     while let Some(front) = queue.front() {
-        if Instant::now().duration_since(front.timestamp) > time_window {
+        if SystemTime::now().duration_since(front.timestamp).unwrap() > time_window {
             queue.pop_front();
         } else {
             break;