feat(audio): compute level meter + device-change recovery (T7.3, FR-CAP-5/6)

Adds AudioLevel (rms/peak per chunk, throttled to ~20Hz) and
CaptureEvent::DeviceChanged emitted through a new EventSink alongside
the existing FrameSink. On a capture read failure, reopens against
whatever is now the default render device and continues the same WAV
file if the new format is compatible; otherwise fails loudly rather
than silently corrupting the recording.
This commit is contained in:
iamdoubz
2026-07-01 21:09:28 -05:00
parent 5a1b5ee0d1
commit 098331d6f5
+217 -29
View File
@@ -24,7 +24,10 @@ use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::SyncSender;
use std::sync::Arc;
use std::thread::{self, JoinHandle};
use wasapi::{Direction, SampleType, ShareMode, WaveFormat};
use std::time::{Duration, Instant};
use wasapi::{
AudioCaptureClient, AudioClient, Direction, Handle, SampleType, ShareMode, WaveFormat,
};
#[derive(Debug, thiserror::Error)]
pub enum AudioError {
@@ -47,6 +50,30 @@ pub struct CaptureHandle {
/// bounded so a slow/absent consumer can never stall the capture thread.
pub type FrameSink = SyncSender<Vec<f32>>;
/// Instantaneous input level for one capture chunk, throttled to ~20Hz by the
/// capture loop (FR-CAP-5 waveform/level meter).
#[derive(Debug, Clone, Copy)]
pub struct AudioLevel {
pub rms: f32,
pub peak: f32,
}
/// Out-of-band capture notices, lower-volume than `FrameSink`/level updates.
#[derive(Debug, Clone)]
pub enum CaptureEvent {
Level(AudioLevel),
/// The capture device stopped responding mid-recording (unplugged, default
/// device switched, etc.) — `recovered = true` once a reconnect to
/// whatever is now the default device succeeded without losing the
/// session (FR-CAP-6); `false` while recovery is in progress/failed.
DeviceChanged {
recovered: bool,
message: String,
},
}
pub type EventSink = SyncSender<CaptureEvent>;
pub struct CaptureSummary {
pub duration_ms: u64,
pub sample_rate: u32,
@@ -54,7 +81,12 @@ pub struct CaptureSummary {
}
pub trait AudioCapture: Send + Sync {
fn start(&self, wav_path: &Path, sink: FrameSink) -> Result<CaptureHandle, AudioError>;
fn start(
&self,
wav_path: &Path,
frame_sink: FrameSink,
event_sink: EventSink,
) -> Result<CaptureHandle, AudioError>;
fn pause(&self, h: &CaptureHandle) -> Result<(), AudioError>;
fn resume(&self, h: &CaptureHandle) -> Result<(), AudioError>;
fn stop(&self, h: CaptureHandle) -> Result<CaptureSummary, AudioError>;
@@ -66,7 +98,12 @@ pub struct WasapiCapture;
#[cfg(feature = "audio")]
impl AudioCapture for WasapiCapture {
fn start(&self, wav_path: &Path, sink: FrameSink) -> Result<CaptureHandle, AudioError> {
fn start(
&self,
wav_path: &Path,
frame_sink: FrameSink,
event_sink: EventSink,
) -> Result<CaptureHandle, AudioError> {
let running = Arc::new(AtomicBool::new(true));
let paused = Arc::new(AtomicBool::new(false));
let running_th = running.clone();
@@ -75,7 +112,9 @@ impl AudioCapture for WasapiCapture {
let thread = thread::Builder::new()
.name("wa-audio-capture".into())
.spawn(move || capture_loop(&wav_path, &sink, &running_th, &paused_th))
.spawn(move || {
capture_loop(&wav_path, &frame_sink, &event_sink, &running_th, &paused_th)
})
.map_err(|e| AudioError::Capture(format!("spawn failed: {e}")))?;
Ok(CaptureHandle {
@@ -103,20 +142,23 @@ impl AudioCapture for WasapiCapture {
}
}
/// Runs on a dedicated OS thread for the lifetime of a `CaptureHandle`. Owns the
/// WASAPI client and the WAV writer; exits (and finalizes the WAV) once `running`
/// is cleared.
/// One open WASAPI capture session — bundled so it can be torn down and
/// re-opened wholesale on device recovery (FR-CAP-6) without duplicating the
/// open sequence at each call site.
#[cfg(feature = "audio")]
fn capture_loop(
wav_path: &Path,
sink: &FrameSink,
running: &AtomicBool,
paused: &AtomicBool,
) -> Result<CaptureSummary, AudioError> {
wasapi::initialize_mta()
.ok()
.map_err(|e| AudioError::Device(format!("COM init failed: {e}")))?;
struct CaptureSession {
audio_client: AudioClient,
event_handle: Handle,
capture_client: AudioCaptureClient,
format: WaveFormat,
}
/// Opens the current default render device for loopback capture. Called both
/// for the initial open and to reconnect after the device disappears
/// mid-recording (FR-CAP-6) — each call re-resolves "the default device",
/// so it naturally picks up whatever the OS switched to.
#[cfg(feature = "audio")]
fn open_capture_session() -> Result<CaptureSession, AudioError> {
let device = wasapi::get_default_device(&Direction::Render)
.map_err(|e| AudioError::Device(format!("no default render device: {e}")))?;
let mut audio_client = device
@@ -147,25 +189,116 @@ fn capture_loop(
.get_audiocaptureclient()
.map_err(|e| AudioError::Capture(format!("GetService(IAudioCaptureClient) failed: {e}")))?;
let spec = wav_spec_for(&format)?;
Ok(CaptureSession {
audio_client,
event_handle,
capture_client,
format,
})
}
/// True if two formats are close enough to keep writing into the same
/// already-open WAV file (channels/rate/bit-depth/sample-type all match).
/// A reconnect landing on an incompatible format can't safely continue the
/// same file, so `capture_loop` treats that as unrecoverable rather than
/// silently corrupting `audio.wav`.
#[cfg(feature = "audio")]
fn format_compatible(a: &WaveFormat, b: &WaveFormat) -> bool {
a.get_nchannels() == b.get_nchannels()
&& a.get_samplespersec() == b.get_samplespersec()
&& a.get_bitspersample() == b.get_bitspersample()
&& matches!(
(a.get_subformat(), b.get_subformat()),
(Ok(x), Ok(y)) if std::mem::discriminant(&x) == std::mem::discriminant(&y)
)
}
/// Amplitude of one chunk of mono samples, for the live level meter (FR-CAP-5).
#[cfg(feature = "audio")]
fn audio_level(mono: &[f32]) -> AudioLevel {
if mono.is_empty() {
return AudioLevel {
rms: 0.0,
peak: 0.0,
};
}
let sum_sq: f32 = mono.iter().map(|s| s * s).sum();
let peak = mono.iter().fold(0.0f32, |m, s| m.max(s.abs()));
AudioLevel {
rms: (sum_sq / mono.len() as f32).sqrt(),
peak,
}
}
/// How often level updates are forwarded — plenty for a waveform/meter UI
/// without flooding the frontend with an event per ~10ms WASAPI callback.
const LEVEL_EMIT_INTERVAL: Duration = Duration::from_millis(50);
/// Runs on a dedicated OS thread for the lifetime of a `CaptureHandle`. Owns the
/// WASAPI client and the WAV writer; exits (and finalizes the WAV) once `running`
/// is cleared.
#[cfg(feature = "audio")]
fn capture_loop(
wav_path: &Path,
frame_sink: &FrameSink,
event_sink: &EventSink,
running: &AtomicBool,
paused: &AtomicBool,
) -> Result<CaptureSummary, AudioError> {
wasapi::initialize_mta()
.ok()
.map_err(|e| AudioError::Device(format!("COM init failed: {e}")))?;
let mut session = open_capture_session()?;
let spec = wav_spec_for(&session.format)?;
let mut writer = WavWriter::create(wav_path, spec).map_err(|e| {
AudioError::Capture(format!("could not create {}: {e}", wav_path.display()))
})?;
audio_client
session
.audio_client
.start_stream()
.map_err(|e| AudioError::Capture(format!("start_stream failed: {e}")))?;
let mut resampler = Resampler::new(format.get_samplespersec());
let mut resampler = Resampler::new(session.format.get_samplespersec());
let mut queue: std::collections::VecDeque<u8> = std::collections::VecDeque::new();
let mut frames_written: u64 = 0;
let mut last_level_emit = Instant::now() - LEVEL_EMIT_INTERVAL;
while running.load(Ordering::Relaxed) {
// Short timeout so we periodically re-check `running` even with no data.
let _ = event_handle.wait_for_event(100);
capture_client
.read_from_device_to_deque(&mut queue)
.map_err(|e| AudioError::Capture(format!("GetBuffer failed: {e}")))?;
let _ = session.event_handle.wait_for_event(100);
if let Err(e) = session.capture_client.read_from_device_to_deque(&mut queue) {
// The device likely disappeared mid-recording (unplugged, default
// output switched, etc.) — reconnect to whatever is now the
// default device instead of losing the rest of the meeting
// (FR-CAP-6). Only continue on the same WAV file if the new
// device's format still matches; otherwise this is unrecoverable.
let _ = event_sink.try_send(CaptureEvent::DeviceChanged {
recovered: false,
message: e.to_string(),
});
session.audio_client.stop_stream().ok();
let new_session = open_capture_session().map_err(|re| {
AudioError::Capture(format!("device lost, reconnect failed: {re}"))
})?;
if !format_compatible(&session.format, &new_session.format) {
return Err(AudioError::Capture(
"reconnected to a device with an incompatible format; cannot continue the same recording".into(),
));
}
new_session
.audio_client
.start_stream()
.map_err(|e| AudioError::Capture(format!("restart_stream failed: {e}")))?;
resampler = Resampler::new(new_session.format.get_samplespersec());
session = new_session;
let _ = event_sink.try_send(CaptureEvent::DeviceChanged {
recovered: true,
message: "reconnected to the default audio device".into(),
});
continue;
}
if queue.is_empty() {
continue;
}
@@ -177,24 +310,30 @@ fn capture_loop(
continue;
}
frames_written += write_wav_bytes(&mut writer, &bytes, &format)?;
let mono = decode_mono_f32(&bytes, &format)?;
frames_written += write_wav_bytes(&mut writer, &bytes, &session.format)?;
let mono = decode_mono_f32(&bytes, &session.format)?;
if last_level_emit.elapsed() >= LEVEL_EMIT_INTERVAL {
let _ = event_sink.try_send(CaptureEvent::Level(audio_level(&mono)));
last_level_emit = Instant::now();
}
let resampled = resampler.process(&mono);
if !resampled.is_empty() {
let _ = sink.try_send(resampled); // drop on backpressure; disk write is unaffected
let _ = frame_sink.try_send(resampled); // drop on backpressure; disk write is unaffected
}
}
audio_client.stop_stream().ok();
session.audio_client.stop_stream().ok();
writer
.finalize()
.map_err(|e| AudioError::Capture(format!("wav finalize failed: {e}")))?;
let sample_rate = format.get_samplespersec();
let sample_rate = session.format.get_samplespersec();
Ok(CaptureSummary {
duration_ms: (frames_written * 1000) / sample_rate.max(1) as u64,
sample_rate,
channels: format.get_nchannels(),
channels: session.format.get_nchannels(),
})
}
@@ -547,4 +686,53 @@ mod tests {
let input = vec![0.1f32, 0.2, 0.3];
assert_eq!(r.process(&input), input);
}
#[test]
fn audio_level_of_empty_chunk_is_silence() {
let level = audio_level(&[]);
assert_eq!(level.rms, 0.0);
assert_eq!(level.peak, 0.0);
}
#[test]
fn audio_level_computes_rms_and_peak() {
// Two samples of equal magnitude: RMS equals that magnitude, peak too.
let level = audio_level(&[0.5, -0.5]);
assert!((level.rms - 0.5).abs() < 1e-6);
assert!((level.peak - 0.5).abs() < 1e-6);
}
#[test]
fn audio_level_peak_tracks_the_largest_magnitude_sample() {
let level = audio_level(&[0.1, -0.9, 0.3]);
assert!((level.peak - 0.9).abs() < 1e-6);
}
#[test]
fn format_compatible_matches_identical_formats() {
let a = WaveFormat::new(32, 32, &SampleType::Float, 48_000, 2, None);
let b = WaveFormat::new(32, 32, &SampleType::Float, 48_000, 2, None);
assert!(format_compatible(&a, &b));
}
#[test]
fn format_compatible_rejects_a_different_sample_rate() {
let a = WaveFormat::new(32, 32, &SampleType::Float, 48_000, 2, None);
let b = WaveFormat::new(32, 32, &SampleType::Float, 44_100, 2, None);
assert!(!format_compatible(&a, &b));
}
#[test]
fn format_compatible_rejects_a_different_channel_count() {
let a = WaveFormat::new(32, 32, &SampleType::Float, 48_000, 2, None);
let b = WaveFormat::new(32, 32, &SampleType::Float, 48_000, 1, None);
assert!(!format_compatible(&a, &b));
}
#[test]
fn format_compatible_rejects_a_different_sample_type() {
let a = WaveFormat::new(32, 32, &SampleType::Float, 48_000, 2, None);
let b = WaveFormat::new(16, 16, &SampleType::Int, 48_000, 2, None);
assert!(!format_compatible(&a, &b));
}
}