Add audio-player PoC

Cargo.toml

@@ -1,2 +1,8 @@
 [workspace]
-members = ["cbd-tui", "crabidy-core", "crabidy-server", "tidaldy"]
+members = [
+  "audio-player",
+  "cbd-tui",
+  "crabidy-core",
+  "crabidy-server",
+  "tidaldy",
+]

audio-player/Cargo.toml

@@ -0,0 +1,11 @@
+[package]
+name = "audio-player"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+rodio = { version = "0.17.1", default-features = false, features = ["symphonia-all"] }
+symphonia = { version = "0.5.3", features = ["all"] }
+stream-download = { git = "https://github.com/aschey/stream-download-rs.git" }
+anyhow = "1.0.71"
+url = "2.4.0"

audio-player/src/decoder.rs

@@ -0,0 +1,302 @@
+use std::error::Error;
+use std::fmt;
+use std::time::Duration;
+use symphonia::{
+    core::{
+        audio::{AudioBufferRef, SampleBuffer, SignalSpec},
+        codecs::{Decoder, DecoderOptions},
+        errors::Error as SymphoniaError,
+        formats::{FormatOptions, FormatReader, SeekMode, SeekTo, Track},
+        io::MediaSourceStream,
+        meta::{MetadataOptions, MetadataRevision},
+        probe::Hint,
+        units::{self, Time, TimeBase},
+    },
+    default::get_probe,
+};
+
+use rodio::Source;
+
+// Decoder errors are not considered fatal.
+// The correct action is to just get a new packet and try again.
+// But a decode error in more than 3 consecutive packets is fatal.
+const MAX_DECODE_ERRORS: usize = 3;
+
+pub struct MediaInfo {
+    pub duration: Option<Duration>,
+    pub metadata: Option<MetadataRevision>,
+    pub track: Track,
+}
+
+pub struct SymphoniaDecoder {
+    decoder: Box<dyn Decoder>,
+    current_frame_offset: usize,
+    format: Box<dyn FormatReader>,
+    buffer: SampleBuffer<i16>,
+    spec: SignalSpec,
+    time_base: Option<TimeBase>,
+    duration: Option<Duration>,
+    elapsed: u64,
+    metadata: Option<MetadataRevision>,
+    track: Track,
+}
+
+impl SymphoniaDecoder {
+    pub fn new(mss: MediaSourceStream, hint: Hint) -> Result<Self, DecoderError> {
+        match SymphoniaDecoder::init(mss, hint) {
+            Err(e) => match e {
+                SymphoniaError::IoError(e) => Err(DecoderError::IoError(e.to_string())),
+                SymphoniaError::DecodeError(e) => Err(DecoderError::DecodeError(e)),
+                SymphoniaError::SeekError(_) => {
+                    unreachable!("Seek errors should not occur during initialization")
+                }
+                SymphoniaError::Unsupported(_) => Err(DecoderError::UnrecognizedFormat),
+                SymphoniaError::LimitError(e) => Err(DecoderError::LimitError(e)),
+                SymphoniaError::ResetRequired => Err(DecoderError::ResetRequired),
+            },
+            Ok(Some(decoder)) => Ok(decoder),
+            Ok(None) => Err(DecoderError::NoStreams),
+        }
+    }
+
+    pub fn into_inner(self) -> MediaSourceStream {
+        self.format.into_inner()
+    }
+
+    fn init(
+        mss: MediaSourceStream,
+        hint: Hint,
+    ) -> symphonia::core::errors::Result<Option<SymphoniaDecoder>> {
+        let format_opts: FormatOptions = FormatOptions {
+            enable_gapless: true,
+            ..Default::default()
+        };
+        let metadata_opts: MetadataOptions = Default::default();
+        let mut probed = get_probe().format(&hint, mss, &format_opts, &metadata_opts)?;
+
+        let track = match probed.format.default_track() {
+            Some(stream) => stream,
+            None => return Ok(None),
+        }
+        .clone();
+
+        let time_base = track.codec_params.time_base;
+
+        let dur = track
+            .codec_params
+            .n_frames
+            .map(|frames| track.codec_params.start_ts + frames)
+            .unwrap_or_default();
+
+        let duration = match time_base {
+            Some(tb) => {
+                let time = tb.calc_time(dur);
+                Some(Duration::from_secs_f64(time.seconds as f64 + time.frac))
+            }
+            None => None,
+        };
+
+        let mut elapsed = 0;
+
+        let mut decoder = symphonia::default::get_codecs()
+            .make(&track.codec_params, &DecoderOptions { verify: true })?;
+
+        let mut decode_errors: usize = 0;
+        let decoded = loop {
+            let current_frame = probed.format.next_packet()?;
+            elapsed = current_frame.ts();
+            match decoder.decode(&current_frame) {
+                Ok(decoded) => break decoded,
+                Err(e) => match e {
+                    SymphoniaError::DecodeError(_) => {
+                        decode_errors += 1;
+                        if decode_errors > MAX_DECODE_ERRORS {
+                            return Err(e);
+                        } else {
+                            continue;
+                        }
+                    }
+                    _ => return Err(e),
+                },
+            }
+        };
+        let spec = decoded.spec().to_owned();
+        let buffer = SymphoniaDecoder::get_buffer(decoded, &spec);
+
+        // Prefer metadata that's provided in the container format, over other tags found during the
+        // probe operation.
+        let metadata = probed.format.metadata().current().cloned().or_else(|| {
+            probed
+                .metadata
+                .get()
+                .as_ref()
+                .and_then(|m| m.current().cloned())
+        });
+
+        Ok(Some(SymphoniaDecoder {
+            decoder,
+            current_frame_offset: 0,
+            format: probed.format,
+            buffer,
+            spec,
+            time_base,
+            duration,
+            elapsed,
+            metadata,
+            track,
+        }))
+    }
+
+    #[inline]
+    pub fn media_info(&self) -> MediaInfo {
+        MediaInfo {
+            duration: self.duration,
+            metadata: self.metadata.clone(),
+            track: self.track.clone(),
+        }
+    }
+
+    #[inline]
+    pub fn elapsed(&self) -> Duration {
+        if let Some(tb) = self.time_base {
+            let time = tb.calc_time(self.elapsed);
+            return Duration::from_secs_f64(time.seconds as f64 + time.frac);
+        };
+        Duration::default()
+    }
+
+    #[inline]
+    pub fn seek(&mut self, time: Duration) -> Option<Duration> {
+        let nanos_per_sec = 1_000_000_000.0;
+        match self.format.seek(
+            SeekMode::Coarse,
+            SeekTo::Time {
+                time: Time::new(
+                    time.as_secs(),
+                    f64::from(time.subsec_nanos()) / nanos_per_sec,
+                ),
+                track_id: None,
+            },
+        ) {
+            Ok(seeked_to) => {
+                let base = TimeBase::new(1, self.sample_rate());
+                let time = base.calc_time(seeked_to.actual_ts);
+
+                Some(Duration::from_millis(
+                    time.seconds * 1000 + ((time.frac * 60. * 1000.).round() as u64),
+                ))
+            }
+            Err(_) => None,
+        }
+    }
+
+    #[inline]
+    fn get_buffer(decoded: AudioBufferRef, spec: &SignalSpec) -> SampleBuffer<i16> {
+        let duration = decoded.capacity() as u64;
+        let mut buffer = SampleBuffer::<i16>::new(duration, *spec);
+        buffer.copy_interleaved_ref(decoded);
+        buffer
+    }
+}
+
+impl Source for SymphoniaDecoder {
+    #[inline]
+    fn current_frame_len(&self) -> Option<usize> {
+        Some(self.buffer.samples().len())
+    }
+
+    #[inline]
+    fn channels(&self) -> u16 {
+        self.spec.channels.count() as u16
+    }
+
+    #[inline]
+    fn sample_rate(&self) -> u32 {
+        self.spec.rate
+    }
+
+    #[inline]
+    fn total_duration(&self) -> Option<Duration> {
+        self.duration
+    }
+}
+
+impl Iterator for SymphoniaDecoder {
+    type Item = i16;
+
+    #[inline]
+    fn next(&mut self) -> Option<i16> {
+        if self.current_frame_offset == self.buffer.len() {
+            let mut decode_errors: usize = 0;
+            let decoded = loop {
+                match self.format.next_packet() {
+                    Ok(packet) => {
+                        self.elapsed = packet.ts();
+                        match self.decoder.decode(&packet) {
+                            Ok(decoded) => break decoded,
+                            Err(e) => match e {
+                                SymphoniaError::DecodeError(_) => {
+                                    decode_errors += 1;
+                                    if decode_errors > MAX_DECODE_ERRORS {
+                                        return None;
+                                    } else {
+                                        continue;
+                                    }
+                                }
+                                _ => return None,
+                            },
+                        }
+                    }
+                    Err(_) => return None,
+                }
+            };
+            self.spec = decoded.spec().to_owned();
+            self.buffer = SymphoniaDecoder::get_buffer(decoded, &self.spec);
+            self.current_frame_offset = 0;
+        }
+
+        let sample = *self.buffer.samples().get(self.current_frame_offset)?;
+        self.current_frame_offset += 1;
+
+        Some(sample)
+    }
+}
+
+/// Error that can happen when creating a decoder.
+#[derive(Debug, Clone)]
+pub enum DecoderError {
+    /// The format of the data has not been recognized.
+    UnrecognizedFormat,
+
+    /// An IO error occurred while reading, writing, or seeking the stream.
+    IoError(String),
+
+    /// The stream contained malformed data and could not be decoded or demuxed.
+    DecodeError(&'static str),
+
+    /// A default or user-defined limit was reached while decoding or demuxing the stream. Limits
+    /// are used to prevent denial-of-service attacks from malicious streams.
+    LimitError(&'static str),
+
+    /// The demuxer or decoder needs to be reset before continuing.
+    ResetRequired,
+
+    /// No streams were found by the decoder
+    NoStreams,
+}
+
+impl fmt::Display for DecoderError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let text = match self {
+            DecoderError::UnrecognizedFormat => "Unrecognized format",
+            DecoderError::IoError(msg) => &msg[..],
+            DecoderError::DecodeError(msg) => msg,
+            DecoderError::LimitError(msg) => msg,
+            DecoderError::ResetRequired => "Reset required",
+            DecoderError::NoStreams => "No streams",
+        };
+        write!(f, "{}", text)
+    }
+}
+
+impl Error for DecoderError {}

audio-player/src/main.rs

@@ -0,0 +1,141 @@
+mod decoder;
+
+use std::fs::File;
+use std::io::BufReader;
+use std::path::Path;
+use std::thread;
+use std::time::Duration;
+use symphonia::core::probe::Hint;
+use url::Url;
+
+use anyhow::{anyhow, Result};
+use decoder::SymphoniaDecoder;
+use rodio::source::{PeriodicAccess, SineWave};
+use rodio::{OutputStream, OutputStreamHandle, Sink, Source};
+use stream_download::StreamDownload;
+use symphonia::core::io::{
+    MediaSource, MediaSourceStream, MediaSourceStreamOptions, ReadOnlySource,
+};
+
+struct Player {
+    sink: Option<Sink>,
+    stream: Option<OutputStream>,
+}
+
+// TODO:
+// * Emit Metadata
+// * Emit duration
+// * Emit track data
+// * Emit EOS
+// * Emit buffering
+
+impl Player {
+    pub fn default() -> Self {
+        Self {
+            sink: None,
+            stream: None,
+        }
+    }
+
+    pub fn play(&mut self, source_str: &str) -> Result<()> {
+        let (stream, handle) = OutputStream::try_default()?;
+        let mut sink = Sink::try_new(&handle)?;
+        let (source, hint) = self.get_source(source_str)?;
+        let mss = MediaSourceStream::new(source, MediaSourceStreamOptions::default());
+
+        let decoder = SymphoniaDecoder::new(mss, hint)?;
+
+        let media_info = decoder.media_info();
+
+        let decoder = decoder.periodic_access(Duration::from_millis(500), |src| {
+            println!("ELAPSED: {:?}", src.elapsed());
+
+            if src.elapsed().as_secs() > 10 {
+                src.seek(Duration::from_secs(2));
+            }
+        });
+
+        sink.append(decoder);
+
+        // We need to keep the stream around, otherwise it gets dropped outside of this scope
+        self.stream = Some(stream);
+        // The sink is used to control the stream
+        self.sink = Some(sink);
+
+        Ok(())
+    }
+
+    pub fn pause(&mut self) {
+        if let Some(sink) = &self.sink {
+            sink.pause();
+        }
+    }
+
+    pub fn unpause(&mut self) {
+        if let Some(sink) = &self.sink {
+            sink.play();
+        }
+    }
+
+    pub fn stop(&mut self) {
+        if let Some(sink) = &self.sink {
+            sink.stop();
+            self.sink.take();
+            self.stream.take();
+        }
+    }
+
+    pub fn is_playing(&self) -> bool {
+        self.sink.as_ref().map(|s| !s.is_paused()).unwrap_or(false)
+    }
+
+    pub fn is_paused(&self) -> bool {
+        self.sink.as_ref().map(|s| s.is_paused()).unwrap_or(false)
+    }
+
+    pub fn is_stopped(&self) -> bool {
+        self.sink.is_none()
+    }
+
+    fn get_source(&self, source_str: &str) -> Result<(Box<dyn MediaSource>, Hint)> {
+        match Url::parse(source_str) {
+            Ok(url) => {
+                if let "http" | "https" = url.scheme() {
+                    let reader = StreamDownload::new_http(source_str.parse().unwrap());
+                    let path = Path::new(url.path());
+                    let hint = self.get_hint(path);
+
+                    Ok((Box::new(ReadOnlySource::new(reader)), hint))
+                } else {
+                    Err(anyhow!("Not a valid URL scheme: {}", url.scheme()))
+                }
+            }
+            Err(_) => {
+                let path = Path::new(source_str);
+                let hint = self.get_hint(path);
+                Ok((Box::new(File::open(path)?), hint))
+            }
+        }
+    }
+
+    fn get_hint(&self, path: &Path) -> Hint {
+        // Create a hint to help the format registry guess what format reader is appropriate.
+        let mut hint = Hint::new();
+        // Provide the file extension as a hint.
+        if let Some(extension) = path.extension() {
+            if let Some(extension_str) = extension.to_str() {
+                hint.with_extension(extension_str);
+            }
+        }
+        hint
+    }
+}
+
+fn main() {
+    let mut player = Player::default();
+    player.play("./Slip.m4a");
+
+    thread::sleep(Duration::from_millis(5000));
+
+    player.stop();
+}