From 12fb17c9c3899d215fe229b334d3f4e3ef442dc6 Mon Sep 17 00:00:00 2001 From: Klesti Selimaj Date: Fri, 28 Aug 2026 14:24:20 -0400 Subject: [PATCH] Fixed linux audio --- src-tauri/src/commands/audio.rs | 300 +++++++++++++++++++++++++++----- 1 file changed, 254 insertions(+), 46 deletions(-) diff --git a/src-tauri/src/commands/audio.rs b/src-tauri/src/commands/audio.rs index 7e6b5cf..1b3a3c5 100644 --- a/src-tauri/src/commands/audio.rs +++ b/src-tauri/src/commands/audio.rs @@ -10,7 +10,10 @@ use std::{ }; use chacha20poly1305::{aead::KeyInit, ChaCha20Poly1305, Key}; -use cpal::traits::{DeviceTrait, HostTrait, StreamTrait}; +use cpal::{ + traits::{DeviceTrait, HostTrait, StreamTrait}, + FromSample, SampleFormat, SizedSample, +}; use ringbuf::{ storage::Heap, traits::{Consumer, Observer, Producer, Split}, @@ -178,24 +181,67 @@ fn build_input_stream( producer: AudioProducer, state_inner: Arc, ) -> Result { + let sample_format = device + .default_input_config() + .map_err(|e| format!("Failed to get input config: {e}"))? + .sample_format(); + + match sample_format { + SampleFormat::F32 => build_input_stream_with::(device, config, producer, state_inner), + SampleFormat::I16 => build_input_stream_with::(device, config, producer, state_inner), + SampleFormat::I32 => build_input_stream_with::(device, config, producer, state_inner), + SampleFormat::I64 => build_input_stream_with::(device, config, producer, state_inner), + SampleFormat::U8 => build_input_stream_with::(device, config, producer, state_inner), + SampleFormat::U16 => build_input_stream_with::(device, config, producer, state_inner), + SampleFormat::U32 => build_input_stream_with::(device, config, producer, state_inner), + SampleFormat::U64 => build_input_stream_with::(device, config, producer, state_inner), + format => Err(format!("Unsupported input sample format: {format}")), + } +} + +fn build_input_stream_with( + device: &cpal::Device, + config: cpal::StreamConfig, + producer: AudioProducer, + state_inner: Arc, +) -> Result +where + T: SizedSample + Send + 'static, + f32: FromSample, +{ let native_sample_rate = config.sample_rate; let channels = config.channels as usize; let mut resampler = LinearResampler::new(); let mut mono_buffer = Vec::with_capacity(2048); let mut resampled_buffer = Vec::with_capacity(2048); + let mut callback_count: u64 = 0; let inner_input_err = Arc::clone(&state_inner); device .build_input_stream( config, - move |data: &[f32], _| { + move |data: &[T], _| { + callback_count += 1; + if callback_count % 200 == 0 { + let peak: f32 = data + .iter() + .map(|&s| s.to_sample::().abs()) + .fold(0.0f32, f32::max); + eprintln!( + "[vc] input callback #{callback_count}: {} frames, {} ch, peak {:.4}", + data.len() / channels, + channels, + peak + ); + } + mono_buffer.clear(); resampled_buffer.clear(); - // Downmix channels to mono + // Downmix channels to mono, converting to f32 for chunk in data.chunks_exact(channels) { - let sum: f32 = chunk.iter().sum(); + let sum: f32 = chunk.iter().map(|&s| s.to_sample::()).sum(); mono_buffer.push(sum / channels as f32); } @@ -213,13 +259,21 @@ fn build_input_stream( }, move |err| { eprintln!("[vc] Input error: {err}. Attempting recovery..."); - if let Ok(mut lock) = inner_input_err.session.lock() { + let rec = Arc::clone(&inner_input_err); + thread::spawn(move || { + // Give the errored stream a moment to unwind before rebuilding. + thread::sleep(Duration::from_millis(10)); + let Ok(mut lock) = rec.session.lock() else { + return; + }; if let Some(session) = lock.as_mut() { + if session.shutdown.load(Ordering::SeqCst) { + return; + } let target_device = session.current_input_device.clone(); - let _ = session - .update_input_device(target_device, Arc::clone(&inner_input_err)); + let _ = session.update_input_device(target_device, Arc::clone(&rec)); } - } + }); }, None, ) @@ -233,25 +287,64 @@ fn build_output_stream( consumer: AudioConsumer, state_inner: Arc, ) -> Result { + let sample_format = device + .default_output_config() + .map_err(|e| format!("Failed to get output config: {e}"))? + .sample_format(); + + match sample_format { + SampleFormat::F32 => build_output_stream_with::(device, config, consumer, state_inner), + SampleFormat::I16 => build_output_stream_with::(device, config, consumer, state_inner), + SampleFormat::I32 => build_output_stream_with::(device, config, consumer, state_inner), + SampleFormat::I64 => build_output_stream_with::(device, config, consumer, state_inner), + SampleFormat::U8 => build_output_stream_with::(device, config, consumer, state_inner), + SampleFormat::U16 => build_output_stream_with::(device, config, consumer, state_inner), + SampleFormat::U32 => build_output_stream_with::(device, config, consumer, state_inner), + SampleFormat::U64 => build_output_stream_with::(device, config, consumer, state_inner), + format => Err(format!("Unsupported output sample format: {format}")), + } +} + +fn build_output_stream_with( + device: &cpal::Device, + config: cpal::StreamConfig, + consumer: AudioConsumer, + state_inner: Arc, +) -> Result +where + T: SizedSample + Send + 'static, + T: FromSample, +{ let native_sample_rate = config.sample_rate; let channels = config.channels as usize; let mut resampler = LinearResampler::new(); let mut raw_mono_samples = Vec::with_capacity(2048); let mut resampled_mono = Vec::with_capacity(2048); let mut last_sample = 0.0f32; + let mut callback_count: u64 = 0; let inner_output_err = Arc::clone(&state_inner); device .build_output_stream( config, - move |data: &mut [f32], _| { + move |data: &mut [T], _| { + callback_count += 1; + if callback_count % 200 == 0 { + eprintln!( + "[vc] output callback #{callback_count}: {} frames, {} ch", + data.len() / channels, + channels + ); + } + let required_mono_samples = (data.len() / channels) * TARGET_SAMPLE_RATE as usize / native_sample_rate as usize; raw_mono_samples.clear(); resampled_mono.clear(); + let mut underflow_count = 0usize; if let Ok(mut cons) = consumer.lock() { for _ in 0..required_mono_samples { if let Some(s) = cons.try_pop() { @@ -261,9 +354,18 @@ fn build_output_stream( // Exponential decay to prevent clicking when underflowing last_sample *= 0.92; raw_mono_samples.push(last_sample); + underflow_count += 1; } } } + if callback_count % 200 == 0 { + eprintln!( + "[vc] output: got {}/{} mono samples ({} underflow)", + required_mono_samples - underflow_count, + required_mono_samples, + underflow_count + ); + } // Resample from 48kHz mono to target native output rate resampler.process( @@ -280,7 +382,7 @@ fn build_output_stream( let mono_val = resampled_mono[res_idx]; for ch in 0..channels { if out_idx + ch < data.len() { - data[out_idx + ch] = mono_val; + data[out_idx + ch] = T::from_sample(mono_val); } } out_idx += channels; @@ -289,13 +391,20 @@ fn build_output_stream( }, move |err| { eprintln!("[vc] Output error: {err}. Attempting recovery..."); - if let Ok(mut lock) = inner_output_err.session.lock() { + let rec = Arc::clone(&inner_output_err); + thread::spawn(move || { + thread::sleep(Duration::from_millis(10)); + let Ok(mut lock) = rec.session.lock() else { + return; + }; if let Some(session) = lock.as_mut() { + if session.shutdown.load(Ordering::SeqCst) { + return; + } let target_device = session.current_output_device.clone(); - let _ = session - .update_output_device(target_device, Arc::clone(&inner_output_err)); + let _ = session.update_output_device(target_device, Arc::clone(&rec)); } - } + }); }, None, ) @@ -324,18 +433,22 @@ pub fn list_output_devices() -> Result, String> { #[tauri::command] pub fn disconnect_from_vc(voice_state: State<'_, VoiceState>) -> Result<(), String> { - let mut lock = voice_state - .inner - .session - .lock() - .map_err(|e| e.to_string())?; + let session = { + let mut lock = voice_state + .inner + .session + .lock() + .map_err(|e| e.to_string())?; + lock.take() + }; - if let Some(session) = lock.take() { + if let Some(session) = session { session.shutdown.store(true, Ordering::SeqCst); let _ = session.input_stream.pause(); if let Ok(output) = session.output_stream.lock() { let _ = output.pause(); } + eprintln!("[vc] disconnected and paused streams"); } Ok(()) @@ -349,59 +462,102 @@ pub fn connect_to_vc( config_state: State<'_, ConfigState>, voice_state: State<'_, VoiceState>, ) -> Result<(), String> { - disconnect_from_vc(voice_state.clone())?; + eprintln!("[vc] connect_to_vc called hostname={hostname} pin={pin}"); - let key = Key::try_from(shared_secret.as_slice()).map_err(|v| v.to_string())?; + if let Err(e) = disconnect_from_vc(voice_state.clone()) { + eprintln!("[vc] disconnect_from_vc failed: {e}"); + } + + let key = match Key::try_from(shared_secret.as_slice()) { + Ok(k) => k, + Err(v) => { + let msg = v.to_string(); + eprintln!("[vc] invalid key length: {msg}"); + return Err(msg); + } + }; let cipher = Arc::new(Mutex::new(SessionCipher::new(ChaCha20Poly1305::new(&key)))); let config = config_state.0.lock().unwrap().clone(); let state_inner = Arc::clone(&voice_state.inner); - let socket = Arc::new( - UdpSocket::bind("0.0.0.0:0").map_err(|e| format!("Failed to bind UDP socket: {e}"))?, - ); - socket - .connect(&hostname) - .map_err(|e| format!("Failed to connect UDP socket: {e}"))?; - socket - .set_read_timeout(Some(Duration::from_millis(5))) - .map_err(|e| e.to_string())?; - socket - .send(&pin.to_be_bytes()) - .map_err(|e| format!("Failed to send pin: {e}"))?; + let socket = match UdpSocket::bind("0.0.0.0:0") { + Ok(s) => s, + Err(e) => { + let msg = format!("Failed to bind UDP socket: {e}"); + eprintln!("[vc] {msg}"); + return Err(msg); + } + }; + let socket = Arc::new(socket); + if let Err(e) = socket.connect(&hostname) { + let msg = format!("Failed to connect UDP socket: {e}"); + eprintln!("[vc] {msg}"); + return Err(msg); + } + if let Err(e) = socket.set_read_timeout(Some(Duration::from_millis(5))) { + let msg = e.to_string(); + eprintln!("[vc] set_read_timeout failed: {msg}"); + return Err(msg); + } + if let Err(e) = socket.send(&pin.to_be_bytes()) { + let msg = format!("Failed to send pin: {e}"); + eprintln!("[vc] {msg}"); + return Err(msg); + } let host = cpal::default_host(); let input_device = match &config.input_device_name { Some(name) => host .input_devices() - .map_err(|e| e.to_string())? + .map_err(|e| { + let msg = e.to_string(); + eprintln!("[vc] failed to enumerate input devices: {msg}"); + msg + })? .find(|d| { d.description() .ok() .map(|x| x.name() == *name) .unwrap_or(false) }) - .ok_or("Input device not found")?, - None => host - .default_input_device() - .ok_or("No default input device")?, + .ok_or_else(|| { + let msg = format!("Input device not found: {name}"); + eprintln!("[vc] {msg}"); + msg + })?, + None => host.default_input_device().ok_or_else(|| { + let msg = "No default input device".to_string(); + eprintln!("[vc] {msg}"); + msg + })?, }; let output_device = match &config.output_device_name { Some(name) => host .output_devices() - .map_err(|e| e.to_string())? + .map_err(|e| { + let msg = e.to_string(); + eprintln!("[vc] failed to enumerate output devices: {msg}"); + msg + })? .find(|d| { d.description() .ok() .map(|x| x.name() == *name) .unwrap_or(false) }) - .ok_or("Output device not found")?, - None => host - .default_output_device() - .ok_or("No default output device")?, + .ok_or_else(|| { + let msg = format!("Output device not found: {name}"); + eprintln!("[vc] {msg}"); + msg + })?, + None => host.default_output_device().ok_or_else(|| { + let msg = "No default output device".to_string(); + eprintln!("[vc] {msg}"); + msg + })?, }; let shutdown = Arc::new(AtomicBool::new(false)); @@ -417,6 +573,12 @@ pub fn connect_to_vc( .map_err(|e| format!("Failed to get default output config: {e}"))? .config(); + eprintln!( + "[vc] Output device {:?} config: {} Hz, {} ch", + output_device.description().map(|d| d.name().to_string()), + output_config.sample_rate, + output_config.channels + ); let output_stream = build_output_stream( &output_device, output_config, @@ -437,6 +599,12 @@ pub fn connect_to_vc( .map_err(|e| format!("Failed to get default input config: {e}"))? .config(); + eprintln!( + "[vc] Input device {:?} config: {} Hz, {} ch", + input_device.description().map(|d| d.name().to_string()), + input_config.sample_rate, + input_config.channels + ); let input_stream = build_input_stream( &input_device, input_config, @@ -454,6 +622,9 @@ pub fn connect_to_vc( let mut sequence = 0u32; let mut frame_buf = vec![0.0f32; PACKET_SAMPLES]; let mut hangover_counter = 0; + let mut loop_count: u64 = 0; + // Adaptive VAD: track a slow-moving noise floor so quiet mics still trigger. + let mut noise_floor: f32 = 0.0; while !shutdown.load(Ordering::Relaxed) { if consumer_in.occupied_len() >= PACKET_SAMPLES { @@ -462,7 +633,17 @@ pub fn connect_to_vc( let sum_squares: f32 = frame_buf.iter().map(|&s| s * s).sum(); let rms = (sum_squares / PACKET_SAMPLES as f32).sqrt(); - let is_speaking = if rms >= VAD_THRESHOLD { + // Update noise floor (attack fast, release slow). + if noise_floor == 0.0 { + noise_floor = rms; + } else if rms < noise_floor { + noise_floor = noise_floor * 0.9 + rms * 0.1; + } else { + noise_floor = noise_floor * 0.999; + } + let threshold = (noise_floor * 4.0).max(VAD_THRESHOLD); + + let is_speaking = if rms >= threshold { hangover_counter = VAD_HANGOVER_FRAMES; true } else if hangover_counter > 0 { @@ -472,6 +653,14 @@ pub fn connect_to_vc( false }; + loop_count += 1; + if loop_count % 200 == 0 { + eprintln!( + "[vc] sender: buffered {} samples, frame rms {rms:.4}, floor {noise_floor:.4}, thr {threshold:.4}, speaking {is_speaking}, seq {sequence}", + consumer_in.occupied_len() + ); + } + if is_speaking { // sequence goes INSIDE the plaintext now, prefixed before the PCM let mut plaintext = Vec::with_capacity(4 + PACKET_SAMPLES * 2); @@ -488,7 +677,15 @@ pub fn connect_to_vc( continue; }; - let _ = input_socket.send(&net_packet); + let sent = input_socket.send(&net_packet); + if let Err(e) = sent { + eprintln!("[vc] sender: send failed: {e}"); + } else if loop_count % 200 == 0 { + eprintln!( + "[vc] sender: sent packet seq {sequence} ({} bytes)", + net_packet.len() + ); + } sequence = sequence.wrapping_add(1); } } else { @@ -511,9 +708,13 @@ pub fn connect_to_vc( let mut last_good_frame = vec![0.0f32; PACKET_SAMPLES]; let mut udp_buffer = [0u8; MAX_PACKET_SIZE]; let mut next_frame_time = Instant::now(); + let mut recv_count: u64 = 0; + let mut loop_count: u64 = 0; while !shutdown.load(Ordering::Relaxed) { + loop_count += 1; if let Ok(len) = socket.recv(&mut udp_buffer) { + recv_count += 1; match cipher.lock().unwrap().decrypt(&udp_buffer[..len]) { Ok(plaintext) => { if plaintext.len() < 4 { @@ -536,6 +737,13 @@ pub fn connect_to_vc( } } + if loop_count % 2000 == 0 { + eprintln!( + "[vc] receiver: received {recv_count} packets total, {} buffered, prebuffering {is_prebuffering}, expected {expected:?}", + packets.len() + ); + } + if is_prebuffering { if packets.len() >= INITIAL_PACKET_CUSHION { expected = packets.keys().next().copied();