pkgs/mpdpopm: Add a (basic) dj mode

7 files changed, 429 insertions, 4 deletions

diff --git a/pkgs/by-name/mp/mpdpopm/Cargo.lock b/pkgs/by-name/mp/mpdpopm/Cargo.lock
index 96909646..8b61799a 100644
--- a/pkgs/by-name/mp/mpdpopm/Cargo.lock
+++ b/pkgs/by-name/mp/mpdpopm/Cargo.lock
@@ -403,6 +403,18 @@ dependencies = [
 ]
 
 [[package]]
+name = "getrandom"
+version = "0.3.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "r-efi",
+ "wasip2",
+]
+
+[[package]]
 name = "hashbrown"
 version = "0.16.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -589,9 +601,11 @@ dependencies = [
  "lazy_static",
  "os_str_bytes",
  "pin-project",
+ "rand",
  "regex",
  "serde",
  "serde_json",
+ "shlex",
  "tokio",
  "toml",
  "tracing",
@@ -724,6 +738,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
 
 [[package]]
+name = "ppv-lite86"
+version = "0.2.21"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9"
+dependencies = [
+ "zerocopy",
+]
+
+[[package]]
 name = "precomputed-hash"
 version = "0.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -748,6 +771,41 @@ dependencies = [
 ]
 
 [[package]]
+name = "r-efi"
+version = "5.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f"
+
+[[package]]
+name = "rand"
+version = "0.9.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6db2770f06117d490610c7488547d543617b21bfa07796d7a12f6f1bd53850d1"
+dependencies = [
+ "rand_chacha",
+ "rand_core",
+]
+
+[[package]]
+name = "rand_chacha"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb"
+dependencies = [
+ "ppv-lite86",
+ "rand_core",
+]
+
+[[package]]
+name = "rand_core"
+version = "0.9.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c"
+dependencies = [
+ "getrandom",
+]
+
+[[package]]
 name = "redox_syscall"
 version = "0.5.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1148,6 +1206,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
 
 [[package]]
+name = "wasip2"
+version = "1.0.2+wasi-0.2.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9517f9239f02c069db75e65f174b3da828fe5f5b945c4dd26bd25d89c03ebcf5"
+dependencies = [
+ "wit-bindgen",
+]
+
+[[package]]
 name = "wasm-bindgen"
 version = "0.2.108"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1350,6 +1417,32 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5a5364e9d77fcdeeaa6062ced926ee3381faa2ee02d3eb83a5c27a8825540829"
 
 [[package]]
+name = "wit-bindgen"
+version = "0.51.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d7249219f66ced02969388cf2bb044a09756a083d0fab1e566056b04d9fbcaa5"
+
+[[package]]
+name = "zerocopy"
+version = "0.8.37"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7456cf00f0685ad319c5b1693f291a650eaf345e941d082fc4e03df8a03996ac"
+dependencies = [
+ "zerocopy-derive",
+]
+
+[[package]]
+name = "zerocopy-derive"
+version = "0.8.37"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1328722bbf2115db7e19d69ebcc15e795719e2d66b60827c6a69a117365e37a0"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
 name = "zmij"
 version = "1.0.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
diff --git a/pkgs/by-name/mp/mpdpopm/Cargo.toml b/pkgs/by-name/mp/mpdpopm/Cargo.toml
index c82537e6..71232236 100644
--- a/pkgs/by-name/mp/mpdpopm/Cargo.toml
+++ b/pkgs/by-name/mp/mpdpopm/Cargo.toml
@@ -42,3 +42,5 @@ tokio = { version = "1.49", features = ["io-util", "macros", "net", "process", "
 tracing = "0.1.44"
 tracing-subscriber = { version = "0.3.22", features = ["env-filter"]}
 anyhow = "1.0.100"
+shlex = "1.3.0"
+rand = "0.9.2"
diff --git a/pkgs/by-name/mp/mpdpopm/src/bin/mpdpopm.rs b/pkgs/by-name/mp/mpdpopm/src/bin/mpdpopm.rs
index 5d4eeae4..faa651bf 100644
--- a/pkgs/by-name/mp/mpdpopm/src/bin/mpdpopm.rs
+++ b/pkgs/by-name/mp/mpdpopm/src/bin/mpdpopm.rs
@@ -31,6 +31,7 @@ use mpdpopm::{
     config::{self, Config},
     filters::ExpressionParser,
     filters_ast::{FilterStickerNames, evaluate},
+    messanges::COMMAND_CHANNEL,
     storage::{last_played, play_count, rating},
 };
 
@@ -425,6 +426,23 @@ enum PlaylistsCommand {
 }
 
 #[derive(Subcommand)]
+enum DjCommand {
+    /// Activate the automatic DJ mode on the mpdpopmd daemon.
+    ///
+    /// In this mode, the daemon will automatically add new tracks to the playlist based on a
+    /// recommendation algorithm.
+    #[clap(verbatim_doc_comment)]
+    Start {},
+
+    /// Deactivate the automatic DJ mode on the mpdpopmd daemon.
+    ///
+    /// In this mode, the daemon will automatically add new tracks to the playlist based on a
+    /// recommendation algorithm.
+    #[clap(verbatim_doc_comment)]
+    Stop {},
+}
+
+#[derive(Subcommand)]
 enum SubCommand {
     /// Change details about rating.
     Rating {
@@ -477,9 +495,18 @@ enum SubCommand {
         filter: String,
 
         /// Respect the casing, when performing the filter evaluation.
-        #[arg(short, long, default_value_t = true)]
+        #[arg(short, long, default_value_t = false)]
         case_sensitive: bool,
     },
+
+    /// Modify the automatic DJ mode on the mpdpopmd daemon.
+    ///
+    /// In this mode, the daemon will automatically add new tracks to the playlist based on a
+    /// recommendation algorithm.
+    Dj {
+        #[command(subcommand)]
+        command: DjCommand,
+    },
 }
 
 #[tokio::main]
@@ -569,5 +596,9 @@ async fn main() -> Result<()> {
             filter,
             case_sensitive,
         } => searchadd(&mut client, &filter, case_sensitive).await,
+        SubCommand::Dj { command } => match command {
+            DjCommand::Start {} => client.send_message(COMMAND_CHANNEL, "dj start").await,
+            DjCommand::Stop {} => client.send_message(COMMAND_CHANNEL, "dj stop").await,
+        },
     }
 }
diff --git a/pkgs/by-name/mp/mpdpopm/src/dj/algorithms.rs b/pkgs/by-name/mp/mpdpopm/src/dj/algorithms.rs
new file mode 100644
index 00000000..fcb05817
--- /dev/null
+++ b/pkgs/by-name/mp/mpdpopm/src/dj/algorithms.rs
@@ -0,0 +1,139 @@
+use std::collections::HashSet;
+
+use anyhow::{Context, Result};
+use rand::{Rng, distr};
+use tracing::info;
+
+use crate::{clients::Client, storage};
+
+pub(crate) trait Algorithm {
+    async fn next_track(&mut self, client: &mut Client) -> Result<String>;
+}
+
+/// Generates generic discovery playlist, that fulfills following requirements:
+///  - Will (eventually) include every not-played song. (So it can be used to rank a library)
+///  - Returns liked songs more often then not-played or negative songs.
+pub(crate) struct Discovery {
+    already_done: HashSet<String>,
+}
+
+impl Algorithm for Discovery {
+    async fn next_track(&mut self, client: &mut Client) -> Result<String> {
+        macro_rules! take {
+            ($first:expr, $second:expr, $third:expr) => {{
+                tracing::info!(
+                    "Selecting a `{}` track for the next entry in the queue",
+                    stringify!($first)
+                );
+
+                $first.pop().map_or_else(
+                    || {
+                        $second.pop().map_or_else(
+                            || {
+                                $third.pop().map_or_else(
+                                    || {
+                                        unreachable!(
+                                            "This means that there are no songs in the libary"
+                                        )
+                                    },
+                                    Ok::<_, anyhow::Error>,
+                                )
+                            },
+                            Ok::<_, anyhow::Error>,
+                        )
+                    },
+                    Ok::<_, anyhow::Error>,
+                )
+            }};
+        }
+
+        let (mut positive, mut neutral, mut negative) = {
+            let tracks = {
+                let mut base = client
+                    .get_all_songs()
+                    .await?
+                    .into_iter()
+                    .filter(|song| !self.already_done.contains(song))
+                    .collect::<Vec<_>>();
+
+                if base.is_empty() {
+                    // We could either have no tracks in the library,
+                    // or we actually already listed to everything.
+                    self.already_done = HashSet::new();
+
+                    info!("Resetting already done songs, as we have no more to choose from");
+
+                    base = client.get_all_songs().await?;
+                }
+
+                base
+            };
+
+            let mut positive = vec![];
+            let mut neutral = vec![];
+            let mut negative = vec![];
+
+            for track in tracks {
+                let weight = Self::weight_track(client, &track).await?;
+
+                match weight {
+                    1..=i64::MAX => positive.push(track),
+                    0 => neutral.push(track),
+                    i64::MIN..0 => negative.push(track),
+                }
+            }
+
+            (positive, neutral, negative)
+        };
+
+        let pick = {
+            let mut rng = rand::rng();
+            rng.sample(
+                distr::weighted::WeightedIndex::new([0.65, 0.5, 0.2].iter())
+                    .expect("to be valid, as hardcoded"),
+            )
+        };
+
+        let next = match pick {
+            0 => take!(positive, neutral, negative),
+            1 => take!(neutral, positive, negative),
+            2 => take!(negative, neutral, positive),
+            _ => unreachable!("These indexes are not possible"),
+        }?;
+
+        self.already_done.insert(next.clone());
+
+        Ok(next)
+    }
+}
+
+impl Discovery {
+    pub(crate) fn new() -> Self {
+        Self {
+            already_done: HashSet::new(),
+        }
+    }
+
+    /// Calculate a recommendation score for a track.
+    ///
+    /// The algorithm maps tracks, that the user likes to a high score and songs that the user
+    /// dislikes to a lower number.
+    /// Currently, only the rating, skip count and play count are considered. Similarity scores,
+    /// fetched from e.g. last.fm should be included in the future.
+    async fn weight_track(client: &mut Client, track: &str) -> Result<i64> {
+        let rating = i32::from(storage::rating::get(client, track).await?.unwrap_or(0));
+        let play_count = i32::try_from(storage::play_count::get(client, track).await?.unwrap_or(0))
+            .context("`play_count` too big")?;
+        let skip_count = i32::try_from(storage::skip_count::get(client, track).await?.unwrap_or(0))
+            .context("`skip_count` too big")?;
+
+        let output: f64 =
+            1.0 * f64::from(rating) + 0.3 * f64::from(play_count) + -0.6 * f64::from(skip_count);
+
+        let weight = output.round() as i64;
+
+        // info!("`{track}`: {weight}");
+
+        Ok(weight)
+    }
+}
diff --git a/pkgs/by-name/mp/mpdpopm/src/dj/mod.rs b/pkgs/by-name/mp/mpdpopm/src/dj/mod.rs
new file mode 100644
index 00000000..a211a571
--- /dev/null
+++ b/pkgs/by-name/mp/mpdpopm/src/dj/mod.rs
@@ -0,0 +1,28 @@
+use anyhow::Result;
+use tracing::info;
+
+use crate::{clients::Client, dj::algorithms::Algorithm};
+
+pub(crate) mod algorithms;
+
+pub(crate) struct Dj<A: Algorithm> {
+    algo: A,
+}
+
+impl<A: Algorithm> Dj<A> {
+    pub(crate) fn new(algo: A) -> Self {
+        Self { algo }
+    }
+
+    /// Add the next track to the playlist.
+    ///
+    /// This should be called after the previous track is finished, to avoid unbounded growth.
+    pub(crate) async fn add_track(&mut self, client: &mut Client) -> Result<()> {
+        let next = self.algo.next_track(client).await?;
+
+        info!("Adding `{next}`, due to active dj mode");
+        client.add(&next).await?;
+
+        Ok(())
+    }
+}
diff --git a/pkgs/by-name/mp/mpdpopm/src/lib.rs b/pkgs/by-name/mp/mpdpopm/src/lib.rs
index d5db57b4..cc2765dc 100644
--- a/pkgs/by-name/mp/mpdpopm/src/lib.rs
+++ b/pkgs/by-name/mp/mpdpopm/src/lib.rs
@@ -34,7 +34,9 @@
 
 pub mod clients;
 pub mod config;
+pub mod dj;
 pub mod filters_ast;
+pub mod messanges;
 pub mod playcounts;
 pub mod storage;
 pub mod vars;
@@ -51,6 +53,7 @@ pub mod filters {
 use crate::{
     clients::{Client, IdleClient, IdleSubSystem},
     config::{Config, Connection},
+    messanges::MessageQueue,
     playcounts::PlayState,
 };
 
@@ -90,6 +93,8 @@ pub async fn mpdpopm(cfg: Config) -> std::result::Result<(), Error> {
             .context("Failed to connect to TCP idle client")?,
     };
 
+    let mut mqueue = MessageQueue::new();
+
     idle_client
         .subscribe(&cfg.commands_chan)
         .await
@@ -106,6 +111,7 @@ pub async fn mpdpopm(cfg: Config) -> std::result::Result<(), Error> {
     pin_mut!(ctrl_c, sighup, sigkill, tick);
 
     let mut done = false;
+    let mut msg_check_needed = false;
     while !done {
         debug!("selecting...");
         {
@@ -132,7 +138,7 @@ pub async fn mpdpopm(cfg: Config) -> std::result::Result<(), Error> {
                         tick.set(sleep(Duration::from_millis(cfg.poll_interval_ms)).fuse());
                         state.update(&mut client)
                                 .await
-                                .context("PlayState update failed")?
+                                .context("PlayState update failed")?;
                     },
                     res = idle => match res {
                         Ok(subsys) => {
@@ -140,9 +146,14 @@ pub async fn mpdpopm(cfg: Config) -> std::result::Result<(), Error> {
                             if subsys == IdleSubSystem::Player {
                                 state.update(&mut client)
                                     .await
-                                    .context("PlayState update failed")?
+                                    .context("PlayState update failed")?;
+
+                                mqueue
+                                    .advance_dj(&mut client)
+                                    .await
+                                    .context("MessageQueue tick failed")?;
                             } else if subsys == IdleSubSystem::Message {
-                                error!("Message handling is not supported!");
+                                msg_check_needed = true;
                             }
                             break;
                         },
@@ -155,6 +166,17 @@ pub async fn mpdpopm(cfg: Config) -> std::result::Result<(), Error> {
                 }
             }
         }
+
+        if msg_check_needed {
+            msg_check_needed = false;
+
+            // Check for any messages that have come in; if there's an error there's not a lot we
+            // can do about it (suppose some client fat-fingers a command name, e.g.)-- just log it
+            // & move on.
+            if let Err(err) = mqueue.check_messages(&mut client, &mut idle_client).await {
+                error!("Error while processing messages: {err:#?}");
+            }
+        }
     }
 
     info!("mpdpopm exiting.");
diff --git a/pkgs/by-name/mp/mpdpopm/src/messanges/mod.rs b/pkgs/by-name/mp/mpdpopm/src/messanges/mod.rs
new file mode 100644
index 00000000..c5320dd9
--- /dev/null
+++ b/pkgs/by-name/mp/mpdpopm/src/messanges/mod.rs
@@ -0,0 +1,110 @@
+use anyhow::{Context, Result, anyhow, bail, ensure};
+use clap::{Parser, Subcommand};
+use shlex::Shlex;
+use tracing::info;
+
+use crate::{
+    clients::{Client, IdleClient},
+    dj::{Dj, algorithms::Discovery},
+};
+
+pub const COMMAND_CHANNEL: &str = "unwoundstack.com:commands";
+
+#[derive(Parser)]
+struct Commands {
+    #[command(subcommand)]
+    command: SubCommand,
+}
+
+#[derive(Parser)]
+enum SubCommand {
+    Dj {
+        #[command(subcommand)]
+        command: DjCommand,
+    },
+}
+
+#[derive(Subcommand)]
+enum DjCommand {
+    Start {},
+    Stop {},
+}
+
+pub(crate) struct MessageQueue {
+    dj: Option<Dj<Discovery>>,
+}
+
+impl MessageQueue {
+    pub(crate) fn new() -> Self {
+        Self { dj: None }
+    }
+
+    pub(crate) async fn advance_dj(&mut self, client: &mut Client) -> Result<()> {
+        if let Some(dj) = self.dj.as_mut() {
+            dj.add_track(client).await?;
+        }
+
+        Ok(())
+    }
+
+    /// Read messages off the commands channel & dispatch 'em
+    pub(crate) async fn check_messages(
+        &mut self,
+        client: &mut Client,
+        idle_client: &mut IdleClient,
+    ) -> Result<()> {
+        let m = idle_client
+            .get_messages()
+            .await
+            .context("Failed to `get_messages` from client")?;
+
+        for (chan, msgs) in m {
+            ensure!(chan == COMMAND_CHANNEL, "Unknown channel: `{}`", chan);
+
+            for msg in msgs {
+                self.process(client, msg).await?;
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Process a single command
+    pub(crate) async fn process(&mut self, client: &mut Client, msg: String) -> Result<()> {
+        let split = {
+            let mut shl = Shlex::new(&msg);
+            let res: Vec<_> = shl.by_ref().collect();
+
+            if shl.had_error {
+                bail!("Failed to parse command '{msg}'")
+            }
+
+            assert_eq!(shl.line_no, 1, "A unexpected newline appeared");
+            assert!(!res.is_empty());
+
+            let mut base = vec!["base".to_owned()];
+            base.extend(res);
+            base
+        };
+
+        let args = Commands::parse_from(split);
+
+        match args.command {
+            SubCommand::Dj { command } => match command {
+                DjCommand::Start {} => {
+                    info!("Dj started");
+                    self.dj = Some(Dj::new(Discovery::new()));
+                    self.advance_dj(client).await?;
+                }
+                DjCommand::Stop {} => {
+                    self.dj
+                        .take()
+                        .ok_or_else(|| anyhow!("Tried to disable already disabled dj mode"))?;
+                    info!("Dj stopped");
+                }
+            },
+        }
+
+        Ok(())
+    }
+}