streamtools / core / src / lib.rs

use anyhow::{Result, anyhow};
use cpal::{
    SampleFormat, StreamConfig,
    traits::{DeviceTrait, HostTrait, StreamTrait},
};
use std::{
    collections::VecDeque,
    sync::{
        Arc, Mutex,
        atomic::{self, AtomicBool, AtomicU64},
    },
    time::{self, SystemTime, UNIX_EPOCH},
};

#[derive(Debug, Clone)]
pub struct MicInfo {
    pub name: String,
    pub sample_rate: u32,
    pub channels: u16,
}

impl Default for MicInfo {
    fn default() -> Self {
        Self::new()
    }
}

impl MicInfo {
    pub fn new() -> Self {
        let host = cpal::default_host();
        let device = host.default_input_device().expect("no default input device");
        let supported_config = device.default_input_config().unwrap();
        let config: cpal::StreamConfig = supported_config.clone().into();

        Self {
            name: device.name().unwrap_or_else(|_| "Unknown device".into()),
            sample_rate: config.sample_rate.0,
            channels: config.channels,
        }
    }
}

pub type SharedLevels = Arc<Mutex<VecDeque<f32>>>;
type LastActiveTime = Arc<AtomicU64>;

const HOLD_TIME_MS: u64 = 300;

fn get_current_time_ms() -> u64 {
    SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis() as u64
}

fn process_input_f32(data: &[f32], active: &AtomicBool, levels: &SharedLevels, last_active: &LastActiveTime) {
    if data.is_empty() {
        return;
    }

    let mut sum = 0.0;
    for &s in data {
        sum += s * s;
    }
    let rms = (sum / data.len() as f32).sqrt();

    let threshold = 0.01;
    let now = get_current_time_ms();

    if rms > threshold {
        last_active.store(now, atomic::Ordering::Relaxed);
        active.store(true, atomic::Ordering::Relaxed);
    } else {
        let last_time = last_active.load(atomic::Ordering::Relaxed);
        let is_active = (now - last_time) < HOLD_TIME_MS;
        active.store(is_active, atomic::Ordering::Relaxed);
    }

    push_level(levels, rms);
}

fn process_input_i16(data: &[i16], active: &AtomicBool, levels: &SharedLevels, last_active: &LastActiveTime) {
    if data.is_empty() {
        return;
    }
    let norm = i16::MAX as f32;
    let mut sum = 0.0;
    for &s in data {
        let v = s as f32 / norm;
        sum += v * v;
    }
    let rms = (sum / data.len() as f32).sqrt();

    let threshold = 0.01;
    let now = get_current_time_ms();

    if rms > threshold {
        last_active.store(now, atomic::Ordering::Relaxed);
        active.store(true, atomic::Ordering::Relaxed);
    } else {
        let last_time = last_active.load(atomic::Ordering::Relaxed);
        let is_active = (now - last_time) < HOLD_TIME_MS;
        active.store(is_active, atomic::Ordering::Relaxed);
    }

    push_level(levels, rms);
}

fn process_input_u16(data: &[u16], active: &AtomicBool, levels: &SharedLevels, last_active: &LastActiveTime) {
    if data.is_empty() {
        return;
    }
    let max = u16::MAX as f32;
    let mid = max / 2.0;
    let mut sum = 0.0;
    for &s in data {
        let v = (s as f32 - mid) / mid;
        sum += v * v;
    }
    let rms = (sum / data.len() as f32).sqrt();

    let threshold = 0.01;
    let now = get_current_time_ms();

    if rms > threshold {
        last_active.store(now, atomic::Ordering::Relaxed);
        active.store(true, atomic::Ordering::Relaxed);
    } else {
        let last_time = last_active.load(atomic::Ordering::Relaxed);
        let is_active = (now - last_time) < HOLD_TIME_MS;
        active.store(is_active, atomic::Ordering::Relaxed);
    }

    push_level(levels, rms);
}

fn push_level(levels: &SharedLevels, rms: f32) {
    const MAX_SAMPLES: usize = 64;

    if let Ok(mut buf) = levels.lock() {
        buf.push_back(rms);
        if buf.len() > MAX_SAMPLES {
            buf.pop_front();
        }
    }
}

pub fn spawn_mic_listener(active: Arc<AtomicBool>, levels: SharedLevels) -> Result<()> {
    std::thread::spawn(move || {
        if let Err(e) = mic_loop(active, levels) {
            eprintln!("mic loop error: {e:?}");
        }
    });
    Ok(())
}

pub fn mic_loop(active: Arc<AtomicBool>, levels: SharedLevels) -> Result<()> {
    let host = cpal::default_host();
    let device = host
        .default_input_device()
        .ok_or_else(|| anyhow::anyhow!("no default input device"))?;
    println!("Using input device: {}", device.name()?);

    let supported_config = device
        .default_input_config()
        .map_err(|e| anyhow::anyhow!("failed to get default input config: {e}"))?;
    let sample_format = supported_config.sample_format();
    let config: StreamConfig = supported_config.into();

    let last_active: LastActiveTime = Arc::new(AtomicU64::new(0));

    let stream = match sample_format {
        SampleFormat::F32 => {
            let active = active.clone();
            let levels = levels.clone();
            let last_active = last_active.clone();
            let err_fn = |err| eprintln!("cpal input stream error: {err}");
            device.build_input_stream(
                &config,
                move |data: &[f32], _| process_input_f32(data, &active, &levels, &last_active),
                err_fn,
                None,
            )?
        }
        SampleFormat::I16 => {
            let active = active.clone();
            let levels = levels.clone();
            let last_active = last_active.clone();
            let err_fn = |err| eprintln!("cpal input stream error: {err}");
            device.build_input_stream(
                &config,
                move |data: &[i16], _| process_input_i16(data, &active, &levels, &last_active),
                err_fn,
                None,
            )?
        }
        SampleFormat::U16 => {
            let active = active.clone();
            let levels = levels.clone();
            let last_active = last_active.clone();
            let err_fn = |err| eprintln!("cpal input stream error: {err}");
            device.build_input_stream(
                &config,
                move |data: &[u16], _| process_input_u16(data, &active, &levels, &last_active),
                err_fn,
                None,
            )?
        }

        other => {
            return Err(anyhow!("Unsupported sample format: {other:?}"));
        }
    };

    stream.play()?;

    loop {
        std::thread::sleep(time::Duration::from_secs(1));
    }
}