oak-team/oak-gpui
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix experimental audio, add denoise, auto volume.Prep migration (#38874)

Browse Source 
Uses the previously merged denoising crate (and fixes a bug in it that
snug in during refactoring) to add denoising to the microphone input. 

Adds automatic volume control for microphone and output.

Prepares for migrating to 16kHz SR mono:
The experimental audio path now picks the samplerate and channel count depending on a setting. It can handle incoming streams with both the current (future legacy) and new samplerate & channel count. These are url-encoded into the livekit track name

Release Notes:

- N/A
This commit is contained in:
David Kleingeld
2025-09-25 15:11:12 +02:00
committed by GitHub
parent 783ba389f7
commit 3c5e683fbe
15 changed files with 498 additions and 161 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/livekit_client/src/livekit_client.rs
+16 -4
View File
@@ -1,6 +1,6 @@
use std::sync::Arc;
use anyhow::{Context as _, Result};
use anyhow::{Context as _, Result, anyhow};
use audio::AudioSettings;
use collections::HashMap;
use futures::{SinkExt, channel::mpsc};
@@ -12,7 +12,10 @@ use settings::Settings;
mod playback;
use crate::{LocalTrack, Participant, RemoteTrack, RoomEvent, TrackPublication};
use crate::{
LocalTrack, Participant, RemoteTrack, RoomEvent, TrackPublication,
livekit_client::playback::Speaker,
};
pub use playback::AudioStream;
pub(crate) use playback::{RemoteVideoFrame, play_remote_video_track};
@@ -132,11 +135,20 @@ impl Room {
track: &RemoteAudioTrack,
cx: &mut App,
) -> Result<playback::AudioStream> {
let speaker: Speaker =
serde_urlencoded::from_str(&track.0.name()).unwrap_or_else(|_| Speaker {
name: track.0.name(),
is_staff: false,
sends_legacy_audio: true,
});
if AudioSettings::get_global(cx).rodio_audio {
info!("Using experimental.rodio_audio audio pipeline for output");
playback::play_remote_audio_track(&track.0, cx)
} else {
playback::play_remote_audio_track(&track.0, speaker, cx)
} else if speaker.sends_legacy_audio {
Ok(self.playback.play_remote_audio_track(&track.0))
} else {
Err(anyhow!("Client version too old to play audio in call"))
}
}
}
crates/livekit_client/src/livekit_client/playback.rs
+76 -38
View File
@@ -1,6 +1,6 @@
use anyhow::{Context as _, Result};
use audio::{AudioSettings, CHANNEL_COUNT, SAMPLE_RATE};
use audio::{AudioSettings, CHANNEL_COUNT, LEGACY_CHANNEL_COUNT, LEGACY_SAMPLE_RATE, SAMPLE_RATE};
use cpal::traits::{DeviceTrait, StreamTrait as _};
use futures::channel::mpsc::UnboundedSender;
use futures::{Stream, StreamExt as _};
@@ -43,12 +43,15 @@ pub(crate) struct AudioStack {
pub(crate) fn play_remote_audio_track(
track: &livekit::track::RemoteAudioTrack,
speaker: Speaker,
cx: &mut gpui::App,
) -> Result<AudioStream> {
info!("speaker: {speaker:?}");
let stream =
source::LiveKitStream::new(cx.background_executor(), track, speaker.sends_legacy_audio);
let stop_handle = Arc::new(AtomicBool::new(false));
let stop_handle_clone = stop_handle.clone();
let stream = source::LiveKitStream::new(cx.background_executor(), track);
let stream = stream
.stoppable()
.periodic_access(Duration::from_millis(50), move |s| {
@@ -57,10 +60,8 @@ pub(crate) fn play_remote_audio_track(
}
});
let speaker: Speaker = serde_urlencoded::from_str(&track.name()).unwrap_or_else(|_| Speaker {
name: track.name(),
is_staff: false,
});
info!("sample_rate: {:?}", stream.sample_rate());
info!("channel_count: {:?}", stream.channels());
audio::Audio::play_voip_stream(stream, speaker.name, speaker.is_staff, cx)
.context("Could not play audio")?;
@@ -96,8 +97,8 @@ impl AudioStack {
let next_ssrc = self.next_ssrc.fetch_add(1, Ordering::Relaxed);
let source = AudioMixerSource {
ssrc: next_ssrc,
sample_rate: SAMPLE_RATE.get(),
num_channels: CHANNEL_COUNT.get() as u32,
sample_rate: LEGACY_SAMPLE_RATE.get(),
num_channels: LEGACY_CHANNEL_COUNT.get() as u32,
buffer: Arc::default(),
};
self.mixer.lock().add_source(source.clone());
@@ -137,9 +138,14 @@ impl AudioStack {
let apm = self.apm.clone();
let mixer = self.mixer.clone();
async move {
Self::play_output(apm, mixer, SAMPLE_RATE.get(), CHANNEL_COUNT.get().into())
.await
.log_err();
Self::play_output(
apm,
mixer,
LEGACY_SAMPLE_RATE.get(),
LEGACY_CHANNEL_COUNT.get().into(),
)
.await
.log_err();
}
}));
*self._output_task.borrow_mut() = Arc::downgrade(&task);
@@ -152,19 +158,36 @@ impl AudioStack {
is_staff: bool,
cx: &AsyncApp,
) -> Result<(crate::LocalAudioTrack, AudioStream)> {
let source = NativeAudioSource::new(
// n.b. this struct's options are always ignored, noise cancellation is provided by apm.
AudioSourceOptions::default(),
SAMPLE_RATE.get(),
CHANNEL_COUNT.get().into(),
10,
);
let legacy_audio_compatible =
AudioSettings::try_read_global(cx, |setting| setting.legacy_audio_compatible)
.unwrap_or(true);
let track_name = serde_urlencoded::to_string(Speaker {
let source = if legacy_audio_compatible {
NativeAudioSource::new(
// n.b. this struct's options are always ignored, noise cancellation is provided by apm.
AudioSourceOptions::default(),
LEGACY_SAMPLE_RATE.get(),
LEGACY_CHANNEL_COUNT.get().into(),
10,
)
} else {
NativeAudioSource::new(
// n.b. this struct's options are always ignored, noise cancellation is provided by apm.
AudioSourceOptions::default(),
SAMPLE_RATE.get(),
CHANNEL_COUNT.get().into(),
10,
)
};
let speaker = Speaker {
name: user_name,
is_staff,
})
.context("Could not encode user information in track name")?;
sends_legacy_audio: legacy_audio_compatible,
};
log::info!("Microphone speaker: {speaker:?}");
let track_name = serde_urlencoded::to_string(speaker)
.context("Could not encode user information in track name")?;
let track = track::LocalAudioTrack::create_audio_track(
&track_name,
@@ -186,22 +209,32 @@ impl AudioStack {
let capture_task = if rodio_pipeline {
info!("Using experimental.rodio_audio audio pipeline");
let voip_parts = audio::VoipParts::new(cx)?;
// Audio needs to run real-time and should never be paused. That is why we are using a
// normal std::thread and not a background task
// Audio needs to run real-time and should never be paused. That is
// why we are using a normal std::thread and not a background task
thread::Builder::new()
.name("AudioCapture".to_string())
.name("MicrophoneToLivekit".to_string())
.spawn(move || {
// microphone is non send on mac
let microphone = audio::Audio::open_microphone(voip_parts)?;
let microphone = match audio::Audio::open_microphone(voip_parts) {
Ok(m) => m,
Err(e) => {
log::error!("Could not open microphone: {e}");
return;
}
};
send_to_livekit(frame_tx, microphone);
Ok::<(), anyhow::Error>(())
})
.unwrap();
.expect("should be able to spawn threads");
Task::ready(Ok(()))
} else {
self.executor.spawn(async move {
Self::capture_input(apm, frame_tx, SAMPLE_RATE.get(), CHANNEL_COUNT.get().into())
.await
Self::capture_input(
apm,
frame_tx,
LEGACY_SAMPLE_RATE.get(),
LEGACY_CHANNEL_COUNT.get().into(),
)
.await
})
};
@@ -388,26 +421,31 @@ impl AudioStack {
}
}
#[derive(Serialize, Deserialize)]
struct Speaker {
name: String,
is_staff: bool,
#[derive(Serialize, Deserialize, Debug)]
pub struct Speaker {
pub name: String,
pub is_staff: bool,
pub sends_legacy_audio: bool,
}
fn send_to_livekit(frame_tx: UnboundedSender<AudioFrame<'static>>, mut microphone: impl Source) {
use cpal::Sample;
let sample_rate = microphone.sample_rate().get();
let num_channels = microphone.channels().get() as u32;
let buffer_size = sample_rate / 100 * num_channels;
loop {
let sampled: Vec<_> = microphone
.by_ref()
.take(audio::BUFFER_SIZE)
.take(buffer_size as usize)
.map(|s| s.to_sample())
.collect();
if frame_tx
.unbounded_send(AudioFrame {
sample_rate: SAMPLE_RATE.get(),
num_channels: CHANNEL_COUNT.get() as u32,
samples_per_channel: sampled.len() as u32 / CHANNEL_COUNT.get() as u32,
sample_rate,
num_channels,
samples_per_channel: sampled.len() as u32 / num_channels,
data: Cow::Owned(sampled),
})
.is_err()
crates/livekit_client/src/livekit_client/playback/source.rs
+25 -15
View File
@@ -3,17 +3,19 @@ use std::num::NonZero;
use futures::StreamExt;
use libwebrtc::{audio_stream::native::NativeAudioStream, prelude::AudioFrame};
use livekit::track::RemoteAudioTrack;
use rodio::{Source, buffer::SamplesBuffer, conversions::SampleTypeConverter, nz};
use rodio::{
ChannelCount, SampleRate, Source, buffer::SamplesBuffer, conversions::SampleTypeConverter,
};
use audio::{CHANNEL_COUNT, SAMPLE_RATE};
use audio::{CHANNEL_COUNT, LEGACY_CHANNEL_COUNT, LEGACY_SAMPLE_RATE, SAMPLE_RATE};
fn frame_to_samplesbuffer(frame: AudioFrame) -> SamplesBuffer {
let samples = frame.data.iter().copied();
let samples = SampleTypeConverter::<_, _>::new(samples);
let samples: Vec<f32> = samples.collect();
SamplesBuffer::new(
nz!(2), // frame always has two channels
NonZero::new(frame.sample_rate).expect("audio frame sample rate is nonzero"),
NonZero::new(frame.num_channels as u16).expect("zero channels is nonsense"),
NonZero::new(frame.sample_rate).expect("samplerate zero is nonsense"),
samples,
)
}
@@ -22,14 +24,26 @@ pub struct LiveKitStream {
// shared_buffer: SharedBuffer,
inner: rodio::queue::SourcesQueueOutput,
_receiver_task: gpui::Task<()>,
channel_count: ChannelCount,
sample_rate: SampleRate,
}
impl LiveKitStream {
pub fn new(executor: &gpui::BackgroundExecutor, track: &RemoteAudioTrack) -> Self {
pub fn new(
executor: &gpui::BackgroundExecutor,
track: &RemoteAudioTrack,
legacy: bool,
) -> Self {
let (channel_count, sample_rate) = if legacy {
(LEGACY_CHANNEL_COUNT, LEGACY_SAMPLE_RATE)
} else {
(CHANNEL_COUNT, SAMPLE_RATE)
};
let mut stream = NativeAudioStream::new(
track.rtc_track(),
SAMPLE_RATE.get() as i32,
CHANNEL_COUNT.get().into(),
sample_rate.get() as i32,
channel_count.get().into(),
);
let (queue_input, queue_output) = rodio::queue::queue(true);
// spawn rtc stream
@@ -45,6 +59,8 @@ impl LiveKitStream {
LiveKitStream {
_receiver_task: receiver_task,
inner: queue_output,
sample_rate,
channel_count,
}
}
}
@@ -63,17 +79,11 @@ impl Source for LiveKitStream {
}
fn channels(&self) -> rodio::ChannelCount {
// This must be hardcoded because the playback source assumes constant
// sample rate and channel count. The queue upon which this is build
// will however report different counts and rates. Even though we put in
// only items with our (constant) CHANNEL_COUNT & SAMPLE_RATE this will
// play silence on one channel and at 44100 which is not what our
// constants are.
CHANNEL_COUNT
self.channel_count
}
fn sample_rate(&self) -> rodio::SampleRate {
SAMPLE_RATE // see comment on channels
self.sample_rate
}
fn total_duration(&self) -> Option<std::time::Duration> {