aede_core/playback/
decoder.rs

1//! Progressive file decoding for local and remote playback drivers.
2//!
3//! The decoder yields complete, finite, interleaved `f32` frames in the
4//! source sample rate and channel layout. It does not own an audio device.
5//! [`IntegerFileDecoder`] separately preserves source-depth integer PCM for
6//! native 16/24-bit mono/stereo FLAC and WAV, without changing that float API.
7
8use std::fmt;
9use std::io::Read;
10use std::path::Path;
11use std::process::{Child, ChildStdout, Command, Stdio};
12use std::thread::JoinHandle;
13
14use aede_audio::decode::PcmStreamDecoder;
15use aede_dsp::ChannelLayout;
16use flaccompagnon_core::AnalysisError;
17use symphonia::core::formats::FormatOptions;
18use symphonia::core::io::MediaSourceStream;
19use symphonia::core::meta::MetadataOptions;
20use symphonia::core::probe::Hint;
21
22mod flac;
23mod flac_source;
24mod integer;
25mod integer_wav;
26mod vorbis;
27
28pub use integer::IntegerFileDecoder;
29
30/// Verification of a FLAC's original integer PCM against its stored MD5.
31///
32/// This describes this decoder run, not a persisted audit or audio quality.
33#[derive(Clone, Copy, Debug, Eq, PartialEq)]
34pub enum FlacMd5Status {
35    /// STREAMINFO has no signature; successful decoding cannot verify MD5.
36    NoSignature,
37    /// A signature exists, but the complete source has not reached EOF.
38    Pending,
39    /// All decoded source samples match the signature at successful EOF.
40    Verified,
41    /// Complete decoding disagrees with the signature; reads return an error.
42    Mismatch,
43}
44
45/// A file or buffer that cannot satisfy the playback PCM contract.
46#[derive(Debug)]
47pub enum Error {
48    /// The underlying file could not be opened or decoded.
49    Decode(AnalysisError),
50    /// The file header describes an unusable PCM format.
51    InvalidFormat,
52    /// The caller's buffer is empty or cannot hold complete frames.
53    InvalidBuffer,
54    /// A decoded packet does not contain complete frames.
55    IncompleteFrame,
56    /// A decoded packet contains a non-finite sample.
57    NonFiniteSample,
58    /// A multichannel stream's speaker positions could not be established.
59    ChannelLayout(String),
60    /// An external decoder could not be started or did not finish cleanly.
61    External(String),
62    /// The requested position exceeds the representable source-frame count.
63    InvalidPosition,
64    /// Seeking requires a newly opened stream, before any PCM is read.
65    SeekAfterStart,
66    /// The caller stopped a progressive seek before reaching its target.
67    SeekCancelled,
68    /// The completely decoded FLAC audio disagrees with its STREAMINFO MD5.
69    FlacMd5Mismatch,
70    /// The source lies outside the initial exact integer PCM profile.
71    UnsupportedIntegerSource(&'static str),
72}
73
74impl fmt::Display for Error {
75    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
76        match self {
77            Self::Decode(error) => write!(f, "audio decode failed: {error}"),
78            Self::InvalidFormat => f.write_str("audio file has an invalid PCM format"),
79            Self::InvalidBuffer => f.write_str("output buffer must hold complete PCM frames"),
80            Self::IncompleteFrame => f.write_str("decoded packet has an incomplete PCM frame"),
81            Self::NonFiniteSample => f.write_str("decoded packet contains a non-finite sample"),
82            Self::ChannelLayout(error) => write!(f, "cannot identify channel positions: {error}"),
83            Self::External(error) => write!(f, "external audio decoder failed: {error}"),
84            Self::InvalidPosition => f.write_str("seek position exceeds the source-frame range"),
85            Self::SeekAfterStart => f.write_str("seek requires a newly opened PCM stream"),
86            Self::SeekCancelled => f.write_str("audio seek was cancelled"),
87            Self::FlacMd5Mismatch => {
88                f.write_str("decoded FLAC audio does not match its STREAMINFO MD5")
89            }
90            Self::UnsupportedIntegerSource(reason) => f.write_str(reason),
91        }
92    }
93}
94
95impl std::error::Error for Error {
96    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
97        match self {
98            Self::Decode(error) => Some(error),
99            _ => None,
100        }
101    }
102}
103
104impl From<AnalysisError> for Error {
105    fn from(error: AnalysisError) -> Self {
106        Self::Decode(error)
107    }
108}
109
110/// Decodes one local audio file progressively into caller-owned PCM buffers.
111///
112/// Each successful read produces at most one decoded packet. Short reads are
113/// normal; only `Ok(0)` means end of stream. The caller can feed the returned
114/// portion of the buffer to `aede-dsp` after creating a matching PCM format.
115pub struct FileDecoder {
116    inner: Source,
117    sample_rate: u32,
118    channels: u16,
119    layout: ChannelLayout,
120    pending: Vec<f32>,
121    pending_offset: usize,
122    finished: bool,
123}
124
125enum Source {
126    Flac(flac::FlacStream),
127    Native(PcmStreamDecoder),
128    Vorbis(vorbis::VorbisStream),
129    Ffmpeg(FfmpegStream),
130}
131
132/// Source frames traversed by a progressive seek.
133#[derive(Clone, Copy, Debug, Eq, PartialEq)]
134pub struct SeekResult {
135    /// Complete source frames discarded before the next readable frame.
136    pub frames: u64,
137    /// The source ended before the requested position could be reached.
138    pub reached_eof: bool,
139}
140
141impl FileDecoder {
142    /// Open a local file and read its format without decoding the whole track.
143    pub fn open(path: &Path) -> Result<Self, Error> {
144        Self::open_with_fallback(path, true)
145    }
146
147    /// Background analysis never launches an external decoder with uncontrolled producer work.
148    pub fn open_native(path: &Path) -> Result<Self, Error> {
149        Self::open_with_fallback(path, false)
150    }
151
152    fn open_with_fallback(path: &Path, fallback: bool) -> Result<Self, Error> {
153        let (inner, sample_rate, channels) = if let Some(native) = vorbis::VorbisStream::open(path)?
154        {
155            let sample_rate = native.sample_rate;
156            let channels = native.channels;
157            (Source::Vorbis(native), sample_rate, channels)
158        } else if let Some(native) = flac::FlacStream::open(path)? {
159            let sample_rate = native.sample_rate;
160            let channels = native.channels;
161            (Source::Flac(native), sample_rate, channels)
162        } else {
163            match PcmStreamDecoder::open(path) {
164                Ok(native) => {
165                    let channels =
166                        u16::try_from(native.channels).map_err(|_| Error::InvalidFormat)?;
167                    let sample_rate = native.sample_rate;
168                    (Source::Native(native), sample_rate, channels)
169                }
170                Err(native_error) => {
171                    let native_error = AnalysisError::Decode(native_error.to_string());
172                    if !fallback {
173                        return Err(Error::Decode(native_error));
174                    }
175                    let tags = match crate::tags::read(path).ok() {
176                        Some(tags)
177                            if matches!(
178                                tags.properties.codec.as_str(),
179                                "opus" | "aac" | "alac"
180                            ) =>
181                        {
182                            tags
183                        }
184                        _ => return Err(Error::Decode(native_error)),
185                    };
186                    let sample_rate = tags.properties.sample_rate.ok_or(Error::InvalidFormat)?;
187                    let channels = tags.properties.channels.ok_or(Error::InvalidFormat)?;
188                    let external = FfmpegStream::open(path, sample_rate, channels)?;
189                    (Source::Ffmpeg(external), sample_rate, channels)
190                }
191            }
192        };
193        if sample_rate == 0 || channels == 0 {
194            return Err(Error::InvalidFormat);
195        }
196        let layout = match (channels, &inner) {
197            (1, _) => ChannelLayout::MONO,
198            (2, _) => ChannelLayout::STEREO,
199            (_, Source::Vorbis(native)) => native.layout,
200            (_, Source::Flac(native)) => native.layout,
201            (_, Source::Native(_)) => {
202                ChannelLayout::from_mask(probe_channel_mask(path, sample_rate, channels)?)
203                    .map_err(|_| Error::InvalidFormat)?
204            }
205            (_, Source::Ffmpeg(_)) => ChannelLayout::Unknown(channels),
206        };
207        Ok(Self {
208            inner,
209            sample_rate,
210            channels,
211            layout,
212            pending: Vec::new(),
213            pending_offset: 0,
214            finished: false,
215        })
216    }
217
218    /// Sample rate of the decoded PCM, in Hz.
219    pub fn sample_rate(&self) -> u32 {
220        self.sample_rate
221    }
222
223    /// Number of interleaved channels in each frame.
224    pub fn channels(&self) -> u16 {
225        self.channels
226    }
227
228    /// Speaker positions, when the decoder exposes them.
229    pub fn channel_layout(&self) -> ChannelLayout {
230        self.layout
231    }
232
233    /// FLAC source MD5 status, or `None` for another codec.
234    ///
235    /// Verification uses original integer samples before float conversion or
236    /// DSP and becomes conclusive only at EOF. Stopping early leaves it pending.
237    /// Progressive seeking also decodes the discarded prefix and therefore
238    /// still permits whole-source verification when the remainder reaches EOF.
239    pub fn flac_md5_status(&self) -> Option<FlacMd5Status> {
240        match &self.inner {
241            Source::Flac(native) => Some(native.md5_status()),
242            _ => None,
243        }
244    }
245
246    /// Fill a buffer with complete frames and return the number of frames.
247    ///
248    /// The buffer must be nonempty and its length divisible by `channels()`.
249    /// A rejected buffer is unchanged and does not advance decoding. A valid
250    /// buffer may receive fewer frames than it can hold, even before EOF.
251    pub fn read_frames(&mut self, output: &mut [f32]) -> Result<usize, Error> {
252        let channels = usize::from(self.channels);
253        if output.is_empty() || !output.len().is_multiple_of(channels) {
254            return Err(Error::InvalidBuffer);
255        }
256        if !self.fill_pending()? {
257            return Ok(0);
258        }
259
260        let count = output.len().min(self.pending.len() - self.pending_offset);
261        output[..count]
262            .copy_from_slice(&self.pending[self.pending_offset..self.pending_offset + count]);
263        self.pending_offset += count;
264        Ok(count / channels)
265    }
266
267    /// Discard up to `frames` complete source frames from the current position.
268    ///
269    /// This uses the ordinary decoder, preserving encoder-delay and end-trim
270    /// handling exactly. Memory stays bounded by one decoded packet, but work
271    /// grows with the traversed audio. Cancellation is checked before decoding
272    /// and after at most 4096 discarded frames; an individual decoder call may
273    /// still block on I/O. On cancellation or a decode error, discard this
274    /// decoder: its partially advanced position is not a successful seek.
275    pub fn skip_frames(
276        &mut self,
277        frames: u64,
278        mut cancelled: impl FnMut() -> bool,
279    ) -> Result<SeekResult, Error> {
280        let channels = usize::from(self.channels);
281        let mut skipped = 0;
282        loop {
283            if cancelled() {
284                return Err(Error::SeekCancelled);
285            }
286            if skipped == frames {
287                return Ok(SeekResult {
288                    frames: skipped,
289                    reached_eof: false,
290                });
291            }
292            if !self.fill_pending()? {
293                return Ok(SeekResult {
294                    frames: skipped,
295                    reached_eof: true,
296                });
297            }
298            let available = (self.pending.len() - self.pending_offset) / channels;
299            let count = available.min((frames - skipped).min(4096) as usize);
300            self.pending_offset += count * channels;
301            skipped += count as u64;
302        }
303    }
304
305    fn fill_pending(&mut self) -> Result<bool, Error> {
306        if self.finished {
307            return Ok(false);
308        }
309        let channels = usize::from(self.channels);
310        while self.pending_offset == self.pending.len() {
311            let chunk = match &mut self.inner {
312                Source::Flac(native) => native.next_chunk()?,
313                Source::Native(native) => native
314                    .next_chunk()
315                    .map_err(|e| Error::Decode(AnalysisError::Decode(e.to_string())))?,
316                Source::Vorbis(native) => native.next_chunk()?,
317                Source::Ffmpeg(external) => external.next_chunk()?,
318            };
319            let Some(chunk) = chunk else {
320                self.finished = true;
321                return Ok(false);
322            };
323            if !chunk.len().is_multiple_of(channels) {
324                return Err(Error::IncompleteFrame);
325            }
326            if chunk.iter().any(|sample| !sample.is_finite()) {
327                return Err(Error::NonFiniteSample);
328            }
329            self.pending = chunk;
330            self.pending_offset = 0;
331        }
332        Ok(true)
333    }
334}
335
336fn probe_channel_mask(path: &Path, sample_rate: u32, channels: u16) -> Result<u32, Error> {
337    let file =
338        std::fs::File::open(path).map_err(|error| Error::ChannelLayout(error.to_string()))?;
339    let source = MediaSourceStream::new(Box::new(file), Default::default());
340    let mut hint = Hint::new();
341    if let Some(extension) = path.extension().and_then(|extension| extension.to_str()) {
342        hint.with_extension(extension);
343    }
344    let probed = symphonia::default::get_probe()
345        .format(
346            &hint,
347            source,
348            &FormatOptions::default(),
349            &MetadataOptions::default(),
350        )
351        .map_err(|error| Error::ChannelLayout(error.to_string()))?;
352    let track = probed
353        .format
354        .default_track()
355        .ok_or_else(|| Error::ChannelLayout("no default audio track".into()))?;
356    let mask = track
357        .codec_params
358        .channels
359        .ok_or_else(|| Error::ChannelLayout("no channel mask in the stream header".into()))?
360        .bits();
361    if mask.count_ones() != u32::from(channels)
362        || track.codec_params.sample_rate != Some(sample_rate)
363    {
364        return Err(Error::ChannelLayout(
365            "channel mask or sample rate disagrees with the decoder".into(),
366        ));
367    }
368    Ok(mask)
369}
370
371/// FFmpeg handles Opus, AAC and ALAC when the native decoder cannot open them.
372/// Stream data and errors are kept separate, so a failed decode never looks
373/// like an ordinary end of file.
374struct FfmpegStream {
375    child: Child,
376    stdout: ChildStdout,
377    stderr: Option<JoinHandle<std::io::Result<Vec<u8>>>>,
378    pending_bytes: Vec<u8>,
379    frame_bytes: usize,
380    done: bool,
381}
382
383impl FfmpegStream {
384    fn open(path: &Path, sample_rate: u32, channels: u16) -> Result<Self, Error> {
385        let ffmpeg = crate::ffmpeg::find()
386            .ok_or_else(|| Error::External(crate::ffmpeg::missing("playback")))?;
387        Self::open_with_program(path, sample_rate, channels, &ffmpeg)
388    }
389
390    fn open_with_program(
391        path: &Path,
392        sample_rate: u32,
393        channels: u16,
394        ffmpeg: &str,
395    ) -> Result<Self, Error> {
396        if sample_rate == 0 || channels == 0 {
397            return Err(Error::InvalidFormat);
398        }
399        // FFmpeg parses relative colon-prefixed names as URLs. A canonical
400        // local path and restricted protocols keep fallback decoding local.
401        let path =
402            std::fs::canonicalize(path).map_err(|error| Error::External(error.to_string()))?;
403        let metadata =
404            std::fs::metadata(&path).map_err(|error| Error::External(error.to_string()))?;
405        if !metadata.is_file() {
406            return Err(Error::External(
407                "playback requires a local regular file".into(),
408            ));
409        }
410        let mut child = Command::new(ffmpeg)
411            .args([
412                "-hide_banner",
413                "-loglevel",
414                "error",
415                "-nostdin",
416                "-protocol_whitelist",
417                "file,pipe",
418                "-i",
419            ])
420            .arg(&path)
421            .args([
422                "-map",
423                "0:a:0",
424                "-vn",
425                "-sn",
426                "-dn",
427                "-f",
428                "f32le",
429                "-acodec",
430                "pcm_f32le",
431            ])
432            .args([
433                "-ar",
434                &sample_rate.to_string(),
435                "-ac",
436                &channels.to_string(),
437                "pipe:1",
438            ])
439            .stdin(Stdio::null())
440            .stdout(Stdio::piped())
441            .stderr(Stdio::piped())
442            .spawn()
443            .map_err(|error| Error::External(error.to_string()))?;
444        let Some(stdout) = child.stdout.take() else {
445            let _ = child.kill();
446            let _ = child.wait();
447            return Err(Error::External("ffmpeg did not open its PCM pipe".into()));
448        };
449        let Some(stderr) = child.stderr.take() else {
450            let _ = child.kill();
451            let _ = child.wait();
452            return Err(Error::External("ffmpeg did not open its error pipe".into()));
453        };
454        let stderr = match std::thread::Builder::new()
455            .name("aede-ffmpeg-errors".into())
456            .spawn(move || crate::ffmpeg::diagnostic_tail(stderr))
457        {
458            Ok(handle) => handle,
459            Err(error) => {
460                let _ = child.kill();
461                let _ = child.wait();
462                return Err(Error::External(error.to_string()));
463            }
464        };
465        Ok(Self {
466            child,
467            stdout,
468            stderr: Some(stderr),
469            pending_bytes: Vec::new(),
470            frame_bytes: usize::from(channels) * 4,
471            done: false,
472        })
473    }
474
475    fn next_chunk(&mut self) -> Result<Option<Vec<f32>>, Error> {
476        let mut buffer = [0; 16_384];
477        loop {
478            let complete = self.pending_bytes.len() / self.frame_bytes * self.frame_bytes;
479            if complete > 0 {
480                let bytes = self.pending_bytes.drain(..complete).collect::<Vec<_>>();
481                let chunk = bytes
482                    .as_chunks::<4>()
483                    .0
484                    .iter()
485                    .map(|bytes| f32::from_le_bytes(*bytes))
486                    .collect();
487                return Ok(Some(chunk));
488            }
489            let count = self
490                .stdout
491                .read(&mut buffer)
492                .map_err(|error| Error::External(error.to_string()))?;
493            if count == 0 {
494                let status = self
495                    .child
496                    .wait()
497                    .map_err(|error| Error::External(error.to_string()))?;
498                self.done = true;
499                let tail = match self.stderr.take() {
500                    Some(handle) => handle
501                        .join()
502                        .map_err(|_| Error::External("ffmpeg error reader stopped".into()))?
503                        .map_err(|error| Error::External(error.to_string()))?,
504                    None => Vec::new(),
505                };
506                if !status.success() {
507                    let message = String::from_utf8_lossy(&tail);
508                    return Err(Error::External(format!(
509                        "ffmpeg exited with {status}: {}",
510                        message.trim()
511                    )));
512                }
513                if !self.pending_bytes.is_empty() {
514                    return Err(Error::IncompleteFrame);
515                }
516                return Ok(None);
517            }
518            self.pending_bytes.extend_from_slice(&buffer[..count]);
519        }
520    }
521}
522
523impl Drop for FfmpegStream {
524    fn drop(&mut self) {
525        if !self.done {
526            let _ = self.child.kill();
527            let _ = self.child.wait();
528        }
529        if let Some(handle) = self.stderr.take() {
530            let _ = handle.join();
531        }
532    }
533}
534
535#[cfg(test)]
536#[path = "decoder_tests.rs"]
537mod tests;
538
539#[cfg(test)]
540#[path = "decoder_vorbis_tests.rs"]
541mod vorbis_tests;
aede_core/playback/decoder.rs

aede_core/playback/
decoder.rs
