Merge pull request #102 from mum-rs/output-rework

Input rework

1 files changed, 65 insertions, 12 deletions

diff --git a/mumd/src/audio/input.rs b/mumd/src/audio/input.rs
index e45ff27..a1227e3 100644
--- a/mumd/src/audio/input.rs
+++ b/mumd/src/audio/input.rs
@@ -4,24 +4,42 @@ use log::*;
 use tokio::sync::watch;
 
 use crate::audio::SAMPLE_RATE;
+use crate::audio::transformers::{NoiseGate, Transformer};
 use crate::error::{AudioError, AudioStream};
 use crate::state::StatePhase;
 
 pub fn callback<T: Sample>(
-    mut input_sender: futures_channel::mpsc::Sender<f32>,
+    mut input_sender: futures_channel::mpsc::Sender<Vec<u8>>,
+    mut transformers: Vec<Box<dyn Transformer + Send + 'static>>,
+    mut opus_encoder: opus::Encoder,
+    buffer_size: usize,
     input_volume_receiver: watch::Receiver<f32>,
     phase_watcher: watch::Receiver<StatePhase>,
 ) -> impl FnMut(&[T], &InputCallbackInfo) + Send + 'static {
+    let mut buffer = Vec::with_capacity(buffer_size);
+
     move |data: &[T], _info: &InputCallbackInfo| {
         if !matches!(&*phase_watcher.borrow(), StatePhase::Connected(_)) {
             return;
         }
         let input_volume = *input_volume_receiver.borrow();
-        for sample in data.iter().map(|e| e.to_f32()).map(|e| e * input_volume) {
-            if let Err(_e) = input_sender.try_send(sample) {
-                warn!("Error sending audio: {}", _e);
+        let mut data = data.iter().map(|e| e.to_f32()).map(|e| e * input_volume);
+
+        while buffer.len() + data.len() > buffer_size {
+            buffer.extend(data.by_ref().take(buffer_size - buffer.len()));
+            let encoded = transformers
+                .iter_mut()
+                .try_fold(&mut buffer[..], |acc, e| e.transform(acc))
+                .map(|buf| opus_encoder.encode_vec_float(&*buf, buffer_size).unwrap());
+
+            if let Some(encoded) = encoded {
+                if let Err(e) = input_sender.try_send(encoded) {
+                    warn!("Error sending audio: {}", e);
+                }
             }
+            buffer.clear();
         }
+        buffer.extend(data);
     }
 }
 
@@ -29,13 +47,13 @@ pub trait AudioInputDevice {
     fn play(&self) -> Result<(), AudioError>;
     fn pause(&self) -> Result<(), AudioError>;
     fn set_volume(&self, volume: f32);
-    fn sample_receiver(&mut self) -> futures_channel::mpsc::Receiver<f32>;
+    fn sample_receiver(&mut self) -> Option<futures_channel::mpsc::Receiver<Vec<u8>>>;
     fn num_channels(&self) -> usize;
 }
 
 pub struct DefaultAudioInputDevice {
     stream: cpal::Stream,
-    sample_receiver: Option<futures_channel::mpsc::Receiver<f32>>,
+    sample_receiver: Option<futures_channel::mpsc::Receiver<Vec<u8>>>,
     volume_sender: watch::Sender<f32>,
     channels: u16,
 }
@@ -44,6 +62,7 @@ impl DefaultAudioInputDevice {
     pub fn new(
         input_volume: f32,
         phase_watcher: watch::Receiver<StatePhase>,
+        frame_size: u32, // blocks of 2.5 ms
     ) -> Result<Self, AudioError> {
         let sample_rate = SampleRate(SAMPLE_RATE);
 
@@ -73,20 +92,55 @@ impl DefaultAudioInputDevice {
 
         let (volume_sender, input_volume_receiver) = watch::channel::<f32>(input_volume);
 
+        let opus_encoder = opus::Encoder::new(
+            sample_rate.0,
+            match input_config.channels {
+                1 => opus::Channels::Mono,
+                2 => opus::Channels::Stereo,
+                _ => unimplemented!("Only 1 or 2 channels supported, got {}", input_config.channels),
+            },
+            opus::Application::Voip,
+        )
+        .unwrap();
+        let buffer_size = (sample_rate.0 * frame_size / 400) as usize;
+
+        let transformers = vec![Box::new(NoiseGate::new(50)) as Box<dyn Transformer + Send + 'static>];
+
         let input_stream = match input_supported_sample_format {
             SampleFormat::F32 => input_device.build_input_stream(
                 &input_config,
-                callback::<f32>(sample_sender, input_volume_receiver, phase_watcher),
+                callback::<f32>(
+                    sample_sender, 
+                    transformers, 
+                    opus_encoder,
+                    buffer_size,
+                    input_volume_receiver, 
+                    phase_watcher
+                ),
                 err_fn,
             ),
             SampleFormat::I16 => input_device.build_input_stream(
                 &input_config,
-                callback::<i16>(sample_sender, input_volume_receiver, phase_watcher),
+                callback::<i16>(
+                    sample_sender, 
+                    transformers, 
+                    opus_encoder,
+                    buffer_size,
+                    input_volume_receiver, 
+                    phase_watcher
+                ),
                 err_fn,
             ),
             SampleFormat::U16 => input_device.build_input_stream(
                 &input_config,
-                callback::<u16>(sample_sender, input_volume_receiver, phase_watcher),
+                callback::<u16>(
+                    sample_sender, 
+                    transformers, 
+                    opus_encoder,
+                    buffer_size,
+                    input_volume_receiver, 
+                    phase_watcher
+                ),
                 err_fn,
             ),
         }
@@ -116,9 +170,8 @@ impl AudioInputDevice for DefaultAudioInputDevice {
     fn set_volume(&self, volume: f32) {
         self.volume_sender.send(volume).unwrap();
     }
-    fn sample_receiver(&mut self) -> futures_channel::mpsc::Receiver<f32> {
-        let ret = self.sample_receiver.take();
-        ret.unwrap()
+    fn sample_receiver(&mut self) -> Option<futures_channel::mpsc::Receiver<Vec<u8>>> {
+        self.sample_receiver.take()
     }
     fn num_channels(&self) -> usize {
         self.channels as usize