aede_core/playback/
stream.rs

1//! Reusable decoded PCM blocks for local and server playback drivers.
2//!
3//! This layer owns decoding and `f32le` encoding. The caller supplies optional
4//! sample processing, an output sink, and any transport or progress controls.
5//! It never opens a device or sends data over a network.
6
7use std::error::Error;
8use std::fmt;
9use std::path::Path;
10use std::time::Duration;
11
12use super::decoder::{self, FileDecoder};
13use super::format::PcmStreamFormat;
14use aede_dsp::{DspError, RateConverter, RateError, StereoDownmixer};
15
16const BLOCK_FRAMES: usize = 4096;
17
18/// A decoded block after caller-supplied processing.
19pub struct PcmBlock<'a> {
20    /// Interleaved floating-point samples for meters and visualization.
21    pub samples: &'a [f32],
22    /// The same samples encoded as little-endian `f32` for an output sink.
23    pub f32le: &'a [u8],
24    /// Complete PCM frames in the block.
25    pub frames: usize,
26}
27
28/// A failure while decoding or processing a PCM block.
29#[derive(Debug)]
30pub enum StreamError<E> {
31    /// The source file could not supply valid PCM.
32    Decode(decoder::Error),
33    /// The source channel layout or a PCM frame cannot be downmixed.
34    Downmix(DspError),
35    /// A requested output rate could not be produced.
36    Resample(RateError),
37    /// The caller's DSP stage rejected the block.
38    Process(E),
39    /// The caller's stage produced a non-finite output sample.
40    NonFiniteSample,
41}
42
43impl<E: fmt::Display> fmt::Display for StreamError<E> {
44    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
45        match self {
46            Self::Decode(error) => write!(f, "{error}"),
47            Self::Downmix(error) => write!(f, "stereo downmix failed: {error}"),
48            Self::Resample(error) => write!(f, "{error}"),
49            Self::Process(error) => write!(f, "audio processing failed: {error}"),
50            Self::NonFiniteSample => f.write_str("audio processing produced a non-finite sample"),
51        }
52    }
53}
54
55impl<E: Error + 'static> Error for StreamError<E> {
56    fn source(&self) -> Option<&(dyn Error + 'static)> {
57        match self {
58            Self::Decode(error) => Some(error),
59            Self::Downmix(error) => Some(error),
60            Self::Resample(error) => Some(error),
61            Self::Process(error) => Some(error),
62            Self::NonFiniteSample => None,
63        }
64    }
65}
66
67/// A progressive PCM stream with one fixed source format.
68pub struct PcmTrack {
69    decoder: FileDecoder,
70    format: PcmStreamFormat,
71    source_format: PcmStreamFormat,
72    decode_format: PcmStreamFormat,
73    downmix: Option<StereoDownmixer>,
74    resampler: Option<RateConverter>,
75    started: bool,
76    ended: bool,
77    samples: Vec<f32>,
78    stereo: Vec<f32>,
79    converted: Vec<f32>,
80    bytes: Vec<u8>,
81}
82
83impl PcmTrack {
84    /// Open a local audio file without decoding it in full.
85    pub fn open(path: &Path) -> Result<Self, decoder::Error> {
86        Self::from_decoder(FileDecoder::open(path)?)
87    }
88
89    /// Native-only progressive decode for cooperative background extraction, without a child process.
90    pub fn open_native(path: &Path) -> Result<Self, decoder::Error> {
91        Self::from_decoder(FileDecoder::open_native(path)?)
92    }
93
94    fn from_decoder(decoder: FileDecoder) -> Result<Self, decoder::Error> {
95        let format = PcmStreamFormat::with_layout(decoder.sample_rate(), decoder.channel_layout())
96            .map_err(|_| decoder::Error::InvalidFormat)?;
97        let sample_capacity = BLOCK_FRAMES * usize::from(format.channels());
98        Ok(Self {
99            decoder,
100            format,
101            source_format: format,
102            decode_format: format,
103            downmix: None,
104            resampler: None,
105            started: false,
106            ended: false,
107            samples: vec![0.0; sample_capacity],
108            stereo: Vec::new(),
109            converted: Vec::new(),
110            bytes: vec![0; sample_capacity * 4],
111        })
112    }
113
114    /// Prepare stereo output for known multichannel material. Mono and
115    /// stereo keep their original samples and channel positions.
116    pub fn open_stereo(path: &Path) -> Result<Self, Box<dyn Error>> {
117        let mut track = Self::open(path)?;
118        if track.format.channels() > 2 {
119            let downmix = StereoDownmixer::new(track.format.layout())?;
120            track.format = PcmStreamFormat::new(track.format.sample_rate(), 2)?;
121            track.decode_format = track.format;
122            track.stereo = vec![0.0; BLOCK_FRAMES * 2];
123            track.bytes = vec![0; BLOCK_FRAMES * 2 * 4];
124            track.downmix = Some(downmix);
125        }
126        Ok(track)
127    }
128
129    /// The PCM format a sink must accept for this track.
130    pub fn format(&self) -> PcmStreamFormat {
131        self.format
132    }
133
134    /// Decoded source format before an optional stereo downmix.
135    pub fn source_format(&self) -> PcmStreamFormat {
136        self.source_format
137    }
138
139    /// FLAC source verification for this run, before downmix, resampling or DSP.
140    ///
141    /// Only a fully decoded source reaching EOF can be verified; another codec
142    /// returns `None`. This status does not alter persisted integrity results.
143    pub fn flac_md5_status(&self) -> Option<decoder::FlacMd5Status> {
144        self.decoder.flac_md5_status()
145    }
146
147    /// Choose the sink's sample rate before seeking or reading the first block. When
148    /// rates match, samples retain the exact decoded path.
149    pub fn set_output_rate(&mut self, rate: u32) -> Result<(), RateError> {
150        if self.started || self.resampler.is_some() {
151            return Err(RateError::Finished);
152        }
153        if rate != self.decode_format.sample_rate() {
154            let converter = RateConverter::new(self.decode_format, rate)?;
155            self.format = PcmStreamFormat::with_layout(rate, self.decode_format.layout())
156                .map_err(|_| RateError::InvalidFormat)?;
157            self.resampler = Some(converter);
158        }
159        Ok(())
160    }
161
162    /// Position a newly opened track at the source frame at or before `position`.
163    ///
164    /// Seeking progressively decodes and discards the prefix, without sending
165    /// it through the source observer, downmix, resampler or caller's DSP. It
166    /// preserves the ordinary decoder's delay, padding and error semantics and
167    /// uses no whole-file buffer. Work grows with the requested position;
168    /// cancellation is checked between bounded frame steps, though one decoder
169    /// call may block on I/O. A position beyond EOF returns the actual frame
170    /// count with `reached_eof` set, rather than trusting tag duration.
171    ///
172    /// Call at most once, before reading PCM. Set the output rate first when
173    /// needed. A backward or subsequent seek requires reopening the track and
174    /// resetting the driver's output and DSP session. After cancellation or a
175    /// decode error, discard this track. Measurements of the suffix are not a
176    /// full-track loudness result and must not be cached as one.
177    pub fn seek_from_start(
178        &mut self,
179        position: Duration,
180        cancelled: impl FnMut() -> bool,
181    ) -> Result<decoder::SeekResult, decoder::Error> {
182        if self.started {
183            return Err(decoder::Error::SeekAfterStart);
184        }
185        let rate = u128::from(self.source_format.sample_rate());
186        let frames = u128::from(position.as_secs()) * rate
187            + u128::from(position.subsec_nanos()) * rate / 1_000_000_000;
188        let frames = u64::try_from(frames).map_err(|_| decoder::Error::InvalidPosition)?;
189        self.started = true;
190        let result = self.decoder.skip_frames(frames, cancelled)?;
191        self.ended = result.reached_eof;
192        Ok(result)
193    }
194
195    /// Read, optionally process, and encode the next block.
196    ///
197    /// Returns `None` only at end of file. The processing function is called
198    /// once for each nonempty block and may retain state between calls. The
199    /// returned views are valid until the next call to `read_block`.
200    pub fn read_block<E>(
201        &mut self,
202        process: impl FnOnce(&mut [f32]) -> Result<(), E>,
203    ) -> Result<Option<PcmBlock<'_>>, StreamError<E>> {
204        self.read_block_observed(|_, _| {}, process)
205    }
206
207    /// Read a block while observing decoded source PCM before any downmix,
208    /// rate conversion or caller-supplied processing.
209    ///
210    /// The observer receives each nonempty source block exactly once with its
211    /// original format. One output block may require several source blocks,
212    /// or contain only a converter tail; the tail and EOF are never observed.
213    /// Observers that perform optional measurements should retain their own
214    /// errors without interrupting playback.
215    pub fn read_block_observed<E>(
216        &mut self,
217        mut observe: impl FnMut(PcmStreamFormat, &[f32]),
218        process: impl FnOnce(&mut [f32]) -> Result<(), E>,
219    ) -> Result<Option<PcmBlock<'_>>, StreamError<E>> {
220        if self.ended {
221            return Ok(None);
222        }
223        self.started = true;
224        if let Some(converter) = &mut self.resampler {
225            loop {
226                let frames = self
227                    .decoder
228                    .read_frames(&mut self.samples)
229                    .map_err(StreamError::Decode)?;
230                let source_count = frames * usize::from(self.source_format.channels());
231                if frames > 0 {
232                    observe(self.source_format, &self.samples[..source_count]);
233                }
234                let count = frames * usize::from(self.decode_format.channels());
235                let samples = if let Some(downmix) = &self.downmix {
236                    let stereo = &mut self.stereo[..count];
237                    downmix
238                        .process(&self.samples[..source_count], stereo)
239                        .map_err(StreamError::Downmix)?;
240                    stereo
241                } else {
242                    &mut self.samples[..count]
243                };
244                let output = converter
245                    .push(samples, frames == 0)
246                    .map_err(StreamError::Resample)?;
247                if frames == 0 {
248                    self.ended = true;
249                }
250                if !output.is_empty() {
251                    self.converted.clear();
252                    self.converted.extend_from_slice(output);
253                    break;
254                }
255                if self.ended {
256                    return Ok(None);
257                }
258            }
259            return finish_block(
260                &mut self.converted,
261                &mut self.bytes,
262                self.format.channels(),
263                process,
264            )
265            .map(Some);
266        }
267        let frames = self
268            .decoder
269            .read_frames(&mut self.samples)
270            .map_err(StreamError::Decode)?;
271        if frames == 0 {
272            self.ended = true;
273            return Ok(None);
274        }
275        let source_count = frames * usize::from(self.source_format.channels());
276        observe(self.source_format, &self.samples[..source_count]);
277        let count = frames * usize::from(self.decode_format.channels());
278        let samples = if let Some(downmix) = &self.downmix {
279            let stereo = &mut self.stereo[..count];
280            downmix
281                .process(&self.samples[..source_count], stereo)
282                .map_err(StreamError::Downmix)?;
283            stereo
284        } else {
285            &mut self.samples[..count]
286        };
287        finish_block(samples, &mut self.bytes, self.format.channels(), process).map(Some)
288    }
289}
290
291fn finish_block<'a, E>(
292    samples: &'a mut [f32],
293    bytes: &'a mut Vec<u8>,
294    channels: u16,
295    process: impl FnOnce(&mut [f32]) -> Result<(), E>,
296) -> Result<PcmBlock<'a>, StreamError<E>> {
297    process(samples).map_err(StreamError::Process)?;
298    if samples.iter().any(|sample| !sample.is_finite()) {
299        return Err(StreamError::NonFiniteSample);
300    }
301    bytes.resize(samples.len() * 4, 0);
302    for (sample, encoded) in samples.iter().zip(bytes.as_chunks_mut::<4>().0.iter_mut()) {
303        encoded.copy_from_slice(&sample.to_le_bytes());
304    }
305    Ok(PcmBlock {
306        samples,
307        f32le: bytes,
308        frames: samples.len() / usize::from(channels),
309    })
310}
311
312#[cfg(test)]
313#[path = "stream_tests.rs"]
314mod tests;
aede_core/playback/stream.rs

aede_core/playback/
stream.rs
