jake/Lumotia
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

audio: wire user's microphone choice through start_native_capture + live session

Browse Source 
Day 1 follow-up to 96980c7. The device-picker UI in Settings now
actually takes effect: settings.microphoneDevice flows from the Svelte
store, through the Tauri invoke, into MicrophoneCapture::start_with_device
on the Rust side.

Touched paths (back-to-front):
- src-tauri/src/commands/audio.rs:start_native_capture — new optional
  `device_name: Option<String>` parameter; routes to start_with_device
  when set, falls back to auto-select start() when None or empty.
- src-tauri/src/commands/live.rs:StartLiveTranscriptionConfig — new
  optional `microphone_device: Option<String>` field with same
  semantics (rename_all = "camelCase" maps it to microphoneDevice on
  the wire).
- src-tauri/src/commands/live.rs:run_live_session — picks
  start_with_device when an explicit name is provided.
- src/lib/pages/DictationPage.svelte — passes
  microphoneDevice: settings.microphoneDevice || null in the invoke.

Behaviour:
- "Auto" in the picker (empty string) -> backend auto-selects, skipping
  monitor sources and validating by RMS energy.
- Specific device -> backend opens that device by exact name; if it has
  been disconnected the user gets a clear error pointing them back at
  Settings.

cargo check -p kon-audio passes clean. Tauri-crate cargo check requires
cmake (pre-existing infra dependency for whisper-rs-sys); install via
`sudo dnf install cmake clang-devel`.
This commit is contained in:
Jake
2026-04-17 12:45:19 +01:00
parent 96980c7d5c
commit 41db162041
3 changed files with 1002 additions and 225 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
src-tauri/src/commands/audio.rs
View File

@@ -38,23 +38,38 @@ impl NativeCaptureState {
/// Start native microphone capture via cpal.
/// Streams 16kHz mono PCM chunks to the frontend via `native-pcm` events.
///
/// `device_name`: explicit device name (from `list_audio_devices`) or None / ""
/// to auto-select. The frontend passes `settings.microphoneDevice` here so the
/// user's pick from Settings → Audio → Microphone takes effect.
#[tauri::command]
pub async fn start_native_capture(
app: tauri::AppHandle,
state: tauri::State<'_, NativeCaptureState>,
device_name: Option<String>,
) -> Result<(), String> {
eprintln!("[native-capture] start_native_capture called");
eprintln!(
"[native-capture] start_native_capture called (device='{}')",
device_name.as_deref().unwrap_or("<auto>")
);
// Stop any existing capture
if let Some(tx) = state.stop_tx.lock().unwrap().take() {
drop(tx);
}
let (capture, rx) = MicrophoneCapture::start().map_err(|e| {
let (capture, rx) = match device_name.as_deref() {
Some(name) if !name.is_empty() => MicrophoneCapture::start_with_device(name),
_ => MicrophoneCapture::start(),
}
.map_err(|e| {
eprintln!("[native-capture] MicrophoneCapture::start failed: {e}");
e.to_string()
})?;
eprintln!("[native-capture] cpal capture started successfully");
eprintln!(
"[native-capture] cpal capture started successfully on '{}'",
capture.device_name
);
// Wrap capture in Arc<Mutex> so it can be moved into the blocking task
let capture = Arc::new(Mutex::new(Some(capture)));

614
src-tauri/src/commands/live.rs Normal file
View File

@@ -0,0 +1,614 @@
#![allow(clippy::too_many_arguments)]
use std::sync::{
atomic::{AtomicBool, AtomicU64, Ordering},
Arc, Mutex,
};
use std::thread;
use std::time::{Duration, Instant};
use serde::{Deserialize, Serialize};
use tauri::ipc::Channel;
use crate::commands::audio::persist_audio_samples;
use crate::commands::models::{default_model_id_for_engine, ensure_model_loaded};
use crate::AppState;
use kon_ai_formatting::{
post_process_segments, FormatMode, PostProcessOptions,
};
use kon_audio::{MicrophoneCapture, StreamingResampler};
use kon_core::constants::WHISPER_SAMPLE_RATE;
use kon_core::types::{AudioSamples, Segment, TranscriptionOptions};
use kon_transcription::LocalEngine;
const CHUNK_SAMPLES: usize = 32_000; // 2s at 16kHz
const OVERLAP_SAMPLES: usize = 4_000; // 0.25s at 16kHz
const FINAL_CHUNK_MIN_SAMPLES: usize = 4_000; // 0.25s
const MAX_PENDING_SAMPLES: usize = CHUNK_SAMPLES;
const SPEECH_FRAME_SAMPLES: usize = 800; // 50ms
const MIN_SPEECH_FRAMES: usize = 1; // any plausible speech-like frame
const RMS_SPEECH_THRESHOLD: f32 = 0.001;
const PEAK_SPEECH_THRESHOLD: f32 = 0.004;
const FLATLINE_PEAK_THRESHOLD: f32 = 0.0005;
#[derive(Default)]
pub struct LiveTranscriptionState {
next_session_id: AtomicU64,
running: Mutex<Option<RunningLiveSession>>,
}
struct RunningLiveSession {
id: u64,
output_folder: Option<String>,
stop_flag: Arc<AtomicBool>,
handle: tokio::task::JoinHandle<Result<LiveSessionSummary, String>>,
status_channel: Channel<LiveStatusMessage>,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct StartLiveTranscriptionConfig {
pub engine: String,
pub model_id: Option<String>,
pub language: Option<String>,
pub initial_prompt: Option<String>,
pub save_audio: bool,
pub output_folder: Option<String>,
pub remove_fillers: bool,
pub british_english: bool,
pub anti_hallucination: bool,
pub format_mode: String,
/// Optional explicit microphone device name (from `list_audio_devices`).
/// None or empty string = let `MicrophoneCapture::start` auto-select.
pub microphone_device: Option<String>,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct StartLiveTranscriptionResponse {
pub session_id: u64,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct StopLiveTranscriptionResponse {
pub session_id: u64,
pub audio_path: Option<String>,
pub dropped_audio_ms: u64,
}
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct LiveResultMessage {
pub session_id: u64,
pub chunk_id: u32,
pub chunk_start_secs: f64,
pub duration: f64,
pub language: String,
pub inference_ms: u64,
pub segments: Vec<Segment>,
}
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase", tag = "type")]
#[allow(dead_code)]
pub enum LiveStatusMessage {
Warning {
session_id: u64,
message: String,
},
Overload {
session_id: u64,
dropped_audio_ms: u64,
message: String,
},
Error {
session_id: u64,
message: String,
},
Finished {
session_id: u64,
audio_path: Option<String>,
dropped_audio_ms: u64,
},
}
struct LiveSessionSummary {
session_id: u64,
dropped_audio_ms: u64,
audio_samples: Option<Vec<f32>>,
}
struct InferenceTask {
chunk_id: u32,
chunk_start_sample: u64,
trim_before_secs: f64,
duration_secs: f64,
rx: std::sync::mpsc::Receiver<Result<kon_transcription::TimedTranscript, String>>,
}
#[tauri::command]
pub async fn start_live_transcription_session(
state: tauri::State<'_, AppState>,
live_state: tauri::State<'_, LiveTranscriptionState>,
config: StartLiveTranscriptionConfig,
result_channel: Channel<LiveResultMessage>,
status_channel: Channel<LiveStatusMessage>,
) -> Result<StartLiveTranscriptionResponse, String> {
{
let running = live_state.running.lock().unwrap();
if running.is_some() {
return Err("A live transcription session is already running".into());
}
}
let model_id = config
.model_id
.clone()
.unwrap_or_else(|| default_model_id_for_engine(&config.engine).to_string());
eprintln!(
"[live] starting session: engine={}, model={}, language={:?}, save_audio={}",
config.engine,
model_id,
config.language,
config.save_audio
);
ensure_model_loaded(&state, &config.engine, &model_id).await?;
let session_id = live_state
.next_session_id
.fetch_add(1, Ordering::Relaxed)
.saturating_add(1);
let stop_flag = Arc::new(AtomicBool::new(false));
let engine = pick_engine(&state, &config.engine)?;
let output_folder = config.output_folder.clone();
let worker_stop = stop_flag.clone();
let worker_status = status_channel.clone();
let worker_results = result_channel.clone();
let handle = tokio::task::spawn_blocking(move || {
run_live_session(
session_id,
engine,
config,
worker_results,
worker_status,
worker_stop,
)
});
*live_state.running.lock().unwrap() = Some(RunningLiveSession {
id: session_id,
output_folder,
stop_flag,
handle,
status_channel,
});
Ok(StartLiveTranscriptionResponse { session_id })
}
#[tauri::command]
pub async fn stop_live_transcription_session(
app: tauri::AppHandle,
live_state: tauri::State<'_, LiveTranscriptionState>,
session_id: u64,
) -> Result<StopLiveTranscriptionResponse, String> {
let running = live_state.running.lock().unwrap().take();
let Some(running) = running else {
return Err("No live transcription session is running".into());
};
if running.id != session_id {
*live_state.running.lock().unwrap() = Some(running);
return Err(format!("Session {session_id} is not active"));
}
running.stop_flag.store(true, Ordering::Relaxed);
let summary = running
.handle
.await
.map_err(|e| format!("Live session task failed: {e}"))??;
let audio_path = if let Some(samples) = summary.audio_samples {
Some(
persist_audio_samples(&app, samples, running.output_folder.clone())
.await?,
)
} else {
None
};
let response = StopLiveTranscriptionResponse {
session_id: summary.session_id,
audio_path: audio_path.clone(),
dropped_audio_ms: summary.dropped_audio_ms,
};
let _ = running.status_channel.send(LiveStatusMessage::Finished {
session_id: summary.session_id,
audio_path,
dropped_audio_ms: summary.dropped_audio_ms,
});
Ok(response)
}
fn pick_engine(
state: &AppState,
engine: &str,
) -> Result<Arc<LocalEngine>, String> {
match engine {
"whisper" => Ok(state.whisper_engine.clone()),
"parakeet" => Ok(state.parakeet_engine.clone()),
other => Err(format!("Unknown engine: {other}")),
}
}
fn run_live_session(
session_id: u64,
engine: Arc<LocalEngine>,
config: StartLiveTranscriptionConfig,
result_channel: Channel<LiveResultMessage>,
status_channel: Channel<LiveStatusMessage>,
stop_flag: Arc<AtomicBool>,
) -> Result<LiveSessionSummary, String> {
let (capture, rx) = match config.microphone_device.as_deref() {
Some(name) if !name.is_empty() => MicrophoneCapture::start_with_device(name),
_ => MicrophoneCapture::start(),
}
.map_err(|e| e.to_string())?;
let _capture = capture;
let mut resampler: Option<StreamingResampler> = None;
let mut capture_buffer: Vec<f32> = Vec::new();
let mut kept_audio = if config.save_audio {
Some(Vec::new())
} else {
None
};
let mut buffer_start_sample: u64 = 0;
let mut dropped_audio_ms: u64 = 0;
let mut chunk_id: u32 = 0;
let mut inflight: Option<InferenceTask> = None;
let mut resampler_flushed = false;
loop {
if let Some(_done) = poll_inference(
&mut inflight,
session_id,
&config,
&result_channel,
&status_channel,
)? {}
match rx.recv_timeout(Duration::from_millis(25)) {
Ok(chunk) => {
let mono = downmix_chunk(chunk.samples, chunk.channels as usize);
let resampler = match &mut resampler {
Some(resampler) => resampler,
None => {
resampler = Some(
StreamingResampler::new(chunk.sample_rate)
.map_err(|e| e.to_string())?,
);
resampler.as_mut().expect("resampler just set")
}
};
let resampled =
resampler.push_samples(&mono).map_err(|e| e.to_string())?;
append_resampled_audio(
&mut capture_buffer,
&mut kept_audio,
&resampled,
);
}
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {}
Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
let message =
"Microphone capture disconnected unexpectedly".to_string();
let _ = status_channel.send(LiveStatusMessage::Error {
session_id,
message: message.clone(),
});
return Err(message);
}
}
if inflight.is_some() && capture_buffer.len() > MAX_PENDING_SAMPLES {
let overflow = capture_buffer.len() - MAX_PENDING_SAMPLES;
capture_buffer.drain(..overflow);
buffer_start_sample = buffer_start_sample.saturating_add(overflow as u64);
dropped_audio_ms = dropped_audio_ms
.saturating_add((overflow as u64 * 1000) / WHISPER_SAMPLE_RATE as u64);
let _ = status_channel.send(LiveStatusMessage::Overload {
session_id,
dropped_audio_ms,
message: "Kon dropped older audio to keep live dictation responsive".into(),
});
}
let stopping = stop_flag.load(Ordering::Relaxed);
if stopping && !resampler_flushed {
if let Some(resampler) = &mut resampler {
let tail = resampler.flush().map_err(|e| e.to_string())?;
append_resampled_audio(&mut capture_buffer, &mut kept_audio, &tail);
}
resampler_flushed = true;
}
if inflight.is_none() {
if let Some(task) = maybe_dispatch_chunk(
&engine,
&config,
&mut capture_buffer,
&mut buffer_start_sample,
&mut chunk_id,
stopping,
&status_channel,
session_id,
) {
inflight = Some(task);
continue;
}
if stopping && resampler_flushed {
break;
}
}
}
while inflight.is_some() {
poll_inference(
&mut inflight,
session_id,
&config,
&result_channel,
&status_channel,
)?;
thread::sleep(Duration::from_millis(10));
}
Ok(LiveSessionSummary {
session_id,
dropped_audio_ms,
audio_samples: kept_audio,
})
}
fn append_resampled_audio(
capture_buffer: &mut Vec<f32>,
kept_audio: &mut Option<Vec<f32>>,
resampled: &[f32],
) {
if resampled.is_empty() {
return;
}
capture_buffer.extend_from_slice(resampled);
if let Some(kept_audio) = kept_audio {
kept_audio.extend_from_slice(resampled);
}
}
fn maybe_dispatch_chunk(
engine: &Arc<LocalEngine>,
config: &StartLiveTranscriptionConfig,
capture_buffer: &mut Vec<f32>,
buffer_start_sample: &mut u64,
chunk_id: &mut u32,
stopping: bool,
status_channel: &Channel<LiveStatusMessage>,
session_id: u64,
) -> Option<InferenceTask> {
let target_len = if capture_buffer.len() >= CHUNK_SAMPLES {
CHUNK_SAMPLES
} else if stopping && capture_buffer.len() >= FINAL_CHUNK_MIN_SAMPLES {
capture_buffer.len()
} else {
return None;
};
let trim_before_secs = if *chunk_id > 0 && !stopping && target_len > OVERLAP_SAMPLES {
OVERLAP_SAMPLES as f64 / WHISPER_SAMPLE_RATE as f64
} else {
0.0
};
let speech_window = if trim_before_secs > 0.0 {
&capture_buffer[OVERLAP_SAMPLES..target_len]
} else {
&capture_buffer[..target_len]
};
if !has_enough_speech(speech_window) {
let skipped_ms =
(target_len as u64 * 1000) / WHISPER_SAMPLE_RATE as u64;
eprintln!(
"[live] session {session_id}: skipped {skipped_ms}ms chunk as near-silence"
);
let _ = status_channel.send(LiveStatusMessage::Warning {
session_id,
message: format!(
"Skipped {skipped_ms}ms of near-silent audio. If this keeps happening, try a louder mic level or move closer to the microphone."
),
});
capture_buffer.drain(..target_len);
*buffer_start_sample = buffer_start_sample.saturating_add(target_len as u64);
return None;
}
*chunk_id = chunk_id.saturating_add(1);
let current_chunk_id = *chunk_id;
let chunk_start_sample = *buffer_start_sample;
let duration_secs = target_len as f64 / WHISPER_SAMPLE_RATE as f64;
let chunk_samples = capture_buffer[..target_len].to_vec();
eprintln!(
"[live] session {session_id}: dispatching chunk {} ({duration_secs:.2}s, {} samples)",
current_chunk_id,
chunk_samples.len()
);
let advance_by = if stopping || target_len < CHUNK_SAMPLES {
target_len
} else {
target_len.saturating_sub(OVERLAP_SAMPLES)
};
capture_buffer.drain(..advance_by);
*buffer_start_sample = buffer_start_sample.saturating_add(advance_by as u64);
let options = TranscriptionOptions {
language: config.language.clone(),
initial_prompt: config.initial_prompt.clone(),
};
let engine = engine.clone();
let (tx, rx) = std::sync::mpsc::channel();
thread::spawn(move || {
let audio = AudioSamples::mono_16khz(chunk_samples);
let started = Instant::now();
let result = engine
.transcribe_sync(&audio, &options)
.map(|mut timed| {
timed.inference_ms = started.elapsed().as_millis() as u64;
timed
})
.map_err(|e| e.to_string());
let _ = tx.send(result);
});
Some(InferenceTask {
chunk_id: current_chunk_id,
chunk_start_sample,
trim_before_secs,
duration_secs,
rx,
})
}
fn poll_inference(
inflight: &mut Option<InferenceTask>,
session_id: u64,
config: &StartLiveTranscriptionConfig,
result_channel: &Channel<LiveResultMessage>,
status_channel: &Channel<LiveStatusMessage>,
) -> Result<Option<bool>, String> {
let Some(task) = inflight else {
return Ok(None);
};
match task.rx.try_recv() {
Ok(Ok(timed)) => {
let mut segments: Vec<Segment> =
timed.transcript.segments().to_vec();
trim_overlap_segments(&mut segments, task.trim_before_secs);
post_process_segments(
&mut segments,
&PostProcessOptions {
remove_fillers: config.remove_fillers,
british_english: config.british_english,
anti_hallucination: config.anti_hallucination,
format_mode: FormatMode::parse(&config.format_mode),
},
);
let segment_count = segments.len();
result_channel
.send(LiveResultMessage {
session_id,
chunk_id: task.chunk_id,
chunk_start_secs: task.chunk_start_sample as f64
/ WHISPER_SAMPLE_RATE as f64,
duration: task.duration_secs,
language: timed.transcript.language().to_string(),
inference_ms: timed.inference_ms,
segments,
})
.map_err(|e| e.to_string())?;
eprintln!(
"[live] session {session_id}: delivered chunk {} with {} segments in {}ms",
task.chunk_id,
segment_count,
timed.inference_ms
);
*inflight = None;
Ok(Some(true))
}
Ok(Err(err)) => {
eprintln!("[live] session {session_id}: inference error: {err}");
*inflight = None;
let _ = status_channel.send(LiveStatusMessage::Error {
session_id,
message: err.clone(),
});
Err(err)
}
Err(std::sync::mpsc::TryRecvError::Empty) => Ok(Some(false)),
Err(std::sync::mpsc::TryRecvError::Disconnected) => {
*inflight = None;
let message = "Inference worker disconnected unexpectedly".to_string();
eprintln!("[live] session {session_id}: {message}");
let _ = status_channel.send(LiveStatusMessage::Error {
session_id,
message: message.clone(),
});
Err(message)
}
}
}
fn trim_overlap_segments(segments: &mut Vec<Segment>, trim_before_secs: f64) {
if trim_before_secs <= 0.0 {
return;
}
segments.retain(|segment| segment.end > trim_before_secs);
for segment in segments.iter_mut() {
if segment.start < trim_before_secs {
segment.start = trim_before_secs;
}
}
}
fn has_enough_speech(samples: &[f32]) -> bool {
if samples.is_empty() {
return false;
}
let chunk_peak = samples
.iter()
.map(|sample| sample.abs())
.fold(0.0_f32, f32::max);
if chunk_peak < FLATLINE_PEAK_THRESHOLD {
return false;
}
let mut speech_frames = 0usize;
for frame in samples.chunks(SPEECH_FRAME_SAMPLES) {
let len = frame.len().max(1) as f32;
let rms = (frame.iter().map(|sample| sample * sample).sum::<f32>() / len)
.sqrt();
let peak = frame
.iter()
.map(|sample| sample.abs())
.fold(0.0_f32, f32::max);
if rms >= RMS_SPEECH_THRESHOLD || peak >= PEAK_SPEECH_THRESHOLD {
speech_frames += 1;
}
}
speech_frames >= MIN_SPEECH_FRAMES
}
fn downmix_chunk(samples: Vec<f32>, channels: usize) -> Vec<f32> {
if channels <= 1 {
return samples;
}
samples
.chunks(channels)
.map(|frame| frame.iter().sum::<f32>() / channels as f32)
.collect()
}