Merge pull request #152 from babblevoice/wideband-reader

Native 16k audio tap: wideband reader (captions) + native-rate recordings

Author: tinpotnick

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "@babblevoice/projectrtp",
-  "version": "3.1.0",
+  "version": "3.2.0",
   "description": "A scalable Node addon RTP server",
   "main": "index.js",
   "directories": {

package.json

@@ -628,6 +628,9 @@ async fn handle_command_local(
         }
 
         Command::Record { cfg, ack } => {
+            // Record at the channel's native rate (16k for G.722, else 8k).
+            let mut cfg = cfg;
+            cfg.sample_rate = state.codecx.native_samplerate();
             // A bare `record` at a different path leaves a pending recorder
             // dangling — clear it so the play-end activation doesn't later
             // open it alongside this fresh one.
@@ -773,9 +776,12 @@ async fn handle_command_local(
             // play-end (or barge-in). The pre-buffer accumulates inbound
             // samples while the player runs; on activation it is flushed
             // via `write_raw`. C++ `pendingrecorder` equivalent.
-            let file_str = cfg.recorder.file.to_string_lossy().into_owned();
+            // Record at the channel's native rate (16k for G.722, else 8k).
+            let mut recorder_cfg = cfg.recorder;
+            recorder_cfg.sample_rate = state.codecx.native_samplerate();
+            let file_str = recorder_cfg.file.to_string_lossy().into_owned();
             subs.pending_recorder = Some(PendingRecorder {
-                cfg: cfg.recorder,
+                cfg: recorder_cfg,
                 file_str,
             });
 

rust/src/channel/actor.rs

@@ -119,9 +119,11 @@ impl AudioReader {
     pub fn is_closed(&self) -> bool { self.sender.is_closed() }
 
     /// Feed one 20 ms frame. Called from the recorder's feed point in the
-    /// tick (same cache state, same timing). Samples are already narrowband
-    /// 8 kHz linear — wideband and wire-byte formats are pulled directly
-    /// from `codecx`.
+    /// tick (same cache state, same timing). Samples are narrowband 8 kHz
+    /// linear; wideband (16 kHz) and wire-byte formats are pulled from
+    /// `codecx` for the inbound side. The outbound side has no `codecx`, so a
+    /// caller that holds wideband for the out side (the mixer, where out is
+    /// the peer's inbound) supplies it via `feed_with`.
     ///
     /// Never blocks. On full queue or closed consumer the frame is silently
     /// dropped (bumping `drops`). Returns `true` if the frame was queued.
@@ -131,7 +133,21 @@ impl AudioReader {
         in_samples_8k: Option<&[i16]>,
         out_samples_8k: Option<&[i16]>,
     ) -> bool {
-        let Some(bytes) = self.build_frame(codecx, in_samples_8k, out_samples_8k) else {
+        self.feed_with(codecx, in_samples_8k, out_samples_8k, None)
+    }
+
+    /// As [`feed`](Self::feed), but with the outbound side's wideband (16 kHz)
+    /// samples supplied by the caller. Used by the mixer so an `out`/`both`
+    /// reader at 16 kHz emits the peer's true wideband instead of silence.
+    /// Ignored by 8 kHz readers and wire-byte formats.
+    pub fn feed_with(
+        &mut self,
+        codecx: &mut CodecBundle,
+        in_samples_8k: Option<&[i16]>,
+        out_samples_8k: Option<&[i16]>,
+        out_samples_16k: Option<&[i16]>,
+    ) -> bool {
+        let Some(bytes) = self.build_frame(codecx, in_samples_8k, out_samples_8k, out_samples_16k) else {
             return false;
         };
         match self.sender.try_send(bytes) {
@@ -145,9 +161,10 @@ impl AudioReader {
         codecx: &mut CodecBundle,
         in_samples_8k: Option<&[i16]>,
         out_samples_8k: Option<&[i16]>,
+        out_samples_16k: Option<&[i16]>,
     ) -> Option<Vec<u8>> {
         match self.cfg.format {
-            ReaderFormat::L16 => self.build_l16(codecx, in_samples_8k, out_samples_8k),
+            ReaderFormat::L16 => self.build_l16(codecx, in_samples_8k, out_samples_8k, out_samples_16k),
             ReaderFormat::Pcma => codecx.require_wire_as(8).map(|b| b.to_vec()),
             ReaderFormat::Pcmu => codecx.require_wire_as(0).map(|b| b.to_vec()),
             ReaderFormat::G722 => codecx.require_wire_as(9).map(|b| b.to_vec()),
@@ -162,16 +179,20 @@ impl AudioReader {
         codecx: &mut CodecBundle,
         in_samples_8k: Option<&[i16]>,
         out_samples_8k: Option<&[i16]>,
+        out_samples_16k: Option<&[i16]>,
     ) -> Option<Vec<u8>> {
         // Resolve the two sides at the requested sample rate.
         let (in_samples, out_samples): (Option<Vec<i16>>, Option<Vec<i16>>) = match self.cfg.samplerate {
             16000 => {
-                // Wideband is cached on codecx; upsamples from narrowband if the
-                // wire is a narrowband codec. Only the "in" side lives on codecx
-                // — the "out" side can't sensibly become wideband in v1 (no
-                // upsample cache for player frames). Treat out=None at 16k.
-                let wb = codecx.require_wideband_16k().map(|s| s.to_vec());
-                (wb, None)
+                // Inbound wideband is cached on codecx (decoded from G.722, or
+                // upsampled from narrowband). The outbound side has no codecx
+                // here, so its wideband must be supplied by the caller — the
+                // mixer passes the peer's wideband for a bridged call. When it
+                // isn't supplied (e.g. a non-bridged channel whose out is an
+                // 8 kHz player) the out side is silent at 16k.
+                let wb_in = codecx.require_wideband_16k().map(|s| s.to_vec());
+                let wb_out = out_samples_16k.map(|s| s.to_vec());
+                (wb_in, wb_out)
             }
             _ => (
                 in_samples_8k.map(|s| s.to_vec()),
@@ -257,3 +278,50 @@ pub struct ForwarderHandle {
     pub id: u64,
     pub cancel: Arc<tokio::sync::Notify>,
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::codec::CodecBundle;
+
+    fn reader(direction: ReaderDirection, samplerate: u32) -> AudioReader {
+        let (tx, _rx) = make_channel();
+        let cfg = ReaderConfig { direction, samplerate, ..Default::default() };
+        AudioReader::new(1, cfg, tx)
+    }
+
+    // The fix: at 16 kHz an `out` reader emits the caller-supplied peer
+    // wideband. Previously the out side was hard-coded to silence at 16 kHz.
+    #[test]
+    fn out_reader_16k_uses_supplied_wideband() {
+        let r = reader(ReaderDirection::Out, 16000);
+        let mut cx = CodecBundle::new();
+        let out_wb: Vec<i16> = vec![ 1234; 320 ]; // 20 ms @ 16 kHz mono
+        let in_8k = vec![ 0i16; 160 ];
+        let out_8k = vec![ 0i16; 160 ];
+
+        let bytes = r
+            .build_frame(&mut cx, Some(&in_8k), Some(&out_8k), Some(&out_wb))
+            .expect("frame produced");
+
+        assert_eq!(bytes.len(), 320 * 2, "16k mono 20ms = 640 bytes");
+        assert_eq!(i16::from_le_bytes([ bytes[0], bytes[1] ]), 1234);
+    }
+
+    // The 8 kHz path is unchanged: the out side uses the 8 kHz samples directly
+    // and ignores any supplied wideband.
+    #[test]
+    fn out_reader_8k_uses_narrowband() {
+        let r = reader(ReaderDirection::Out, 8000);
+        let mut cx = CodecBundle::new();
+        let in_8k = vec![ 0i16; 160 ];
+        let out_8k = vec![ 321i16; 160 ];
+
+        let bytes = r
+            .build_frame(&mut cx, Some(&in_8k), Some(&out_8k), None)
+            .expect("frame produced");
+
+        assert_eq!(bytes.len(), 160 * 2, "8k mono 20ms = 320 bytes");
+        assert_eq!(i16::from_le_bytes([ bytes[0], bytes[1] ]), 321);
+    }
+}

rust/src/channel/audio_reader.rs

@@ -577,19 +577,36 @@ impl Member {
     /// R=far leg) — matches C++ mix recording semantics. When `None`
     /// (Local mode, N≥3, or peer had no inbound this tick) stereo
     /// falls back to duplicate-mono.
-    async fn write_recordings(&mut self, peer_samples: Option<&[i16]>) {
+    async fn write_recordings(&mut self, peer_samples: Option<&[i16]>, peer_wideband: Option<&[i16]>) {
         if !self.inbound_this_tick { return; }
         let Some(samples) = self.state.codecx.require_narrowband_8k().map(|s| s.to_vec()) else { return; };
+        let self_wb = if self.state.codecx.is_wideband() {
+            self.state.codecx.require_wideband_16k().map(|s| s.to_vec())
+        } else {
+            None
+        };
         let ch_count = self.state.in_count.load(Ordering::Relaxed);
+
+        // Recorders capture at the channel's native rate: wideband self + peer
+        // when the codec is wideband (recorder WAV rate is stamped to match),
+        // else 8 kHz narrowband. Power runs on whichever self leg is used
+        // (RMS is amplitude-normalised, so thresholds hold at either rate).
+        let (rec_self, rec_peer): (&[i16], Option<&[i16]>) = match self_wb.as_deref() {
+            Some(wb) => (wb, peer_wideband),
+            None => (&samples, peer_samples),
+        };
         feed_recorders(
-            &mut self.subs.recorders, &samples, peer_samples, ch_count,
+            &mut self.subs.recorders, rec_self, rec_peer, ch_count,
             &mut self.state.pending_events,
         ).await;
-        // AudioReaders share the recorder's L/R convention: L=self inbound,
-        // R=peer inbound (in mix mode that's the "outbound" side — what
-        // we'd send out). Same cache, same timing as the recorder.
+
+        // AudioReaders share the recorder's L/R convention (L=self inbound,
+        // R=peer inbound — in mix mode that's the outbound side). Readers
+        // manage their own rate: the in side is fed 8 kHz (16k readers pull
+        // wideband from codecx), and a 16k reader's out side gets the peer's
+        // wideband via feed_with.
         for reader in self.subs.readers.iter_mut() {
-            reader.feed(&mut self.state.codecx, Some(&samples), peer_samples);
+            reader.feed_with(&mut self.state.codecx, Some(&samples), peer_samples, peer_wideband);
         }
         self.subs.readers.retain(|r| !r.is_closed());
     }
@@ -720,6 +737,7 @@ async fn run_post_mix_phase(
     n_alive: usize,
 ) {
     let peer_samples_by_id = compute_peer_samples_by_id(members, n_alive);
+    let peer_wideband_by_id = compute_peer_wideband_by_id(members, n_alive);
 
     // Clone the id list so we can look up each member's peer samples
     // from the sibling map while holding a `&mut` to the member.
@@ -735,8 +753,16 @@ async fn run_post_mix_phase(
         } else {
             None
         };
+        // 16k wideband counterpart for the reader tap's out side (320 samples).
+        let peer_wideband: Option<Vec<i16>> = if n_alive == 2 {
+            peer_wideband_by_id.iter()
+                .find_map(|(&pid, s)| if pid != id { Some(s.clone()) } else { None })
+                .or_else(|| Some(vec![ 0i16; MIX_FRAME_SAMPLES * 2 ]))
+        } else {
+            None
+        };
         let Some(m) = members.get_mut(&id) else { continue; };
-        m.write_recordings(peer_samples.as_deref()).await;
+        m.write_recordings(peer_samples.as_deref(), peer_wideband.as_deref()).await;
         m.send_dtmf_outbound().await;
         m.drain_pending_events();
     }
@@ -768,6 +794,32 @@ fn compute_peer_samples_by_id(
         .collect()
 }
 
+/// Build the per-member "peer wideband" (16 kHz) map for the N=2 reader tap —
+/// the far party's true wideband (decoded from G.722, or upsampled from
+/// narrowband) so an `out`/`both` reader at 16 kHz isn't fed a downsampled
+/// copy. Mirrors `compute_peer_samples_by_id` but at 16 kHz (320 samples /
+/// 20 ms). Empty for N≠2; only the reader uses this (the recorder stays 8 kHz).
+fn compute_peer_wideband_by_id(
+    members: &mut HashMap<ChannelId, Box<Member>>,
+    n_alive: usize,
+) -> HashMap<ChannelId, Vec<i16>> {
+    if n_alive != 2 {
+        return HashMap::new();
+    }
+    members.iter_mut()
+        .map(|(&id, m)| {
+            let samples = if m.inbound_this_tick {
+                m.state.codecx.require_wideband_16k()
+                    .map(|s| s.to_vec())
+                    .unwrap_or_else(|| vec![ 0i16; MIX_FRAME_SAMPLES * 2 ])
+            } else {
+                vec![ 0i16; MIX_FRAME_SAMPLES * 2 ]
+            };
+            (id, samples)
+        })
+        .collect()
+}
+
 /// For N=2, tell each member's codec bundle the peer's iLBC wire PT
 /// (when dynamic, i.e. ≥96) so the bridge can decode/encode at the
 /// right PT on both sides of an asymmetric-dynamic pair.
@@ -1001,6 +1053,9 @@ async fn apply_forwarded(m: &mut Member, cmd: Command) {
             let _ = ack.send(());
         }
         Command::Record { cfg, ack } => {
+            // Record at the channel's native rate (16k for G.722, else 8k).
+            let mut cfg = cfg;
+            cfg.sample_rate = m.state.codecx.native_samplerate();
             m.subs.pending_recorder = None;
             m.subs.prebuffer.clear();
             let file_str = cfg.file.to_string_lossy().into_owned();
@@ -1131,9 +1186,12 @@ async fn apply_forwarded(m: &mut Member, cmd: Command) {
             }
             // Queue the recorder — opened on play-end or barge-in via
             // `activate_pending_recorder`. Parity with local-mode handler.
-            let file_str = cfg.recorder.file.to_string_lossy().into_owned();
+            // Record at the channel's native rate (16k for G.722, else 8k).
+            let mut recorder_cfg = cfg.recorder;
+            recorder_cfg.sample_rate = m.state.codecx.native_samplerate();
+            let file_str = recorder_cfg.file.to_string_lossy().into_owned();
             m.subs.pending_recorder = Some(super::actor::PendingRecorder {
-                cfg: cfg.recorder,
+                cfg: recorder_cfg,
                 file_str,
             });
             let _ = ack.send(());

rust/src/channel/mixer.rs

@@ -306,11 +306,41 @@ async fn feed_recorders_and_readers(
     // inbound produces no samples, so it doesn't qualify.
     let has_recordable_audio = decoded.is_some() || player_frame.is_some();
     if has_recordable_audio {
-        write_recorder_frames(state, subs, in_s, out_s).await;
+        if state.codecx.is_wideband() {
+            // Native-rate recording on a wideband channel: true wideband
+            // inbound (G.722 decode) + the outbound (player/echo/silence)
+            // upsampled to 16 kHz so the WAV stays coherent. The recorder's
+            // WAV rate is stamped to 16k at open to match.
+            let in_wb = state.codecx.require_wideband_16k()
+                .map(|s| s.to_vec())
+                .unwrap_or_else(|| upsample_2x(in_s));
+            let out_wb = upsample_2x(out_s);
+            write_recorder_frames(state, subs, &in_wb, &out_wb).await;
+        } else {
+            write_recorder_frames(state, subs, in_s, out_s).await;
+        }
     }
+    // Readers manage their own rate (a 16k reader pulls wideband from codecx);
+    // always hand them the 8 kHz narrowband here.
     feed_readers(state, subs, in_s, out_s);
 }
 
+/// Linear 2x upsample (8 kHz -> 16 kHz) used for the recorder's outbound leg
+/// on a wideband channel, where the player/echo source is narrowband. The
+/// inbound leg uses the codec's true wideband decode; this only covers the
+/// out side so a stereo native recording stays sample-aligned.
+fn upsample_2x(input: &[i16]) -> Vec<i16> {
+    if input.is_empty() { return Vec::new(); }
+    let mut out = Vec::with_capacity(input.len() * 2);
+    for i in 0..input.len() {
+        let cur = input[i] as i32;
+        let next = if i + 1 < input.len() { input[i + 1] as i32 } else { cur };
+        out.push(cur as i16);
+        out.push(((cur + next) / 2) as i16);
+    }
+    out
+}
+
 async fn write_recorder_frames(
     state: &mut ChannelState,
     subs: &mut Subsystems,
@@ -763,4 +793,18 @@ mod tests {
             "expected full IVR digit sequence; got {:?}", digits
         );
     }
+
+    #[test]
+    fn upsample_2x_doubles_and_interpolates() {
+        let input = [100i16, 200, 300, 400];
+        let out = upsample_2x(&input);
+        assert_eq!(out.len(), 8);
+        assert_eq!(out[0], 100);
+        assert_eq!(out[1], 150); // (100+200)/2
+        assert_eq!(out[2], 200);
+        assert_eq!(out[3], 250); // (200+300)/2
+        // last sample duplicates (no next to interpolate toward)
+        assert_eq!(out[7], 400);
+        assert!(upsample_2x(&[]).is_empty());
+    }
 }