Текстовый файл из учебного комплекта, не свидетельство установленного runtime. Вернуться к учебнику · Исходный файл
1 //! Streaming device-rate conversion, echo/noise/gain processing, and 20 ms Opus encoding. 2 //! Mute transitions and capture gaps reset history; delayed pre-unmute buffers are discarded. 3 4 use std::collections::VecDeque; 5 use std::time::Duration; 6 use std::time::Instant; 7 8 use rubato::Resampler; 9 use rubato::audioadapter_buffers::direct::InterleavedSlice; 10 use sonora::AudioProcessing; 11 use sonora::StreamConfig; 12 use sonora::config::AdaptiveDigital; 13 use sonora::config::EchoCanceller; 14 use sonora::config::GainController2; 15 use sonora::config::NoiseSuppression; 16 17 use super::buffers::CAPTURE_GAP_TOLERANCE; 18 use super::buffers::Frame; 19 use crate::audio_track::EncodedAudio; 20 21 type Result<T> = std::result::Result<T, &'static str>; 22 const BLOCK: usize = 480; 23 pub(super) const MAX_PROCESSING_DELAY: Duration = Duration::from_millis(/*millis*/ 500); 24 pub(super) const PROCESSING_LATE: &str = "voice processing fell behind"; 25 26 struct Converter { 27 resampler: rubato::Async<f32>, 28 input: VecDeque<f32>, 29 output: VecDeque<f32>, 30 scratch: Vec<f32>, 31 rate: u32, 32 end: Option<Instant>, 33 } 34 35 impl Converter { 36 fn new(rate: u32) -> Result<Self> { 37 let resampler = rubato::Async::new_sinc( 38 48_000.0 / f64::from(rate), 39 /*max_resample_ratio_relative*/ 1.0, 40 &rubato::SincInterpolationParameters::default(), 41 rate.div_ceil(100) as usize, 42 /*nbr_channels*/ 1, 43 rubato::FixedAsync::Input, 44 ) 45 .map_err(|_| "failed to create voice resampler")?; 46 let scratch = vec![0.0; resampler.output_frames_max()]; 47 Ok(Self { 48 resampler, 49 input: VecDeque::new(), 50 output: VecDeque::new(), 51 scratch, 52 rate, 53 end: None, 54 }) 55 } 56 57 fn push(&mut self, frame: &Frame) -> Result<()> { 58 if frame.len > frame.samples.len() 59 || self.input.len() + frame.len > self.rate as usize 60 || self.output.len() > 48_000 61 { 62 return Err("voice resampler backlog exceeded"); 63 } 64 self.end = 65 Some(frame.at + Duration::from_secs_f64(frame.len as f64 / f64::from(self.rate))); 66 self.input.extend(&frame.samples[..frame.len]); 67 while self.input.len() >= self.resampler.input_frames_next() { 68 let samples = self.input.make_contiguous(); 69 let input = InterleavedSlice::new(samples, /*channels*/ 1, samples.len()) 70 .map_err(|_| "invalid voice resampler input")?; 71 let capacity = self.scratch.len(); 72 let mut output = 73 InterleavedSlice::new_mut(&mut self.scratch, /*channels*/ 1, capacity) 74 .map_err(|_| "invalid voice resampler output")?; 75 let (consumed, produced) = self 76 .resampler 77 .process_into_buffer(&input, &mut output, /*indexing*/ None) 78 .map_err(|_| "failed to resample voice audio")?; 79 self.input.drain(..consumed); 80 self.output.extend(&self.scratch[..produced]); 81 } 82 Ok(()) 83 } 84 85 fn next(&mut self) -> Option<(Instant, [f32; BLOCK])> { 86 if self.output.len() < BLOCK { 87 return None; 88 } 89 let delay = self.input.len() as f64 / f64::from(self.rate) 90 + (self.output.len() + self.resampler.output_delay()) as f64 / 48_000.0; 91 let end = self.end?; 92 let at = end 93 .checked_sub(Duration::from_secs_f64(delay)) 94 .unwrap_or(end); 95 let mut output = [0.0; BLOCK]; 96 for (out, sample) in output.iter_mut().zip(self.output.drain(..BLOCK)) { 97 *out = sample; 98 } 99 Some((at, output)) 100 } 101 } 102 103 pub(super) struct Processor { 104 capture: Converter, 105 render: Converter, 106 apm: AudioProcessing, 107 encoder: opus::Encoder, 108 pending: Vec<f32>, 109 at: Instant, 110 cutoff: Instant, 111 render_delay: i64, 112 } 113 114 impl Processor { 115 pub(super) fn new(input_rate: u32, output_rate: u32) -> Result<Self> { 116 Ok(Self { 117 capture: Converter::new(input_rate)?, 118 render: Converter::new(output_rate)?, 119 apm: AudioProcessing::builder() 120 .config(sonora::Config { 121 echo_canceller: Some(EchoCanceller::default()), 122 noise_suppression: Some(NoiseSuppression::default()), 123 gain_controller2: Some(GainController2 { 124 adaptive_digital: Some(AdaptiveDigital::default()), 125 ..Default::default() 126 }), 127 ..Default::default() 128 }) 129 .capture_config(StreamConfig::new( 130 /*sample_rate_hz*/ 48_000, /*num_channels*/ 1, 131 )) 132 .render_config(StreamConfig::new( 133 /*sample_rate_hz*/ 48_000, /*num_channels*/ 1, 134 )) 135 .build(), 136 encoder: opus::Encoder::new( 137 /*sample_rate*/ 48_000, 138 opus::Channels::Mono, 139 opus::Application::Voip, 140 ) 141 .map_err(|_| "failed to create voice encoder")?, 142 pending: Vec::with_capacity(960), 143 at: Instant::now(), 144 cutoff: Instant::now(), 145 render_delay: 0, 146 }) 147 } 148 149 pub(super) fn reset(&mut self) -> Result<()> { 150 *self = Self::new(self.capture.rate, self.render.rate)?; 151 Ok(()) 152 } 153 154 pub(super) fn set_render_rate(&mut self, rate: u32) -> Result<()> { 155 // Capture keeps its own rate and history across speaker-only changes. 156 if self.render.rate != rate { 157 self.render = Converter::new(rate)?; 158 } 159 self.reset_render(); 160 Ok(()) 161 } 162 163 pub(super) fn reset_render(&mut self) { 164 // Speaker changes discard old references without resetting capture or APM state. 165 self.render.resampler.reset(); 166 self.render.input.clear(); 167 self.render.output.clear(); 168 self.render.end = None; 169 self.render_delay = 0; 170 } 171 172 pub(super) fn validate_callback_timing( 173 &self, 174 input_frames: u32, 175 output_frames: u32, 176 ) -> Result<()> { 177 let mut delay = super::super::DEVICE_SERVICE_INTERVAL; 178 for (converter, frames) in [(&self.capture, input_frames), (&self.render, output_frames)] { 179 // Include callback and queue packing latency, plus the resampler's 180 // actual delay and one processing block, for both device directions. 181 delay += Duration::from_secs_f64( 182 (f64::from(frames) + super::buffers::BLOCK as f64) / f64::from(converter.rate) 183 + (converter.resampler.output_delay() + BLOCK) as f64 / 48_000.0, 184 ); 185 } 186 if delay >= MAX_PROCESSING_DELAY { 187 return Err("audio callback timing exceeds processing deadline"); 188 } 189 Ok(()) 190 } 191 192 pub(super) fn render(&mut self, frame: &Frame) -> Result<()> { 193 self.render.push(frame)?; 194 while let Some((at, input)) = self.render.next() { 195 let mut output = [0.0; BLOCK]; 196 self.apm 197 .process_render_f32(&[&input], &mut [&mut output]) 198 .map_err(|_| "voice echo reference failed")?; 199 let now = Instant::now(); 200 self.render_delay = if at > now { 201 at.duration_since(now).as_millis() as i64 202 } else { 203 -(now.duration_since(at).as_millis() as i64) 204 }; 205 } 206 Ok(()) 207 } 208 209 pub(super) fn capture( 210 &mut self, 211 frame: &Frame, 212 now: impl Fn() -> Instant, 213 ) -> Result<Vec<EncodedAudio>> { 214 if frame.at < self.cutoff { 215 return Ok(vec![]); 216 } 217 if self 218 .capture 219 .end 220 .is_some_and(|end| frame.at.saturating_duration_since(end) > CAPTURE_GAP_TOLERANCE) 221 { 222 // Drop resampler, APM and half-packet history without advancing the 223 // unmute boundary: this frame may already have waited in the queue. 224 let cutoff = self.cutoff; 225 self.reset()?; 226 self.cutoff = cutoff; 227 } 228 self.capture.push(frame)?; 229 let mut encoded = Vec::new(); 230 while let Some((at, input)) = self.capture.next() { 231 let capture_age = now().saturating_duration_since(at); 232 if capture_age > MAX_PROCESSING_DELAY { 233 return Err(PROCESSING_LATE); 234 } 235 let delay = (capture_age.as_millis() as i64 + self.render_delay).max(0); 236 if delay > MAX_PROCESSING_DELAY.as_millis() as i64 { 237 return Err(PROCESSING_LATE); 238 } 239 self.apm 240 .set_stream_delay_ms(delay as i32) 241 .map_err(|_| "invalid voice echo delay")?; 242 let mut output = [0.0; BLOCK]; 243 self.apm 244 .process_capture_f32(&[&input], &mut [&mut output]) 245 .map_err(|_| "voice capture processing failed")?; 246 if output.iter().any(|sample| !sample.is_finite()) { 247 return Err("invalid processed voice audio"); 248 } 249 if self.pending.is_empty() { 250 self.at = at; 251 } 252 self.pending.extend(output); 253 if self.pending.len() == 960 { 254 let mut data = vec![0; 1275]; 255 let len = self 256 .encoder 257 .encode_float(&self.pending, &mut data) 258 .map_err(|_| "failed to encode voice audio")?; 259 data.truncate(len); 260 encoded.push(EncodedAudio { data, at: self.at }); 261 self.pending.clear(); 262 } 263 } 264 Ok(encoded) 265 } 266 } 267 268 #[cfg(test)] 269 #[path = "processing_tests.rs"] 270 mod tests;