From bc65445af44a335a0586a393c792614330258249 Mon Sep 17 00:00:00 2001 From: Eskil Queseth Date: Sat, 12 Jun 2021 02:30:01 +0200 Subject: simplify audio output infrastructure --- mumd/src/audio.rs | 33 ++-- mumd/src/audio/input.rs | 79 ++++++++-- mumd/src/audio/noise_gate.rs | 350 ------------------------------------------- 3 files changed, 78 insertions(+), 384 deletions(-) (limited to 'mumd/src') diff --git a/mumd/src/audio.rs b/mumd/src/audio.rs index 63adcc6..67d91c9 100644 --- a/mumd/src/audio.rs +++ b/mumd/src/audio.rs @@ -3,9 +3,6 @@ mod noise_gate; pub mod output; use crate::audio::input::{AudioInputDevice, DefaultAudioInputDevice}; -use crate::audio::noise_gate::{ - from_interleaved_samples_stream, OpusEncoder, StreamingNoiseGate, StreamingSignalExt, -}; use crate::audio::output::{AudioOutputDevice, ClientStream, DefaultAudioOutputDevice}; use crate::error::AudioError; use crate::network::VoiceStreamType; @@ -81,27 +78,19 @@ impl AudioInput { input_volume: f32, phase_watcher: watch::Receiver, ) -> Result { - let mut default = DefaultAudioInputDevice::new(input_volume, phase_watcher)?; + let mut default = DefaultAudioInputDevice::new(input_volume, phase_watcher, 4)?; let sample_rate = SampleRate(SAMPLE_RATE); - let opus_stream = OpusEncoder::new( - 4, - sample_rate.0, - default.num_channels(), - StreamingSignalExt::into_interleaved_samples(StreamingNoiseGate::new( - from_interleaved_samples_stream::<_, f32>(default.sample_receiver()), //TODO group frames correctly - 10_000, - )), - ) - .enumerate() - .map(|(i, e)| VoicePacket::Audio { - _dst: std::marker::PhantomData, - target: 0, // normal speech - session_id: (), // unused for server-bound packets - seq_num: i as u64, - payload: VoicePacketPayload::Opus(e.into(), false), - position_info: None, - }); + let opus_stream = default.sample_receiver().unwrap() + .enumerate() + .map(|(i, e)| VoicePacket::Audio { + _dst: std::marker::PhantomData, + target: 0, // normal speech + session_id: (), // unused for server-bound packets + seq_num: i as u64, + payload: VoicePacketPayload::Opus(e.into(), false), + position_info: None, + }); default.play()?; diff --git a/mumd/src/audio/input.rs b/mumd/src/audio/input.rs index e45ff27..162dd2c 100644 --- a/mumd/src/audio/input.rs +++ b/mumd/src/audio/input.rs @@ -8,20 +8,49 @@ use crate::error::{AudioError, AudioStream}; use crate::state::StatePhase; pub fn callback( - mut input_sender: futures_channel::mpsc::Sender, + mut input_sender: futures_channel::mpsc::Sender>, + transformers: Vec Option<&mut [f32]> + Send + 'static>>, + frame_size: u32, + sample_rate: u32, + channels: u16, input_volume_receiver: watch::Receiver, phase_watcher: watch::Receiver, ) -> impl FnMut(&[T], &InputCallbackInfo) + Send + 'static { + let buffer_size = (sample_rate * frame_size / 400) as usize; + let mut opus_encoder = opus::Encoder::new( + sample_rate, + match channels { + 1 => opus::Channels::Mono, + 2 => opus::Channels::Stereo, + _ => unimplemented!("Only 1 or 2 channels supported, got {}", channels), + }, + opus::Application::Voip, + ) + .unwrap(); + 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() + .try_fold(&mut buffer[..], |acc, e| e(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 +58,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; + fn sample_receiver(&mut self) -> Option>>; fn num_channels(&self) -> usize; } pub struct DefaultAudioInputDevice { stream: cpal::Stream, - sample_receiver: Option>, + sample_receiver: Option>>, volume_sender: watch::Sender, channels: u16, } @@ -44,6 +73,7 @@ impl DefaultAudioInputDevice { pub fn new( input_volume: f32, phase_watcher: watch::Receiver, + frame_size: u32, ) -> Result { let sample_rate = SampleRate(SAMPLE_RATE); @@ -73,20 +103,46 @@ impl DefaultAudioInputDevice { let (volume_sender, input_volume_receiver) = watch::channel::(input_volume); + let transformers = vec![]; + let input_stream = match input_supported_sample_format { SampleFormat::F32 => input_device.build_input_stream( &input_config, - callback::(sample_sender, input_volume_receiver, phase_watcher), + callback::( + sample_sender, + transformers, + frame_size, + SAMPLE_RATE, + input_config.channels, + input_volume_receiver, + phase_watcher + ), err_fn, ), SampleFormat::I16 => input_device.build_input_stream( &input_config, - callback::(sample_sender, input_volume_receiver, phase_watcher), + callback::( + sample_sender, + transformers, + frame_size, + SAMPLE_RATE, + input_config.channels, + input_volume_receiver, + phase_watcher + ), err_fn, ), SampleFormat::U16 => input_device.build_input_stream( &input_config, - callback::(sample_sender, input_volume_receiver, phase_watcher), + callback::( + sample_sender, + transformers, + frame_size, + SAMPLE_RATE, + input_config.channels, + input_volume_receiver, + phase_watcher + ), err_fn, ), } @@ -116,9 +172,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 { - let ret = self.sample_receiver.take(); - ret.unwrap() + fn sample_receiver(&mut self) -> Option>> { + self.sample_receiver.take() } fn num_channels(&self) -> usize { self.channels as usize diff --git a/mumd/src/audio/noise_gate.rs b/mumd/src/audio/noise_gate.rs index bd1a262..e69de29 100644 --- a/mumd/src/audio/noise_gate.rs +++ b/mumd/src/audio/noise_gate.rs @@ -1,350 +0,0 @@ -use dasp_frame::Frame; -use dasp_sample::{Sample, SignedSample, ToSample}; -use dasp_signal::Signal; -use futures_util::stream::Stream; -use opus::Channels; -use std::future::Future; -use std::pin::Pin; -use std::task::{Context, Poll}; - -type FloatSample = <::Sample as Sample>::Float; - -pub struct StreamingNoiseGate { - open: usize, - signal: S, - deactivation_delay: usize, - alltime_high: FloatSample, -} - -impl StreamingNoiseGate { - pub fn new(signal: S, deactivation_delay: usize) -> StreamingNoiseGate { - Self { - open: 0, - signal, - deactivation_delay, - alltime_high: FloatSample::::EQUILIBRIUM, - } - } -} - -impl StreamingSignal for StreamingNoiseGate -where - S: StreamingSignal + Unpin, - FloatSample: Unpin, - ::Frame: Unpin, -{ - type Frame = S::Frame; - - fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - const MUTE_PERCENTAGE: f32 = 0.1; - - let s = self.get_mut(); - - let frame = match S::poll_next(Pin::new(&mut s.signal), cx) { - Poll::Ready(v) => v, - Poll::Pending => return Poll::Pending, - }; - - if let Some(highest) = frame - .to_float_frame() - .channels() - .find(|e| abs(e.clone()) > s.alltime_high) - { - s.alltime_high = highest; - } - - match s.open { - 0 => { - if frame - .to_float_frame() - .channels() - .any(|e| abs(e) >= s.alltime_high.mul_amp(MUTE_PERCENTAGE.to_sample())) - { - s.open = s.deactivation_delay; - } - } - _ => { - if frame - .to_float_frame() - .channels() - .any(|e| abs(e) >= s.alltime_high.mul_amp(MUTE_PERCENTAGE.to_sample())) - { - s.open = s.deactivation_delay; - } else { - s.open -= 1; - } - } - } - - if s.open != 0 { - Poll::Ready(frame) - } else { - Poll::Ready(::EQUILIBRIUM) - } - } - - fn is_exhausted(&self) -> bool { - self.signal.is_exhausted() - } -} - -fn abs(sample: S) -> S { - let zero = S::EQUILIBRIUM; - if sample >= zero { - sample - } else { - -sample - } -} - -pub trait StreamingSignal { - type Frame: Frame; - - fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll; - - fn is_exhausted(&self) -> bool { - false - } -} - -impl StreamingSignal for S -where - S: Signal + Unpin, -{ - type Frame = S::Frame; - - fn poll_next(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll { - Poll::Ready(self.get_mut().next()) - } -} - -pub trait StreamingSignalExt: StreamingSignal { - fn next(&mut self) -> Next<'_, Self> { - Next { stream: self } - } - - fn into_interleaved_samples(self) -> IntoInterleavedSamples - where - Self: Sized, - { - IntoInterleavedSamples { - signal: self, - current_frame: None, - } - } -} - -impl StreamingSignalExt for S where S: StreamingSignal {} - -pub struct Next<'a, S: ?Sized> { - stream: &'a mut S, -} - -impl<'a, S: StreamingSignal + Unpin> Future for Next<'a, S> { - type Output = S::Frame; - - fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - match S::poll_next(Pin::new(self.stream), cx) { - Poll::Ready(val) => Poll::Ready(val), - Poll::Pending => Poll::Pending, - } - } -} - -pub struct IntoInterleavedSamples { - signal: S, - current_frame: Option<::Channels>, -} - -impl Stream for IntoInterleavedSamples -where - S: StreamingSignal + Unpin, - <::Frame as Frame>::Channels: Unpin, -{ - type Item = ::Sample; - - fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - let s = self.get_mut(); - loop { - if s.current_frame.is_some() { - if let Some(channel) = s.current_frame.as_mut().unwrap().next() { - return Poll::Ready(Some(channel)); - } - } - match S::poll_next(Pin::new(&mut s.signal), cx) { - Poll::Ready(val) => { - s.current_frame = Some(val.channels()); - } - Poll::Pending => return Poll::Pending, - } - } - } -} - -struct FromStream { - stream: S, - underlying_exhausted: bool, -} - -impl StreamingSignal for FromStream -where - S: Stream + Unpin, - S::Item: Frame + Unpin, -{ - type Frame = S::Item; - - fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - let s = self.get_mut(); - if s.underlying_exhausted { - return Poll::Ready(::EQUILIBRIUM); - } - match S::poll_next(Pin::new(&mut s.stream), cx) { - Poll::Ready(Some(val)) => Poll::Ready(val), - Poll::Ready(None) => { - s.underlying_exhausted = true; - return Poll::Ready(::EQUILIBRIUM); - } - Poll::Pending => Poll::Pending, - } - } - - fn is_exhausted(&self) -> bool { - self.underlying_exhausted - } -} - -pub struct FromInterleavedSamplesStream -where - F: Frame, -{ - stream: S, - next_buf: Vec, - underlying_exhausted: bool, -} - -pub fn from_interleaved_samples_stream(stream: S) -> FromInterleavedSamplesStream -where - S: Stream + Unpin, - S::Item: Sample, - F: Frame, -{ - FromInterleavedSamplesStream { - stream, - next_buf: Vec::new(), - underlying_exhausted: false, - } -} - -impl StreamingSignal for FromInterleavedSamplesStream -where - S: Stream + Unpin, - S::Item: Sample + Unpin, - F: Frame + Unpin, -{ - type Frame = F; - - fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - let s = self.get_mut(); - if s.underlying_exhausted { - return Poll::Ready(F::EQUILIBRIUM); - } - while s.next_buf.len() < F::CHANNELS { - match S::poll_next(Pin::new(&mut s.stream), cx) { - Poll::Ready(Some(v)) => { - s.next_buf.push(v); - } - Poll::Ready(None) => { - s.underlying_exhausted = true; - return Poll::Ready(F::EQUILIBRIUM); - } - Poll::Pending => return Poll::Pending, - } - } - - let mut data = s.next_buf.iter().cloned(); - let n = F::from_samples(&mut data).unwrap(); - s.next_buf.clear(); - Poll::Ready(n) - } - - fn is_exhausted(&self) -> bool { - self.underlying_exhausted - } -} - -pub struct OpusEncoder { - encoder: opus::Encoder, - frame_size: u32, - sample_rate: u32, - stream: S, - input_buffer: Vec, - exhausted: bool, -} - -impl OpusEncoder -where - S: Stream, - I: ToSample, -{ - pub fn new(frame_size: u32, sample_rate: u32, channels: usize, stream: S) -> Self { - let encoder = opus::Encoder::new( - sample_rate, - match channels { - 1 => Channels::Mono, - 2 => Channels::Stereo, - _ => unimplemented!("Only 1 or 2 channels supported, got {})", channels), - }, - opus::Application::Voip, - ) - .unwrap(); - Self { - encoder, - frame_size, - sample_rate, - stream, - input_buffer: Vec::new(), - exhausted: false, - } - } -} - -impl Stream for OpusEncoder -where - S: Stream + Unpin, - I: Sample + ToSample, -{ - type Item = Vec; - - fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - let s = self.get_mut(); - if s.exhausted { - return Poll::Ready(None); - } - let opus_frame_size = (s.frame_size * s.sample_rate / 400) as usize; - loop { - while s.input_buffer.len() < opus_frame_size { - match S::poll_next(Pin::new(&mut s.stream), cx) { - Poll::Ready(Some(v)) => { - s.input_buffer.push(v.to_sample::()); - } - Poll::Ready(None) => { - s.exhausted = true; - return Poll::Ready(None); - } - Poll::Pending => return Poll::Pending, - } - } - if s.input_buffer.iter().any(|&e| e != 0.0) { - break; - } - s.input_buffer.clear(); - } - - let encoded = s - .encoder - .encode_vec_float(&s.input_buffer, opus_frame_size) - .unwrap(); - s.input_buffer.clear(); - Poll::Ready(Some(encoded)) - } -} -- cgit v1.2.1