own_assist/cli/src/human_interface/audio.rs

use crate::human_interface::{HumanInterface, HumanInterfaceError};
use crate::tlt;
use crate::translate;
use cpal::traits::DeviceTrait;
use own_mcp::audio::models::VoiceRequest;
use own_mcp::audio::{AudioClient, AudioClientTrait};
use own_mcp::mcp::chat::PermissionAnswer;
use std::io;
use std::path::Path;
use tokio::fs::create_dir_all;
use tokio::io::AsyncBufReadExt;

const TEMPORARY_AUDIO_PATH: &str = "temporary_audio";

pub struct Audio {
    client: AudioClient,
    device: cpal::Device,
}

impl Audio {
    pub async fn new(
        client: AudioClient,
        device: cpal::Device,
    ) -> Result<Self, HumanInterfaceError> {
        create_dir_all(TEMPORARY_AUDIO_PATH)
            .await
            .map_err(HumanInterfaceError::IoError)?;
        Ok(Self { client, device })
    }

    async fn wait_for_input_line(expected_line: &str) -> Result<(), io::Error> {
        loop {
            let stdin_reader = tokio::io::BufReader::new(tokio::io::stdin());
            log::trace!("waiting for '{expected_line}' to continue");
            if let Some(line) = stdin_reader.lines().next_line().await? {
                log::debug!("Received line: {line:?}");
                if line == expected_line {
                    log::trace!("stopped waiting");
                    return Ok(());
                }
            }
        }
    }

    async fn wait_for_decision() -> Result<PermissionAnswer, io::Error> {
        const GRANTED_LINE: &str = "y";
        const DENIED_LINE: &str = "n";

        loop {
            let stdin_reader = tokio::io::BufReader::new(tokio::io::stdin());
            log::trace!("wating for '{GRANTED_LINE}' or '{DENIED_LINE}' to continue");
            if let Some(line) = stdin_reader.lines().next_line().await? {
                log::debug!("Received line: {line:?}");

                match line.as_str() {
                    GRANTED_LINE => return Ok(PermissionAnswer::Granted),
                    DENIED_LINE => return Ok(PermissionAnswer::Denied),
                    _ => {}
                };
            }
        }
    }

    async fn play_text(&self, text: String) -> Result<(), HumanInterfaceError> {
        let bytes = self
            .client
            .tts(VoiceRequest { text, config: None })
            .await
            .map_err(|e| HumanInterfaceError::Other(format!("Failed getting tts: {e}")))?;

        let wav_file_path = Path::new(TEMPORARY_AUDIO_PATH).join("tts.wav");

        tokio::fs::write(&wav_file_path, &bytes).await?;

        let handle =
            rodio::DeviceSinkBuilder::open_default_sink().expect("open default audio stream");
        let file = std::fs::File::open(wav_file_path)?;
        let player = rodio::play(handle.mixer(), file)
            .map_err(|e| HumanInterfaceError::Other(format!("Failed playing audio: {e}")))?;

        player.sleep_until_end();

        Ok(())
    }
}

impl HumanInterface for Audio {
    async fn agent_message(&mut self, message: String) -> Result<(), HumanInterfaceError> {
        self.play_text(message).await?;

        Ok(())
    }

    async fn expect_user_message(&mut self) -> Result<String, HumanInterfaceError> {
        use own_mcp::audio::recording::{samples_to_wav, start, stop_and_take_data};

        const RECORDING_START: &str = "r";
        const RECORDING_STOP: &str = "s";

        let input_configs = self.device.default_input_config().map_err(|e| {
            HumanInterfaceError::Other(format!("Could not get supported input configs: {e}"))
        })?;

        Self::wait_for_input_line(RECORDING_START).await?;
        let recording_handler = start(self.device.clone(), input_configs);
        Self::wait_for_input_line(RECORDING_STOP).await?;

        let (samples, wav_spec) = stop_and_take_data(recording_handler)
            .map_err(|e| HumanInterfaceError::Other(format!("failed recording: {e}")))?;

        let wav_file_path = Path::new(TEMPORARY_AUDIO_PATH).join("recording.wav");

        samples_to_wav(samples, &wav_spec, &wav_file_path)
            .map_err(|e| HumanInterfaceError::Other(format!("hound error: {e}")))?;

        let transcription = self
            .client
            .transcribe(wav_file_path)
            .await
            .map_err(|e| HumanInterfaceError::Other(format!("failed to transcribe: {e}")))?;

        log::info!("transcription: {transcription:?}");

        Ok(transcription.text)
    }

    async fn ask_for_permission(
        &self,
        mcp_server_name: String,
        tool_name: String,
    ) -> Result<PermissionAnswer, HumanInterfaceError> {
        self.play_text(tlt!(
            "tool_allow_question_audio",
            mcp_server_name,
            tool_name
        ))
        .await?;

        let decision = Self::wait_for_decision().await?;
        log::debug!("decision: {decision:?}");
        Ok(decision)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use cpal::traits::HostTrait;

    const AUTH_HEADER: &str = include_str!(".AUTH_HEADER");

    async fn get_test_human_interface() -> Audio {
        Audio::new(
            AudioClient::new(url::Url::parse("https://audio.mboemer.de").unwrap())
                .with_authorization(AUTH_HEADER.to_string()),
            cpal::Host::default().default_input_device().unwrap(),
        )
        .await
        .unwrap()
    }

    #[tokio::test]
    async fn test_tts_fetching() {
        let mut human_interface = get_test_human_interface().await;

        human_interface
            .agent_message(String::from("Peter ist jetzt in deinem PC."))
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn test_tts_playing() {
        let human_interface = get_test_human_interface().await;
        human_interface
            .play_text(String::from("Peter ist jetzt in deinem PC."))
            .await
            .unwrap();
    }
}