chatsounds/

lib.rs

#![warn(clippy::nursery, clippy::pedantic)]
#![allow(clippy::missing_errors_doc, clippy::missing_panics_doc)]

mod cache;
#[cfg(feature = "playback")]
mod channel_volume;
mod error;
mod fetching;
mod parsing;
mod types;

use std::{
    collections::{HashMap, VecDeque},
    io::{BufReader, Cursor},
    path::{Path, PathBuf},
    sync::Arc,
};

pub use bytes::Bytes;
use rand::prelude::*;
pub use rodio;
use rodio::{Decoder, Player, SpatialPlayer};
#[cfg(feature = "playback")]
use rodio::{
    DeviceSinkBuilder, MixerDeviceSink,
    cpal::traits::{DeviceTrait, HostTrait},
};

#[cfg(feature = "playback")]
pub use self::channel_volume::ChannelVolumeSink;
#[cfg(feature = "playback")]
use self::types::ChatsoundsSink;
pub use self::{
    error::Error,
    fetching::{GitHubApiFileEntry, GitHubApiTrees, GitHubMsgpackEntries},
    parsing::normalize_sentence,
    types::Chatsound,
};
use self::{
    error::Result,
    parsing::{ModifierTrait, parse},
    types::BoxSource,
};

#[cfg(feature = "memory")]
type FsMemory = Arc<async_lock::RwLock<HashMap<String, Bytes>>>;

pub struct Chatsounds {
    #[cfg(feature = "fs")]
    cache_path: PathBuf,
    // [sentence]: Chatsound[]
    map_store: HashMap<String, Vec<Chatsound>>,

    #[cfg(feature = "playback")]
    max_sinks: usize,
    #[cfg(feature = "playback")]
    volume: f32,

    #[cfg(feature = "playback")]
    output_stream: MixerDeviceSink,
    #[cfg(feature = "playback")]
    sinks: VecDeque<Box<dyn ChatsoundsSink>>,

    #[cfg(feature = "memory")]
    fs_memory: FsMemory,
}

// TODO ???
#[allow(clippy::non_send_fields_in_send_ty)]
unsafe impl Send for Chatsounds {}
unsafe impl Sync for Chatsounds {}

impl Chatsounds {
    pub fn new(#[cfg(feature = "fs")] cache_path: &Path) -> Result<Self> {
        #[cfg(feature = "fs")]
        let cache_path = cache_path.canonicalize().map_err(|err| Error::Io {
            err,
            path: cache_path.into(),
        })?;
        #[cfg(feature = "fs")]
        ensure!(cache_path.is_dir(), Error::DirMissing { path: cache_path });
        #[cfg(feature = "fs")]
        tracing::debug!(cache_path = %cache_path.display(), "using cache directory");

        // `open_default_sink()` discards the `cpal::Device`, so the returned
        // sink can't tell us what it opened. Open the default device ourselves
        // (exactly what `open_default_sink` tries first, same buffer tuning via
        // `from_device`) so we hold the device and can log its name. Fall back
        // to `open_default_sink` -- which also retries alternate configs and
        // other devices -- if the direct attempt fails; the device is unknown
        // in that case.
        #[cfg(feature = "playback")]
        let (mut output_stream, device_name) =
            match rodio::cpal::default_host().default_output_device() {
                Some(device) => {
                    let name = device.description().ok().map(|d| d.name().to_string());
                    match DeviceSinkBuilder::from_device(device)
                        .and_then(DeviceSinkBuilder::open_stream)
                    {
                        Ok(stream) => (stream, name),
                        Err(_) => (DeviceSinkBuilder::open_default_sink()?, None),
                    }
                }
                None => (DeviceSinkBuilder::open_default_sink()?, None),
            };
        #[cfg(feature = "playback")]
        output_stream.log_on_drop(false);
        #[cfg(feature = "playback")]
        tracing::debug!(
            device = device_name.as_deref().unwrap_or("unknown"),
            config = ?output_stream.config(),
            "opened audio output device"
        );

        Ok(Self {
            #[cfg(feature = "fs")]
            cache_path,
            map_store: HashMap::new(),

            #[cfg(feature = "playback")]
            max_sinks: 64,
            #[cfg(feature = "playback")]
            volume: 0.1,

            #[cfg(feature = "playback")]
            output_stream,
            #[cfg(feature = "playback")]
            sinks: VecDeque::new(),

            #[cfg(feature = "memory")]
            fs_memory: Default::default(),
        })
    }

    #[must_use]
    pub fn get(&self, sentence: &str) -> Option<&Vec<Chatsound>> {
        self.map_store.get(&normalize_sentence(sentence))
    }

    #[must_use]
    pub fn search(&self, search: &str) -> Vec<(usize, &String)> {
        #[cfg(feature = "rayon")]
        use rayon::prelude::*;

        #[cfg(feature = "rayon")]
        let iter = HashMap::par_iter;
        #[cfg(not(feature = "rayon"))]
        let iter = HashMap::iter;

        #[cfg(feature = "rayon")]
        let sort_by = <[(usize, &String)]>::par_sort_unstable_by;
        #[cfg(not(feature = "rayon"))]
        let sort_by = <[(usize, &String)]>::sort_unstable_by;

        let search = normalize_sentence(search);

        let mut positions: Vec<_> = iter(&self.map_store)
            .map(|(key, _value)| key)
            .filter_map(|sentence| twoway::find_str(sentence, &search).map(|pos| (pos, sentence)))
            .collect();

        sort_by(
            positions.as_mut_slice(),
            |(pos1, str1): &(usize, &String), (pos2, str2): &(usize, &String)| {
                pos1.partial_cmp(pos2)
                    .unwrap()
                    .then_with(|| str1.len().partial_cmp(&str2.len()).unwrap())
            },
        );

        tracing::trace!(search, results = positions.len(), "search");
        positions
    }

    #[cfg(feature = "playback")]
    pub fn stop_all(&mut self) {
        tracing::debug!(sinks = self.sinks.len(), "stopping all sinks");
        for sink in self.sinks.drain(..) {
            sink.stop();
        }
    }

    #[cfg(feature = "playback")]
    pub fn set_volume(&mut self, volume: f32) {
        let volume = volume.max(0.0);

        self.volume = volume;

        for sink in &mut self.sinks {
            sink.set_volume(volume);
        }
    }

    #[cfg(feature = "playback")]
    #[must_use]
    pub const fn volume(&self) -> f32 {
        self.volume
    }

    #[cfg(feature = "playback")]
    pub async fn play<R: Rng>(
        &mut self,
        text: &str,
        rng: R,
    ) -> Result<(Arc<Player>, Vec<Chatsound>)> {
        tracing::debug!(text, "play");
        let sink = Arc::new(Player::connect_new(self.output_stream.mixer()));

        sink.set_volume(self.volume);

        let (sources, chatsounds): (Vec<_>, Vec<_>) =
            self.get_sources(text, rng).await?.into_iter().unzip();
        for source in sources {
            sink.append(source);
        }

        self.sinks.push_back(Box::new(sink.clone()));
        if self.sinks.len() == self.max_sinks {
            tracing::debug!(
                max_sinks = self.max_sinks,
                "sink limit reached; dropping oldest sink"
            );
            self.sinks.pop_front();
        }

        Ok((sink, chatsounds))
    }

    #[cfg(feature = "playback")]
    pub async fn play_spatial<R: Rng>(
        &mut self,
        text: &str,
        rng: R,
        emitter_pos: [f32; 3],
        left_ear_pos: [f32; 3],
        right_ear_pos: [f32; 3],
    ) -> Result<(Arc<SpatialPlayer>, Vec<Chatsound>)> {
        tracing::debug!(text, ?emitter_pos, "play_spatial");
        let sink = Arc::new(SpatialPlayer::connect_new(
            self.output_stream.mixer(),
            emitter_pos,
            left_ear_pos,
            right_ear_pos,
        ));

        sink.set_volume(self.volume);

        let (sources, chatsounds): (Vec<_>, Vec<_>) =
            self.get_sources(text, rng).await?.into_iter().unzip();
        for source in sources {
            sink.append(source);
        }

        self.sinks.push_back(Box::new(sink.clone()));
        if self.sinks.len() == self.max_sinks {
            tracing::debug!(
                max_sinks = self.max_sinks,
                "sink limit reached; dropping oldest sink"
            );
            self.sinks.pop_front();
        }

        Ok((sink, chatsounds))
    }

    #[cfg(feature = "playback")]
    pub async fn play_channel_volume<R: Rng>(
        &mut self,
        text: &str,
        rng: R,
        channel_volumes: Vec<f32>,
    ) -> Result<(Arc<ChannelVolumeSink>, Vec<Chatsound>)> {
        tracing::debug!(text, ?channel_volumes, "play_channel_volume");
        let sink = Arc::new(ChannelVolumeSink::connect_new(
            self.output_stream.mixer(),
            channel_volumes,
        )?);

        sink.sink.set_volume(self.volume);

        let (sources, chatsounds): (Vec<_>, Vec<_>) =
            self.get_sources(text, rng).await?.into_iter().unzip();
        for source in sources {
            sink.append(source);
        }

        self.sinks.push_back(Box::new(sink.clone()));
        if self.sinks.len() == self.max_sinks {
            tracing::debug!(
                max_sinks = self.max_sinks,
                "sink limit reached; dropping oldest sink"
            );
            self.sinks.pop_front();
        }

        Ok((sink, chatsounds))
    }

    pub async fn get_sources<R: Rng>(
        &mut self,
        text: &str,
        mut rng: R,
    ) -> Result<Vec<(BoxSource, Chatsound)>> {
        let mut sources = Vec::new();

        let parsed_chatsounds = parse(text)?;
        tracing::debug!(
            text,
            count = parsed_chatsounds.len(),
            "resolving chatsounds"
        );
        for parsed_chatsound in parsed_chatsounds {
            let chatsound = if let Some(chatsounds) = self.get(&parsed_chatsound.sentence) {
                // TODO random hashed number passed in?
                parsed_chatsound
                    .choose(chatsounds, &mut rng)
                    .ok_or_else(|| Error::EmptyChoose { text: text.into() })?
                    .to_owned()
            } else {
                tracing::debug!(sentence = %parsed_chatsound.sentence, "no chatsound matched sentence");
                continue;
            };
            tracing::trace!(
                sentence = %parsed_chatsound.sentence,
                url = %chatsound.get_url(),
                "chose chatsound"
            );

            let mut source: BoxSource = {
                #[cfg(feature = "fs")]
                let cache = &self.cache_path;
                #[cfg(feature = "memory")]
                let cache = self.fs_memory.clone();

                let bytes = chatsound.load(cache).await?;

                let reader = BufReader::new(Cursor::new(bytes));
                let sound_path = chatsound.sound_path.clone();
                Box::new(
                    Decoder::new(reader).map_err(|err| Error::RodioDecoder { err, sound_path })?,
                )
            };
            for modifier in parsed_chatsound.modifiers {
                source = modifier.modify(source);
            }

            sources.push((source, chatsound));
        }

        Ok(sources)
    }

    #[cfg(feature = "playback")]
    pub fn sleep_until_end(&mut self) {
        for sink in self.sinks.drain(..) {
            sink.sleep_until_end();
        }
    }
}