From 9fba9708bf89020136af027ed6cc8cad2ec09bac Mon Sep 17 00:00:00 2001 From: Test User Date: Fri, 24 Jul 2026 02:10:02 +0200 Subject: [PATCH] Audio: decode Ogg-Opus via libopus so opus streams play rodio decodes through symphonia 0.5, which ships no Opus decoder, so raw .opus sources (audiobookshelf files, and opus from any provider) failed Decoder::build(). Add OpusSource, a rodio Source that demuxes Ogg with symphonia own Ogg reader and decodes with libopus (via symphonia-adapter-libopus, registered into an explicit codec registry), and route to it by content-sniffing OggS+OpusHead in the player -- the abs stream URL has no file extension, so the extension hint is not enough. The bundled libopus builds with cmake/ninja (added to devenv). Verified end-to-end with ffmpeg mono/stereo opus fixtures including seeking; 11 audio-player tests pass, clippy/fmt clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- Cargo.lock | 22 ++ Cargo.toml | 5 + audio-player/Cargo.toml | 2 + audio-player/src/lib.rs | 1 + audio-player/src/opus_source.rs | 342 ++++++++++++++++++++++++++++++ audio-player/src/player_engine.rs | 122 ++++++++--- devenv.nix | 4 + 7 files changed, 463 insertions(+), 35 deletions(-) create mode 100644 audio-player/src/opus_source.rs diff --git a/Cargo.lock b/Cargo.lock index ed531eb..55403f6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -388,6 +388,8 @@ dependencies = [ "reqwest 0.13.1", "rodio", "stream-download", + "symphonia", + "symphonia-adapter-libopus", "thiserror 2.0.19", "tokio", "tracing", @@ -3215,6 +3217,15 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d" +[[package]] +name = "opusic-sys" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2804e694ef0de3b4cbb254de565053b7cb48d3398df7fd60c6c62bed40c5372a" +dependencies = [ + "cmake", +] + [[package]] name = "or_poisoned" version = "0.1.0" @@ -4869,6 +4880,17 @@ dependencies = [ "symphonia-metadata", ] +[[package]] +name = "symphonia-adapter-libopus" +version = "0.2.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2bfc8e95f95c23ed1b5328eb66920ad28d9968c797f9c7aa755d4b45a5f47a41" +dependencies = [ + "log", + "opusic-sys", + "symphonia-core", +] + [[package]] name = "symphonia-bundle-flac" version = "0.5.5" diff --git a/Cargo.toml b/Cargo.toml index 58d94a6..6963d4e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -67,6 +67,11 @@ rustypipe = { version = "0.11", default-features = false, features = [ serde = { version = "1", features = ["derive"] } serde_json = "1" serde_urlencoded = "0.7" +# Opus decoding: symphonia has no Opus decoder, so we demux Ogg-Opus with +# symphonia's Ogg reader and decode via the libopus adapter (matches +# rodio's symphonia 0.5). The adapter bundles libopus (needs cmake). +symphonia = { version = "0.5", default-features = false, features = ["ogg"] } +symphonia-adapter-libopus = "0.2" stream-download = { version = "0.24", default-features = false, features = [ "reqwest", "reqwest-rustls", diff --git a/audio-player/Cargo.toml b/audio-player/Cargo.toml index 9f829af..fdf4b30 100644 --- a/audio-player/Cargo.toml +++ b/audio-player/Cargo.toml @@ -11,6 +11,8 @@ futures.workspace = true reqwest.workspace = true rodio.workspace = true stream-download.workspace = true +symphonia.workspace = true +symphonia-adapter-libopus.workspace = true thiserror.workspace = true tokio = { workspace = true, features = ["rt", "rt-multi-thread"] } tracing.workspace = true diff --git a/audio-player/src/lib.rs b/audio-player/src/lib.rs index 3b0ed52..61cc659 100644 --- a/audio-player/src/lib.rs +++ b/audio-player/src/lib.rs @@ -1,3 +1,4 @@ +mod opus_source; mod player; mod player_engine; mod spectrum_tap; diff --git a/audio-player/src/opus_source.rs b/audio-player/src/opus_source.rs new file mode 100644 index 0000000..ea54918 --- /dev/null +++ b/audio-player/src/opus_source.rs @@ -0,0 +1,342 @@ +//! Ogg-Opus decoding for the rodio pipeline. +//! +//! rodio decodes through symphonia, and symphonia 0.5 ships no Opus +//! decoder — so raw `.opus` streams (notably audiobookshelf files, but any +//! provider serving Opus) fail rodio's `Decoder::build()`. This module fills +//! that gap: it demuxes the container with symphonia's own Ogg reader and +//! decodes the Opus packets with libopus via [`symphonia_adapter_libopus`], +//! then exposes the result as a rodio [`Source`] so the rest of the player +//! (sink, spectrum tap, seeking) is unchanged. +//! +//! Only Ogg-*Opus* is handled here; every other format (including Ogg-Vorbis +//! and Ogg-FLAC, which symphonia decodes natively) stays on rodio's decoder. +//! [`is_ogg_opus`] does the sniffing and the player routes accordingly. +//! +//! The decode loop, buffering, and seek refinement mirror rodio 0.22's own +//! `SymphoniaDecoder`; the only substantive difference is that we build the +//! format reader and the codec from an explicit registry that includes the +//! libopus adapter, instead of symphonia's default `get_codecs()`. + +use std::io::{Read, Result as IoResult, Seek, SeekFrom}; +use std::sync::Arc; +use std::time::Duration; + +use anyhow::{anyhow, Context, Result}; +use symphonia::core::audio::{AudioBufferRef, SampleBuffer, SignalSpec}; +use symphonia::core::codecs::{ + CodecRegistry, Decoder as SymphoniaDecoderTrait, DecoderOptions, CODEC_TYPE_OPUS, +}; +use symphonia::core::errors::Error as SymphoniaError; +use symphonia::core::formats::{FormatOptions, FormatReader, SeekMode, SeekTo, SeekedTo}; +use symphonia::core::io::{MediaSource, MediaSourceStream, MediaSourceStreamOptions}; +use symphonia::core::units; +use symphonia::default::formats::OggReader; +use symphonia_adapter_libopus::OpusDecoder; + +use rodio::source::SeekError; +use rodio::{ChannelCount, Sample, SampleRate, Source}; + +/// The Ogg page capture pattern (every page starts with it). +const OGG_MAGIC: &[u8] = b"OggS"; +/// The magic signature of the Opus identification header packet, which is +/// the first packet of an Ogg-Opus stream (RFC 7845 §5.1). +const OPUS_HEAD_MAGIC: &[u8] = b"OpusHead"; + +/// Returns true if `header` looks like the start of an Ogg-Opus stream: an +/// Ogg page whose first packet is the Opus identification header. `header` +/// should be at least the first Ogg page (~64 bytes is plenty; `OpusHead` +/// sits at offset 28 in a well-formed stream). Ogg-Vorbis/FLAC deliberately +/// do *not* match, so they keep flowing through rodio's native decoder. +pub fn is_ogg_opus(header: &[u8]) -> bool { + header.starts_with(OGG_MAGIC) && contains(header, OPUS_HEAD_MAGIC) +} + +fn contains(haystack: &[u8], needle: &[u8]) -> bool { + haystack + .windows(needle.len()) + .any(|window| window == needle) +} + +/// A rodio [`Source`] that decodes an Ogg-Opus stream to interleaved `f32` +/// samples at libopus's native 48 kHz. +pub struct OpusSource { + decoder: Box, + format: Box, + track_id: u32, + /// Interleaved samples of the most recently decoded packet. + buffer: SampleBuffer, + /// Read cursor into `buffer`, in samples (not frames). + span_offset: usize, + spec: SignalSpec, + total_duration: Option, +} + +impl OpusSource { + /// Builds an Opus source from a seekable byte stream. `byte_len` (the + /// content length) and `is_seekable` mirror the values the player already + /// computes for rodio; supplying them lets symphonia's Ogg reader learn + /// the final granule position (hence duration) and seek by time. + pub fn new(reader: R, byte_len: Option, is_seekable: bool) -> Result + where + R: Read + Seek + Send + Sync + 'static, + { + let source = Box::new(ReadSeekSource { + inner: reader, + byte_len, + is_seekable, + }); + let mss = MediaSourceStream::new(source, MediaSourceStreamOptions::default()); + + let format = OggReader::try_new(mss, &FormatOptions::default()) + .context("failed to open Ogg-Opus container")?; + let format: Box = Box::new(format); + + let track = format + .tracks() + .iter() + .find(|t| t.codec_params.codec == CODEC_TYPE_OPUS) + .context("Ogg stream has no Opus track")?; + let track_id = track.id; + + // A registry holding only the libopus adapter — the container is + // Opus-only, so no other codec can appear. + let mut registry = CodecRegistry::new(); + registry.register_all::(); + let decoder = registry + .make(&track.codec_params, &DecoderOptions::default()) + .context("failed to create libopus decoder")?; + + let total_duration = track + .codec_params + .time_base + .zip(track.codec_params.n_frames) + .map(|(base, frames)| Duration::from(base.calc_time(frames))) + .filter(|d| !d.is_zero()); + + // Decode the first packet up front so the struct always holds a real + // buffer and spec. symphonia's `SampleBuffer::new` divides by the + // channel count, so a zero-channel placeholder would panic; this also + // makes channel/rate queries valid before the first `next()` (rodio + // inspects the source right after construction). + let mut format = format; + let mut decoder = decoder; + let (spec, buffer) = decode_next_buffer(&mut format, &mut decoder, track_id)? + .context("Opus stream produced no audio")?; + + Ok(Self { + decoder, + format, + track_id, + buffer, + span_offset: 0, + spec, + total_duration, + }) + } +} + +/// Decodes packets — skipping ones from other logical streams and benign +/// decode errors — until one yields audio frames, returning that frame's +/// interleaved samples and signal spec. `Ok(None)` is a clean end of stream. +fn decode_next_buffer( + format: &mut Box, + decoder: &mut Box, + track_id: u32, +) -> Result)>> { + loop { + let packet = match format.next_packet() { + Ok(packet) => packet, + // An IO error at a page boundary is how symphonia's Ogg reader + // signals end-of-stream for a bounded file. + Err(SymphoniaError::IoError(_)) => return Ok(None), + Err(e) => return Err(anyhow!(e)).context("reading Opus packet"), + }; + if packet.track_id() != track_id { + continue; + } + match decoder.decode(&packet) { + Ok(decoded) if decoded.frames() > 0 => { + let spec = *decoded.spec(); + return Ok(Some((spec, copy_to_buffer(decoded, &spec)))); + } + // A metadata-only packet (0 frames) or a recoverable decode error: + // skip it and keep going, matching rodio's behaviour. + Ok(_) => continue, + Err(SymphoniaError::DecodeError(_)) => continue, + Err(e) => return Err(anyhow!(e)).context("decoding Opus packet"), + } + } +} + +/// Copies a decoded symphonia buffer into an interleaved `f32` sample buffer. +fn copy_to_buffer(decoded: AudioBufferRef, spec: &SignalSpec) -> SampleBuffer { + let capacity = units::Duration::from(decoded.capacity() as u64); + let mut buffer = SampleBuffer::::new(capacity, *spec); + buffer.copy_interleaved_ref(decoded); + buffer +} + +impl Iterator for OpusSource { + type Item = Sample; + + fn next(&mut self) -> Option { + if self.span_offset >= self.buffer.len() { + match decode_next_buffer(&mut self.format, &mut self.decoder, self.track_id) { + Ok(Some((spec, buffer))) => { + self.spec = spec; + self.buffer = buffer; + self.span_offset = 0; + } + // Clean EOS or a fatal error both end the source. + _ => return None, + } + } + let sample = *self.buffer.samples().get(self.span_offset)?; + self.span_offset += 1; + Some(sample) + } +} + +impl Source for OpusSource { + fn current_span_len(&self) -> Option { + Some(self.buffer.len()) + } + + fn channels(&self) -> ChannelCount { + let count = u16::try_from(self.spec.channels.count().max(1)).unwrap_or(2); + ChannelCount::new(count).unwrap_or(ChannelCount::new(2).expect("2 is nonzero")) + } + + fn sample_rate(&self) -> SampleRate { + SampleRate::new(self.spec.rate) + .unwrap_or(SampleRate::new(48_000).expect("48000 is nonzero")) + } + + fn total_duration(&self) -> Option { + self.total_duration + } + + fn try_seek(&mut self, pos: Duration) -> Result<(), SeekError> { + // Clamp beyond-end seeks to the end (saturating), like rodio. + let mut target = pos; + if let Some(total) = self.total_duration { + if target > total { + target = total; + } + } + + // Preserve the active channel so we resume on a frame boundary. + let channels = self.channels().get() as usize; + let active_channel = self.span_offset % channels; + + let seeked = self + .format + .seek( + SeekMode::Accurate, + SeekTo::Time { + time: target.into(), + track_id: Some(self.track_id), + }, + ) + .map_err(|e| SeekError::Other(Arc::new(e)))?; + + // The demuxer moved without telling the decoder; reset it and force + // the next `next()` to refill from the new position (`span_offset` + // past the current buffer's end triggers a decode in `next()`). + self.decoder.reset(); + self.span_offset = usize::MAX; + + // Ogg seeks land on a page boundary before the target; fast-forward + // the residual so playback resumes at the requested instant. + self.refine_position(seeked); + + // Re-align to the channel we were on before seeking. + for _ in 0..active_channel { + self.next(); + } + Ok(()) + } +} + +impl OpusSource { + /// Skips the samples between the keyframe symphonia seeked to and the + /// exact requested timestamp. No-op when the demuxer hit the target. + fn refine_position(&mut self, seeked: SeekedTo) { + let Some(base) = self.decoder.codec_params().time_base else { + return; + }; + let residual = seeked.required_ts.saturating_sub(seeked.actual_ts); + if residual == 0 { + return; + } + let seconds = Duration::from(base.calc_time(residual)).as_secs_f64(); + let channels = self.channels().get() as f64; + let mut samples = (seconds * self.sample_rate().get() as f64 * channels).ceil() as usize; + samples -= samples % self.channels().get() as usize; + for _ in 0..samples { + if self.next().is_none() { + break; + } + } + } +} + +/// Adapts a `Read + Seek` byte stream into symphonia's [`MediaSource`] +/// (equivalent to rodio's private `ReadSeekSource`). +struct ReadSeekSource { + inner: R, + byte_len: Option, + is_seekable: bool, +} + +impl MediaSource for ReadSeekSource { + fn is_seekable(&self) -> bool { + self.is_seekable + } + fn byte_len(&self) -> Option { + self.byte_len + } +} + +impl Read for ReadSeekSource { + fn read(&mut self, buf: &mut [u8]) -> IoResult { + self.inner.read(buf) + } +} + +impl Seek for ReadSeekSource { + fn seek(&mut self, pos: SeekFrom) -> IoResult { + self.inner.seek(pos) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn detects_ogg_opus_header() { + // "OggS" page header with an "OpusHead" first packet. + let mut header = Vec::new(); + header.extend_from_slice(b"OggS"); + header.extend_from_slice(&[0u8; 24]); // rest of the 28-byte page header + header.extend_from_slice(b"OpusHead"); + assert!(is_ogg_opus(&header)); + } + + #[test] + fn rejects_ogg_vorbis() { + let mut header = Vec::new(); + header.extend_from_slice(b"OggS"); + header.extend_from_slice(&[0u8; 24]); + header.extend_from_slice(b"\x01vorbis"); + assert!(!is_ogg_opus(&header)); + } + + #[test] + fn rejects_non_ogg() { + assert!(!is_ogg_opus(b"ID3\x04OpusHead")); // mp3 with a coincidental needle + assert!(!is_ogg_opus(b"fLaC")); + assert!(!is_ogg_opus(b"")); + } +} diff --git a/audio-player/src/player_engine.rs b/audio-player/src/player_engine.rs index 84ebe41..92a7838 100644 --- a/audio-player/src/player_engine.rs +++ b/audio-player/src/player_engine.rs @@ -1,5 +1,5 @@ use std::fs::File; -use std::io::BufReader; +use std::io::{BufReader, Read, Seek, SeekFrom}; use std::path::Path; use std::time::Duration; @@ -11,6 +11,7 @@ use rodio::{Decoder, Source}; use stream_download::storage::temp::TempStorageProvider; use stream_download::{Settings, StreamDownload}; +use crate::opus_source::{is_ogg_opus, OpusSource}; use crate::spectrum_tap::{SpectrumTap, TappingSource}; use crate::windowed_http::{WindowedHttpParams, WindowedHttpStream}; use std::sync::Arc; @@ -325,26 +326,17 @@ impl PlayerEngine { .map_err(|_| anyhow!("timed out opening stream after {STREAM_OPEN_TIMEOUT:?}"))? .context("failed to open http stream") })?; - // Symphonia probes the container length during init; without a - // known byte length it seeks from the end, which rodio 0.22 turns - // into an `unreachable!` panic on a streamed source. Handing it the - // content length up front avoids that seek entirely. let byte_len = reader.content_length(); - let mut builder = Decoder::builder().with_data(reader).with_seekable(true); - if let Some(len) = byte_len { - builder = builder.with_byte_len(len); - } - if let Some(extension) = Path::new(url.path()).extension().and_then(|e| e.to_str()) - { - builder = builder.with_hint(extension); - } - let decoder = builder.build().context("failed to decode http stream")?; - let duration = decoder.total_duration(); - // Mirror the played audio into the spectrum tap (it only - // observes; playback is unaffected). - let source: Box = - Box::new(TappingSource::new(decoder, self.spectrum.clone())); - Ok((source, duration)) + let hint = Path::new(url.path()) + .extension() + .and_then(|e| e.to_str()) + .map(str::to_owned); + self.build_source( + reader, + byte_len, + hint.as_deref(), + "failed to decode http stream", + ) } Ok(url) => Err(anyhow!("Not a valid URL scheme: {}", url.scheme())), Err(_) => { @@ -352,25 +344,69 @@ impl PlayerEngine { let file = File::open(source_str) .with_context(|| format!("failed to open file {source_str}"))?; let byte_len = file.metadata().ok().map(|m| m.len()); - let mut builder = Decoder::builder() - .with_data(BufReader::new(file)) - .with_seekable(true); - if let Some(len) = byte_len { - builder = builder.with_byte_len(len); - } - if let Some(extension) = Path::new(source_str).extension().and_then(|e| e.to_str()) - { - builder = builder.with_hint(extension); - } - let decoder = builder.build().context("failed to decode file")?; - let duration = decoder.total_duration(); - let source: Box = - Box::new(TappingSource::new(decoder, self.spectrum.clone())); - Ok((source, duration)) + let hint = Path::new(source_str) + .extension() + .and_then(|e| e.to_str()) + .map(str::to_owned); + self.build_source( + BufReader::new(file), + byte_len, + hint.as_deref(), + "failed to decode file", + ) } } } + /// Sniffs the stream's header and decodes it. Ogg-Opus — which symphonia + /// (hence rodio) cannot decode — is routed to [`OpusSource`]; every other + /// format goes through rodio's decoder. The reader is rewound after + /// sniffing so the chosen decoder sees the whole stream, and both paths + /// are wrapped in the spectrum tap (it only observes; playback is + /// unaffected). + fn build_source( + &self, + mut reader: R, + byte_len: Option, + hint: Option<&str>, + decode_err: &'static str, + ) -> Result<(Box, Option)> + where + R: Read + Seek + Send + Sync + 'static, + { + let mut header = [0u8; 64]; + let n = read_header(&mut reader, &mut header)?; + reader + .seek(SeekFrom::Start(0)) + .context("failed to rewind after sniffing the header")?; + + if is_ogg_opus(&header[..n]) { + debug!("decoding Ogg-Opus via libopus"); + let source = OpusSource::new(reader, byte_len, true)?; + let duration = source.total_duration(); + let tapped: Box = + Box::new(TappingSource::new(source, self.spectrum.clone())); + return Ok((tapped, duration)); + } + + // Symphonia probes the container length during init; without a known + // byte length it seeks from the end, which rodio 0.22 turns into an + // `unreachable!` panic on a streamed source. Handing it the content + // length up front avoids that seek entirely. + let mut builder = Decoder::builder().with_data(reader).with_seekable(true); + if let Some(len) = byte_len { + builder = builder.with_byte_len(len); + } + if let Some(extension) = hint { + builder = builder.with_hint(extension); + } + let decoder = builder.build().context(decode_err)?; + let duration = decoder.total_duration(); + let tapped: Box = + Box::new(TappingSource::new(decoder, self.spectrum.clone())); + Ok((tapped, duration)) + } + /// Appends an already-decoded source to the freshly-reset sink, followed /// by an end-of-stream callback tagged with the current generation. The /// callback fires only when this source finishes naturally; a later @@ -538,6 +574,22 @@ fn send_reply(tx: Sender, value: T) { } } +/// Reads up to `buf.len()` bytes for format sniffing, tolerating short reads +/// and `Interrupted`. Returns how many bytes were read (fewer than the buffer +/// only at end-of-stream), so a stream shorter than the buffer is not an error. +fn read_header(reader: &mut R, buf: &mut [u8]) -> Result { + let mut filled = 0; + while filled < buf.len() { + match reader.read(&mut buf[filled..]) { + Ok(0) => break, + Ok(n) => filled += n, + Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue, + Err(e) => return Err(e).context("failed to read stream header"), + } + } + Ok(filled) +} + #[cfg(test)] mod tests { use super::*; diff --git a/devenv.nix b/devenv.nix index 05b0ad1..3afdbc5 100644 --- a/devenv.nix +++ b/devenv.nix @@ -15,6 +15,10 @@ let pkg-config protobuf cargo-cross + # opusic-sys (via symphonia-adapter-libopus) compiles libopus from + # source with CMake; ninja is picked up automatically if present. + cmake + ninja # Stream-URL sidecar for the ytdy provider: YouTube caps tokenless # stream URLs at ~1 MiB and yt-dlp is the only maintained cipher # solver (architecture/youtube-rustypipe.md, D2-revised).