use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;
use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
use cpal::{Stream, StreamConfig};
use super::device::find_audio_device;
use crate::constants::{
AUDIO_BUFFER_FRAMES, AUDIO_DEVICE_DEFAULT, AUDIO_SAMPLE_RATE_44K, AUDIO_SAMPLE_RATE_48K,
};
use crate::errors::CameraError;
use crate::timing::PTSClock;
const MAX_BUFFER_FRAMES: usize = AUDIO_BUFFER_FRAMES;
#[derive(Debug, Clone)]
pub struct AudioFrame {
pub samples: Vec<f32>,
pub sample_rate: u32,
pub channels: u16,
pub timestamp: f64,
}
pub struct AudioCapture {
stream: Option<Stream>,
receiver: crossbeam_channel::Receiver<AudioFrame>,
is_running: Arc<AtomicBool>,
sample_rate: u32,
channels: u16,
clock: PTSClock,
}
impl AudioCapture {
pub fn new(
device_id: Option<&str>,
sample_rate: u32,
channels: u16,
clock: PTSClock,
) -> Result<Self, CameraError> {
let device_id_str = device_id.unwrap_or(AUDIO_DEVICE_DEFAULT);
let device_info = find_audio_device(device_id_str)?;
let host = cpal::default_host();
let device = if device_id_str.is_empty() || device_id_str == AUDIO_DEVICE_DEFAULT {
host.default_input_device()
.ok_or_else(|| CameraError::AudioError("No default audio device".to_string()))?
} else {
host.input_devices()
.map_err(|e| CameraError::AudioError(format!("Failed to enumerate devices: {e}")))?
.find(|d| d.name().ok().as_ref() == Some(&device_info.name))
.ok_or_else(|| {
CameraError::AudioError(format!("Device not found: {device_id_str}"))
})?
};
let supported_config = device
.default_input_config()
.map_err(|e| CameraError::AudioError(format!("No supported config: {e}")))?;
let actual_sample_rate =
if sample_rate == AUDIO_SAMPLE_RATE_48K || sample_rate == AUDIO_SAMPLE_RATE_44K {
sample_rate
} else {
supported_config.sample_rate().0
};
let actual_channels = if channels == 1 || channels == 2 {
channels
} else {
supported_config.channels()
};
let config = StreamConfig {
channels: actual_channels,
sample_rate: cpal::SampleRate(actual_sample_rate),
buffer_size: cpal::BufferSize::Default,
};
let (sender, receiver) = crossbeam_channel::bounded(MAX_BUFFER_FRAMES);
let is_running = Arc::new(AtomicBool::new(false));
let is_running_clone = is_running.clone();
let clock_clone = clock.clone();
let config_sample_rate = config.sample_rate.0;
let config_channels = config.channels;
let stream = device
.build_input_stream(
&config,
move |data: &[f32], _: &cpal::InputCallbackInfo| {
if !is_running_clone.load(Ordering::Relaxed) {
return;
}
let frame = AudioFrame {
samples: data.to_vec(),
sample_rate: config_sample_rate,
channels: config_channels,
timestamp: clock_clone.pts(),
};
let _ = sender.try_send(frame);
},
move |err| {
log::error!("Audio capture error: {err}");
},
None,
)
.map_err(|e| CameraError::AudioError(format!("Failed to build stream: {e}")))?;
Ok(Self {
stream: Some(stream),
receiver,
is_running,
sample_rate: config.sample_rate.0,
channels: config.channels,
clock,
})
}
pub fn start(&mut self) -> Result<(), CameraError> {
if self.is_running.load(Ordering::Relaxed) {
return Ok(()); }
if let Some(ref stream) = self.stream {
stream
.play()
.map_err(|e| CameraError::AudioError(format!("Failed to start stream: {e}")))?;
self.is_running.store(true, Ordering::Relaxed);
}
Ok(())
}
pub fn stop(&mut self) -> Result<(), CameraError> {
if !self.is_running.load(Ordering::Relaxed) {
return Ok(()); }
if let Some(ref stream) = self.stream {
stream
.pause()
.map_err(|e| CameraError::AudioError(format!("Failed to stop stream: {e}")))?;
self.is_running.store(false, Ordering::Relaxed);
}
Ok(())
}
pub fn try_read(&self) -> Option<AudioFrame> {
self.receiver.try_recv().ok()
}
pub fn recv_timeout(
&self,
timeout: Duration,
) -> Result<AudioFrame, crossbeam_channel::RecvTimeoutError> {
self.receiver.recv_timeout(timeout)
}
pub fn drain(&self) -> Vec<AudioFrame> {
let mut frames = Vec::new();
while let Ok(frame) = self.receiver.try_recv() {
frames.push(frame);
}
frames
}
pub fn is_running(&self) -> bool {
self.is_running.load(Ordering::Relaxed)
}
pub fn sample_rate(&self) -> u32 {
self.sample_rate
}
pub fn channels(&self) -> u16 {
self.channels
}
pub fn clock(&self) -> &PTSClock {
&self.clock
}
}
impl Drop for AudioCapture {
fn drop(&mut self) {
let _ = self.stop();
self.stream = None;
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_audio_frame_structure() {
let frame = AudioFrame {
samples: vec![0.0, 0.1, 0.2, 0.3],
sample_rate: 48000,
channels: 2,
timestamp: 1.5,
};
assert_eq!(frame.samples.len(), 4);
assert_eq!(frame.sample_rate, 48000);
assert_eq!(frame.channels, 2);
}
#[test]
#[cfg_attr(
target_os = "windows",
ignore = "Opens a real audio device via cpal/WASAPI; COM stream setup can hard-abort (STATUS_ACCESS_VIOLATION) on headless CI runners - run manually"
)]
fn test_start_stop_idempotent() {
let clock = PTSClock::new();
if let Ok(mut capture) = AudioCapture::new(None, 48000, 2, clock) {
assert!(capture.start().is_ok());
assert!(capture.start().is_ok());
assert!(capture.stop().is_ok());
assert!(capture.stop().is_ok());
}
}
}