Merge pull request #55 from HEnquist/cpaladjust

Cpaladjust

.github/workflows/publish.yml

@@ -140,12 +140,12 @@ jobs:
           args: --release --no-default-features --features cpal-backend --features socketserver
 
       - name: Compress
-        run: zip -j camilladsp.zip target/release/camilladsp
+        run: tar -zcvf camilladsp.tar.gz -C target/release camilladsp
 
       - name: Upload binaries to release
         uses: svenstaro/upload-release-action@v1-release
         with:
           repo_token: ${{ secrets.GITHUB_TOKEN }}
-          file: camilladsp.zip
-          asset_name: camilladsp-macos-amd64.zip
+          file: camilladsp.tar.gz
+          asset_name: camilladsp-macos-amd64.tar.gz
           tag: ${{ github.ref }}

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "camilladsp"
-version = "0.3.0"
+version = "0.3.1"
 authors = ["Henrik Enquist <[email protected]>"]
 description = "A flexible tool for processing audio"
 

README.md

@@ -110,7 +110,7 @@ The following configurations are provided:
 |----------|-------------|----------|
 | `camilladsp-linux-amd64.tar.gz` | Linux on 64-bit Intel or AMD CPU | Alsa, Pulseaudio |
 | `camilladsp-linux-armv7.tar.gz` | Linux on Armv7 with Neon, intended for Raspberry Pi 2 and up but should also work on others | Alsa |
-| `camilladsp-macos-amd64.zip` | macOS on 64-bit Intel CPU | CoreAudio |
+| `camilladsp-macos-amd64.tar.gz` | macOS on 64-bit Intel CPU | CoreAudio |
 | `camilladsp-windows-amd64.zip` | Windows on 64-bit Intel or AMD CPU | Wasapi |
 
 All builds include the Websocket server.
@@ -390,15 +390,16 @@ devices:
 * `enable_rate_adjust` (optional, defaults to false)
 
   This enables the playback device to control the rate of the capture device, 
-  in order to avoid buffer underruns of a slowly increasing latency. This is currently only supported when using an Alsa playback device.
+  in order to avoid buffer underruns of a slowly increasing latency. This is currently supported when using an Alsa, Wasapi or CoreAudio playback device.
   Setting the rate can be done in two ways.
   * If the capture device is an Alsa Loopback device, the adjustment is done by tuning the virtual sample clock of the Loopback device. This avoids any need for resampling.
   * If resampling is enabled, the adjustment is done by tuning the resampling ratio. The `resampler_type` must then be one of the "Async" variants.
 
 
 * `target_level` (optional, defaults to the `chunksize` value)
+
   The value is the number of samples that should be left in the buffer of the playback device
-  when the next chunk arrives. It works by fine tuning the sample rate of the virtual Loopback device.
+  when the next chunk arrives. Only applies when `enable_rate_adjust` is set to `true`.
   It will take some experimentation to find the right number. 
   If it's too small there will be buffer underruns from time to time, 
   and making it too large might lead to a longer input-output delay than what is acceptable. 
@@ -407,7 +408,7 @@ devices:
 * `adjust_period` (optional, defaults to 10)
 
   The `adjust_period` parameter is used to set the interval between corrections, in seconds. 
-  The default is 10 seconds.
+  The default is 10 seconds. Only applies when `enable_rate_adjust` is set to `true`.
 
 * `silence_threshold` & `silence_timeout` (optional)
   The fields `silence_threshold` and `silence_timeout` are optional 
@@ -556,10 +557,6 @@ and any rate adjust request will be ignored.
 See the library documentation for more details. [Rubato on docs.rs](https://docs.rs/rubato/0.1.0/rubato/)
 
 
-
-
-
-
 ## Mixers
 A mixer is used to route audio between channels, and to increase or decrease the number of channels in the pipeline.
 Example for a mixer that copies two channels into four:

src/audiodevice.rs

@@ -154,6 +154,9 @@ pub fn get_playback_device(conf: config::Devices) -> Box<dyn PlaybackDevice> {
             chunksize: conf.chunksize,
             channels,
             format,
+            target_level: conf.target_level,
+            adjust_period: conf.adjust_period,
+            enable_rate_adjust: conf.enable_rate_adjust,
         }),
         #[cfg(all(feature = "cpal-backend", target_os = "windows"))]
         config::PlaybackDevice::Wasapi {
@@ -167,6 +170,9 @@ pub fn get_playback_device(conf: config::Devices) -> Box<dyn PlaybackDevice> {
             chunksize: conf.chunksize,
             channels,
             format,
+            target_level: conf.target_level,
+            adjust_period: conf.adjust_period,
+            enable_rate_adjust: conf.enable_rate_adjust,
         }),
     }
 }

src/cpaldevice.rs

@@ -10,9 +10,11 @@ use cpal::{ChannelCount, Format, HostId, SampleRate};
 use cpal::{Device, EventLoop, Host};
 use rubato::Resampler;
 use std::collections::VecDeque;
+use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::mpsc;
 use std::sync::{Arc, Barrier};
 use std::thread;
+use std::time::SystemTime;
 
 use CommandMessage;
 use PrcFmt;
@@ -35,6 +37,9 @@ pub struct CpalPlaybackDevice {
     pub chunksize: usize,
     pub channels: usize,
     pub format: SampleFormat,
+    pub target_level: usize,
+    pub adjust_period: f32,
+    pub enable_rate_adjust: bool,
 }
 
 #[derive(Clone, Debug)]
@@ -163,6 +168,16 @@ impl PlaybackDevice for CpalPlaybackDevice {
         let samplerate = self.samplerate;
         let chunksize = self.chunksize;
         let channels = self.channels;
+        let target_level = if self.target_level > 0 {
+            self.target_level
+        } else {
+            self.chunksize
+        };
+        let adjust_period = self.adjust_period;
+        let adjust = self.adjust_period > 0.0 && self.enable_rate_adjust;
+        let chunksize_clone = chunksize;
+        let channels_clone = channels;
+
         let bits = match self.format {
             SampleFormat::S16LE => 16,
             SampleFormat::S24LE => 24,
@@ -185,10 +200,19 @@ impl PlaybackDevice for CpalPlaybackDevice {
                         barrier.wait();
                         debug!("Starting playback loop");
                         let (tx_dev, rx_dev) = mpsc::sync_channel(1);
+                        let buffer_fill = Arc::new(AtomicUsize::new(0));
+                        let buffer_fill_clone = buffer_fill.clone();
+                        let mut start = SystemTime::now();
+                        let mut now;
+                        let mut delay = 0;
+                        let mut ndelays = 0;
+                        let mut speed;
+                        let mut diff: isize;
+
                         match format {
                             SampleFormat::S16LE => {
                                 let mut sample_queue: VecDeque<i16> =
-                                    VecDeque::with_capacity(4 * chunksize * channels);
+                                    VecDeque::with_capacity(4 * chunksize_clone * channels_clone);
                                 std::thread::spawn(move || {
                                     event_loop.run(move |id, result| {
                                         let data = match result {
@@ -223,6 +247,8 @@ impl PlaybackDevice for CpalPlaybackDevice {
                                                     &mut buffer,
                                                     &mut sample_queue,
                                                 );
+                                                buffer_fill_clone
+                                                    .store(sample_queue.len(), Ordering::Relaxed);
                                             }
                                             _ => (),
                                         };
@@ -231,7 +257,7 @@ impl PlaybackDevice for CpalPlaybackDevice {
                             }
                             SampleFormat::FLOAT32LE => {
                                 let mut sample_queue: VecDeque<f32> =
-                                    VecDeque::with_capacity(4 * chunksize * channels);
+                                    VecDeque::with_capacity(4 * chunksize_clone * channels_clone);
                                 std::thread::spawn(move || {
                                     event_loop.run(move |id, result| {
                                         let data = match result {
@@ -262,6 +288,8 @@ impl PlaybackDevice for CpalPlaybackDevice {
                                                     &mut buffer,
                                                     &mut sample_queue,
                                                 );
+                                                buffer_fill_clone
+                                                    .store(sample_queue.len(), Ordering::Relaxed);
                                             }
                                             _ => (),
                                         };
@@ -273,6 +301,29 @@ impl PlaybackDevice for CpalPlaybackDevice {
                         loop {
                             match channel.recv() {
                                 Ok(AudioMessage::Audio(chunk)) => {
+                                    now = SystemTime::now();
+                                    delay += buffer_fill.load(Ordering::Relaxed) as isize;
+                                    ndelays += 1;
+                                    if adjust
+                                        && (now.duration_since(start).unwrap().as_millis()
+                                            > ((1000.0 * adjust_period) as u128))
+                                    {
+                                        let av_delay = delay / ndelays;
+                                        diff = av_delay - target_level as isize;
+                                        let rel_diff = (diff as f64) / (samplerate as f64);
+                                        speed = 1.0 - 0.5 * rel_diff / adjust_period as f64;
+                                        debug!(
+                                            "Current buffer level {}, set capture rate to {}%",
+                                            av_delay,
+                                            100.0 * speed
+                                        );
+                                        start = now;
+                                        delay = 0;
+                                        ndelays = 0;
+                                        status_channel
+                                            .send(StatusMessage::SetSpeed { speed })
+                                            .unwrap();
+                                    }
                                     tx_dev.send(chunk).unwrap();
                                 }
                                 Ok(AudioMessage::EndOfStream) => {
@@ -443,6 +494,7 @@ impl CaptureDevice for CpalCaptureDevice {
                                 }
                                 Ok(CommandMessage::SetSpeed { speed }) => {
                                     if let Some(resampl) = &mut resampler {
+                                        debug!("Adjusting resampler rate to {}", speed);
                                         if async_src {
                                             if resampl.set_resample_ratio_relative(speed).is_err() {
                                                 debug!("Failed to set resampling speed to {}", speed);