From 8bec23c184c2ce5d2c29366859139bce3fbcdd92 Mon Sep 17 00:00:00 2001 From: Sergei Solo Date: Thu, 23 Jul 2026 14:50:39 +0400 Subject: [PATCH] feat(audio): native PipeWire capture backend on Linux On Linux, capture is pinned to cpal's ALSA host (`get_cpal_host` -> `HostId::Alsa`), so Handy's microphone stream is an ALSA "default" client. PipeWire cannot see it as a first-class node: there is no per-application capture volume, no routing, and device enumeration only ever exposes "default" instead of the graph's real sources. Add a native PipeWire capture backend (pipewire-rs 0.10) that registers the microphone stream as a real PipeWire node, so it appears in `wpctl status` with its own settable volume and is routable per-app. It runs the PipeWire main loop on a dedicated thread (all non-Send pw objects stay on it), downmixes interleaved F32 to mono, and feeds the existing resampler -> VAD -> consumer pipeline unchanged via the shared Cmd/AudioChunk protocol. A small `Recorder` seam selects the backend: on Linux it prefers PipeWire and falls back to the cpal/ALSA path if PipeWire setup fails; non-Linux targets compile cpal-only and are unaffected. cpal stays fully compiled and available as the fallback. The MVP captures the default source (reproducing today's "default" behaviour). PipeWire-native device enumeration/selection is left as a follow-up (TODO in recorder_backend.rs, where a selected mic would map to a node.name passed through TARGET_OBJECT). --- src-tauri/Cargo.lock | 159 ++++++- src-tauri/Cargo.toml | 5 + src-tauri/src/audio_toolkit/audio/mod.rs | 6 + .../audio_toolkit/audio/pipewire_recorder.rs | 421 ++++++++++++++++++ src-tauri/src/audio_toolkit/audio/recorder.rs | 70 ++- .../audio_toolkit/audio/recorder_backend.rs | 151 +++++++ src-tauri/src/audio_toolkit/mod.rs | 5 +- src-tauri/src/managers/audio.rs | 56 +-- 8 files changed, 834 insertions(+), 39 deletions(-) create mode 100644 src-tauri/src/audio_toolkit/audio/pipewire_recorder.rs create mode 100644 src-tauri/src/audio_toolkit/audio/recorder_backend.rs diff --git a/src-tauri/Cargo.lock b/src-tauri/Cargo.lock index 8d122e3c90..7f3aaba12c 100644 --- a/src-tauri/Cargo.lock +++ b/src-tauri/Cargo.lock @@ -97,6 +97,16 @@ dependencies = [ "libc", ] +[[package]] +name = "annotate-snippets" +version = "0.11.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "710e8eae58854cdc1790fcb56cca04d712a17be849eeb81da2a724bf4bae2bc4" +dependencies = [ + "anstyle", + "unicode-width", +] + [[package]] name = "anstream" version = "0.6.21" @@ -404,6 +414,25 @@ dependencies = [ "virtue", ] +[[package]] +name = "bindgen" +version = "0.72.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "993776b509cfb49c750f11b8f07a46fa23e0a1386ffc01fb1e7d343efc387895" +dependencies = [ + "annotate-snippets", + "bitflags 2.11.0", + "cexpr", + "clang-sys", + "itertools", + "proc-macro2", + "quote", + "regex", + "rustc-hash", + "shlex", + "syn 2.0.117", +] + [[package]] name = "bit-set" version = "0.8.0" @@ -702,6 +731,15 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6d43a04d8753f35258c91f8ec639f792891f748a1edbd759cf1dcea3382ad83c" +[[package]] +name = "cexpr" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6fac387a98bb7c37292057cffc56d62ecb629900026402633ae9160df93a8766" +dependencies = [ + "nom 7.1.3", +] + [[package]] name = "cfb" version = "0.7.3" @@ -759,6 +797,17 @@ dependencies = [ "windows-link 0.2.1", ] +[[package]] +name = "clang-sys" +version = "1.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b023947811758c97c59bf9d1c188fd619ad4718dcaa767947df1cadb14f39f4" +dependencies = [ + "glob", + "libc", + "libloading 0.8.9", +] + [[package]] name = "clap" version = "4.5.60" @@ -901,6 +950,12 @@ dependencies = [ "version_check", ] +[[package]] +name = "cookie-factory" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9885fa71e26b8ab7855e2ec7cae6e9b380edff76cd052e07c683a0319d51b3a2" + [[package]] name = "cookie_store" version = "0.22.1" @@ -1505,6 +1560,12 @@ version = "1.0.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555" +[[package]] +name = "either" +version = "1.16.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e" + [[package]] name = "embed-resource" version = "3.0.6" @@ -1557,7 +1618,7 @@ dependencies = [ "foreign-types-shared 0.3.1", "libc", "log", - "nom", + "nom 8.0.0", "objc2", "objc2-app-kit", "objc2-foundation", @@ -2466,6 +2527,7 @@ dependencies = [ "log", "natural", "once_cell", + "pipewire", "rdev", "regex", "reqwest 0.12.28", @@ -3058,6 +3120,15 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" +[[package]] +name = "itertools" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "413ee7dfc52ee1a4949ceeb7dbc8a33f2d6c088194d9f922fb8318faf1f01186" +dependencies = [ + "either", +] + [[package]] name = "itoa" version = "1.0.17" @@ -3196,7 +3267,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6e9ec52138abedcc58dc17a7c6c0c00a2bdb4f3427c7f63fa97fd0d859155caf" dependencies = [ "gtk-sys", - "libloading", + "libloading 0.7.4", "once_cell", ] @@ -3225,6 +3296,16 @@ dependencies = [ "winapi", ] +[[package]] +name = "libloading" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d7c4b02199fee7c5d21a5ae7d8cfa79a6ef5bb2fc834d6e9058e89c825efdc55" +dependencies = [ + "cfg-if", + "windows-link 0.2.1", +] + [[package]] name = "libm" version = "0.2.16" @@ -3242,6 +3323,33 @@ dependencies = [ "redox_syscall 0.7.1", ] +[[package]] +name = "libspa" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2909f3be29d674e7f10604aff18d1bbe1bb03c4cd61c8a8ba19c0b1d162f7d4e" +dependencies = [ + "bitflags 2.11.0", + "cc", + "cookie-factory", + "libc", + "libspa-sys", + "nom 8.0.0", + "rustix", + "system-deps 7.0.7", +] + +[[package]] +name = "libspa-sys" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69ad52764fca54818486f3cf75afec844d1f1a1568c24dcee25d41b1ab007dda" +dependencies = [ + "bindgen", + "cc", + "system-deps 7.0.7", +] + [[package]] name = "libsqlite3-sys" version = "0.35.0" @@ -3421,6 +3529,12 @@ dependencies = [ "walkdir", ] +[[package]] +name = "minimal-lexical" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" + [[package]] name = "minisign-verify" version = "0.2.4" @@ -3598,6 +3712,16 @@ version = "0.1.14" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72ef4a56884ca558e5ddb05a1d1e7e1bfd9a68d9ed024c21704cc98872dae1bb" +[[package]] +name = "nom" +version = "7.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a" +dependencies = [ + "memchr", + "minimal-lexical", +] + [[package]] name = "nom" version = "8.0.0" @@ -4152,7 +4276,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7d8fae84b431384b68627d0f9b3b1245fcf9f46f6c0e3dc902e9dce64edd1967" dependencies = [ "libc", - "windows-sys 0.48.0", + "windows-sys 0.61.2", ] [[package]] @@ -4463,6 +4587,31 @@ dependencies = [ "futures-io", ] +[[package]] +name = "pipewire" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8585aba8a52ad74ccc633b8e293c1dc4277976bd5d510b925533f34fd6685f38" +dependencies = [ + "bitflags 2.11.0", + "libc", + "libspa", + "libspa-sys", + "pipewire-sys", + "rustix", +] + +[[package]] +name = "pipewire-sys" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2089f245b548723e60325773c27f586b7a2372c79ea941b246cd0d654706adc" +dependencies = [ + "bindgen", + "libspa-sys", + "system-deps 7.0.7", +] + [[package]] name = "pkg-config" version = "0.3.32" @@ -7246,7 +7395,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b8765b90061cba6c22b5831f675da109ae5561588290f9fa2317adab2714d5a6" dependencies = [ "memchr", - "nom", + "nom 8.0.0", "petgraph", ] @@ -7961,7 +8110,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.48.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index ea17e8e6a6..ffcdb1f7e2 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -143,6 +143,11 @@ transcribe-cpp = { version = "0.1.3", default-features = false, features = ["met [target.'cfg(target_os = "linux")'.dependencies] gtk-layer-shell = { version = "0.8", features = ["v0_6"] } gtk = "0.18" +# Native PipeWire capture backend (see audio/pipewire_recorder.rs). The +# `v0_3_44` feature unlocks newer property keys such as TARGET_OBJECT used to +# pin capture to a specific source node. Linux-only: the cpal/ALSA path stays +# the fallback and remains the only backend on Windows/macOS. +pipewire = { version = "0.10.0", features = ["v0_3_44"] } transcribe-cpp = { version = "0.1.3", default-features = false, features = [ "dynamic-backends", "vulkan", diff --git a/src-tauri/src/audio_toolkit/audio/mod.rs b/src-tauri/src/audio_toolkit/audio/mod.rs index 40ef7f47a6..854ed1d6c2 100644 --- a/src-tauri/src/audio_toolkit/audio/mod.rs +++ b/src-tauri/src/audio_toolkit/audio/mod.rs @@ -1,6 +1,9 @@ // Re-export all audio components mod device; +#[cfg(target_os = "linux")] +mod pipewire_recorder; mod recorder; +mod recorder_backend; mod resampler; mod utils; mod visualizer; @@ -9,6 +12,9 @@ pub use device::{list_input_devices, list_output_devices, CpalDeviceInfo}; pub use recorder::{ is_microphone_access_denied, is_no_input_device_error, AudioRecorder, VadPolicy, }; +// Shared parts used by the `Recorder` seam / manager to build backends. +pub(crate) use recorder::{AudioFrameCallback, VadConfig}; +pub use recorder_backend::Recorder; pub use resampler::FrameResampler; pub use utils::{read_wav_samples, save_wav_file, verify_wav_file}; pub use visualizer::AudioVisualiser; diff --git a/src-tauri/src/audio_toolkit/audio/pipewire_recorder.rs b/src-tauri/src/audio_toolkit/audio/pipewire_recorder.rs new file mode 100644 index 0000000000..a43ef2f049 --- /dev/null +++ b/src-tauri/src/audio_toolkit/audio/pipewire_recorder.rs @@ -0,0 +1,421 @@ +//! Native PipeWire microphone capture backend (Linux-only). +//! +//! Why this exists: the cpal path pins capture to the ALSA host +//! (`get_cpal_host()` -> `HostId::Alsa`), so Handy's mic stream is an ALSA +//! "default" client that PipeWire cannot see as a first-class node — no per-app +//! volume, no routing, no node selection. This backend registers Handy's +//! capture as a real PipeWire node instead, so it shows up in `wpctl status` +//! with its own settable volume and is routable per-app. +//! +//! Threading / ownership model (the important part): +//! * PipeWire objects (MainLoop, Context, Core, Stream, listeners) are NOT +//! `Send`. We therefore construct and own ALL of them on ONE dedicated +//! thread (`run_pipewire_loop`) and never move them off it. Cross-thread +//! communication happens only through channels. +//! * A SECOND thread runs the backend-neutral `run_consumer` (resample -> VAD +//! -> buffer) — the exact same consumer the cpal recorder uses. The RT +//! `process` callback (on PipeWire's data thread) downmixes to mono and +//! pushes `AudioChunk::Samples` into an `mpsc` channel that `run_consumer` +//! drains. This is the seam that lets us reuse the whole pipeline below the +//! mono-`Vec` producer without reimplementing any of it. +//! * Start/Stop/Shutdown reuse the shared `Cmd` protocol: `cmd_tx` talks to +//! `run_consumer` identically to the cpal backend. A separate +//! `pipewire::channel` sender wakes the loop thread to quit it on `close()`. +//! +//! Targets the `pipewire` crate 0.10 with the `v0_3_44` feature (for +//! `TARGET_OBJECT`). Non-obvious calls are commented inline. + +use std::io::Cursor; +use std::mem; +use std::sync::{ + atomic::{AtomicBool, Ordering}, + mpsc, Arc, +}; +use std::time::Instant; + +use pipewire as pw; +use pw::{properties::properties, spa}; + +use spa::param::audio::{AudioFormat, AudioInfoRaw}; +use spa::param::format::{MediaSubtype, MediaType}; +use spa::param::format_utils; +use spa::param::ParamType; +use spa::pod::{serialize::PodSerializer, Object, Pod, Value}; +use spa::utils::{Direction, SpaTypes}; + +use super::recorder::{run_consumer, AudioChunk, AudioFrameCallback, Cmd, LevelCallback, VadConfig}; +use super::VadPolicy; + +/// Human-readable identity for the capture node in `wpctl status`. +const APP_NAME: &str = "Handy"; +const NODE_NAME: &str = "handy-capture"; + +/// We pin the negotiated capture rate so the consumer's `FrameResampler` +/// (rate -> 16 kHz) can be constructed up front, before PipeWire's async format +/// negotiation completes. PipeWire's adapter resamples the source to this rate +/// for us. 48 kHz is the near-universal graph rate, so this is usually a no-op +/// conversion. Channels are left unpinned and downmixed in `process`. +const PIPEWIRE_CAPTURE_RATE: u32 = 48_000; + +/// Native PipeWire capture backend. Public surface intentionally mirrors the +/// cpal `AudioRecorder` (`from_parts`/`open`/`start`/`stop`/`close`) so the +/// `Recorder` seam can drive either backend the same way. +pub struct PipeWireRecorder { + /// Shared VAD + callbacks handed to `run_consumer`. Cloned per `open`. + vad: Option, + level_cb: Option, + audio_cb: Option, + + /// Talks to `run_consumer` (Start/Stop/Shutdown) — same protocol as cpal. + cmd_tx: Option>, + /// Wakes the loop thread to quit it on `close()`. `pipewire::channel`'s + /// sender is `Send`; its receiver is attached to the (non-`Send`) loop. + quit_tx: Option>, + /// The dedicated PipeWire loop thread and the consumer thread. + pw_handle: Option>, + consumer_handle: Option>, +} + +impl PipeWireRecorder { + /// Build from already-shared parts. See `AudioRecorder::from_parts`. + pub(crate) fn from_parts( + vad: Option, + level_cb: Option, + audio_cb: Option, + ) -> Self { + PipeWireRecorder { + vad, + level_cb, + audio_cb, + cmd_tx: None, + quit_tx: None, + pw_handle: None, + consumer_handle: None, + } + } + + /// Open the capture stream. `target_node` optionally pins capture to a + /// specific source by its `node.name` (via `TARGET_OBJECT`); `None` + /// autoconnects to the system default source — reproducing today's cpal + /// "default" behaviour. + /// + /// Returns `Err` if the PipeWire connection/stream setup fails (e.g. no + /// PipeWire session running), which is the signal the `Recorder` seam uses + /// to fall back to the cpal/ALSA backend. + pub fn open(&mut self, target_node: Option) -> Result<(), Box> { + if self.pw_handle.is_some() { + return Ok(()); // already open + } + + // Producer (pipewire data thread) -> consumer thread. + let (sample_tx, sample_rx) = mpsc::channel::(); + // Control channel to run_consumer (Start/Stop/Shutdown). + let (cmd_tx, cmd_rx) = mpsc::channel::(); + // One-shot init handshake so open() can report setup success/failure. + let (init_tx, init_rx) = mpsc::sync_channel::>(1); + // Cross-thread wake to quit the loop on close(). Attached to the loop + // inside the pipewire thread; sender kept here. + let (quit_tx, quit_rx) = pw::channel::channel::<()>(); + + // Shared stop flag: run_consumer flips it on Stop so the producer emits + // EndOfStream (mirrors the cpal backend's flag semantics exactly). + let stop_flag = Arc::new(AtomicBool::new(false)); + + // ---- consumer thread: the reused, backend-neutral pipeline --------- // + let vad = self.vad.clone(); + let level_cb = self.level_cb.clone(); + let audio_cb = self.audio_cb.clone(); + let stop_flag_consumer = Arc::clone(&stop_flag); + let consumer_handle = std::thread::spawn(move || { + run_consumer( + PIPEWIRE_CAPTURE_RATE, + vad, + sample_rx, + cmd_rx, + level_cb, + audio_cb, + stop_flag_consumer, + Instant::now(), + ); + }); + + // ---- pipewire loop thread: owns all non-Send pw objects ------------ // + let stop_flag_pw = Arc::clone(&stop_flag); + let pw_handle = std::thread::spawn(move || { + run_pipewire_loop(sample_tx, stop_flag_pw, target_node, init_tx, quit_rx); + }); + + match init_rx.recv() { + Ok(Ok(())) => { + self.cmd_tx = Some(cmd_tx); + self.quit_tx = Some(quit_tx); + self.pw_handle = Some(pw_handle); + self.consumer_handle = Some(consumer_handle); + Ok(()) + } + Ok(Err(error_message)) => { + // Setup failed: the pw thread has already returned, dropping + // sample_tx, which ends the consumer's recv loop. Join both. + let _ = pw_handle.join(); + let _ = consumer_handle.join(); + Err(Box::new(std::io::Error::other(error_message))) + } + Err(recv_error) => { + let _ = pw_handle.join(); + let _ = consumer_handle.join(); + Err(Box::new(std::io::Error::other(format!( + "PipeWire capture worker died during init: {recv_error}" + )))) + } + } + } + + pub fn start(&self, vad_policy: VadPolicy) -> Result<(), Box> { + if let Some(tx) = &self.cmd_tx { + tx.send(Cmd::Start(vad_policy, Instant::now()))?; + } + Ok(()) + } + + pub fn stop(&self) -> Result, Box> { + let (resp_tx, resp_rx) = mpsc::channel(); + if let Some(tx) = &self.cmd_tx { + tx.send(Cmd::Stop(resp_tx))?; + } + Ok(resp_rx.recv()?) // wait for the buffered samples + } + + pub fn close(&mut self) -> Result<(), Box> { + // Best-effort clean shutdown of the consumer first… + if let Some(tx) = self.cmd_tx.take() { + let _ = tx.send(Cmd::Shutdown); + } + // …then quit the loop thread. Once it returns, the stream/listener and + // the producer's sample_tx drop, which also unblocks the consumer if it + // hadn't already seen Shutdown. Either path terminates both threads. + if let Some(quit) = self.quit_tx.take() { + let _ = quit.send(()); + } + if let Some(h) = self.pw_handle.take() { + let _ = h.join(); + } + if let Some(h) = self.consumer_handle.take() { + let _ = h.join(); + } + Ok(()) + } +} + +impl Drop for PipeWireRecorder { + fn drop(&mut self) { + let _ = self.close(); + } +} + +/// Per-stream state owned by the listener on the pipewire loop thread. Holds the +/// producer end of the sample channel and the shared stop flag. Not `Send` is +/// fine — it never leaves the loop thread. +struct CaptureState { + /// Negotiated raw-audio format; channels filled in by `param_changed`. + format: AudioInfoRaw, + /// Producer end into `run_consumer`. + sample_tx: mpsc::Sender, + /// Shared with run_consumer; true means "recording stopped, drain". + stop_flag: Arc, + /// Whether we've already emitted the EndOfStream sentinel for this stop. + eos_sent: bool, + /// Reused mono scratch buffer to avoid per-cycle allocation. + scratch: Vec, +} + +/// Body of the dedicated pipewire thread. Constructs the full MainLoop -> Core +/// -> Stream graph HERE (nothing pipewire crosses a thread boundary), reports +/// setup success/failure through `init_tx`, then blocks in `mainloop.run()` +/// until `close()` sends a quit. +fn run_pipewire_loop( + sample_tx: mpsc::Sender, + stop_flag: Arc, + target_node: Option, + init_tx: mpsc::SyncSender>, + quit_rx: pw::channel::Receiver<()>, +) { + // All fallible setup runs in this closure so a single `?` chain can report + // failure via init_tx (the seam's fallback trigger). + let setup = (|| -> Result<(), pw::Error> { + pw::init(); + + // MainLoop drives this thread. 0.10 exposes the ref-counted variants. + let mainloop = pw::main_loop::MainLoopRc::new(None)?; + let context = pw::context::ContextRc::new(&mainloop, None)?; + let core = context.connect_rc(None)?; + + // Attach the quit receiver to this loop. The callback runs ON the loop + // thread, so it can safely stop the (non-Send) loop. Keep the returned + // guard alive for the whole run — dropping it detaches the receiver. + let _quit_guard = quit_rx.attach(mainloop.loop_(), { + let mainloop = mainloop.clone(); + move |_| mainloop.quit() + }); + + // Advertise ourselves as a Communication-role audio capture stream with + // a stable, human-readable identity. The Communication role matches a + // voice-capture use case and lets session policy route it accordingly. + let mut props = properties! { + *pw::keys::MEDIA_TYPE => "Audio", + *pw::keys::MEDIA_CATEGORY => "Capture", + *pw::keys::MEDIA_ROLE => "Communication", + *pw::keys::APP_NAME => APP_NAME, + *pw::keys::NODE_NAME => NODE_NAME, + }; + + // Optionally pin to a specific source by node.name. TARGET_OBJECT is the + // modern targeting property (needs the v0_3_44 feature); as a stream prop + // it is equivalent to passing a target to connect(). + if let Some(target) = target_node { + if !target.is_empty() { + props.insert(*pw::keys::TARGET_OBJECT, target); + } + } + + let stream = pw::stream::StreamBox::new(&core, APP_NAME, props)?; + + // The listener owns our CaptureState; both callbacks get `&mut` to it. + // `_listener` must outlive the run — dropping it unregisters callbacks. + let _listener = stream + .add_local_listener_with_user_data(CaptureState { + format: AudioInfoRaw::default(), + sample_tx, + stop_flag, + eos_sent: false, + scratch: Vec::new(), + }) + // Read back the server-chosen concrete format (we only pinned the + // sample format + rate; channels are negotiated). + .param_changed(|_stream, state, id, param| { + let Some(param) = param else { + return; + }; + if id != ParamType::Format.as_raw() { + return; + } + let Ok((media_type, media_subtype)) = format_utils::parse_format(param) else { + return; + }; + if media_type != MediaType::Audio || media_subtype != MediaSubtype::Raw { + return; + } + if state.format.parse(param).is_err() { + log::warn!("PipeWire: failed to parse negotiated audio format"); + return; + } + log::info!( + "PipeWire capture negotiated: rate={} Hz, channels={}", + state.format.rate(), + state.format.channels() + ); + }) + // RT callback (on the data thread): downmix interleaved F32 frames to + // mono and push the SAME AudioChunk the cpal producer pushes. + .process(|stream, state| { + let Some(mut buffer) = stream.dequeue_buffer() else { + return; + }; + let datas = buffer.datas_mut(); + if datas.is_empty() { + return; + } + + // Honour the shared stop flag exactly like the cpal producer: + // once stopped, emit EndOfStream once and then go silent so the + // consumer's drain terminates deterministically. + if state.stop_flag.load(Ordering::Relaxed) { + if !state.eos_sent { + let _ = state.sample_tx.send(AudioChunk::EndOfStream); + state.eos_sent = true; + } + return; + } + state.eos_sent = false; + + let channels = state.format.channels().max(1) as usize; + let data = &mut datas[0]; + let stride = mem::size_of::(); + // chunk().size() is the VALID byte count (may be < allocated). + let byte_len = data.chunk().size() as usize; + let sample_count = byte_len / stride; + + if let Some(raw) = data.data() { + let frames = sample_count / channels; + state.scratch.clear(); + state.scratch.reserve(frames); + for frame in 0..frames { + // Downmix this frame's interleaved channels by averaging. + let mut acc = 0.0f32; + for ch in 0..channels { + let sample_index = frame * channels + ch; + let start = sample_index * stride; + let end = start + stride; + // Little-endian F32 (we negotiated F32LE). + let bytes: [u8; 4] = match raw[start..end].try_into() { + Ok(b) => b, + Err(_) => return, + }; + acc += f32::from_le_bytes(bytes); + } + state.scratch.push(acc / channels as f32); + } + if !state.scratch.is_empty() { + let _ = state + .sample_tx + .send(AudioChunk::Samples(state.scratch.clone())); + } + } + }) + .register()?; + + // EnumFormat POD: accept F32LE at our pinned rate; leave channels free. + // (AudioInfoRaw -> Vec `Into` + the Object/Value/PodSerializer + // shape match the pipewire-rs upstream audio-capture example.) + let mut audio_info = AudioInfoRaw::new(); + audio_info.set_format(AudioFormat::F32LE); + audio_info.set_rate(PIPEWIRE_CAPTURE_RATE); + let obj = Object { + type_: SpaTypes::ObjectParamFormat.as_raw(), + id: ParamType::EnumFormat.as_raw(), + properties: audio_info.into(), + }; + let values: Vec = + PodSerializer::serialize(Cursor::new(Vec::new()), &Value::Object(obj)) + .expect("failed to serialize EnumFormat POD") + .0 + .into_inner(); + let mut params = [Pod::from_bytes(&values).expect("serialized POD is malformed")]; + + // Connect as INPUT (we consume audio from a source): + // - AUTOCONNECT: let the session manager wire us to the source + // (default, or our TARGET_OBJECT if set). + // - MAP_BUFFERS: CPU-mapped buffers so `data()` returns real memory. + // - RT_PROCESS: run `process` on the realtime data thread. + stream.connect( + Direction::Input, + None, + pw::stream::StreamFlags::AUTOCONNECT + | pw::stream::StreamFlags::MAP_BUFFERS + | pw::stream::StreamFlags::RT_PROCESS, + &mut params, + )?; + + // Setup succeeded — hand control back to open(), then block processing + // audio until close() sends a quit. `_listener`, `stream`, `_quit_guard` + // and `core` stay owned in this scope for the whole run. + let _ = init_tx.send(Ok(())); + mainloop.run(); + Ok(()) + })(); + + // Only reached-with-Err before init_tx was sent Ok (setup failure). After a + // successful setup the closure returns Ok(()) once the loop quits. + if let Err(e) = setup { + let _ = init_tx.send(Err(format!("PipeWire capture setup failed: {e}"))); + } +} diff --git a/src-tauri/src/audio_toolkit/audio/recorder.rs b/src-tauri/src/audio_toolkit/audio/recorder.rs index 82297cb672..e1be408c87 100644 --- a/src-tauri/src/audio_toolkit/audio/recorder.rs +++ b/src-tauri/src/audio_toolkit/audio/recorder.rs @@ -19,7 +19,11 @@ use crate::audio_toolkit::{ VoiceActivityDetector, }; -enum Cmd { +/// Control protocol shared by every capture backend. The cpal `AudioRecorder` +/// and the native `PipeWireRecorder` both drive the SAME `run_consumer` loop, so +/// this type is `pub(crate)` to let the pipewire backend speak the same protocol +/// instead of duplicating the consumer/VAD/resampler pipeline. +pub(crate) enum Cmd { /// Begin capturing. Carries the send timestamp so the consumer can log how /// long the command sat in the channel (and how much audio was dropped /// before it was seen). @@ -28,7 +32,10 @@ enum Cmd { Shutdown, } -enum AudioChunk { +/// Mono `f32` chunk produced by a capture backend and consumed by +/// `run_consumer`. `pub(crate)` so the pipewire backend can push the exact same +/// chunk type the cpal producer does (see `PipeWireRecorder`). +pub(crate) enum AudioChunk { Samples(Vec), EndOfStream, } @@ -48,14 +55,33 @@ pub enum VadPolicy { /// should use. The offline and streaming policies are never active /// concurrently, so one detector is reconfigured per session (see `Cmd::Start`) /// rather than kept as two resident engines. +/// A single VAD engine plus its two hangover-tail lengths, shared by both +/// capture backends. `pub(crate)` + a public `new` so the pipewire backend and +/// the `Recorder` seam can hold and clone the same config (the detector lives +/// behind `Arc>`, so one ONNX session is shared, never duplicated). #[derive(Clone)] -struct VadConfig { +pub(crate) struct VadConfig { detector: Arc>>, offline_hangover_frames: usize, streaming_hangover_frames: usize, } impl VadConfig { + /// Build a shared VAD config from a detector and the offline/streaming + /// hangover tails. The detector is wrapped in `Arc>` so a single + /// engine can back multiple recorder backends without re-instantiating it. + pub(crate) fn new( + detector: Box, + offline_hangover_frames: usize, + streaming_hangover_frames: usize, + ) -> Self { + VadConfig { + detector: Arc::new(Mutex::new(detector)), + offline_hangover_frames, + streaming_hangover_frames, + } + } + /// Post-speech hangover tail (in 30 ms frames) for the given policy. /// `Disabled` never reaches the detector, so it maps to the offline value. fn hangover_for(&self, policy: VadPolicy) -> usize { @@ -70,6 +96,11 @@ impl VadConfig { /// policy while recording. Used to feed a live streaming transcription as audio arrives. pub type AudioFrameCallback = Arc; +/// Spectrum-level callback type (per-frame frequency buckets forwarded to the +/// UI). Aliased so both capture backends and the `Recorder` seam can pass the +/// exact same boxed callback without re-spelling the signature. +pub(crate) type LevelCallback = Arc) + Send + Sync + 'static>; + pub struct AudioRecorder { device: Option, cmd_tx: Option>, @@ -108,14 +139,35 @@ impl AudioRecorder { offline_hangover_frames: usize, streaming_hangover_frames: usize, ) -> Self { - self.vad = Some(VadConfig { - detector: Arc::new(Mutex::new(detector)), + self.vad = Some(VadConfig::new( + detector, offline_hangover_frames, streaming_hangover_frames, - }); + )); self } + /// Construct a recorder directly from already-built shared parts (VAD + + /// callbacks). This is the seam used by `Recorder` so the cpal backend and + /// the native pipewire backend can share ONE VAD engine and ONE set of + /// callbacks instead of each building its own. Mirrors what the `with_*` + /// builder chain assembles, minus the device (resolved later in `open`). + pub(crate) fn from_parts( + vad: Option, + level_cb: Option, + audio_cb: Option, + ) -> Self { + AudioRecorder { + device: None, + cmd_tx: None, + worker_handle: None, + vad, + level_cb, + audio_cb, + config_cache: Arc::new(Mutex::new(None)), + } + } + pub fn with_level_callback(mut self, cb: F) -> Self where F: Fn(Vec) + Send + Sync + 'static, @@ -513,8 +565,12 @@ mod tests { } } +/// Backend-neutral consumer: resample -> VAD -> buffer, driven by the shared +/// `Cmd`/`AudioChunk` protocol. Both the cpal `AudioRecorder` worker and the +/// native `PipeWireRecorder` spawn this on their own thread, so it is +/// `pub(crate)` and must NOT be duplicated per backend. #[allow(clippy::too_many_arguments)] -fn run_consumer( +pub(crate) fn run_consumer( in_sample_rate: u32, vad: Option, sample_rx: mpsc::Receiver, diff --git a/src-tauri/src/audio_toolkit/audio/recorder_backend.rs b/src-tauri/src/audio_toolkit/audio/recorder_backend.rs new file mode 100644 index 0000000000..af76572366 --- /dev/null +++ b/src-tauri/src/audio_toolkit/audio/recorder_backend.rs @@ -0,0 +1,151 @@ +//! `Recorder` — the thin seam that lets `AudioRecordingManager` drive either the +//! native PipeWire backend (Linux) or the cpal/ALSA backend (everywhere) through +//! one type with the SAME method surface (`open`/`start`/`stop`/`close`). +//! +//! Selection strategy (Linux): try PipeWire at `open()`; if its +//! connection/stream setup fails (no session, etc.), fall back to the existing +//! cpal path and log it. Both backends are built up front from the SAME shared +//! VAD + callbacks (VAD lives behind `Arc>`, so only one ONNX session +//! exists), but only the selected one is ever opened at a time. +//! +//! Non-Linux builds compile the cpal backend ONLY — no PipeWire code, no +//! behavioural change. cpal remains fully compiled and working on Linux too. + +use super::recorder::{AudioFrameCallback, LevelCallback, VadConfig}; +use super::{AudioRecorder, VadPolicy}; + +#[cfg(target_os = "linux")] +use super::pipewire_recorder::PipeWireRecorder; + +/// Which backend is currently open (Linux only — non-Linux is always cpal). +#[cfg(target_os = "linux")] +#[derive(Clone, Copy, PartialEq, Eq)] +enum Backend { + /// Nothing open yet, or closed. + None, + PipeWire, + Cpal, +} + +pub struct Recorder { + /// Always present: the fallback and the only backend off Linux. + cpal: AudioRecorder, + #[cfg(target_os = "linux")] + pipewire: PipeWireRecorder, + #[cfg(target_os = "linux")] + active: Backend, +} + +impl Recorder { + /// Build both backends from shared parts. See `AudioRecorder::from_parts`. + pub(crate) fn from_parts( + vad: Option, + level_cb: Option, + audio_cb: Option, + ) -> Self { + #[cfg(target_os = "linux")] + { + // Share one VAD engine + callbacks across both backends (all cheap + // to clone: VAD is Arc>, callbacks are Arc). Only one + // backend is opened at a time, so they never run concurrently. + let cpal = + AudioRecorder::from_parts(vad.clone(), level_cb.clone(), audio_cb.clone()); + let pipewire = PipeWireRecorder::from_parts(vad, level_cb, audio_cb); + Recorder { + cpal, + pipewire, + active: Backend::None, + } + } + #[cfg(not(target_os = "linux"))] + { + Recorder { + cpal: AudioRecorder::from_parts(vad, level_cb, audio_cb), + } + } + } + + /// Open the microphone. On Linux, prefer native PipeWire and fall back to + /// cpal/ALSA if PipeWire setup fails. The cpal-resolved `device` is used for + /// the cpal path; the PipeWire path currently ignores it and captures the + /// default source (see TODO). + pub fn open( + &mut self, + device: Option, + ) -> Result<(), Box> { + #[cfg(target_os = "linux")] + { + // TODO(pipewire device selection): translate a user-selected mic to + // a PipeWire `node.name` and pass it here instead of `None`. For the + // MVP we capture the default source, which reproduces today's cpal + // "default" behaviour. PipeWire-native device enumeration/selection + // (the de-cpal refactor) would slot in at this call site and in + // `device.rs`/`managers/audio.rs`. + match self.pipewire.open(None) { + Ok(()) => { + self.active = Backend::PipeWire; + log::info!("Microphone capture using native PipeWire backend"); + return Ok(()); + } + Err(e) => { + log::warn!( + "PipeWire capture unavailable ({e}); falling back to cpal/ALSA backend" + ); + } + } + self.cpal.open(device)?; + self.active = Backend::Cpal; + Ok(()) + } + #[cfg(not(target_os = "linux"))] + { + self.cpal.open(device) + } + } + + pub fn start(&self, vad_policy: VadPolicy) -> Result<(), Box> { + // Expression-block-per-cfg idiom (mirrors `get_cpal_host`): exactly one + // block survives cfg-stripping and becomes the tail expression. + #[cfg(target_os = "linux")] + { + match self.active { + Backend::PipeWire => self.pipewire.start(vad_policy), + _ => self.cpal.start(vad_policy), + } + } + #[cfg(not(target_os = "linux"))] + { + self.cpal.start(vad_policy) + } + } + + pub fn stop(&self) -> Result, Box> { + #[cfg(target_os = "linux")] + { + match self.active { + Backend::PipeWire => self.pipewire.stop(), + _ => self.cpal.stop(), + } + } + #[cfg(not(target_os = "linux"))] + { + self.cpal.stop() + } + } + + pub fn close(&mut self) -> Result<(), Box> { + #[cfg(target_os = "linux")] + { + let result = match self.active { + Backend::PipeWire => self.pipewire.close(), + _ => self.cpal.close(), + }; + self.active = Backend::None; + result + } + #[cfg(not(target_os = "linux"))] + { + self.cpal.close() + } + } +} diff --git a/src-tauri/src/audio_toolkit/mod.rs b/src-tauri/src/audio_toolkit/mod.rs index 291753a22b..4af2b97193 100644 --- a/src-tauri/src/audio_toolkit/mod.rs +++ b/src-tauri/src/audio_toolkit/mod.rs @@ -6,8 +6,11 @@ pub mod vad; pub use audio::{ is_microphone_access_denied, is_no_input_device_error, list_input_devices, list_output_devices, - read_wav_samples, save_wav_file, verify_wav_file, AudioRecorder, CpalDeviceInfo, VadPolicy, + read_wav_samples, save_wav_file, verify_wav_file, AudioRecorder, CpalDeviceInfo, Recorder, + VadPolicy, }; +// Shared parts the manager uses to build the `Recorder` backends directly. +pub(crate) use audio::{AudioFrameCallback, VadConfig}; pub use text::{apply_custom_words, filter_transcription_output}; pub use utils::get_cpal_host; pub use vad::{SileroVad, VoiceActivityDetector}; diff --git a/src-tauri/src/managers/audio.rs b/src-tauri/src/managers/audio.rs index dc8ef9f189..1d49ec7e44 100644 --- a/src-tauri/src/managers/audio.rs +++ b/src-tauri/src/managers/audio.rs @@ -4,7 +4,7 @@ use crate::audio_toolkit::{ SmoothedVad, VAD_OFFLINE_HANGOVER_FRAMES, VAD_ONSET_FRAMES, VAD_PREFILL_FRAMES, VAD_STREAMING_HANGOVER_FRAMES, }, - AudioRecorder, SileroVad, VadPolicy, + AudioFrameCallback, Recorder, SileroVad, VadConfig, VadPolicy, }; use crate::helpers::clamshell; use crate::managers::transcription::StreamRouter; @@ -263,7 +263,7 @@ fn create_audio_recorder( vad_path: &Path, app_handle: &tauri::AppHandle, stream_router: Arc, -) -> Result { +) -> Result { // A single Silero engine covers both the offline and streaming policies (never // active at once within a recording), so the recorder reconfigures its // hangover tail per session rather than keeping two ONNX sessions resident. @@ -276,30 +276,34 @@ fn create_audio_recorder( VAD_ONSET_FRAMES, ); - // Recorder with VAD, a spectrum-level callback that forwards level updates to - // the frontend, and an audio-frame callback that feeds live streaming via a - // shared `StreamRouter` (captured directly, not via Tauri state — see its docs). - let recorder = AudioRecorder::new() - .map_err(|e| anyhow::anyhow!("Failed to create AudioRecorder: {}", e))? - .with_vad( - Box::new(smoothed_vad), - VAD_OFFLINE_HANGOVER_FRAMES, - VAD_STREAMING_HANGOVER_FRAMES, - ) - .with_level_callback({ - let app_handle = app_handle.clone(); - move |levels| { - utils::emit_levels(&app_handle, &levels); - } - }) - .with_audio_callback({ - let router = stream_router; - move |frame| { - router.feed(frame); - } - }); + // Build the shared VAD + callbacks once, then hand them to the `Recorder` + // seam which distributes them to whichever backend it selects (native + // PipeWire on Linux, cpal/ALSA fallback / other OSes). The single Silero + // engine is shared across backends (only one is ever open at a time). + let vad = VadConfig::new( + Box::new(smoothed_vad), + VAD_OFFLINE_HANGOVER_FRAMES, + VAD_STREAMING_HANGOVER_FRAMES, + ); + + // Spectrum-level callback forwards level updates to the frontend. + let level_cb: Arc) + Send + Sync + 'static> = Arc::new({ + let app_handle = app_handle.clone(); + move |levels| { + utils::emit_levels(&app_handle, &levels); + } + }); + + // Audio-frame callback feeds live streaming via a shared `StreamRouter` + // (captured directly, not via Tauri state — see its docs). + let audio_cb: AudioFrameCallback = Arc::new({ + let router = stream_router; + move |frame| { + router.feed(frame); + } + }); - Ok(recorder) + Ok(Recorder::from_parts(Some(vad), Some(level_cb), Some(audio_cb))) } /* ──────────────────────────────────────────────────────────────── */ @@ -310,7 +314,7 @@ pub struct AudioRecordingManager { mode: Arc>, app_handle: tauri::AppHandle, - recorder: Arc>>, + recorder: Arc>>, is_open: Arc>, is_recording: Arc>, mute_state: Arc>,