use crate::builder::{NativePipelineBuilder, PipelineBuildRequest};
use crate::contracts::{CaptureBackend, DeviceEnumerator, ProcessorBuildRequest, ProcessorFactory};
use crate::core::{
ActiveListening, ActiveRecording, AudioError, AudioResult, CaptureDiagnostics, InputDeviceInfo,
NativeCaptureConfig,
};
use crate::factory::ProfileProcessorFactory;
use crate::metrics::LevelMetrics;
use crate::mic_sim::MicrophoneSimulatorFactory;
use crate::pipeline::{AudioPipeline, PlaybackBuffer, PlaybackSink};
use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
use cpal::{SampleFormat, StreamConfig};
use serde_json::Value;
use std::process::Command;
use std::sync::mpsc;
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};
use std::time::Instant;
pub struct NativeCaptureBackend;
impl DeviceEnumerator for NativeCaptureBackend {
fn list_input_devices(&self) -> AudioResult<Vec<InputDeviceInfo>> {
list_input_devices()
}
fn set_default_input_device(&self, device_id: &str) -> AudioResult<()> {
set_default_input_device(device_id)
}
}
pub fn list_input_devices() -> AudioResult<Vec<InputDeviceInfo>> {
#[cfg(target_os = "linux")]
if let Ok(devices) = list_pipewire_input_devices() {
if !devices.is_empty() {
return Ok(devices);
}
}
list_cpal_input_devices()
}
pub fn set_default_input_device(device_id: &str) -> AudioResult<()> {
#[cfg(target_os = "linux")]
{
let output = Command::new("pactl")
.args(["set-default-source", device_id])
.output()
.map_err(|err| {
AudioError::new(format!(
"Failed to invoke pactl set-default-source for '{}': {}",
device_id, err
))
})?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
return Err(AudioError::new(format!(
"pactl set-default-source '{}' failed: {}",
device_id,
stderr.trim()
)));
}
return Ok(());
}
#[allow(unreachable_code)]
Err(AudioError::new(
"Setting the default input device is only supported on Linux",
))
}
fn list_cpal_input_devices() -> AudioResult<Vec<InputDeviceInfo>> {
let host = cpal::default_host();
let default_name = host
.default_input_device()
.and_then(|device| device.description().ok())
.map(|d| d.to_string());
let devices = host
.input_devices()
.map_err(|err| AudioError::new(format!("Failed to enumerate input devices: {}", err)))?;
let mut result = Vec::new();
for device in devices {
let name = device
.description()
.map_err(|err| AudioError::new(format!("Failed to read input device name: {}", err)))?
.to_string();
let is_default = default_name.as_deref() == Some(name.as_str());
result.push(InputDeviceInfo {
id: name.clone(),
name,
is_default,
});
}
result.sort_by(|a, b| a.name.cmp(&b.name));
Ok(result)
}
#[cfg(target_os = "linux")]
fn list_pipewire_input_devices() -> AudioResult<Vec<InputDeviceInfo>> {
let default_source = pactl_default_source().ok().flatten();
let output = Command::new("pactl")
.args(["-f", "json", "list", "sources"])
.output()
.map_err(|err| AudioError::new(format!("Failed to invoke pactl list sources: {}", err)))?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
return Err(AudioError::new(format!(
"pactl list sources failed: {}",
stderr.trim()
)));
}
let value: Value = serde_json::from_slice(&output.stdout)
.map_err(|err| AudioError::new(format!("Failed to parse pactl source list: {}", err)))?;
let sources = value
.as_array()
.ok_or_else(|| AudioError::new("Unexpected pactl source list payload"))?;
let mut result = Vec::new();
for source in sources {
if !is_selectable_pipewire_source(source) {
continue;
}
let Some(id) = source.get("name").and_then(Value::as_str) else {
continue;
};
let display_name = friendly_pipewire_source_name(source).unwrap_or_else(|| id.to_string());
let is_default = default_source.as_deref() == Some(id);
result.push(InputDeviceInfo {
id: id.to_string(),
name: display_name,
is_default,
});
}
result.sort_by(|a, b| a.name.cmp(&b.name));
Ok(result)
}
#[cfg(target_os = "linux")]
fn pactl_default_source() -> AudioResult<Option<String>> {
let output = Command::new("pactl")
.arg("info")
.output()
.map_err(|err| AudioError::new(format!("Failed to invoke pactl info: {}", err)))?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
return Err(AudioError::new(format!(
"pactl info failed: {}",
stderr.trim()
)));
}
let stdout = String::from_utf8_lossy(&output.stdout);
Ok(stdout
.lines()
.find_map(|line| line.strip_prefix("Default Source: "))
.map(|value| value.trim().to_string()))
}
#[cfg(target_os = "linux")]
fn is_selectable_pipewire_source(source: &Value) -> bool {
let Some(id) = source.get("name").and_then(Value::as_str) else {
return false;
};
if id.ends_with(".monitor") {
return false;
}
if source
.get("monitor_source")
.and_then(Value::as_str)
.is_some_and(|value| !value.is_empty())
{
return false;
}
let properties = source.get("properties").and_then(Value::as_object);
if properties
.and_then(|props| props.get("media.class"))
.and_then(Value::as_str)
.is_some_and(|class| class != "Audio/Source")
{
return false;
}
let description = source
.get("description")
.and_then(Value::as_str)
.unwrap_or_default();
!description.starts_with("Monitor of ")
}
#[cfg(target_os = "linux")]
fn friendly_pipewire_source_name(source: &Value) -> Option<String> {
let properties = source.get("properties").and_then(Value::as_object);
let device_description = properties
.and_then(|props| props.get("device.description"))
.and_then(Value::as_str)
.or_else(|| source.get("description").and_then(Value::as_str))?;
let active_port_name = source.get("active_port").and_then(Value::as_str);
let active_port_description = source
.get("ports")
.and_then(Value::as_array)
.and_then(|ports| {
ports.iter().find_map(|port| {
let name = port.get("name").and_then(Value::as_str)?;
if Some(name) == active_port_name {
port.get("description").and_then(Value::as_str)
} else {
None
}
})
});
if let Some(port_description) = active_port_description {
return Some(format!("{} - {}", port_description, device_description));
}
Some(device_description.to_string())
}
impl NativeCaptureBackend {
pub fn new() -> Self {
Self
}
fn build_pipeline(
config: &NativeCaptureConfig,
sample_rate: u32,
device_channels: u16,
device_name: Option<String>,
) -> AudioResult<AudioPipeline> {
NativePipelineBuilder::new().build(PipelineBuildRequest {
output_path: config.output_path.clone(),
profile: config.profile,
sample_rate,
device_channels,
device_name,
gain_db: config.input_gain_db,
limiter_threshold: config.limiter_threshold,
high_pass_hz: config.high_pass_hz,
noise_suppression_amount: config.noise_suppression_amount,
noise_calibration_ms: config.noise_calibration_ms,
delay_effect: config.delay_effect,
stage_overrides: config.stage_overrides.clone(),
microphone_sim: config.microphone_sim,
})
}
fn build_live_pipeline(
config: &NativeCaptureConfig,
sample_rate: u32,
device_channels: u16,
device_name: Option<String>,
playback_buffer: Arc<Mutex<PlaybackBuffer>>,
) -> AudioResult<AudioPipeline> {
let output_spec = crate::core::AudioSpec {
sample_rate,
channels: 1,
};
let processors =
ProfileProcessorFactory::new().build_processors(ProcessorBuildRequest {
profile: config.profile,
gain_db: config.input_gain_db,
limiter_threshold: config.limiter_threshold,
high_pass_hz: config.high_pass_hz,
noise_suppression_amount: config.noise_suppression_amount,
noise_calibration_ms: config.noise_calibration_ms,
delay_effect: config.delay_effect,
stage_overrides: config.stage_overrides.clone(),
})?;
let processor_names = processors
.iter()
.map(|processor| processor.name().to_string())
.collect::<Vec<_>>();
let microphone_sim_processors =
MicrophoneSimulatorFactory::new().build_processors(config.microphone_sim)?;
let microphone_sim_processor_names = microphone_sim_processors
.iter()
.map(|processor| processor.name().to_string())
.collect::<Vec<_>>();
let microphone_sim_model = config
.microphone_sim
.active_model()
.map(|model| model.as_str().to_string());
let mut notes = config
.delay_effect
.map(|effect| vec![format!("delay_effect={}", effect.preset.as_str())])
.unwrap_or_default();
notes.push("mode=live_listening".to_string());
notes.push("output=default".to_string());
notes.push("pipeline_order=microphone_sim->voice_processing->playback".to_string());
if let Some(model) = µphone_sim_model {
notes.push(format!("microphone_sim={}", model));
}
AudioPipeline::new(
output_spec,
CaptureDiagnostics {
backend: "native_listening".to_string(),
profile: config.profile.as_str().to_string(),
profile_base: Some(config.profile.as_str().to_string()),
device_name,
sample_rate: Some(sample_rate),
channels: Some(device_channels),
processor_names,
processor_stage_overrides: config
.stage_overrides
.iter()
.map(|(stage, mode)| format!("{}={}", stage, mode.as_str()))
.collect(),
resolved_delay_preset: config
.delay_effect
.map(|effect| effect.preset.as_str().to_string()),
microphone_sim_model,
microphone_sim_processor_names,
notes,
..CaptureDiagnostics::default()
},
microphone_sim_processors,
processors,
Box::new(PlaybackSink::from_buffer(playback_buffer)),
Box::new(LevelMetrics::default()),
)
}
pub fn start_live(&self, config: NativeCaptureConfig) -> AudioResult<Box<dyn ActiveListening>> {
let (ready_tx, ready_rx) = mpsc::channel();
let (cmd_tx, cmd_rx) = mpsc::channel();
let handle = thread::spawn(move || {
native_listening_thread(config, cmd_rx, ready_tx);
});
match ready_rx
.recv()
.map_err(|_| AudioError::new("Native listening thread failed to report readiness"))?
{
Ok(()) => Ok(Box::new(NativeListeningHandle {
cmd_tx,
join_handle: Some(handle),
})),
Err(err) => {
let _ = handle.join();
Err(err)
}
}
}
}
impl CaptureBackend for NativeCaptureBackend {
fn start(&self, config: NativeCaptureConfig) -> AudioResult<Box<dyn ActiveRecording>> {
let (ready_tx, ready_rx) = mpsc::channel();
let (cmd_tx, cmd_rx) = mpsc::channel();
let handle = thread::spawn(move || {
native_capture_thread(config, cmd_rx, ready_tx);
});
match ready_rx
.recv()
.map_err(|_| AudioError::new("Native capture thread failed to report readiness"))?
{
Ok(()) => Ok(Box::new(NativeRecordingHandle {
cmd_tx,
join_handle: Some(handle),
})),
Err(err) => {
let _ = handle.join();
Err(err)
}
}
}
}
enum NativeCommand {
Stop(mpsc::Sender<AudioResult<CaptureDiagnostics>>),
}
enum NativeListeningCommand {
Stop(mpsc::Sender<AudioResult<CaptureDiagnostics>>),
UpdateConfig(NativeCaptureConfig, mpsc::Sender<AudioResult<()>>),
}
fn native_capture_thread(
config: NativeCaptureConfig,
cmd_rx: mpsc::Receiver<NativeCommand>,
ready_tx: mpsc::Sender<AudioResult<()>>,
) {
let result = start_native_stream(config);
let (stream, pipeline, start_time, notes) = match result {
Ok(values) => {
let _ = ready_tx.send(Ok(()));
values
}
Err(err) => {
let _ = ready_tx.send(Err(err));
return;
}
};
let _keep_stream_alive = stream;
match cmd_rx.recv() {
Ok(NativeCommand::Stop(reply_tx)) => {
drop(_keep_stream_alive);
let duration_ms = Some(start_time.elapsed().as_millis() as i64);
let diagnostics = pipeline
.lock()
.map_err(|_| AudioError::new("Audio pipeline lock poisoned"))
.and_then(|mut pipeline| pipeline.finalize(duration_ms))
.map(|mut diagnostics| {
diagnostics.notes.extend(notes);
diagnostics
});
let _ = reply_tx.send(diagnostics);
}
Err(_) => {}
}
}
fn native_listening_thread(
config: NativeCaptureConfig,
cmd_rx: mpsc::Receiver<NativeListeningCommand>,
ready_tx: mpsc::Sender<AudioResult<()>>,
) {
let result = start_live_monitor_stream(config);
let (input_stream, output_stream, pipeline, playback_buffer, start_time, notes) = match result {
Ok(values) => {
let _ = ready_tx.send(Ok(()));
values
}
Err(err) => {
let _ = ready_tx.send(Err(err));
return;
}
};
let _keep_input_stream_alive = input_stream;
let _keep_output_stream_alive = output_stream;
loop {
match cmd_rx.recv() {
Ok(NativeListeningCommand::Stop(reply_tx)) => {
drop(_keep_input_stream_alive);
drop(_keep_output_stream_alive);
let duration_ms = Some(start_time.elapsed().as_millis() as i64);
let diagnostics = pipeline
.lock()
.map_err(|_| AudioError::new("Audio pipeline lock poisoned"))
.and_then(|mut pipeline| pipeline.finalize(duration_ms))
.map(|mut diagnostics| {
diagnostics.notes.extend(notes.clone());
diagnostics
});
let _ = reply_tx.send(diagnostics);
break;
}
Ok(NativeListeningCommand::UpdateConfig(config, reply_tx)) => {
let result = rebuild_live_pipeline(&pipeline, &playback_buffer, config);
let _ = reply_tx.send(result);
}
Err(_) => break,
}
}
}
type NativeThreadState = (
cpal::Stream,
Arc<Mutex<AudioPipeline>>,
Instant,
Vec<String>,
);
type NativeListeningThreadState = (
cpal::Stream,
cpal::Stream,
Arc<Mutex<AudioPipeline>>,
Arc<Mutex<PlaybackBuffer>>,
Instant,
Vec<String>,
);
fn start_native_stream(config: NativeCaptureConfig) -> AudioResult<NativeThreadState> {
let host = cpal::default_host();
let device = select_input_device(&host, config.preferred_input_device.as_deref())?;
let device_name = device.description().ok().map(|d| d.to_string());
let supported_config = device
.default_input_config()
.map_err(|err| AudioError::new(format!("Failed to read input device config: {}", err)))?;
let (sample_rate, channels) =
select_stream_config(&device, config.target_sample_rate, config.target_channels).unwrap_or(
(
supported_config.sample_rate(),
supported_config.channels(),
),
);
let pipeline = Arc::new(Mutex::new(NativeCaptureBackend::build_pipeline(
&config,
sample_rate,
channels,
device_name.clone(),
)?));
let cpal_config = StreamConfig {
channels,
sample_rate: sample_rate,
buffer_size: cpal::BufferSize::Default,
};
let stream = build_input_stream(
&device,
supported_config.sample_format(),
&cpal_config,
channels,
Arc::clone(&pipeline),
native_input_stream_error,
)?;
stream
.play()
.map_err(|err| AudioError::new(format!("Failed to start input stream: {}", err)))?;
Ok((
stream,
pipeline,
Instant::now(),
vec![
format!("device={}", device_name.unwrap_or_else(|| "unknown".into())),
format!(
"configured_device={}",
config
.preferred_input_device
.unwrap_or_else(|| "system_default".to_string())
),
format!("channels={}", channels),
format!("sample_rate={}", sample_rate),
format!("noise_calibration_ms={}", config.noise_calibration_ms),
],
))
}
fn native_input_stream_error(err: cpal::StreamError) {
log::error!("Native audio stream error: {}", err);
}
fn native_listening_input_stream_error(err: cpal::StreamError) {
log::error!("Native listening input stream error: {}", err);
}
fn native_listening_output_stream_error(err: cpal::StreamError) {
log::error!("Native listening output stream error: {}", err);
}
fn start_live_monitor_stream(
config: NativeCaptureConfig,
) -> AudioResult<NativeListeningThreadState> {
let host = cpal::default_host();
let input_device = select_input_device(&host, config.preferred_input_device.as_deref())?;
let input_device_name = input_device.description().ok().map(|d| d.to_string());
let input_config = input_device
.default_input_config()
.map_err(|err| AudioError::new(format!("Failed to read input device config: {}", err)))?;
let output_device = host
.default_output_device()
.ok_or_else(|| AudioError::new("No output device available"))?;
let output_device_name = output_device.description().ok().map(|d| d.to_string());
let output_config = output_device
.default_output_config()
.map_err(|err| AudioError::new(format!("Failed to read output device config: {}", err)))?;
let output_sample_rate = output_config.sample_rate();
let output_channels = output_config.channels().max(1);
let input_channels = select_stream_config(
&input_device,
output_sample_rate,
config.target_channels.max(1),
)
.map(|(_, channels)| channels)
.ok_or_else(|| {
AudioError::new(format!(
"Input device does not support {} Hz for live listening",
output_sample_rate
))
})?;
let cpal_input_config = StreamConfig {
channels: input_channels,
sample_rate: output_sample_rate,
buffer_size: cpal::BufferSize::Default,
};
let cpal_output_config = StreamConfig {
channels: output_channels,
sample_rate: output_sample_rate,
buffer_size: cpal::BufferSize::Default,
};
let max_buffer_samples = (output_sample_rate as usize * 2).max(4096);
let (_, playback_buffer) = PlaybackSink::with_capacity(max_buffer_samples);
let pipeline = Arc::new(Mutex::new(NativeCaptureBackend::build_live_pipeline(
&config,
output_sample_rate,
input_channels,
input_device_name.clone(),
Arc::clone(&playback_buffer),
)?));
let input_stream = build_input_stream(
&input_device,
input_config.sample_format(),
&cpal_input_config,
input_channels,
Arc::clone(&pipeline),
native_listening_input_stream_error,
)?;
let output_stream = build_output_stream(
&output_device,
output_config.sample_format(),
&cpal_output_config,
Arc::clone(&playback_buffer),
native_listening_output_stream_error,
)?;
input_stream
.play()
.map_err(|err| AudioError::new(format!("Failed to start input stream: {}", err)))?;
output_stream
.play()
.map_err(|err| AudioError::new(format!("Failed to start output stream: {}", err)))?;
Ok((
input_stream,
output_stream,
pipeline,
playback_buffer,
Instant::now(),
vec![
format!(
"input_device={}",
input_device_name.unwrap_or_else(|| "unknown".into())
),
format!(
"output_device={}",
output_device_name.unwrap_or_else(|| "unknown".into())
),
format!("input_channels={}", input_channels),
format!("output_channels={}", output_channels),
format!("sample_rate={}", output_sample_rate),
],
))
}
fn rebuild_live_pipeline(
pipeline: &Arc<Mutex<AudioPipeline>>,
playback_buffer: &Arc<Mutex<PlaybackBuffer>>,
config: NativeCaptureConfig,
) -> AudioResult<()> {
if let Ok(mut buffer) = playback_buffer.lock() {
buffer.clear();
}
let (pipeline_sample_rate, pipeline_channels, device_name) = {
let current = pipeline
.lock()
.map_err(|_| AudioError::new("Audio pipeline lock poisoned"))?;
(
current.sample_rate(),
current.input_channels(),
current.device_name().map(ToOwned::to_owned),
)
};
let rebuilt = NativeCaptureBackend::build_live_pipeline(
&config,
pipeline_sample_rate,
pipeline_channels,
device_name,
Arc::clone(playback_buffer),
)?;
let mut current = pipeline
.lock()
.map_err(|_| AudioError::new("Audio pipeline lock poisoned"))?;
*current = rebuilt;
Ok(())
}
fn build_input_stream(
device: &cpal::Device,
sample_format: SampleFormat,
config: &StreamConfig,
input_channels: u16,
pipeline: Arc<Mutex<AudioPipeline>>,
err_fn: fn(cpal::StreamError),
) -> AudioResult<cpal::Stream> {
let stream = match sample_format {
SampleFormat::F32 => {
let pipeline_for_stream = Arc::clone(&pipeline);
device.build_input_stream(
config,
move |data: &[f32], _| {
process_stream_buffer(&pipeline_for_stream, data, input_channels)
},
err_fn,
None,
)
}
SampleFormat::I16 => {
let pipeline_for_stream = Arc::clone(&pipeline);
device.build_input_stream(
config,
move |data: &[i16], _| {
let converted = data
.iter()
.map(|sample| *sample as f32 / i16::MAX as f32)
.collect::<Vec<_>>();
process_stream_buffer(&pipeline_for_stream, &converted, input_channels)
},
err_fn,
None,
)
}
SampleFormat::U16 => {
let pipeline_for_stream = Arc::clone(&pipeline);
device.build_input_stream(
config,
move |data: &[u16], _| {
let converted = data
.iter()
.map(|sample| (*sample as f32 - 32768.0) / 32768.0)
.collect::<Vec<_>>();
process_stream_buffer(&pipeline_for_stream, &converted, input_channels)
},
err_fn,
None,
)
}
other => {
return Err(AudioError::new(format!(
"Unsupported input sample format: {:?}",
other
)))
}
}
.map_err(|err| AudioError::new(format!("Failed to build input stream: {}", err)))?;
Ok(stream)
}
fn build_output_stream(
device: &cpal::Device,
sample_format: SampleFormat,
config: &StreamConfig,
playback_buffer: Arc<Mutex<PlaybackBuffer>>,
err_fn: fn(cpal::StreamError),
) -> AudioResult<cpal::Stream> {
let output_channels = config.channels;
let stream = match sample_format {
SampleFormat::F32 => {
let buffer = Arc::clone(&playback_buffer);
device.build_output_stream(
config,
move |data: &mut [f32], _| fill_output_buffer_f32(data, output_channels, &buffer),
err_fn,
None,
)
}
SampleFormat::I16 => {
let buffer = Arc::clone(&playback_buffer);
device.build_output_stream(
config,
move |data: &mut [i16], _| fill_output_buffer_i16(data, output_channels, &buffer),
err_fn,
None,
)
}
SampleFormat::U16 => {
let buffer = Arc::clone(&playback_buffer);
device.build_output_stream(
config,
move |data: &mut [u16], _| fill_output_buffer_u16(data, output_channels, &buffer),
err_fn,
None,
)
}
other => {
return Err(AudioError::new(format!(
"Unsupported output sample format: {:?}",
other
)))
}
}
.map_err(|err| AudioError::new(format!("Failed to build output stream: {}", err)))?;
Ok(stream)
}
fn fill_output_buffer_f32(
data: &mut [f32],
channels: u16,
playback_buffer: &Arc<Mutex<PlaybackBuffer>>,
) {
let channels = channels.max(1) as usize;
if let Ok(mut buffer) = playback_buffer.lock() {
for frame in data.chunks_mut(channels) {
let sample = buffer.pop_mono_sample().unwrap_or(0.0).clamp(-1.0, 1.0);
for out in frame {
*out = sample;
}
}
} else {
data.fill(0.0);
}
}
fn fill_output_buffer_i16(
data: &mut [i16],
channels: u16,
playback_buffer: &Arc<Mutex<PlaybackBuffer>>,
) {
let channels = channels.max(1) as usize;
if let Ok(mut buffer) = playback_buffer.lock() {
for frame in data.chunks_mut(channels) {
let sample =
(buffer.pop_mono_sample().unwrap_or(0.0).clamp(-1.0, 1.0) * i16::MAX as f32) as i16;
for out in frame {
*out = sample;
}
}
} else {
data.fill(0);
}
}
fn fill_output_buffer_u16(
data: &mut [u16],
channels: u16,
playback_buffer: &Arc<Mutex<PlaybackBuffer>>,
) {
let channels = channels.max(1) as usize;
if let Ok(mut buffer) = playback_buffer.lock() {
for frame in data.chunks_mut(channels) {
let sample = ((buffer.pop_mono_sample().unwrap_or(0.0).clamp(-1.0, 1.0) * 32767.0)
+ 32768.0)
.round()
.clamp(0.0, u16::MAX as f32) as u16;
for out in frame {
*out = sample;
}
}
} else {
data.fill(u16::MIN);
}
}
fn select_input_device(
host: &cpal::Host,
preferred_name: Option<&str>,
) -> AudioResult<cpal::Device> {
#[cfg(target_os = "linux")]
{
if let Some(preferred_name) = preferred_name {
if let Err(err) = set_default_input_device(preferred_name) {
log::warn!(
"Failed to set preferred PipeWire source '{}' as default: {}",
preferred_name,
err
);
}
}
if let Some(device) = find_named_input_device(host, "pipewire")? {
return Ok(device);
}
}
if let Some(preferred_name) = preferred_name {
let devices = host.input_devices().map_err(|err| {
AudioError::new(format!("Failed to enumerate input devices: {}", err))
})?;
for device in devices {
if device.description().ok().map(|d| d.to_string()).as_deref() == Some(preferred_name) {
return Ok(device);
}
}
log::warn!(
"Preferred input device '{}' not found, falling back to system default",
preferred_name
);
}
host.default_input_device()
.ok_or_else(|| AudioError::new("No input device available"))
}
fn find_named_input_device(
host: &cpal::Host,
target_name: &str,
) -> AudioResult<Option<cpal::Device>> {
let devices = host
.input_devices()
.map_err(|err| AudioError::new(format!("Failed to enumerate input devices: {}", err)))?;
for device in devices {
if device
.description()
.ok()
.is_some_and(|name| name.to_string().eq_ignore_ascii_case(target_name))
{
return Ok(Some(device));
}
}
Ok(None)
}
fn process_stream_buffer(pipeline: &Arc<Mutex<AudioPipeline>>, data: &[f32], channels: u16) {
if let Ok(mut pipeline) = pipeline.lock() {
if let Err(err) = pipeline.process_input(data, channels) {
log::error!("Audio pipeline processing failed: {}", err);
}
}
}
fn select_stream_config(
device: &cpal::Device,
target_sample_rate: u32,
target_channels: u16,
) -> Option<(u32, u16)> {
let supported_configs = device.supported_input_configs().ok()?;
let mut best_match = None;
for cfg in supported_configs {
let min_rate = cfg.min_sample_rate();
let max_rate = cfg.max_sample_rate();
let channels = cfg.channels();
if target_sample_rate >= min_rate
&& target_sample_rate <= max_rate
&& channels == target_channels
{
return Some((target_sample_rate, target_channels));
}
if best_match.is_none() && target_sample_rate >= min_rate && target_sample_rate <= max_rate
{
best_match = Some((target_sample_rate, channels));
}
}
best_match
}
struct NativeRecordingHandle {
cmd_tx: mpsc::Sender<NativeCommand>,
join_handle: Option<JoinHandle<()>>,
}
struct NativeListeningHandle {
cmd_tx: mpsc::Sender<NativeListeningCommand>,
join_handle: Option<JoinHandle<()>>,
}
impl ActiveRecording for NativeRecordingHandle {
fn stop(&mut self) -> AudioResult<CaptureDiagnostics> {
let (reply_tx, reply_rx) = mpsc::channel();
self.cmd_tx
.send(NativeCommand::Stop(reply_tx))
.map_err(|_| AudioError::new("Native capture thread is no longer running"))?;
let diagnostics = reply_rx
.recv()
.map_err(|_| AudioError::new("No response from native capture thread"))??;
if let Some(handle) = self.join_handle.take() {
let _ = handle.join();
}
Ok(diagnostics)
}
}
impl ActiveListening for NativeListeningHandle {
fn stop(&mut self) -> AudioResult<CaptureDiagnostics> {
let (reply_tx, reply_rx) = mpsc::channel();
self.cmd_tx
.send(NativeListeningCommand::Stop(reply_tx))
.map_err(|_| AudioError::new("Native listening thread is no longer running"))?;
let diagnostics = reply_rx
.recv()
.map_err(|_| AudioError::new("No response from native listening thread"))??;
if let Some(handle) = self.join_handle.take() {
let _ = handle.join();
}
Ok(diagnostics)
}
fn update_config(&mut self, config: NativeCaptureConfig) -> AudioResult<()> {
let (reply_tx, reply_rx) = mpsc::channel();
self.cmd_tx
.send(NativeListeningCommand::UpdateConfig(config, reply_tx))
.map_err(|_| AudioError::new("Native listening thread is no longer running"))?;
reply_rx
.recv()
.map_err(|_| AudioError::new("No response from native listening thread"))??;
Ok(())
}
}