use crate::contracts::{AudioProcessor, AudioSink, MetricsCollector};
use crate::core::{
AudioError, AudioFrame, AudioProfile, AudioResult, AudioSpec, CaptureDiagnostics,
};
use crate::metrics::LevelMetrics;
use crate::mic_sim::contracts::MicrophoneSimProcessor;
use crate::runtime_stats::RuntimeStats;
use crate::types::{AudioBlock, ProcessContext};
use hound::{SampleFormat, WavSpec, WavWriter};
use std::collections::VecDeque;
use std::fs::File;
use std::io::BufWriter;
use std::path::Path;
use std::sync::{Arc, Mutex};
pub struct AudioPipeline {
spec: AudioSpec,
diagnostics: CaptureDiagnostics,
microphone_sim_processors: Vec<Box<dyn MicrophoneSimProcessor>>,
processors: Vec<Box<dyn AudioProcessor>>,
sink: Box<dyn AudioSink>,
metrics: Box<dyn MetricsCollector>,
runtime_stats: RuntimeStats,
frame_size: usize,
input_buffer: Vec<f32>,
frame_index: u64,
}
impl AudioPipeline {
pub fn new(
spec: AudioSpec,
diagnostics: CaptureDiagnostics,
mut microphone_sim_processors: Vec<Box<dyn MicrophoneSimProcessor>>,
mut processors: Vec<Box<dyn AudioProcessor>>,
sink: Box<dyn AudioSink>,
metrics: Box<dyn MetricsCollector>,
) -> AudioResult<Self> {
let frame_size = (spec.sample_rate as usize / 100)
.max(160)
.next_power_of_two();
let prepare_ctx = ProcessContext {
spec,
frame_size,
frame_index: 0,
};
for processor in &mut microphone_sim_processors {
let _ = processor.prepare_with_context(&prepare_ctx)?;
}
for processor in &mut processors {
let _ = processor.prepare_with_context(&prepare_ctx)?;
}
let runtime_stats = RuntimeStats::new(AudioProfile::from_str(&diagnostics.profile));
Ok(Self {
spec,
diagnostics,
microphone_sim_processors,
processors,
sink,
metrics,
runtime_stats,
frame_size,
input_buffer: Vec::with_capacity(frame_size * 2),
frame_index: 0,
})
}
pub fn process_input(&mut self, input: &[f32], input_channels: u16) -> AudioResult<()> {
let mono = downmix_to_mono(input, input_channels);
self.input_buffer.extend_from_slice(&mono);
while self.input_buffer.len() >= self.frame_size {
let samples = self
.input_buffer
.drain(..self.frame_size)
.collect::<Vec<_>>();
self.process_frame(samples)?;
}
Ok(())
}
pub fn finalize(&mut self, duration_ms: Option<i64>) -> AudioResult<CaptureDiagnostics> {
if !self.input_buffer.is_empty() {
let mut samples = std::mem::take(&mut self.input_buffer);
samples.resize(self.frame_size, 0.0);
self.process_frame(samples)?;
}
self.sink.finalize()?;
let mut diagnostics = self.diagnostics.clone();
diagnostics.duration_ms = duration_ms;
self.metrics.apply(&mut diagnostics);
self.runtime_stats.apply(&mut diagnostics);
for processor in &self.processors {
diagnostics.notes.extend(processor.diagnostics_notes());
}
Ok(diagnostics)
}
pub fn sample_rate(&self) -> u32 {
self.spec.sample_rate
}
pub fn input_channels(&self) -> u16 {
self.diagnostics.channels.unwrap_or(1)
}
pub fn device_name(&self) -> Option<&str> {
self.diagnostics.device_name.as_deref()
}
fn process_frame(&mut self, samples: Vec<f32>) -> AudioResult<()> {
let mut block = AudioBlock::new(samples, self.spec);
let ctx = ProcessContext {
spec: self.spec,
frame_size: self.frame_size,
frame_index: self.frame_index,
};
for processor in &mut self.microphone_sim_processors {
processor.process_block(&mut block, &ctx)?;
}
for processor in &mut self.processors {
processor.process_block(&mut block, &ctx)?;
}
self.frame_index += 1;
let frame = AudioFrame::from(block);
self.metrics.observe(&frame);
self.runtime_stats.observe(&frame);
self.sink.write_frame(&frame)
}
}
fn downmix_to_mono(input: &[f32], input_channels: u16) -> Vec<f32> {
if input_channels <= 1 {
return input.to_vec();
}
let channels = input_channels as usize;
let mut mono = Vec::with_capacity(input.len() / channels);
for frame in input.chunks(channels) {
let sum: f32 = frame.iter().copied().sum();
mono.push(sum / channels as f32);
}
mono
}
pub struct WavSink {
writer: Option<WavWriter<BufWriter<File>>>,
}
impl WavSink {
pub fn create(path: &Path, spec: AudioSpec) -> AudioResult<Self> {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)
.map_err(|err| AudioError::new(format!("Failed to create audio dir: {}", err)))?;
}
let writer = WavWriter::create(
path,
WavSpec {
channels: spec.channels,
sample_rate: spec.sample_rate,
bits_per_sample: 16,
sample_format: SampleFormat::Int,
},
)
.map_err(|err| AudioError::new(format!("Failed to create WAV file: {}", err)))?;
Ok(Self {
writer: Some(writer),
})
}
}
#[derive(Debug)]
pub struct PlaybackBuffer {
samples: VecDeque<f32>,
max_samples: usize,
}
impl PlaybackBuffer {
pub fn new(max_samples: usize) -> Self {
Self {
samples: VecDeque::with_capacity(max_samples),
max_samples,
}
}
pub fn push_mono(&mut self, input: &[f32]) {
if input.len() >= self.max_samples {
self.samples.clear();
self.samples.extend(
input[input.len().saturating_sub(self.max_samples)..]
.iter()
.copied(),
);
return;
}
let overflow = self
.samples
.len()
.saturating_add(input.len())
.saturating_sub(self.max_samples);
if overflow > 0 {
self.samples.drain(..overflow);
}
self.samples.extend(input.iter().copied());
}
pub fn pop_mono_sample(&mut self) -> Option<f32> {
self.samples.pop_front()
}
pub fn clear(&mut self) {
self.samples.clear();
}
}
pub struct PlaybackSink {
buffer: Arc<Mutex<PlaybackBuffer>>,
}
impl PlaybackSink {
pub fn with_capacity(max_samples: usize) -> (Self, Arc<Mutex<PlaybackBuffer>>) {
let buffer = Arc::new(Mutex::new(PlaybackBuffer::new(max_samples)));
(
Self {
buffer: Arc::clone(&buffer),
},
buffer,
)
}
pub fn from_buffer(buffer: Arc<Mutex<PlaybackBuffer>>) -> Self {
Self { buffer }
}
}
impl AudioSink for WavSink {
fn write_frame(&mut self, frame: &AudioFrame) -> AudioResult<()> {
let writer = self
.writer
.as_mut()
.ok_or_else(|| AudioError::new("WAV writer already finalized"))?;
for sample in &frame.samples {
let scaled = (sample.clamp(-1.0, 1.0) * i16::MAX as f32) as i16;
writer
.write_sample(scaled)
.map_err(|err| AudioError::new(format!("Failed to write WAV sample: {}", err)))?;
}
Ok(())
}
fn finalize(&mut self) -> AudioResult<()> {
if let Some(writer) = self.writer.take() {
writer
.finalize()
.map_err(|err| AudioError::new(format!("Failed to finalize WAV file: {}", err)))?;
}
Ok(())
}
}
impl AudioSink for PlaybackSink {
fn write_frame(&mut self, frame: &AudioFrame) -> AudioResult<()> {
let mut buffer = self
.buffer
.lock()
.map_err(|_| AudioError::new("Playback buffer lock poisoned"))?;
buffer.push_mono(&frame.samples);
Ok(())
}
fn finalize(&mut self) -> AudioResult<()> {
Ok(())
}
}
impl MetricsCollector for LevelMetrics {
fn observe(&mut self, frame: &AudioFrame) {
LevelMetrics::observe(self, frame);
}
fn apply(&self, diagnostics: &mut CaptureDiagnostics) {
LevelMetrics::apply(self, diagnostics);
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::contracts::{AudioProcessor, AudioSink, MetricsCollector};
use crate::mic_sim::contracts::MicrophoneSimProcessor;
use crate::types::{AudioBlock, ProcessContext, ProcessorInfo};
use std::sync::{Arc, Mutex};
struct PassthroughProcessor;
impl AudioProcessor for PassthroughProcessor {
fn name(&self) -> &'static str {
"passthrough"
}
fn prepare(&mut self, _spec: AudioSpec) -> AudioResult<()> {
Ok(())
}
fn process(&mut self, _frame: &mut AudioFrame) -> AudioResult<()> {
Ok(())
}
}
struct DiagnosticNoteProcessor;
impl AudioProcessor for DiagnosticNoteProcessor {
fn name(&self) -> &'static str {
"diagnostic_note"
}
fn prepare(&mut self, _spec: AudioSpec) -> AudioResult<()> {
Ok(())
}
fn process(&mut self, _frame: &mut AudioFrame) -> AudioResult<()> {
Ok(())
}
fn diagnostics_notes(&self) -> Vec<String> {
vec!["startup=ok".to_string()]
}
}
struct ContextTrackingProcessor {
seen_frames: Arc<Mutex<Vec<u64>>>,
}
impl AudioProcessor for ContextTrackingProcessor {
fn name(&self) -> &'static str {
"context_tracking"
}
fn prepare(&mut self, _spec: AudioSpec) -> AudioResult<()> {
Ok(())
}
fn process(&mut self, _frame: &mut AudioFrame) -> AudioResult<()> {
Ok(())
}
fn prepare_with_context(&mut self, ctx: &ProcessContext) -> AudioResult<ProcessorInfo> {
self.prepare(ctx.spec)?;
Ok(ProcessorInfo::default())
}
fn process_block(
&mut self,
_block: &mut AudioBlock,
ctx: &ProcessContext,
) -> AudioResult<()> {
self.seen_frames.lock().unwrap().push(ctx.frame_index);
Ok(())
}
}
struct ContextTrackingMicSimProcessor {
seen_frames: Arc<Mutex<Vec<u64>>>,
}
impl MicrophoneSimProcessor for ContextTrackingMicSimProcessor {
fn name(&self) -> &'static str {
"mic_sim_context_tracking"
}
fn prepare(&mut self, _spec: AudioSpec) -> AudioResult<()> {
Ok(())
}
fn process(&mut self, _frame: &mut AudioFrame) -> AudioResult<()> {
Ok(())
}
fn prepare_with_context(&mut self, ctx: &ProcessContext) -> AudioResult<ProcessorInfo> {
self.prepare(ctx.spec)?;
Ok(ProcessorInfo::default())
}
fn process_block(
&mut self,
_block: &mut AudioBlock,
ctx: &ProcessContext,
) -> AudioResult<()> {
self.seen_frames.lock().unwrap().push(ctx.frame_index);
Ok(())
}
}
#[derive(Default)]
struct CountingSink {
frames: usize,
samples: usize,
}
impl AudioSink for CountingSink {
fn write_frame(&mut self, frame: &AudioFrame) -> AudioResult<()> {
self.frames += 1;
self.samples += frame.samples.len();
Ok(())
}
fn finalize(&mut self) -> AudioResult<()> {
Ok(())
}
}
#[derive(Default)]
struct NoopMetrics;
impl MetricsCollector for NoopMetrics {
fn observe(&mut self, _frame: &AudioFrame) {}
fn apply(&self, _diagnostics: &mut CaptureDiagnostics) {}
}
#[test]
fn pipeline_buffers_variable_input_into_fixed_frames() {
let spec = AudioSpec {
sample_rate: 44_100,
channels: 1,
};
let sink = Box::new(CountingSink::default());
let mut pipeline = AudioPipeline::new(
spec,
CaptureDiagnostics::default(),
Vec::new(),
vec![Box::new(PassthroughProcessor)],
sink,
Box::new(NoopMetrics),
)
.unwrap();
pipeline.process_input(&vec![0.1; 100], 1).unwrap();
pipeline.process_input(&vec![0.1; 200], 1).unwrap();
assert_eq!(pipeline.input_buffer.len(), 300);
pipeline.process_input(&vec![0.1; 200], 1).unwrap();
assert_eq!(pipeline.input_buffer.len(), 500);
pipeline.process_input(&vec![0.1; 20], 1).unwrap();
assert_eq!(pipeline.input_buffer.len(), 8);
pipeline.finalize(Some(10)).unwrap();
}
#[test]
fn pipeline_passes_frame_context_to_processors() {
let spec = AudioSpec {
sample_rate: 44_100,
channels: 1,
};
let seen_frames = Arc::new(Mutex::new(Vec::new()));
let mut pipeline = AudioPipeline::new(
spec,
CaptureDiagnostics::default(),
Vec::new(),
vec![Box::new(ContextTrackingProcessor {
seen_frames: Arc::clone(&seen_frames),
})],
Box::new(CountingSink::default()),
Box::new(NoopMetrics),
)
.unwrap();
pipeline
.process_input(&vec![0.1; pipeline.frame_size * 2], 1)
.unwrap();
pipeline.finalize(Some(10)).unwrap();
assert_eq!(&*seen_frames.lock().unwrap(), &[0, 1]);
}
#[test]
fn pipeline_runs_microphone_sim_before_voice_processors() {
let spec = AudioSpec {
sample_rate: 44_100,
channels: 1,
};
let mic_sim_seen_frames = Arc::new(Mutex::new(Vec::new()));
let voice_seen_frames = Arc::new(Mutex::new(Vec::new()));
let mut pipeline = AudioPipeline::new(
spec,
CaptureDiagnostics::default(),
vec![Box::new(ContextTrackingMicSimProcessor {
seen_frames: Arc::clone(&mic_sim_seen_frames),
})],
vec![Box::new(ContextTrackingProcessor {
seen_frames: Arc::clone(&voice_seen_frames),
})],
Box::new(CountingSink::default()),
Box::new(NoopMetrics),
)
.unwrap();
pipeline
.process_input(&vec![0.1; pipeline.frame_size * 2], 1)
.unwrap();
pipeline.finalize(Some(10)).unwrap();
assert_eq!(&*mic_sim_seen_frames.lock().unwrap(), &[0, 1]);
assert_eq!(&*voice_seen_frames.lock().unwrap(), &[0, 1]);
}
#[test]
fn pipeline_collects_processor_diagnostic_notes() {
let spec = AudioSpec {
sample_rate: 44_100,
channels: 1,
};
let mut pipeline = AudioPipeline::new(
spec,
CaptureDiagnostics::default(),
Vec::new(),
vec![Box::new(DiagnosticNoteProcessor)],
Box::new(CountingSink::default()),
Box::new(NoopMetrics),
)
.unwrap();
pipeline
.process_input(&vec![0.1; pipeline.frame_size], 1)
.unwrap();
let diagnostics = pipeline.finalize(Some(10)).unwrap();
assert!(diagnostics.notes.iter().any(|note| note == "startup=ok"));
}
#[test]
fn playback_buffer_keeps_most_recent_samples_when_over_capacity() {
let mut buffer = PlaybackBuffer::new(4);
buffer.push_mono(&[0.1, 0.2, 0.3]);
buffer.push_mono(&[0.4, 0.5, 0.6]);
assert_eq!(buffer.pop_mono_sample(), Some(0.3));
assert_eq!(buffer.pop_mono_sample(), Some(0.4));
assert_eq!(buffer.pop_mono_sample(), Some(0.5));
assert_eq!(buffer.pop_mono_sample(), Some(0.6));
assert_eq!(buffer.pop_mono_sample(), None);
}
}