aede_core/playback/
session.rs

1//! Continuous processed PCM for successive compatible tracks.
2//!
3//! A session retains its rate converter and tone-filter state at natural track
4//! boundaries. Decoding, source loudness observation, downmix and output timing
5//! remain the driver's responsibility.
6
7use std::collections::VecDeque;
8use std::error::Error;
9use std::fmt;
10use std::ops::Range;
11
12use aede_dsp::{Dsp, DspError, PcmFormat, ProcessStats, RateConverter, RateError, ToneControls};
13
14/// A failure preparing or producing continuous processed PCM.
15#[derive(Debug)]
16pub enum SessionError {
17    /// The input format, gain, tone or PCM was rejected by the DSP.
18    Dsp(DspError),
19    /// Rate conversion could not accept or produce the requested PCM.
20    Resample(RateError),
21    /// The current track must be sealed before starting another or finishing.
22    TrackActive,
23    /// Source PCM and track endings require an active track.
24    NoTrack,
25    /// A finished session cannot accept another track or more source PCM.
26    Finished,
27    /// The stream exceeded the representable cumulative frame count.
28    FrameCountOverflow,
29    /// Converted frames could not be attributed to the decoded source timeline.
30    OutputTimeline,
31}
32
33impl fmt::Display for SessionError {
34    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
35        match self {
36            Self::Dsp(error) => write!(f, "{error}"),
37            Self::Resample(error) => write!(f, "{error}"),
38            Self::TrackActive => f.write_str("seal the active playback track first"),
39            Self::NoTrack => f.write_str("no active playback track"),
40            Self::Finished => f.write_str("continuous playback session has finished"),
41            Self::FrameCountOverflow => f.write_str("continuous playback frame count overflowed"),
42            Self::OutputTimeline => {
43                f.write_str("converted PCM exceeded its playback track timeline")
44            }
45        }
46    }
47}
48
49impl Error for SessionError {
50    fn source(&self) -> Option<&(dyn Error + 'static)> {
51        match self {
52            Self::Dsp(error) => Some(error),
53            Self::Resample(error) => Some(error),
54            _ => None,
55        }
56    }
57}
58
59impl From<DspError> for SessionError {
60    fn from(error: DspError) -> Self {
61        Self::Dsp(error)
62    }
63}
64
65impl From<RateError> for SessionError {
66    fn from(error: RateError) -> Self {
67        Self::Resample(error)
68    }
69}
70
71/// A track's contiguous region of one processed or exact output block.
72#[derive(Clone, Debug, PartialEq)]
73pub struct TrackSpan {
74    /// Driver-supplied identity for the source track or queue entry.
75    pub token: usize,
76    /// Interleaved sample indices in [`SessionBlock::samples`] or
77    /// [`super::exact_session::ExactSessionBlock::samples`].
78    pub samples: Range<usize>,
79    /// Observed normalized peak and guard interventions. Exact spans observe
80    /// the source peak without modifying samples and have zero interventions.
81    pub stats: ProcessStats,
82    /// This span ends the track's produced output, including delayed SRC frames
83    /// in a processed session. It does not acknowledge device consumption.
84    /// An empty range may deliver completion without additional PCM.
85    pub complete: bool,
86}
87
88/// Processed samples and their exact track attribution.
89pub struct SessionBlock<'a> {
90    /// Finite, guarded, interleaved output PCM.
91    pub samples: &'a [f32],
92    /// The same samples encoded as little-endian `f32`.
93    pub f32le: &'a [u8],
94    /// Ordered spans covering every output sample exactly once. Empty spans
95    /// report track completion without introducing samples.
96    pub spans: &'a [TrackSpan],
97}
98
99struct PendingTrack {
100    token: usize,
101    gain_db: f32,
102    end_frame: Option<u64>,
103}
104
105/// A continuous output processor for successive naturally ending tracks.
106///
107/// Input PCM has already been decoded and optionally downmixed. Compatible
108/// tracks share the converter, fractional frame accounting and tone-filter
109/// state. Gain is selected separately for each track after rate conversion.
110/// Output frame boundaries are `ceil(cumulative source frames × rate ratio)`.
111/// A discontinuity, format or tone change requires a new session.
112///
113/// Call `end_track` only after the source was completely decoded, then start
114/// the next compatible track or call `finish` once at the end of the group.
115/// Returned views remain valid until the next mutable session call. Incomplete
116/// frames, non-finite input and lifecycle requests leave state unchanged; after a
117/// processing, conversion or timeline failure, the driver must discard the
118/// session rather than retrying partially consumed PCM.
119pub struct PcmSession {
120    input: PcmFormat,
121    output: PcmFormat,
122    tone: ToneControls,
123    dsp: Dsp,
124    converter: Option<RateConverter>,
125    tracks: VecDeque<PendingTrack>,
126    active: bool,
127    source_frames: u64,
128    output_frames: u64,
129    finished: bool,
130    samples: Vec<f32>,
131    bytes: Vec<u8>,
132    spans: Vec<TrackSpan>,
133}
134
135impl PcmSession {
136    /// Prepare a fixed PCM format, optional rate conversion and persistent tone
137    /// shelves outside the output callback.
138    pub fn new(
139        input: PcmFormat,
140        output_rate: u32,
141        tone: ToneControls,
142    ) -> Result<Self, SessionError> {
143        let output = PcmFormat::with_layout(output_rate, input.layout())?;
144        let converter = if input.sample_rate() == output_rate {
145            None
146        } else {
147            Some(RateConverter::new(input, output_rate)?)
148        };
149        let mut dsp = Dsp::new(output);
150        dsp.set_tone(tone)?;
151        Ok(Self {
152            input,
153            output,
154            tone,
155            dsp,
156            converter,
157            tracks: VecDeque::new(),
158            active: false,
159            source_frames: 0,
160            output_frames: 0,
161            finished: false,
162            samples: Vec::new(),
163            bytes: Vec::new(),
164            spans: Vec::new(),
165        })
166    }
167
168    /// Source PCM format after any downmix and before rate conversion.
169    pub fn input_format(&self) -> PcmFormat {
170        self.input
171    }
172
173    /// Guarded PCM format accepted by the output sink.
174    pub fn output_format(&self) -> PcmFormat {
175        self.output
176    }
177
178    /// Whether the next naturally advancing track can share this unfinished
179    /// processing session. Channel positions are part of compatibility.
180    pub fn compatible(&self, input: PcmFormat, output_rate: u32, tone: ToneControls) -> bool {
181        !self.finished
182            && input == self.input
183            && output_rate == self.output.sample_rate()
184            && tone == self.tone
185    }
186
187    /// Start a new source track using a fixed final gain in dB. The driver
188    /// includes any normalization and tone headroom reserve in this gain.
189    /// Pending output from earlier tracks retains their own gains.
190    pub fn begin_track(&mut self, token: usize, gain_db: f32) -> Result<(), SessionError> {
191        if self.finished {
192            return Err(SessionError::Finished);
193        }
194        if self.active {
195            return Err(SessionError::TrackActive);
196        }
197        let mut validation = Dsp::new(self.output);
198        validation.set_gain_db(gain_db, 0)?;
199        self.tracks.push_back(PendingTrack {
200            token,
201            gain_db,
202            end_frame: None,
203        });
204        self.active = true;
205        Ok(())
206    }
207
208    /// Supply complete finite source frames. A converter may retain source
209    /// frames and return no output until more PCM arrives or the group ends.
210    pub fn push_source(&mut self, samples: &[f32]) -> Result<SessionBlock<'_>, SessionError> {
211        if self.finished {
212            return Err(SessionError::Finished);
213        }
214        if !self.active {
215            return Err(SessionError::NoTrack);
216        }
217        let channels = usize::from(self.input.channels());
218        if !samples.len().is_multiple_of(channels) {
219            return Err(DspError::IncompleteFrame.into());
220        }
221        if samples.iter().any(|sample| !sample.is_finite()) {
222            return Err(DspError::NonFiniteSample.into());
223        }
224        let frames = u64::try_from(samples.len() / channels)
225            .map_err(|_| SessionError::FrameCountOverflow)?;
226        let source_frames = self
227            .source_frames
228            .checked_add(frames)
229            .ok_or(SessionError::FrameCountOverflow)?;
230        self.output_boundary(source_frames)?;
231        self.clear_block();
232        if let Some(converter) = &mut self.converter {
233            self.samples
234                .extend_from_slice(converter.push(samples, false)?);
235        } else {
236            self.samples.extend_from_slice(samples);
237        }
238        self.source_frames = source_frames;
239        self.process_block()?;
240        Ok(self.block())
241    }
242
243    /// Seal the active track's source boundary without flushing the converter
244    /// or resetting filters. Completion is emitted only once every output
245    /// frame assigned to this track has been produced.
246    pub fn end_track(&mut self) -> Result<SessionBlock<'_>, SessionError> {
247        if self.finished {
248            return Err(SessionError::Finished);
249        }
250        if !self.active {
251            return Err(SessionError::NoTrack);
252        }
253        let end_frame = self.output_boundary(self.source_frames)?;
254        let track = self.tracks.back_mut().ok_or(SessionError::OutputTimeline)?;
255        track.end_frame = Some(end_frame);
256        self.active = false;
257        self.clear_block();
258        self.complete_ready_tracks(0)?;
259        Ok(self.block())
260    }
261
262    /// Flush the converter once after every source track has been sealed.
263    /// Repeated calls return an empty block without duplicate completions.
264    pub fn finish(&mut self) -> Result<SessionBlock<'_>, SessionError> {
265        if self.active {
266            return Err(SessionError::TrackActive);
267        }
268        self.clear_block();
269        if self.finished {
270            return Ok(self.block());
271        }
272        if let Some(converter) = &mut self.converter {
273            self.samples.extend_from_slice(converter.push(&[], true)?);
274        }
275        self.process_block()?;
276        if !self.tracks.is_empty()
277            || self.output_frames != self.output_boundary(self.source_frames)?
278        {
279            return Err(SessionError::OutputTimeline);
280        }
281        self.finished = true;
282        Ok(self.block())
283    }
284
285    fn output_boundary(&self, frames: u64) -> Result<u64, SessionError> {
286        let boundary = (u128::from(frames) * u128::from(self.output.sample_rate()))
287            .div_ceil(u128::from(self.input.sample_rate()));
288        u64::try_from(boundary).map_err(|_| SessionError::FrameCountOverflow)
289    }
290
291    fn clear_block(&mut self) {
292        self.samples.clear();
293        self.bytes.clear();
294        self.spans.clear();
295    }
296
297    fn block(&self) -> SessionBlock<'_> {
298        SessionBlock {
299            samples: &self.samples,
300            f32le: &self.bytes,
301            spans: &self.spans,
302        }
303    }
304
305    fn complete_ready_tracks(&mut self, sample_index: usize) -> Result<(), SessionError> {
306        while let Some(track) = self.tracks.front() {
307            let Some(end) = track.end_frame else {
308                break;
309            };
310            if end < self.output_frames {
311                return Err(SessionError::OutputTimeline);
312            }
313            if end != self.output_frames {
314                break;
315            }
316            self.spans.push(TrackSpan {
317                token: track.token,
318                samples: sample_index..sample_index,
319                stats: ProcessStats::default(),
320                complete: true,
321            });
322            self.tracks.pop_front();
323        }
324        Ok(())
325    }
326
327    fn process_block(&mut self) -> Result<(), SessionError> {
328        let channels = usize::from(self.output.channels());
329        if !self.samples.len().is_multiple_of(channels) {
330            return Err(SessionError::OutputTimeline);
331        }
332        let current_boundary = self.output_boundary(self.source_frames)?;
333        let mut offset = 0;
334        self.complete_ready_tracks(offset)?;
335        while offset < self.samples.len() {
336            let track = self.tracks.front().ok_or(SessionError::OutputTimeline)?;
337            let boundary = track.end_frame.unwrap_or(current_boundary);
338            let available = boundary
339                .checked_sub(self.output_frames)
340                .ok_or(SessionError::OutputTimeline)?;
341            let frames = (self.samples.len() - offset) / channels;
342            let accepted = frames.min(usize::try_from(available).unwrap_or(usize::MAX));
343            if accepted == 0 {
344                return Err(SessionError::OutputTimeline);
345            }
346            let end = offset + accepted * channels;
347            let token = track.token;
348            let gain_db = track.gain_db;
349            let sealed = track.end_frame.is_some();
350            self.dsp.set_gain_db(gain_db, 0)?;
351            let stats = self
352                .dsp
353                .process_for_output(&mut self.samples[offset..end])?;
354            self.output_frames = self
355                .output_frames
356                .checked_add(accepted as u64)
357                .ok_or(SessionError::FrameCountOverflow)?;
358            let complete = sealed && self.output_frames == boundary;
359            self.spans.push(TrackSpan {
360                token,
361                samples: offset..end,
362                stats,
363                complete,
364            });
365            if complete {
366                self.tracks.pop_front();
367            }
368            offset = end;
369            self.complete_ready_tracks(offset)?;
370        }
371        self.bytes.reserve(self.samples.len() * 4);
372        for sample in &self.samples {
373            self.bytes.extend_from_slice(&sample.to_le_bytes());
374        }
375        Ok(())
376    }
377}
378
379#[cfg(test)]
380#[path = "session_tests.rs"]
381mod tests;
aede_core/playback/session.rs

aede_core/playback/
session.rs
