add AudioProcessingModule (#580)

Author: theomonnom

libwebrtc/src/lib.rs

@@ -66,7 +66,7 @@ pub mod video_track;
 pub mod native {
     pub use webrtc_sys::webrtc::ffi::create_random_uuid;
 
-    pub use crate::imp::{audio_resampler, frame_cryptor, yuv_helper};
+    pub use crate::imp::{apm, audio_resampler, frame_cryptor, yuv_helper};
 }
 
 #[cfg(target_os = "android")]

libwebrtc/src/native/apm.rs

@@ -0,0 +1,102 @@
+// Copyright 2023 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use cxx::UniquePtr;
+use webrtc_sys::apm::ffi as sys_apm;
+
+use crate::{RtcError, RtcErrorType};
+
+pub struct AudioProcessingModule {
+    sys_handle: UniquePtr<sys_apm::AudioProcessingModule>,
+}
+
+impl AudioProcessingModule {
+    pub fn new(
+        echo_canceller_enabled: bool,
+        gain_controller_enabled: bool,
+        high_pass_filter_enabled: bool,
+        noise_suppression_enabled: bool,
+    ) -> Self {
+        Self {
+            sys_handle: unsafe {
+                sys_apm::create_apm(
+                    echo_canceller_enabled,
+                    gain_controller_enabled,
+                    high_pass_filter_enabled,
+                    noise_suppression_enabled,
+                )
+            },
+        }
+    }
+
+    pub fn process_stream(
+        &mut self,
+        data: &mut [i16],
+        sample_rate: i32,
+        num_channels: i32,
+    ) -> Result<(), RtcError> {
+        let samples_count = (sample_rate as usize / 100) * num_channels as usize;
+        assert_eq!(data.len(), samples_count, "slice must have 10ms worth of samples");
+
+        if unsafe {
+            // using the same slice for src and dst is safe
+            self.sys_handle.pin_mut().process_stream(
+                data.as_mut_ptr(),
+                data.len(),
+                data.as_mut_ptr(),
+                data.len(),
+                sample_rate,
+                num_channels,
+            )
+        } == 0
+        {
+            Ok(())
+        } else {
+            Err(RtcError {
+                error_type: RtcErrorType::Internal,
+                message: "Failed to process stream".to_string(),
+            })
+        }
+    }
+
+    pub fn process_reverse_stream(
+        &mut self,
+        data: &mut [i16],
+        sample_rate: i32,
+        num_channels: i32,
+    ) -> Result<(), RtcError> {
+        let samples_count = (sample_rate as usize / 100) * num_channels as usize;
+        assert_eq!(data.len(), samples_count, "slice must have 10ms worth of samples");
+
+        if unsafe {
+            // using the same slice for src and dst is safe
+            self.sys_handle.pin_mut().process_reverse_stream(
+                data.as_mut_ptr(),
+                data.len(),
+                data.as_mut_ptr(),
+                data.len(),
+                sample_rate,
+                num_channels,
+            )
+        } == 0
+        {
+            Ok(())
+        } else {
+            Err(RtcError {
+                error_type: RtcErrorType::Internal,
+                message: "Failed to process reverse stream".to_string(),
+            })
+        }
+    }
+}

libwebrtc/src/native/mod.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+pub mod apm;
 #[cfg(target_os = "android")]
 pub mod android;
 pub mod audio_resampler;

livekit-api/src/services/egress.rs

@@ -120,6 +120,7 @@ impl EgressClient {
                     segment_outputs,
                     image_outputs,
                     output: None, // Deprecated
+                    ..Default::default()
                 },
                 self.base
                     .auth_header(VideoGrants { room_record: true, ..Default::default() }, None)?,

livekit-api/src/services/sip.rs

@@ -452,6 +452,7 @@ impl SIPClient {
                     headers: Default::default(),
                     include_headers: Default::default(),
                     media_encryption: Default::default(),
+                    ..Default::default()
                 },
                 self.base.auth_header(
                     Default::default(),

livekit-ffi/protocol/audio_frame.proto

@@ -91,6 +91,42 @@ message RemixAndResampleResponse {
   required OwnedAudioFrameBuffer buffer = 1;
 }
 
+// AEC
+
+
+message NewApmRequest {
+  required bool echo_canceller_enabled = 1;
+  required bool gain_controller_enabled = 2;
+  required bool high_pass_filter_enabled = 3;
+  required bool noise_suppression_enabled = 4;
+}
+message NewApmResponse {
+  required OwnedApm apm = 1;
+}
+
+message ApmProcessStreamRequest {
+  required uint64 apm_handle = 1;
+  required uint64 data_ptr = 2; // *mut i16
+  required uint32 size = 3; // in bytes
+  required uint32 sample_rate = 4;
+  required uint32 num_channels = 5;
+}
+
+message ApmProcessStreamResponse {
+  optional string error = 1;
+}
+
+message ApmProcessReverseStreamRequest {
+  required uint64 apm_handle = 1;
+  required uint64 data_ptr = 2; // *mut i16
+  required uint32 size = 3; // in bytes
+  required uint32 sample_rate = 4;
+  required uint32 num_channels = 5;
+}
+
+message ApmProcessReverseStreamResponse {
+  optional string error = 1;
+}
 
 // New resampler using SoX (much better quality)
 
@@ -241,6 +277,13 @@ message OwnedAudioResampler {
 }
 
 
+//
+// AEC
+//
+
+message OwnedApm {
+  required FfiOwnedHandle handle = 1;
+}
 
 //
 // Sox AudioResampler

livekit-ffi/protocol/ffi.proto

@@ -122,6 +122,12 @@ message FfiRequest {
 
     // Audio Filter Plugin
     LoadAudioFilterPluginRequest load_audio_filter_plugin = 49;
+
+    NewApmRequest new_apm = 50;
+    ApmProcessStreamRequest apm_process_stream = 51;
+    ApmProcessReverseStreamRequest apm_process_reverse_stream = 52;
+
+    // NEXT_ID: 53
   }
 }
 
@@ -191,6 +197,12 @@ message FfiResponse {
 
     // Audio Filter Plugin
     LoadAudioFilterPluginResponse load_audio_filter_plugin = 48;
+
+    NewApmResponse new_apm = 49;
+    ApmProcessStreamResponse apm_process_stream = 50;
+    ApmProcessReverseStreamResponse apm_process_reverse_stream = 51;
+
+    // NEXT_ID: 52
   }
 }
 

livekit-ffi/src/livekit.proto.rs

@@ -1,4 +1,5 @@
 // @generated
+// This file is @generated by prost-build.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct FrameCryptor {
@@ -3554,6 +3555,70 @@ pub struct RemixAndResampleResponse {
     #[prost(message, required, tag="1")]
     pub buffer: OwnedAudioFrameBuffer,
 }
+// AEC
+
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct NewApmRequest {
+    #[prost(bool, required, tag="1")]
+    pub echo_canceller_enabled: bool,
+    #[prost(bool, required, tag="2")]
+    pub gain_controller_enabled: bool,
+    #[prost(bool, required, tag="3")]
+    pub high_pass_filter_enabled: bool,
+    #[prost(bool, required, tag="4")]
+    pub noise_suppression_enabled: bool,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct NewApmResponse {
+    #[prost(message, required, tag="1")]
+    pub apm: OwnedApm,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct ApmProcessStreamRequest {
+    #[prost(uint64, required, tag="1")]
+    pub apm_handle: u64,
+    /// *mut i16
+    #[prost(uint64, required, tag="2")]
+    pub data_ptr: u64,
+    /// in bytes
+    #[prost(uint32, required, tag="3")]
+    pub size: u32,
+    #[prost(uint32, required, tag="4")]
+    pub sample_rate: u32,
+    #[prost(uint32, required, tag="5")]
+    pub num_channels: u32,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct ApmProcessStreamResponse {
+    #[prost(string, optional, tag="1")]
+    pub error: ::core::option::Option<::prost::alloc::string::String>,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct ApmProcessReverseStreamRequest {
+    #[prost(uint64, required, tag="1")]
+    pub apm_handle: u64,
+    /// *mut i16
+    #[prost(uint64, required, tag="2")]
+    pub data_ptr: u64,
+    /// in bytes
+    #[prost(uint32, required, tag="3")]
+    pub size: u32,
+    #[prost(uint32, required, tag="4")]
+    pub sample_rate: u32,
+    #[prost(uint32, required, tag="5")]
+    pub num_channels: u32,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct ApmProcessReverseStreamResponse {
+    #[prost(string, optional, tag="1")]
+    pub error: ::core::option::Option<::prost::alloc::string::String>,
+}
 // New resampler using SoX (much better quality)
 
 #[allow(clippy::derive_partial_eq_without_eq)]
@@ -3746,6 +3811,16 @@ pub struct OwnedAudioResampler {
     pub info: AudioResamplerInfo,
 }
 //
+// AEC
+//
+
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct OwnedApm {
+    #[prost(message, required, tag="1")]
+    pub handle: FfiOwnedHandle,
+}
+//
 // Sox AudioResampler
 //
 
@@ -4075,7 +4150,7 @@ pub struct RpcMethodInvocationEvent {
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct FfiRequest {
-    #[prost(oneof="ffi_request::Message", tags="2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 48, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 49")]
+    #[prost(oneof="ffi_request::Message", tags="2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 48, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 49, 50, 51, 52")]
     pub message: ::core::option::Option<ffi_request::Message>,
 }
 /// Nested message and enum types in `FfiRequest`.
@@ -4188,13 +4263,19 @@ pub mod ffi_request {
         /// Audio Filter Plugin
         #[prost(message, tag="49")]
         LoadAudioFilterPlugin(super::LoadAudioFilterPluginRequest),
+        #[prost(message, tag="50")]
+        NewApm(super::NewApmRequest),
+        #[prost(message, tag="51")]
+        ApmProcessStream(super::ApmProcessStreamRequest),
+        #[prost(message, tag="52")]
+        ApmProcessReverseStream(super::ApmProcessReverseStreamRequest),
     }
 }
 /// This is the output of livekit_ffi_request function.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct FfiResponse {
-    #[prost(oneof="ffi_response::Message", tags="2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 47, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 48")]
+    #[prost(oneof="ffi_response::Message", tags="2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 47, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 48, 49, 50, 51")]
     pub message: ::core::option::Option<ffi_response::Message>,
 }
 /// Nested message and enum types in `FfiResponse`.
@@ -4305,6 +4386,12 @@ pub mod ffi_response {
         /// Audio Filter Plugin
         #[prost(message, tag="48")]
         LoadAudioFilterPlugin(super::LoadAudioFilterPluginResponse),
+        #[prost(message, tag="49")]
+        NewApm(super::NewApmResponse),
+        #[prost(message, tag="50")]
+        ApmProcessStream(super::ApmProcessStreamResponse),
+        #[prost(message, tag="51")]
+        ApmProcessReverseStream(super::ApmProcessReverseStreamResponse),
     }
 }
 /// To minimize complexity, participant events are not included in the protocol.

livekit-ffi/src/server/mod.rs

@@ -24,7 +24,7 @@ use std::{
 
 use dashmap::{mapref::one::MappedRef, DashMap};
 use downcast_rs::{impl_downcast, Downcast};
-use livekit::webrtc::{native::audio_resampler::AudioResampler, prelude::*};
+use livekit::webrtc::{native::apm::AudioProcessingModule, native::audio_resampler::AudioResampler, prelude::*};
 use parking_lot::{deadlock, Mutex};
 use tokio::{sync::oneshot, task::JoinHandle};
 
@@ -67,6 +67,7 @@ pub struct FfiDataBuffer {
 
 impl FfiHandle for FfiDataBuffer {}
 impl FfiHandle for Arc<Mutex<AudioResampler>> {}
+impl FfiHandle for Arc<Mutex<AudioProcessingModule>> {}
 impl FfiHandle for Arc<Mutex<resampler::SoxResampler>> {}
 impl FfiHandle for AudioFrame<'static> {}
 impl FfiHandle for BoxVideoBuffer {}