Voice: echo cancellation (AEC3 in pure Rust), on by default

CONTRIBUTING.md

@@ -12,7 +12,7 @@ Issues and pull requests from anyone are welcome here. The rules that hold in ev
 
 ## Build and test
 
-You need Rust 1.88 or newer. These are the commands CI runs on Windows and on Linux. A pull request needs all of them green:
+You need Rust 1.91 or newer. These are the commands CI runs on Windows and on Linux. A pull request needs all of them green:
 
 ```sh
 cargo fmt --all -- --check

Cargo.toml

@@ -7,7 +7,7 @@ version = "0.0.0"
 edition = "2024"
 license = "MIT OR Apache-2.0"
 repository = "https://github.com/anyplaynet/network"
-rust-version = "1.88"
+rust-version = "1.91"
 
 [workspace.dependencies]
 serde = { version = "1", features = ["derive"] }

crates/control/src/lib.rs

@@ -288,6 +288,10 @@ pub enum Command {
     VoiceNoise {
         enabled: bool,
     },
+    /// Echo cancellation on the microphone: on (the default of a new service) or off.
+    VoiceEcho {
+        enabled: bool,
+    },
     VoiceMode {
         mode: VoiceMode,
     },
@@ -1764,6 +1768,10 @@ mod tests {
             parse_command(r#"{"id":7,"do":"voice_noise","enabled":false}"#),
             Ok((7, Command::VoiceNoise { enabled: false }))
         );
+        assert_eq!(
+            parse_command(r#"{"id":8,"do":"voice_echo","enabled":false}"#),
+            Ok((8, Command::VoiceEcho { enabled: false }))
+        );
         assert_eq!(
             parse_command(r#"{"id":2,"do":"push_to_talk","talking":false}"#),
             Ok((2, Command::PushToTalk { talking: false }))

crates/net/src/service.rs

@@ -1479,6 +1479,13 @@ impl Service {
                 }
                 self.ok(id, Done {});
             }
+            Command::VoiceEcho { enabled } => {
+                self.voice.settings.echo_cancellation = Some(enabled);
+                if let Some(hub) = &self.voice.hub {
+                    hub.set_echo_cancellation(enabled);
+                }
+                self.ok(id, Done {});
+            }
             Command::MicMute { muted } => {
                 self.voice.settings.mic_muted = muted;
                 if let Some(hub) = &self.voice.hub {
@@ -3082,13 +3089,15 @@ impl Service {
         let (input, output) = (self.voice.input.clone(), self.voice.output.clone());
         let mode = self.voice.settings.mode;
         let noise = self.voice.settings.noise_suppression;
+        let echo = self.voice.settings.echo_cancellation;
         self.spawn(
             async move {
                 tokio::task::spawn_blocking(move || {
                     let device = audio.open(input.as_deref(), output.as_deref())?;
                     let config = anyplay_voice::Config {
                         mode,
                         noise_suppression: noise.unwrap_or(true),
+                        echo_cancellation: echo.unwrap_or(true),
                         ..anyplay_voice::Config::default()
                     };
                     Engine::new(config, device).map_err(|e| e.to_string())

crates/net/src/voice.rs

@@ -164,6 +164,8 @@ pub struct Settings {
     pub talking: bool,
     /// Noise suppression on the microphone; `None` leaves what the engine was opened with.
     pub noise_suppression: Option<bool>,
+    /// Echo cancellation on the microphone; `None` leaves what the engine was opened with.
+    pub echo_cancellation: Option<bool>,
     /// Volume per account, 0.0 to 2.0; missing is 1.0.
     pub volumes: HashMap<u64, f32>,
     pub muted: HashSet<u64>,
@@ -197,6 +199,7 @@ enum Ctl {
     Talking(bool),
     Mode(TalkMode),
     Noise(bool),
+    Echo(bool),
     Volume(u64, f32),
     Muted(u64, bool),
     MicMuted(bool),
@@ -269,6 +272,10 @@ impl Hub {
         let _ = self.ctl.send(Ctl::Noise(enabled));
     }
 
+    pub fn set_echo_cancellation(&self, enabled: bool) {
+        let _ = self.ctl.send(Ctl::Echo(enabled));
+    }
+
     /// 0.0 (silent) to 2.0.
     pub fn set_volume(&self, account: u64, volume: f32) {
         let _ = self.ctl.send(Ctl::Volume(account, volume));
@@ -335,6 +342,9 @@ impl Task {
         if let Some(enabled) = self.settings.noise_suppression {
             self.engine.set_noise_suppression(enabled);
         }
+        if let Some(enabled) = self.settings.echo_cancellation {
+            self.engine.set_echo_cancellation(enabled);
+        }
         self.engine.set_talking(self.settings.talking);
         self.engine.set_mic_muted(self.settings.mic_muted);
         self.engine.set_deafened(self.settings.deafened);
@@ -387,6 +397,10 @@ impl Task {
                 self.settings.noise_suppression = Some(enabled);
                 self.engine.set_noise_suppression(enabled);
             }
+            Ctl::Echo(enabled) => {
+                self.settings.echo_cancellation = Some(enabled);
+                self.engine.set_echo_cancellation(enabled);
+            }
             Ctl::MicMuted(muted) => {
                 self.settings.mic_muted = muted;
                 self.engine.set_mic_muted(muted);

crates/voice/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "anyplay-voice"
-description = "Party voice for the AnyPlay network: noise suppression, Opus, jitter buffer and mixer behind a small interface"
+description = "Party voice for the AnyPlay network: noise suppression, echo cancellation, Opus, jitter buffer and mixer behind a small interface"
 version.workspace = true
 edition.workspace = true
 license.workspace = true
@@ -15,6 +15,7 @@ devices = ["dep:cpal"]
 [dependencies]
 nnnoiseless = { version = "0.5.2", default-features = false }
 opus = "0.4"
+sonora = "0.2"
 cpal = { version = "0.18", optional = true }
 
 [[example]]

crates/voice/src/echo.rs

@@ -0,0 +1,137 @@
+//! Acoustic echo cancellation: the far end's voice, played by the speakers, is taken out of what
+//! the microphone hears.
+//!
+//! [`EchoCanceller`] is WebRTC's AEC3 (the pure-Rust port `sonora`) on mono 48 kHz in blocks of
+//! 10 ms. It wants two things: every block the speakers play (the reference, "render"), and every
+//! block the microphone captured ("capture"), which comes back with the echo removed.
+//!
+//! The engine writes the reference itself, so it knows it exactly. What it does not know is when
+//! the speakers play it: it queues up to 60 ms ahead of the device. [`Path`] holds the reference
+//! back until the device has taken it off the queue, so the delay the canceller has to find is
+//! only the device's own buffers and the room, which AEC3's delay estimator follows.
+
+use std::collections::VecDeque;
+
+use sonora::config::EchoCanceller as AecConfig;
+use sonora::{AudioProcessing, Config, StreamConfig};
+
+/// Samples in the 10 ms block AEC3 works on, at 48 kHz.
+pub const BLOCK: usize = 480;
+
+/// What the device is guessed to add between the engine and the air, in milliseconds. AEC3
+/// estimates the delay itself; this only tells it where to start.
+const DEVICE_DELAY_HINT_MS: i32 = 40;
+/// The most reference kept waiting for a device that has stopped taking samples: one second.
+const PENDING_LIMIT: usize = 48_000;
+/// Capture blocks (10 ms) without any reference released after which the canceller is bypassed:
+/// a device that plays nothing (stopped, or not draining) makes no echo, and AEC3 without a
+/// reference only colours the microphone.
+const STALLED_BLOCKS: u32 = 50;
+
+/// AEC3 on one mono 48 kHz stream pair.
+pub struct EchoCanceller {
+    apm: AudioProcessing,
+}
+
+impl Default for EchoCanceller {
+    fn default() -> EchoCanceller {
+        EchoCanceller::new()
+    }
+}
+
+impl EchoCanceller {
+    pub fn new() -> EchoCanceller {
+        let config = Config {
+            echo_canceller: Some(AecConfig::default()),
+            ..Config::default()
+        };
+        let mut apm = AudioProcessing::builder()
+            .config(config)
+            .capture_config(StreamConfig::new(48_000, 1))
+            .render_config(StreamConfig::new(48_000, 1))
+            .build();
+        // Out of range is clamped and still processes; nothing to do about it here.
+        let _ = apm.set_stream_delay_ms(DEVICE_DELAY_HINT_MS);
+        EchoCanceller { apm }
+    }
+
+    /// The next 10 ms the speakers play. Anything but [`BLOCK`] samples is ignored.
+    pub fn render(&mut self, block: &[f32]) {
+        if block.len() != BLOCK {
+            return;
+        }
+        let mut sink = [0.0f32; BLOCK];
+        let _ = self.apm.process_render_f32(&[block], &mut [&mut sink[..]]);
+    }
+
+    /// The next 10 ms the microphone captured, in place: the echo of what [`render`] was given
+    /// is taken out. If the canceller refuses the block it stays as it was.
+    ///
+    /// [`render`]: EchoCanceller::render
+    pub fn capture(&mut self, block: &mut [f32]) {
+        if block.len() != BLOCK {
+            return;
+        }
+        let input: [f32; BLOCK] = std::array::from_fn(|i| block[i]);
+        let mut output = [0.0f32; BLOCK];
+        if self
+            .apm
+            .process_capture_f32(&[&input[..]], &mut [&mut output[..]])
+            .is_ok()
+        {
+            block.copy_from_slice(&output);
+        }
+    }
+}
+
+/// An [`EchoCanceller`] plus the reference waiting for the speakers to play it.
+pub(crate) struct Path {
+    canceller: EchoCanceller,
+    /// Reference at 48 kHz that was written to the device and not yet released to the canceller.
+    pending: VecDeque<f32>,
+    /// Capture blocks since the last reference block went to the canceller; `None` before the
+    /// first.
+    since_reference: Option<u32>,
+}
+
+impl Path {
+    pub(crate) fn new() -> Path {
+        Path {
+            canceller: EchoCanceller::new(),
+            pending: VecDeque::new(),
+            since_reference: None,
+        }
+    }
+
+    /// Samples just written to the device, at 48 kHz.
+    pub(crate) fn reference(&mut self, samples: &[f32]) {
+        self.pending.extend(samples);
+        if self.pending.len() > PENDING_LIMIT {
+            let excess = self.pending.len() - PENDING_LIMIT;
+            self.pending.drain(..excess);
+        }
+    }
+
+    /// The device still has `queued` samples (at 48 kHz) of what was written to it waiting; the
+    /// rest has been taken to the speakers, so it goes to the canceller now.
+    pub(crate) fn release(&mut self, queued: usize) {
+        while self.pending.len() >= queued + BLOCK {
+            let block: [f32; BLOCK] =
+                std::array::from_fn(|_| self.pending.pop_front().unwrap_or(0.0));
+            self.canceller.render(&block);
+            self.since_reference = Some(0);
+        }
+    }
+
+    pub(crate) fn capture(&mut self, block: &mut [f32]) {
+        for half in block.as_chunks_mut::<BLOCK>().0 {
+            match &mut self.since_reference {
+                Some(n) if *n < STALLED_BLOCKS => {
+                    *n += 1;
+                    self.canceller.capture(half);
+                }
+                _ => {}
+            }
+        }
+    }
+}