ox_speak_client/src-tauri/src/core/capture.rs

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread;
use std::thread::JoinHandle;
use cpal::{default_host, BufferSize, Device, SampleRate, Stream, StreamConfig, SupportedStreamConfig};
use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
use crate::core::opus::AudioOpus;
use crate::domain::event::{Event, EventBus};
use crate::utils::ringbuf::RingBuffer;

#[derive(Clone)]
pub struct Microphone {
    device: Device,
}

pub struct AudioCapture {
    event_bus: EventBus,
    microphone: Microphone,
    running: Arc<AtomicBool>,
    ring_buffer: RingBuffer<i16>,
    steam: Option<Stream>,
    worker: Option<JoinHandle<()>>,
}

impl Microphone {
    pub fn new(device: Device) -> Self {
        println!("Initializing microphone with device: {}", device.name().unwrap_or_else(|_| "Unknown".to_string()));
        Self { 
            device 
        }
    }

    pub fn default() -> Self {
        println!("Creating default microphone");
        let host = default_host();
        let device = host.default_input_device().unwrap();
        Self::new(device)
    }

    pub fn get_input_config(&self) -> SupportedStreamConfig {
        self.device.default_input_config().unwrap()
    }

    pub fn get_stream_config(&self) -> StreamConfig {
        let config = self.get_input_config();
        let mut stream_config: StreamConfig = config.into();
        stream_config.channels = 1;
        stream_config.sample_rate = SampleRate(48000);
        stream_config.buffer_size = BufferSize::Fixed(960);
        stream_config
    }

    pub fn build_stream<F>(&self, callback: F) -> Stream
    where
        F: FnMut(&[i16], &cpal::InputCallbackInfo) + Send + 'static,
    {
        let config = self.get_stream_config();

        self.device.build_input_stream(
            &config,
            callback,
            |err| println!("Error input stream: {err}"),
            None
        ).unwrap()
    }
}

impl AudioCapture {
    pub fn new(event_bus: EventBus, microphone: Microphone) -> Self {
        println!("Creating new AudioCapture instance");
        Self {
            event_bus,
            microphone,
            running: Arc::new(AtomicBool::new(false)),
            ring_buffer: RingBuffer::new(4096),
            steam: None,
            worker: None,
        }
    }

    pub fn default(event_bus: EventBus) -> Self {
        println!("Creating default AudioCapture");
        Self::new(event_bus, Microphone::default())
    }

    pub async fn start(&mut self) {
        println!("Starting audio capture");
        self.running.store(true, Ordering::Relaxed);

        // stream cpal
        println!("Setting up audio stream");
        let writer = self.ring_buffer.writer();
        let stream_running = self.running.clone();
        let stream = self.microphone.build_stream(move |data, _| {
            if !stream_running.load(Ordering::Relaxed){
                return;
            }
            writer.push_slice_overwrite(data);
        });
        stream.play().unwrap();
        self.steam = Some(stream);
        println!("Audio stream started");

        // Audio processing worker
        println!("Starting audio processing worker");
        self.run_processing_worker();
        println!("Audio capture fully initialized");
    }

    pub async fn stop(&mut self) {
        println!("Stopping audio capture");
        self.running.store(false, Ordering::Relaxed);
        println!("Releasing audio stream");
        self.steam = None;
        self.ring_buffer.force_wake_up();

        // code possiblement bloquant, wrap vers un thread tokio bloquant
        if let Some(worker) = self.worker.take() {
            println!("Waiting for audio processing worker to finish");
            tokio::task::spawn_blocking(move || {
                worker.join().unwrap();
            }).await.unwrap();
        }
        println!("Clearing ring buffer");
        self.ring_buffer.clear();
        println!("Audio capture stopped");
    }

    fn run_processing_worker(&mut self){
        println!("Configuring audio processing worker");
        let worker_running = self.running.clone();
        let event_bus = self.event_bus.clone();
        let input_config = self.microphone.get_input_config();
        println!("Audio input config: sample rate: {}, channels: {}", input_config.sample_rate().0, input_config.channels());
        let opus = AudioOpus::new(input_config.sample_rate().0, input_config.channels(), "voip");
        let mut encoder = opus.create_encoder().unwrap();
        let reader = self.ring_buffer.reader();

        println!("Spawning audio processing thread");
        self.worker = Some(thread::spawn(move || {
            println!("Audio processing thread started");
            let mut frame = [0i16; 960];
            let mut frame_count = 0;

            while worker_running.load(Ordering::Relaxed) {
                let _ = reader.pop_slice_blocking(&mut frame);
                if !worker_running.load(Ordering::Relaxed){
                    println!("Audio processing thread stopping");
                    break;
                }

                frame_count += 1;
                if frame_count % 100 == 0 {
                    println!("Processed {} audio frames", frame_count);
                }

                let raw_data = frame.to_vec();
                event_bus.emit_sync(Event::AudioIn(raw_data));

                match encoder.encode(&frame){
                    Ok(encoded_data) => {
                        event_bus.emit_sync(Event::AudioEncoded(encoded_data))
                    }
                    Err(e) => {
                        println!("Error encoding: {e}");
                    }
                }
            }
            println!("Audio processing thread finished after processing {} frames", frame_count);
        }));
    }
}

impl AudioCapture {
    fn audio_processing(){

    }
}