diff --git a/.github/workflows/app-build.yml b/.github/workflows/app-build.yml index 4e3bfd6dd..b9470abd0 100644 --- a/.github/workflows/app-build.yml +++ b/.github/workflows/app-build.yml @@ -53,6 +53,7 @@ jobs: libayatana-appindicator3-dev \ libgtk-3-dev \ libmpv-dev \ + libpulse-dev \ librsvg2-dev \ libssl-dev \ libudev-dev \ diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bc3fe3547..346a34b24 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -136,6 +136,7 @@ jobs: libayatana-appindicator3-dev \ libgtk-3-dev \ libmpv-dev \ + libpulse-dev \ librsvg2-dev \ libssl-dev \ libudev-dev \ diff --git a/src-tauri/Cargo.lock b/src-tauri/Cargo.lock index f3b4e9eee..9944d1fc0 100644 --- a/src-tauri/Cargo.lock +++ b/src-tauri/Cargo.lock @@ -2597,6 +2597,7 @@ dependencies = [ "libc", "libmpv2", "libmpv2-sys", + "libpulse-binding", "librespot-core", "librespot-metadata", "librespot-oauth", @@ -3657,6 +3658,33 @@ version = "4.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "edb32cabe7176b7d270b0dca6ff418b75187ae8d3854423e3d06fbff376841e4" +[[package]] +name = "libpulse-binding" +version = "2.30.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "909eb3049e16e373680fe65afe6e2a722ace06b671250cc4849557bc57d6a397" +dependencies = [ + "bitflags 2.11.1", + "libc", + "libpulse-sys", + "num-derive", + "num-traits", + "winapi", +] + +[[package]] +name = "libpulse-sys" +version = "1.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d74371848b22e989f829cc1621d2ebd74960711557d8b45cfe740f60d0a05e61" +dependencies = [ + "libc", + "num-derive", + "num-traits", + "pkg-config", + "winapi", +] + [[package]] name = "libredox" version = "0.1.16" diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index 5f8130c53..4b88ccf82 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -207,6 +207,7 @@ webview2-com = "0.39" [target.'cfg(target_os = "linux")'.dependencies] gtk = { version = "0.18", features = ["v3_24"] } zbus = { version = "5", default-features = false, features = ["tokio"] } +libpulse-binding = "2.30" [dev-dependencies] diff --git a/src-tauri/src/mpv.rs b/src-tauri/src/mpv.rs index 521fb5425..a6e49fc96 100644 --- a/src-tauri/src/mpv.rs +++ b/src-tauri/src/mpv.rs @@ -429,6 +429,23 @@ fn mpv_header_field(name: &str, value: &str) -> String { format!("{name}: {value}").replace('\\', "\\\\").replace(',', "\\,") } +// The PulseAudio client name is also how song_id identifies our playback on +// Linux. Flatpak instances can share a sandbox PID, so PID alone cannot +// distinguish their streams. A stable, per-process UUID avoids that collision. +#[cfg(target_os = "linux")] +pub(crate) fn audio_client_name() -> &'static str { + use std::sync::OnceLock; + + static NAME: OnceLock = OnceLock::new(); + NAME.get_or_init(|| format!("Harbor {}", uuid::Uuid::new_v4().simple())) + .as_str() +} + +#[cfg(not(target_os = "linux"))] +pub(crate) fn audio_client_name() -> &'static str { + "Harbor" +} + fn apply_pre_init( init: &MpvInitializer, args: &MpvStartArgs, @@ -459,7 +476,7 @@ fn apply_pre_init( .unwrap_or_default(); let joined_init_only = join_init_only_pairs(&init_only_pairs); set("title", "Harbor"); - set("audio-client-name", "Harbor"); + set("audio-client-name", audio_client_name()); set("terminal", "no"); set("log-file", ""); set("msg-level", "all=warn,vo=v,d3d11=v,gpu=v,win32=v"); diff --git a/src-tauri/src/song_id.rs b/src-tauri/src/song_id.rs index 65043b8bb..491ae7853 100644 --- a/src-tauri/src/song_id.rs +++ b/src-tauri/src/song_id.rs @@ -17,7 +17,7 @@ pub async fn recognize_now_playing( api_token: String, seconds: Option, ) -> Result, String> { - #[cfg(windows)] + #[cfg(any(windows, target_os = "linux"))] { let secs = seconds.unwrap_or(7).clamp(3, 15); let (pcm, sample_rate, channels, bits) = @@ -28,10 +28,10 @@ pub async fn recognize_now_playing( let wav = pcm_to_wav(&pcm, sample_rate, channels, bits)?; audd_recognize(wav, api_token).await } - #[cfg(not(windows))] + #[cfg(not(any(windows, target_os = "linux")))] { let _ = (api_token, seconds); - Err("Song identification is only supported on Windows for now".into()) + Err("Song identification is only supported on Windows and Linux for now".into()) } } @@ -41,7 +41,7 @@ pub async fn recognize_now_playing_ai( model: Option, seconds: Option, ) -> Result, String> { - #[cfg(windows)] + #[cfg(any(windows, target_os = "linux"))] { let secs = seconds.unwrap_or(8).clamp(3, 15); let (pcm, sample_rate, channels, bits) = @@ -52,10 +52,10 @@ pub async fn recognize_now_playing_ai( let wav = pcm_to_wav(&pcm, sample_rate, channels, bits)?; crate::song_id_gemini::recognize(wav, api_key, model).await } - #[cfg(not(windows))] + #[cfg(not(any(windows, target_os = "linux")))] { let _ = (api_key, model, seconds); - Err("Song identification is only supported on Windows for now".into()) + Err("Song identification is only supported on Windows and Linux for now".into()) } } @@ -128,7 +128,230 @@ fn capture_loopback(seconds: u32) -> Result<(Vec, u32, u16, u16), String> { Ok((pcm, sample_rate, channels, bits)) } +// Windows records the whole default output device. Pulse can record a single +// sink input, so on Linux we listen to Harbor's own mpv stream and nothing else: +// a call or a browser tab playing in the background never ends up in the clip. +// pipewire-pulse speaks the same protocol, so this covers PipeWire too. +#[cfg(target_os = "linux")] +fn capture_loopback(seconds: u32) -> Result<(Vec, u32, u16, u16), String> { + use libpulse_binding::callbacks::ListResult; + use libpulse_binding::context::{Context, FlagSet as ContextFlags, State as ContextState}; + use libpulse_binding::def::BufferAttr; + use libpulse_binding::mainloop::standard::Mainloop; + use libpulse_binding::sample::{Format, Spec}; + use libpulse_binding::stream::{ + FlagSet as StreamFlags, PeekResult, State as StreamState, Stream, + }; + use std::cell::{Cell, RefCell}; + use std::rc::Rc; + use std::time::{Duration, Instant}; + + let sample_rate = 44100u32; + let channels = 2u16; + let bits = 16u16; + let bytes_per_sec = sample_rate as usize * channels as usize * (bits as usize / 8); + let target_bytes = bytes_per_sec * seconds as usize; + // pipewire-pulse needs a couple of seconds before a monitor stream starts + // flowing, so the deadline leaves room for that on top of the clip itself. + let deadline = Instant::now() + Duration::from_secs(u64::from(seconds) + 6); + let timed_out = || Instant::now() > deadline; + + let mut mainloop = Mainloop::new().ok_or("Couldn't start PulseAudio")?; + let mut context = + Context::new(&mainloop, "Harbor song id").ok_or("Couldn't reach PulseAudio")?; + context + .connect(None, ContextFlags::NOFLAGS, None) + .map_err(|e| format!("Couldn't connect to PulseAudio: {e}"))?; + loop { + match context.get_state() { + ContextState::Ready => break, + ContextState::Failed | ContextState::Terminated => { + return Err("PulseAudio refused the connection".into()) + } + _ if timed_out() => return Err("PulseAudio took too long to answer".into()), + _ => pump(&mut mainloop)?, + } + } + + // libmpv plays from inside this process, so its Pulse client normally + // carries our PID. Flatpak PID namespaces can reuse the same number, so + // we also require our per-process audio-client name below. + let our_pid = std::process::id().to_string(); + let our_clients = Rc::new(RefCell::new(Vec::::new())); + { + let found = Rc::clone(&our_clients); + let op = context.introspect().get_client_info_list(move |item| { + let ListResult::Item(info) = item else { return }; + if info.proplist.get_str("application.process.id").as_deref() == Some(our_pid.as_str()) + { + found.borrow_mut().push(info.index); + } + }); + drain(&mut mainloop, &op, "list clients", &timed_out)?; + } + + let picked = Rc::new(Cell::new(None::<(u32, u32)>)); + { + let clients = Rc::clone(&our_clients); + let found = Rc::clone(&picked); + let op = context.introspect().get_sink_input_info_list(move |item| { + let ListResult::Item(info) = item else { return }; + if found.get().is_none() + && is_our_stream( + info.proplist.get_str("application.name").as_deref(), + info.client, + info.corked, + &clients.borrow(), + ) + { + found.set(Some((info.index, info.sink))); + } + }); + drain(&mut mainloop, &op, "list playback streams", &timed_out)?; + } + let (index, sink_index) = picked.get().ok_or( + "Harbor isn't playing anything right now. Start the scene with sound, then try again.", + )?; + + // set_monitor_stream only isolates the sink input once the recording stream + // is on the monitor source of that input's own sink, so resolve it by hand: + // the default source is usually a microphone, not a monitor at all. + let monitor = Rc::new(RefCell::new(None::)); + { + let found = Rc::clone(&monitor); + let op = context + .introspect() + .get_sink_info_by_index(sink_index, move |item| { + let ListResult::Item(info) = item else { return }; + *found.borrow_mut() = info.monitor_source_name.as_deref().map(str::to_owned); + }); + drain(&mut mainloop, &op, "resolve the output sink", &timed_out)?; + } + let monitor = monitor + .borrow_mut() + .take() + .ok_or("Harbor's audio output has no monitor source to record from")?; + + let spec = Spec { + format: Format::S16le, + rate: sample_rate, + channels: channels as u8, + }; + let mut stream = Stream::new(&mut context, "Harbor song id", &spec, None) + .ok_or("Couldn't open a PulseAudio stream")?; + stream + .set_monitor_stream(index) + .map_err(|e| format!("Couldn't attach to Harbor's audio: {e}"))?; + // Small fragments, otherwise the server hands the data over in chunks of a + // couple of seconds and the clip ends up late and lopsided. + let attr = BufferAttr { + maxlength: u32::MAX, + tlength: u32::MAX, + prebuf: u32::MAX, + minreq: u32::MAX, + fragsize: (bytes_per_sec / 10) as u32, + }; + stream + .connect_record(Some(&monitor), Some(&attr), StreamFlags::ADJUST_LATENCY) + .map_err(|e| format!("Couldn't start recording Harbor's audio: {e}"))?; + loop { + match stream.get_state() { + StreamState::Ready => break, + StreamState::Failed | StreamState::Terminated => { + return Err("PulseAudio closed the recording stream".into()) + } + _ if timed_out() => return Err("PulseAudio took too long to start recording".into()), + _ => pump(&mut mainloop)?, + } + } + + let mut pcm = Vec::with_capacity(target_bytes); + while pcm.len() < target_bytes && !timed_out() { + pump(&mut mainloop)?; + // PAErr has its own to_string() returning Option, hence format!. + let read_failed = |e| format!("Couldn't read Harbor's audio: {e}"); + loop { + match stream.peek().map_err(read_failed)? { + PeekResult::Empty => break, + PeekResult::Hole(_) => stream.discard().map_err(read_failed)?, + PeekResult::Data(chunk) => { + pcm.extend_from_slice(chunk); + stream.discard().map_err(read_failed)?; + } + } + } + } + let _ = stream.disconnect(); + context.disconnect(); + + let got = pcm.len(); + if got < target_bytes.min(bytes_per_sec * 4) { + let secs = got as f32 / bytes_per_sec as f32; + return Err(format!( + "Harbor only heard {secs:.1}s of its own audio before playback stopped. Keep the scene playing while Harbor listens." + )); + } + pcm.truncate(target_bytes); + Ok((pcm, sample_rate, channels, bits)) +} + +// The mpv PulseAudio name contains a per-process UUID, so two Flatpak +// instances cannot be confused even if their sandbox PIDs coincide. We also +// validate the Pulse client PID as a second check. A corked input is paused. +#[cfg(target_os = "linux")] +fn is_our_stream( + app_name: Option<&str>, + client: Option, + corked: bool, + our_clients: &[u32], +) -> bool { + !corked + && app_name == Some(crate::mpv::audio_client_name()) + && client.is_some_and(|c| our_clients.contains(&c)) +} + +// Introspection answers arrive on the mainloop, so every query needs pumping +// until it settles; the deadline keeps a wedged server from hanging the capture. +#[cfg(target_os = "linux")] +fn drain( + mainloop: &mut libpulse_binding::mainloop::standard::Mainloop, + op: &libpulse_binding::operation::Operation, + what: &str, + timed_out: &impl Fn() -> bool, +) -> Result<(), String> { + use libpulse_binding::operation::State; + while op.get_state() == State::Running { + if timed_out() { + return Err(format!("PulseAudio took too long to {what}")); + } + pump(mainloop)?; + } + Ok(()) +} + +// Never blocks, so a stalled server can't hold the capture past its deadline. +#[cfg(target_os = "linux")] +fn pump(mainloop: &mut libpulse_binding::mainloop::standard::Mainloop) -> Result<(), String> { + use libpulse_binding::mainloop::standard::IterateResult; + match mainloop.iterate(false) { + IterateResult::Success(0) => { + std::thread::sleep(std::time::Duration::from_millis(10)); + Ok(()) + } + IterateResult::Success(_) => Ok(()), + IterateResult::Quit(_) | IterateResult::Err(_) => { + Err("PulseAudio stopped responding".into()) + } + } +} + #[cfg(windows)] +const SILENCE_HINT: &str = "Harbor heard silence on your default playback device. Play the scene with sound, and make sure Windows is playing it through the device set as default."; +#[cfg(target_os = "linux")] +const SILENCE_HINT: &str = + "Harbor heard silence from its own player. Play the scene with sound, and make sure Harbor isn't muted."; + +#[cfg(any(windows, target_os = "linux"))] fn ensure_audible(pcm: &[u8]) -> Result<(), String> { let peak = pcm .chunks_exact(2) @@ -136,12 +359,12 @@ fn ensure_audible(pcm: &[u8]) -> Result<(), String> { .max() .unwrap_or(0); if peak < 150 { - return Err("Harbor heard silence on your default playback device. Play the scene with sound, and make sure Windows is playing it through the device set as default.".to_string()); + return Err(SILENCE_HINT.to_string()); } Ok(()) } -#[cfg(windows)] +#[cfg(any(windows, target_os = "linux"))] fn pcm_to_wav(pcm: &[u8], sample_rate: u32, channels: u16, bits: u16) -> Result, String> { use hound::{SampleFormat, WavSpec, WavWriter}; use std::io::Cursor; @@ -164,7 +387,7 @@ fn pcm_to_wav(pcm: &[u8], sample_rate: u32, channels: u16, bits: u16) -> Result< Ok(cursor.into_inner()) } -#[cfg(windows)] +#[cfg(any(windows, target_os = "linux"))] async fn audd_recognize(wav: Vec, api_token: String) -> Result, String> { use reqwest::multipart::{Form, Part}; @@ -222,3 +445,78 @@ async fn audd_recognize(wav: Vec, api_token: String) -> Result