more fixes

Author: Brendonovich

crates/enc-ffmpeg/src/audio/aac.rs

@@ -91,7 +91,7 @@ impl AACEncoder {
         encoder.set_rate(rate);
         encoder.set_format(output_config.sample_format);
         encoder.set_channel_layout(output_config.channel_layout());
-        encoder.set_time_base(output_config.time_base);
+        encoder.set_time_base(FFRational(1, output_config.rate()));
 
         let encoder = encoder.open()?;
 

crates/enc-ffmpeg/src/audio/buffered_resampler.rs

@@ -11,6 +11,8 @@ pub struct BufferedResampler {
     resampler: ffmpeg::software::resampling::Context,
     buffer: VecDeque<(ffmpeg::frame::Audio, i64)>,
     sample_index: usize,
+    // used to account for cases where pts is rounded down instead of up
+    min_next_pts: Option<i64>,
 }
 
 impl BufferedResampler {
@@ -19,6 +21,7 @@ impl BufferedResampler {
             resampler,
             buffer: VecDeque::new(),
             sample_index: 0,
+            min_next_pts: None,
         }
     }
 
@@ -46,7 +49,13 @@ impl BufferedResampler {
         *self.resampler.output()
     }
 
-    pub fn add_frame(&mut self, frame: ffmpeg::frame::Audio) {
+    pub fn add_frame(&mut self, mut frame: ffmpeg::frame::Audio) {
+        if let Some(min_next_pts) = self.min_next_pts {
+            if let Some(pts) = frame.pts() {
+                frame.set_pts(Some(pts.max(min_next_pts)));
+            }
+        }
+
         let pts = frame.pts().unwrap();
 
         let mut resampled_frame = ffmpeg::frame::Audio::empty();
@@ -76,6 +85,8 @@ impl BufferedResampler {
 
             next_pts = next_pts + samples as i64;
         }
+
+        self.min_next_pts = Some(pts + frame.samples() as i64);
     }
 
     fn get_frame_inner(&mut self, samples: usize) -> Option<ffmpeg::frame::Audio> {

crates/recording/examples/recording-cli.rs

@@ -36,17 +36,17 @@ pub async fn main() {
     //     .await
     //     .unwrap();
 
-    let (error_tx, _) = flume::bounded(1);
-    let mic_feed = MicrophoneFeed::spawn(MicrophoneFeed::new(error_tx));
-
-    mic_feed
-        .ask(microphone::SetInput {
-            label: MicrophoneFeed::default().map(|v| v.0).unwrap(),
-        })
-        .await
-        .unwrap()
-        .await
-        .unwrap();
+    // let (error_tx, _) = flume::bounded(1);
+    // let mic_feed = MicrophoneFeed::spawn(MicrophoneFeed::new(error_tx));
+
+    // mic_feed
+    //     .ask(microphone::SetInput {
+    //         label: MicrophoneFeed::default().map(|v| v.0).unwrap(),
+    //     })
+    //     .await
+    //     .unwrap()
+    //     .await
+    //     .unwrap();
 
     // tokio::time::sleep(Duration::from_millis(10)).await;
 

crates/recording/src/capture_pipeline.rs

@@ -15,7 +15,7 @@ use std::{
     future::Future,
     path::PathBuf,
     sync::{Arc, atomic::AtomicBool},
-    time::{Duration, SystemTime},
+    time::SystemTime,
 };
 
 pub trait MakeCapturePipeline: ScreenCaptureFormat + std::fmt::Debug + 'static {
@@ -395,8 +395,6 @@ impl MakeCapturePipeline for screen_capture::Direct3DCapture {
     where
         Self: Sized,
     {
-        use std::sync::mpsc;
-
         use cap_enc_ffmpeg::{AACEncoder, AudioEncoder};
         use windows::Graphics::SizeInt32;
 
@@ -490,41 +488,35 @@ impl MakeCapturePipeline for screen_capture::Direct3DCapture {
 
         let output = Arc::new(std::sync::Mutex::new(output));
 
-        let (screen_first_tx, screen_first_rx) = mpsc::sync_channel(1);
+        let (first_frame_tx, first_frame_rx) = tokio::sync::oneshot::channel::<Timestamp>();
 
         if let Some(mut audio_encoder) = audio_encoder {
             builder.spawn_source("audio_mixer", audio_mixer);
 
-            // let is_done = is_done.clone();
             let output = output.clone();
             builder.spawn_task("audio_encoding", move |ready| {
-                let screen_first_offset = loop {
-                    match screen_first_rx.recv_timeout(Duration::from_millis(5)) {
-                        Ok(offset) => {
-                            audio_rx.drain().count();
-                            break offset;
-                        }
-                        Err(mpsc::RecvTimeoutError::Timeout) => continue,
-                        Err(mpsc::RecvTimeoutError::Disconnected) => return Ok(()),
-                    }
-                };
-
                 let _ = ready.send(Ok(()));
 
+                let time = first_frame_rx.blocking_recv().unwrap();
+                let screen_first_offset = time.duration_since(start_time);
+
                 while let Ok((mut frame, timestamp)) = audio_rx.recv() {
                     let ts_offset = timestamp.duration_since(start_time);
+                    dbg!(ts_offset);
 
                     let Some(ts_offset) = ts_offset.checked_sub(screen_first_offset) else {
                         continue;
                     };
 
                     let pts = (ts_offset.as_secs_f64() * frame.rate() as f64) as i64;
                     frame.set_pts(Some(pts));
+                    dbg!(pts);
 
                     if let Ok(mut output) = output.lock() {
-                        audio_encoder.queue_frame(frame, &mut *output);
+                        audio_encoder.queue_frame(frame, &mut output)
                     }
                 }
+
                 Ok(())
             });
         }
@@ -540,7 +532,7 @@ impl MakeCapturePipeline for screen_capture::Direct3DCapture {
 
                     let _ = ready.send(Ok(()));
 
-                    let mut screen_first_tx = Some(screen_first_tx);
+                    let mut first_frame_tx = Some(first_frame_tx);
 
                     while let Ok(e) = encoder.get_event() {
                         match e {
@@ -560,9 +552,8 @@ impl MakeCapturePipeline for screen_capture::Direct3DCapture {
                                     PerformanceCounterTimestamp::new(frame_time.Duration),
                                 );
 
-                                if let Some(screen_first_tx) = screen_first_tx.take() {
-                                    let _ = screen_first_tx
-                                        .try_send(timestamp.duration_since(start_time));
+                                if let Some(first_frame_tx) = first_frame_tx.take() {
+                                    let _ = first_frame_tx.send(timestamp);
                                 }
 
                                 encoder

crates/recording/src/sources/audio_mixer.rs

@@ -116,6 +116,8 @@ impl AudioMixerBuilder {
             _filter_graph: filter_graph,
             _amix: amix,
             _aformat: aformat,
+            start_timestamp: None,
+            timestamps: Timestamps::now(),
         })
     }
 }
@@ -131,6 +133,8 @@ pub struct AudioMixer {
     _filter_graph: ffmpeg::filter::Graph,
     _amix: ffmpeg::filter::Context,
     _aformat: ffmpeg::filter::Context,
+    timestamps: Timestamps,
+    start_timestamp: Option<Timestamp>,
 }
 
 impl AudioMixer {
@@ -141,15 +145,15 @@ impl AudioMixer {
     );
     pub const BUFFER_TIMEOUT: Duration = Duration::from_millis(10);
 
-    fn buffer_sources(&mut self, start: Timestamps) {
+    fn buffer_sources(&mut self) {
         for source in &mut self.sources {
             let rate = source.info.rate();
 
             while let Ok((frame, timestamp)) = source.rx.try_recv() {
                 // if gap between incoming and last, insert silence
                 if let Some((buffer_last_timestamp, buffer_last_duration)) = source.buffer_last {
-                    let timestamp_elapsed = timestamp.duration_since(start);
-                    let buffer_last_elapsed = buffer_last_timestamp.duration_since(start);
+                    let timestamp_elapsed = timestamp.duration_since(self.timestamps);
+                    let buffer_last_elapsed = buffer_last_timestamp.duration_since(self.timestamps);
 
                     if timestamp_elapsed > buffer_last_elapsed {
                         let elapsed_since_last_frame = timestamp_elapsed - buffer_last_elapsed;
@@ -158,9 +162,9 @@ impl AudioMixer {
                             && buffer_last_duration - elapsed_since_last_frame
                                 >= Duration::from_millis(1)
                         {
-                            let gap = (buffer_last_timestamp.duration_since(start)
+                            let gap = (buffer_last_timestamp.duration_since(self.timestamps)
                                 + buffer_last_duration)
-                                - timestamp.duration_since(start);
+                                - timestamp.duration_since(self.timestamps);
 
                             debug!("Gap between last buffer frame, inserting {gap:?} of silence");
 
@@ -183,7 +187,7 @@ impl AudioMixer {
                         }
                     }
                 } else {
-                    let gap = timestamp.duration_since(start);
+                    let gap = timestamp.duration_since(self.timestamps);
 
                     if !gap.is_zero() {
                         debug!("Gap from beginning of stream, inserting {gap:?} of silence");
@@ -210,8 +214,9 @@ impl AudioMixer {
 
                             frame.set_rate(rate as u32);
 
-                            let timestamp =
-                                Timestamp::Instant(start.instant() + chunk_duration * i as u32);
+                            let timestamp = Timestamp::Instant(
+                                self.timestamps.instant() + chunk_duration * i as u32,
+                            );
                             source.buffer_last = Some((timestamp, chunk_duration));
                             source.buffer.push_back((frame, timestamp));
                         }
@@ -233,14 +238,15 @@ impl AudioMixer {
                         let duration =
                             Duration::from_secs_f64(leftover_chunk_size as f64 / rate as f64);
                         let timestamp = Timestamp::Instant(
-                            start.instant() + chunk_duration * chunks.floor() as u32 + duration,
+                            self.timestamps.instant()
+                                + chunk_duration * chunks.floor() as u32
+                                + duration,
                         );
                         source.buffer_last = Some((timestamp, duration));
                         source.buffer.push_back((frame, timestamp));
                     }
                 }
 
-                // dbg!(frame.samples());
                 source.buffer_last = Some((
                     timestamp,
                     Duration::from_secs_f64(frame.samples() as f64 / frame.rate() as f64),
@@ -251,7 +257,23 @@ impl AudioMixer {
     }
 
     fn tick(&mut self, start: Timestamps, now: Instant) -> Result<(), ()> {
-        self.buffer_sources(start);
+        self.buffer_sources();
+
+        if self.start_timestamp.is_none() {
+            self.start_timestamp = self
+                .sources
+                .iter()
+                .filter_map(|s| s.buffer.get(0))
+                .min_by(|a, b| {
+                    a.1.duration_since(self.timestamps)
+                        .cmp(&b.1.duration_since(self.timestamps))
+                })
+                .map(|v| v.1);
+        }
+
+        let Some(start_timestamp) = self.start_timestamp else {
+            return Ok(());
+        };
 
         for (i, source) in self.sources.iter_mut().enumerate() {
             for buffer in source.buffer.drain(..) {
@@ -262,12 +284,10 @@ impl AudioMixer {
         let mut filtered = ffmpeg::frame::Audio::empty();
         while self.abuffersink.sink().frame(&mut filtered).is_ok() {
             let elapsed = Duration::from_secs_f64(self.samples_out as f64 / filtered.rate() as f64);
-            let timestamp = start.instant() + elapsed;
+            let timestamp = start.instant() + start_timestamp.duration_since(start) + elapsed;
 
             self.samples_out += filtered.samples();
 
-            // dbg!(filtered.samples(), timestamp);
-
             if self
                 .output
                 .send((filtered, Timestamp::Instant(timestamp)))

crates/recording/src/sources/screen_capture/windows.rs

@@ -487,6 +487,8 @@ pub mod audio {
 
                     let timestamp = Timestamp::from_cpal(info.timestamp().capture);
 
+                    dbg!(timestamp);
+
                     let _ = audio_tx.send((data.as_ffmpeg(config), timestamp));
                 },
                 move |e| {

crates/recording/src/studio_recording.rs

@@ -861,7 +861,7 @@ async fn create_segment_pipeline(
 
                 let elapsed = timestamp.duration_since(start_time)
                     - first_timestamp.duration_since(start_time);
-                frame.set_pts(Some((elapsed.as_secs_f64() * rate) as i64));
+                frame.set_pts(Some(dbg!(elapsed.as_secs_f64() * rate).round() as i64));
 
                 output.queue_frame(frame);
             }