Add speaker capture audio stream

crates/audio/Cargo.toml

@@ -17,6 +17,10 @@ cpal = { workspace = true }
 rodio = { version = "0.20.1", features = ["wav", "hound", "vorbis"] }
 kalosm-sound = { version = "0.3.5", default-features = false }
 
+[dev-dependencies]
+rodio = "*"
+serial_test = "3"
+
 [target.'cfg(target_os = "macos")'.dependencies.cidre]
 git = "https://github.com/yury/cidre"
 rev = "23efaabee6bf75bfb57a9e7739b2beb83cb93942"

crates/audio/src/lib.rs

@@ -2,6 +2,9 @@ mod format;
 mod mic;
 mod speaker;
 
+pub use mic::*;
+pub use speaker::*;
+
 use anyhow::Result;
 use cpal::traits::{DeviceTrait, HostTrait};
 

crates/audio/src/mic.rs

@@ -1,18 +1,23 @@
-use futures::StreamExt;
-use kalosm_sound::rodio::Source;
-use kalosm_sound::MicInput;
-
-async fn main2() {
-    let mic = MicInput::default();
-    let stream = mic.stream().unwrap();
-}
+pub use kalosm_sound::{AsyncSource, MicInput, MicStream};
 
 #[cfg(test)]
 mod tests {
     use super::*;
 
     #[tokio::test]
     async fn test_mic() {
-        main2().await;
+        let mic = MicInput::default();
+        let mut stream = mic.stream().unwrap();
+
+        tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
+
+        let samples = stream
+            .read_samples(48000 * 1)
+            .await
+            .into_iter()
+            .collect::<Vec<_>>();
+
+        assert!(samples.len() > 10 * 1000);
+        assert!(!samples.iter().all(|x| *x == 0.0));
     }
 }

crates/audio/src/speaker/macos.rs

@@ -1,25 +1,127 @@
 use anyhow::Result;
-use ca::aggregate_device_keys as agg_keys;
-use cidre::{arc, av, cat, cf, core_audio as ca, ns, os};
-use std::time::Duration;
-
 use futures_util::StreamExt;
-use rodio::buffer::SamplesBuffer;
-use std::pin::Pin;
 
-use crate::format::Format;
+use ca::aggregate_device_keys as agg_keys;
+use cidre::{arc, av, cat, cf, core_audio as ca, ns, os};
 
 // https://github.com/yury/cidre/blob/23efaabee6bf75bfb57a9e7739b2beb83cb93942/cidre/examples/core-audio-record/main.rs
 // https://github.com/floneum/floneum/blob/92129ec99aac446348f42bc6db326a6d1c2d99ae/interfaces/kalosm-sound/src/source/mic.rs#L41
 #[cfg(target_os = "macos")]
-pub struct SpeakerInput {}
+pub struct SpeakerInput {
+    tap: ca::TapGuard,
+    agg_desc: arc::Retained<cf::DictionaryOf<cf::String, cf::Type>>,
+}
 
-// https://github.com/floneum/floneum/blob/92129ec99aac446348f42bc6db326a6d1c2d99ae/interfaces/kalosm-sound/src/source/mic.rs#L140
 #[cfg(target_os = "macos")]
 pub struct SpeakerStream {
-    format: Format,
     read_data: Vec<f32>,
-    receiver: Pin<Box<dyn futures_core::Stream<Item = f32> + Send + Sync>>,
+    receiver: std::pin::Pin<Box<dyn futures_core::Stream<Item = f32> + Send + Sync>>,
+    device: ca::hardware::StartedDevice<ca::AggregateDevice>,
+    _ctx: Box<Ctx>,
+}
+
+#[derive(Clone)]
+struct Ctx {
+    format: arc::R<av::AudioFormat>,
+    sender: futures_channel::mpsc::UnboundedSender<Vec<f32>>,
+}
+
+impl SpeakerInput {
+    pub fn new() -> Result<Self> {
+        let output_device = ca::System::default_output_device()?;
+        let output_uid = output_device.uid()?;
+
+        let sub_device = cf::DictionaryOf::with_keys_values(
+            &[ca::sub_device_keys::uid()],
+            &[output_uid.as_type_ref()],
+        );
+
+        let tap_desc = ca::TapDesc::with_mono_global_tap_excluding_processes(&ns::Array::new());
+        let tap = tap_desc.create_process_tap()?;
+
+        let sub_tap = cf::DictionaryOf::with_keys_values(
+            &[ca::sub_device_keys::uid()],
+            &[tap.uid().unwrap().as_type_ref()],
+        );
+
+        let agg_desc = cf::DictionaryOf::with_keys_values(
+            &[
+                agg_keys::is_private(),
+                agg_keys::is_stacked(),
+                agg_keys::tap_auto_start(),
+                agg_keys::name(),
+                agg_keys::main_sub_device(),
+                agg_keys::uid(),
+                agg_keys::sub_device_list(),
+                agg_keys::tap_list(),
+            ],
+            &[
+                cf::Boolean::value_true().as_type_ref(),
+                cf::Boolean::value_false(),
+                cf::Boolean::value_true(),
+                cf::str!(c"hypr-audio-tap"),
+                &output_uid,
+                &cf::Uuid::new().to_cf_string(),
+                &cf::ArrayOf::from_slice(&[sub_device.as_ref()]),
+                &cf::ArrayOf::from_slice(&[sub_tap.as_ref()]),
+            ],
+        );
+
+        Ok(Self { tap, agg_desc })
+    }
+
+    fn start_device(
+        &self,
+        ctx: &mut Box<Ctx>,
+    ) -> Result<ca::hardware::StartedDevice<ca::AggregateDevice>> {
+        extern "C" fn proc(
+            _device: ca::Device,
+            _now: &cat::AudioTimeStamp,
+            input_data: &cat::AudioBufList<1>,
+            _input_time: &cat::AudioTimeStamp,
+            _output_data: &mut cat::AudioBufList<1>,
+            _output_time: &cat::AudioTimeStamp,
+            ctx: Option<&mut Ctx>,
+        ) -> os::Status {
+            let ctx = ctx.unwrap();
+
+            assert_eq!(ctx.format.common_format(), av::audio::CommonFormat::PcmF32);
+
+            let view =
+                av::AudioPcmBuf::with_buf_list_no_copy(&ctx.format, input_data, None).unwrap();
+
+            if let Some(data) = view.data_f32_at(0) {
+                let samples = data.to_vec();
+                ctx.sender.start_send(samples).unwrap();
+            }
+
+            os::Status::NO_ERR
+        }
+
+        let agg_device = ca::AggregateDevice::with_desc(&self.agg_desc)?;
+        let proc_id = agg_device.create_io_proc_id(proc, Some(ctx))?;
+        let started_device = ca::device_start(agg_device, Some(proc_id))?;
+
+        Ok(started_device)
+    }
+
+    pub fn stream(&self) -> SpeakerStream {
+        let asbd = self.tap.asbd().unwrap();
+        let format = av::AudioFormat::with_asbd(&asbd).unwrap();
+
+        let (tx, rx) = futures_channel::mpsc::unbounded();
+        let mut ctx = Box::new(Ctx { format, sender: tx });
+
+        let device = self.start_device(&mut ctx).unwrap();
+        let receiver = rx.map(futures_util::stream::iter).flatten();
+
+        SpeakerStream {
+            receiver: Box::pin(receiver),
+            read_data: Vec::new(),
+            device,
+            _ctx: ctx,
+        }
+    }
 }
 
 impl futures_core::Stream for SpeakerStream {
@@ -40,22 +142,14 @@ impl futures_core::Stream for SpeakerStream {
     }
 }
 
-// https://github.com/yury/cidre/blob/23efaabee6bf75bfb57a9e7739b2beb83cb93942/cidre/examples/core-audio-record/main.rs
 impl SpeakerStream {
-    pub fn new(format: Format) {}
-
-    fn create_device(&self) {}
+    pub fn read_sync(&mut self) -> Vec<f32> {
+        let mut ctx = std::task::Context::from_waker(futures_util::task::noop_waker_ref());
 
-    fn start(&self) {}
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_main() {
-        let format = Format::default();
-        SpeakerStream::new(format)
+        while let std::task::Poll::Ready(Some(data_chunk)) = self.receiver.poll_next_unpin(&mut ctx)
+        {
+            self.read_data.push(data_chunk);
+        }
+        self.read_data.clone()
     }
 }

crates/audio/src/speaker/mod.rs

@@ -1,3 +1,6 @@
+use anyhow::Result;
+use futures::StreamExt;
+
 #[cfg(target_os = "macos")]
 mod macos;
 
@@ -12,10 +15,97 @@ pub struct SpeakerInput {
     inner: windows::SpeakerInput,
 }
 
+impl SpeakerInput {
+    #[cfg(target_os = "macos")]
+    pub fn new() -> Result<Self> {
+        let inner = macos::SpeakerInput::new()?;
+        Ok(Self { inner })
+    }
+
+    #[cfg(target_os = "macos")]
+    pub fn stream(&self) -> Result<SpeakerStream> {
+        let inner = self.inner.stream();
+        Ok(SpeakerStream { inner })
+    }
+}
+
 // https://github.com/floneum/floneum/blob/92129ec99aac446348f42bc6db326a6d1c2d99ae/interfaces/kalosm-sound/src/source/mic.rs#L140
 pub struct SpeakerStream {
     #[cfg(target_os = "macos")]
     inner: macos::SpeakerStream,
     #[cfg(target_os = "windows")]
     inner: windows::SpeakerStream,
 }
+
+impl futures_core::Stream for SpeakerStream {
+    type Item = f32;
+
+    #[cfg(target_os = "macos")]
+    fn poll_next(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Option<Self::Item>> {
+        self.inner.poll_next_unpin(cx)
+    }
+}
+
+impl SpeakerStream {
+    pub fn read_sync(&mut self) -> Vec<f32> {
+        self.inner.read_sync()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serial_test::serial;
+
+    fn play_for_sec(seconds: u64) -> std::thread::JoinHandle<()> {
+        use rodio::{
+            cpal::SampleRate,
+            source::{Function::Sine, SignalGenerator, Source},
+            OutputStream,
+        };
+        use std::{
+            thread::{sleep, spawn},
+            time::Duration,
+        };
+
+        spawn(move || {
+            let (_stream, stream_handle) = OutputStream::try_default().unwrap();
+            let source = SignalGenerator::new(SampleRate(44100), 440.0, Sine);
+
+            let source = source
+                .convert_samples()
+                .take_duration(Duration::from_secs(seconds))
+                .amplify(0.01);
+
+            stream_handle.play_raw(source).unwrap();
+            sleep(Duration::from_secs(seconds));
+        })
+    }
+
+    #[cfg(target_os = "macos")]
+    #[test]
+    #[serial]
+    fn test_macos() {
+        let handle = play_for_sec(1);
+        let input = SpeakerInput::new().unwrap();
+        let mut stream = input.stream().unwrap();
+
+        std::thread::sleep(std::time::Duration::from_millis(500));
+
+        let data = stream.read_sync();
+        assert!(data.len() > 10);
+        assert!(!data.iter().all(|x| *x == 0.0));
+
+        handle.join().unwrap();
+    }
+
+    #[cfg(target_os = "windows")]
+    #[test]
+    #[serial]
+    fn test_windows() {
+        assert!(true);
+    }
+}