Add playback progress bar to d-bus

interpreter/src/processing/music_analysis.rs

@@ -682,13 +682,13 @@ fn worker_distance(
                 song: new_song2.clone().map(Box::new),
             })
             .unwrap_or(());
-        if new_song2.is_err() {
+        /*if new_song2.is_err() {
             eprintln!(
                 "Song error on `{}`: {}",
                 path2,
                 new_song2.clone().err().unwrap()
             );
-        }
+        }*/
         new_song2?
     };
     Ok(song1.distance(&song2))

player/src/os_controls.rs

@@ -39,6 +39,7 @@ pub struct SystemControlWrapper {
 enum DbusControl {
     Die,
     SetMetadata(Metadata),
+    SetPosition(i64),
 }
 
 #[cfg(all(target_os = "linux", feature = "os-controls", feature = "mpris-player"))]
@@ -65,10 +66,11 @@ impl SystemControlWrapper {
             //self.dbus_conn.set_supported_mime_types(vec![]);
             //self.dbus_conn.set_supported_uri_schemes(vec![]);
             let mut is_playing = true;
-            dbus_conn.set_playback_status(PlaybackStatus::Playing);
+            dbus_conn.set_playback_status(PlaybackStatus::Stopped);
             dbus_conn.set_can_play(true);
             dbus_conn.set_can_pause(true);
             dbus_conn.set_can_go_next(true);
+            dbus_conn.set_can_seek(false);
 
             let control_clone = control_clone1.clone();
             dbus_conn.connect_next(move || {
@@ -133,6 +135,8 @@ impl SystemControlWrapper {
                     .unwrap_or(())
             });
 
+            dbus_conn.set_playback_status(PlaybackStatus::Playing);
+
             // poll loop, using my custom mpris lib because original did it wrong
             loop {
                 dbus_conn.poll(5);
@@ -141,21 +145,40 @@ impl SystemControlWrapper {
                     Ok(DbusControl::Die) => break,
                     Ok(DbusControl::SetMetadata(meta)) => {
                         dbus_conn.set_metadata(meta);
+                    },
+                    Ok(DbusControl::SetPosition(pos)) => {
+                        dbus_conn.set_position(pos);
                     }
                 }
             }
         }));
         let (tx, rx) = channel();
         self.playback_event_handler_killer = Some(tx);
-        self.playback_event_handler = Some(std::thread::spawn(move || loop {
+        self.playback_event_handler = Some(std::thread::spawn(move || {
-            if let Ok(_) = rx.try_recv() {
+            let mut playback_time = 0;
-                break;
+            let mut duration_cache = None;
-            }
+            loop {
-            match playback.recv() {
+                if let Ok(_) = rx.try_recv() {
-                Err(_) => break,
+                    break;
-                Ok(PlaybackAction::Exit) => break,
+                }
-                Ok(PlaybackAction::Enqueued(item)) => Self::enqueued(item, &dbus_ctrl_tx_clone),
+                match playback.recv() {
-                Ok(PlaybackAction::Empty) => Self::empty(&dbus_ctrl_tx_clone),
+                    Err(_) => break,
+                    Ok(PlaybackAction::Exit) => break,
+                    Ok(PlaybackAction::Enqueued(item)) => {
+                        playback_time = 0;
+                        duration_cache = None;
+                        Self::enqueued(item, &dbus_ctrl_tx_clone);
+                    },
+                    Ok(PlaybackAction::Empty) => Self::empty(&dbus_ctrl_tx_clone),
+                    Ok(PlaybackAction::Time(item, duration)) => {
+                        duration_cache = Some(duration);
+                        Self::time(item, duration, &dbus_ctrl_tx_clone);
+                    },
+                    Ok(PlaybackAction::UpdateTick(item)) => {
+                        Self::time_update(item, playback_time, &duration_cache, &dbus_ctrl_tx_clone);
+                        playback_time += 1;
+                    },
+                }
             }
         }));
     }
@@ -177,32 +200,42 @@ impl SystemControlWrapper {
         }
     }
 
+    fn build_metadata(item: Item) -> Metadata {
+        let file_uri = item.field("filename").and_then(|x| x.to_owned().to_str());
+        Metadata {
+            length: None,
+            art_url: None, //file_uri.clone() TODO do this without having to rip the art image from the file like Elisa
+            album: item.field("album").and_then(|x| x.to_owned().to_str()),
+            album_artist: item
+                .field("albumartist")
+                .map(
+                    |x| x.to_owned()
+                        .to_str()
+                        .map(|x2| vec![x2])
+                ).flatten(),
+            artist: item
+                .field("artist")
+                .and_then(|x| x.to_owned().to_str())
+                .map(|x| x.split(",").map(|s| s.trim().to_owned()).collect()),
+            composer: None,
+            disc_number: None,
+            genre: item
+                .field("genre")
+                .and_then(|x| x.to_owned().to_str())
+                .map(|genre| vec![genre]),
+            title: item.field("title").and_then(|x| x.to_owned().to_str()),
+            track_number: item
+                .field("track")
+                .and_then(|x| x.to_owned().to_i64())
+                .map(|track| track as i32),
+            url: file_uri,
+        }
+    }
+
     fn enqueued(item: Item, dbus_ctrl: &Sender<DbusControl>) {
         //println!("Got enqueued item {}", &item.title);
-        let file_uri = item.field("filename").and_then(|x| x.to_owned().to_str());
         dbus_ctrl
-            .send(DbusControl::SetMetadata(Metadata {
+            .send(DbusControl::SetMetadata(Self::build_metadata(item)))
-                length: None,
-                art_url: None, //file_uri.clone() TODO do this without having to rip the art image from the file like Elisa
-                album: item.field("album").and_then(|x| x.to_owned().to_str()),
-                album_artist: None, // TODO maybe?
-                artist: item
-                    .field("artist")
-                    .and_then(|x| x.to_owned().to_str())
-                    .map(|x| x.split(",").map(|s| s.trim().to_owned()).collect()),
-                composer: None,
-                disc_number: None,
-                genre: item
-                    .field("genre")
-                    .and_then(|x| x.to_owned().to_str())
-                    .map(|genre| vec![genre]),
-                title: item.field("title").and_then(|x| x.to_owned().to_str()),
-                track_number: item
-                    .field("track")
-                    .and_then(|x| x.to_owned().to_i64())
-                    .map(|track| track as i32),
-                url: file_uri,
-            }))
             .unwrap_or(());
     }
 
@@ -212,7 +245,7 @@ impl SystemControlWrapper {
                 length: None,
                 art_url: None,
                 album: None,
-                album_artist: None, // TODO maybe?
+                album_artist: None,
                 artist: None,
                 composer: None,
                 disc_number: None,
@@ -223,6 +256,28 @@ impl SystemControlWrapper {
             }))
             .unwrap_or(());
     }
+
+    fn time(item: Item, duration: std::time::Duration, dbus_ctrl: &Sender<DbusControl>) {
+        let mut meta = Self::build_metadata(item);
+        meta.length = Some(duration.as_secs_f64().round() as i64 * 1_000_000);
+        dbus_ctrl
+            .send(DbusControl::SetMetadata(meta))
+            .unwrap_or(());
+    }
+
+    fn time_update(_item: Item, new_time: i64, duration: &Option<std::time::Duration>, dbus_ctrl: &Sender<DbusControl>) {
+        //println!("Position update tick");
+        if duration.is_some() {
+            /*let mut meta = Self::build_metadata(item);
+            meta.length = Some(new_time + 1);
+            dbus_ctrl
+                .send(DbusControl::SetMetadata(meta))
+                .unwrap_or(());*/
+            dbus_ctrl
+                .send(DbusControl::SetPosition(new_time * 1_000_000))
+                .unwrap_or(());
+        }
+    }
 }
 
 #[cfg(any(

player/src/player.rs

@@ -1,7 +1,7 @@
 use std::fs;
 use std::io;
 
-use rodio::{decoder::Decoder, OutputStream, OutputStreamHandle, Sink};
+use rodio::{decoder::Decoder, OutputStream, OutputStreamHandle, Sink, Source};
 
 use m3u8_rs::{MediaPlaylist, MediaSegment};
 
@@ -133,6 +133,38 @@ impl<I: std::iter::Iterator<Item = Result<Item, InterpreterError>>> Player<I> {
         Ok(enqueued)
     }
 
+    pub fn enqueue_modified(&mut self, count: usize, modify: &dyn Fn(Decoder<io::BufReader<fs::File>>, Item) -> Box<dyn Source<Item=i16> + Send>) -> Result<Vec<Item>, PlayerError> {
+        let mut items_left = count;
+        let mut enqueued = Vec::with_capacity(count);
+        if items_left == 0 {
+            return Ok(enqueued);
+        }
+        while let Some(item) = self.runner.next() {
+            match item {
+                Ok(music) => {
+                    if let Some(filename) =
+                        music.field("filename").and_then(|x| x.to_owned().to_str())
+                    {
+                        enqueued.push(music.clone());
+                        self.append_source_modified(&filename, &|x| modify(x, music.clone()))?;
+                        items_left -= 1;
+                        Ok(())
+                    } else {
+                        Err(PlayerError::from_err_playback(
+                            "Field `filename` does not exist on item",
+                        ))
+                    }
+                }
+                Err(e) => Err(PlayerError::from_err_playback(e)),
+            }?;
+            if items_left == 0 {
+                break;
+            }
+        }
+        //println!("Enqueued {} items", count - items_left);
+        Ok(enqueued)
+    }
+
     pub fn resume(&self) {
         self.sink.play()
     }
@@ -251,6 +283,50 @@ impl<I: std::iter::Iterator<Item = Result<Item, InterpreterError>>> Player<I> {
             }
         }
     }
+
+    fn append_source_modified(&mut self, filename: &str, modify: &dyn Fn(Decoder<io::BufReader<fs::File>>) -> Box<dyn Source<Item=i16> + Send>) -> Result<(), PlayerError> {
+        let uri = Uri::new(filename);
+        match uri.scheme() {
+            Some(s) => match &s.to_lowercase() as &str {
+                "file:" => {
+                    let file =
+                        fs::File::open(uri.path()).map_err(PlayerError::from_err_playback)?;
+                    let stream = io::BufReader::new(file);
+                    let source = Decoder::new(stream).map_err(PlayerError::from_err_playback)?;
+                    self.sink.append(modify(source));
+                    Ok(())
+                }
+                #[cfg(feature = "mpd")]
+                "mpd:" => {
+                    if let Some(mpd_client) = &mut self.mpd_connection {
+                        //println!("Pushing {} into MPD queue", uri.path());
+                        let song = Song {
+                            file: uri.path().to_owned(),
+                            ..Default::default()
+                        };
+                        mpd_client
+                            .push(song)
+                            .map_err(PlayerError::from_err_playback)?;
+                        Ok(())
+                    } else {
+                        Err(PlayerError::from_err_playback(
+                            "Cannot play MPD song: no MPD client connected",
+                        ))
+                    }
+                }
+                scheme => Err(UriError::Unsupported(scheme.to_owned()).into()),
+            },
+            None => {
+                //default
+                // NOTE: Default rodio::Decoder hangs here when decoding large files, but symphonia does not
+                let file = fs::File::open(uri.path()).map_err(PlayerError::from_err_playback)?;
+                let stream = io::BufReader::new(file);
+                let source = Decoder::new(stream).map_err(PlayerError::from_err_playback)?;
+                self.sink.append(source);
+                Ok(())
+            }
+        }
+    }
 }
 
 #[cfg(feature = "mpd")]

player/src/player_wrapper.rs

@@ -1,6 +1,8 @@
 use std::sync::mpsc::{Receiver, Sender};
 use std::{thread, thread::JoinHandle};
 
+use rodio::Source;
+
 //use muss_interpreter::tokens::TokenReader;
 use muss_interpreter::{InterpreterError, Item};
 
@@ -36,9 +38,68 @@ impl<I: std::iter::Iterator<Item = Result<Item, InterpreterError>>> PlayerServer
         }
     }
 
+    fn modify(&self) -> impl Fn(rodio::Decoder<std::io::BufReader<std::fs::File>>, Item) -> Box<dyn Source<Item=i16> + Send> {
+        let event = std::sync::Arc::new(std::sync::Mutex::new(self.playback.clone()));
+        move |source_in, item| {
+            let event2 = event.clone();
+            if let Some(duration) = source_in.total_duration() {
+                event.lock().map(|event|
+                    event.send(
+                        PlaybackAction::Time(item.clone(), duration)
+                    ).unwrap_or(())
+                ).unwrap_or(());
+                Box::new(
+                    source_in.periodic_access(
+                        std::time::Duration::from_secs(1),
+                        move |_|
+                            {
+                                //println!("Debug tick");
+                                event2.lock()
+                                    .map(|x|
+                                        x.send(PlaybackAction::UpdateTick(item.clone())).unwrap_or(())
+                                    )
+                                    .unwrap_or(());
+                            }
+                    )
+                )
+            } else {
+                // manually calculate length
+                let source_in = source_in.buffered();
+                let source_in2 = source_in.clone();
+                let event3 = event.clone();
+                let item2 = item.clone();
+                // Iterator.count() takes a while, so calculate in a different thread
+                std::thread::spawn(move || {
+                    let sample_rate = source_in2.sample_rate();
+                    let channels = source_in2.channels() as u32;
+                    let sample_count = source_in2.clone().count() as f64;
+                    let duration = std::time::Duration::from_secs_f64(sample_count / ((sample_rate * channels) as f64));
+                    event3.lock().map(|event|
+                        event.send(
+                            PlaybackAction::Time(item2.clone(), duration)
+                        ).unwrap_or(())
+                    ).unwrap_or(());
+                });
+                Box::new(
+                    source_in.periodic_access(
+                        std::time::Duration::from_secs(1),
+                        move |_|
+                            {
+                                event2.lock()
+                                    .map(|x|
+                                        x.send(PlaybackAction::UpdateTick(item.clone())).unwrap_or(())
+                                    )
+                                    .unwrap_or(());
+                            }
+                    )
+                )
+            }
+        }
+    }
+
     fn enqeue_some(&mut self, count: usize) {
         //println!("Enqueuing up to {} items", count);
-        match self.player.enqueue(count) {
+        match self.player.enqueue_modified(count, &self.modify()) {
             Err(e) => {
                 self.event.send(PlayerAction::Exception(e)).unwrap();
             }
@@ -221,6 +282,8 @@ pub enum PlayerAction {
 pub enum PlaybackAction {
     Empty,
     Enqueued(Item),
+    Time(Item, std::time::Duration),
+    UpdateTick(Item), // tick sent once every second
     Exit,
 }