Factor out output buffering for KISS sockets

[m17rt] / m17codec2 / src / lib.rs

diff --git a/m17codec2/src/lib.rs b/m17codec2/src/lib.rs
old mode 100755 (executable)
new mode 100644 (file)
index 1d05f27..693cb09
--- a/m17codec2/src/lib.rs
+++ b/m17codec2/src/lib.rs
@@ -1,3 +1,5 @@
+#![doc = include_str!("../README.md")]
+

 use codec2::{Codec2, Codec2Mode};
 use cpal::traits::DeviceTrait;
 use cpal::traits::HostTrait;
 use codec2::{Codec2, Codec2Mode};
 use cpal::traits::DeviceTrait;
 use cpal::traits::HostTrait;


@@ -6,9 +8,13 @@ use cpal::{Sample, SampleFormat, SampleRate};
 use log::debug;
 use m17app::adapter::StreamAdapter;
 use m17app::app::TxHandle;
 use log::debug;
 use m17app::adapter::StreamAdapter;
 use m17app::app::TxHandle;

+use m17app::error::AdapterError;

 use m17app::link_setup::LinkSetup;
 use m17app::link_setup::M17Address;
 use m17app::StreamFrame;
 use m17app::link_setup::LinkSetup;
 use m17app::link_setup::M17Address;
 use m17app::StreamFrame;

+use rubato::Resampler;
+use rubato::SincFixedIn;
+use rubato::SincInterpolationParameters;

 use std::collections::VecDeque;
 use std::fs::File;
 use std::io::Write;
 use std::collections::VecDeque;
 use std::fs::File;
 use std::io::Write;


@@ -20,7 +26,12 @@ use std::sync::{
 };
 use std::time::Duration;
 use std::time::Instant;
 };
 use std::time::Duration;
 use std::time::Instant;

+use thiserror::Error;
+
+pub mod soundcards;

 
 

+/// Write one or more 8-byte chunks of 3200-bit Codec2 to a raw S16LE file
+/// and return the samples.

 pub fn decode_codec2<P: AsRef<Path>>(data: &[u8], out_path: P) -> Vec<i16> {
     let codec2 = Codec2::new(Codec2Mode::MODE_3200);
     let var_name = codec2;
 pub fn decode_codec2<P: AsRef<Path>>(data: &[u8], out_path: P) -> Vec<i16> {
     let codec2 = Codec2::new(Codec2Mode::MODE_3200);
     let var_name = codec2;


@@ -31,8 +42,6 @@ pub fn decode_codec2<P: AsRef<Path>>(data: &[u8], out_path: P) -> Vec<i16> {
         codec.decode(&mut samples, &data[i * 8..((i + 1) * 8)]);
         all_samples.append(&mut samples);
     }
         codec.decode(&mut samples, &data[i * 8..((i + 1) * 8)]);
         all_samples.append(&mut samples);
     }

-
-    // dude this works

     let mut speech_out = File::create(out_path).unwrap();
     for b in &all_samples {
         speech_out.write_all(&b.to_le_bytes()).unwrap();
     let mut speech_out = File::create(out_path).unwrap();
     for b in &all_samples {
         speech_out.write_all(&b.to_le_bytes()).unwrap();


@@ -40,10 +49,10 @@ pub fn decode_codec2<P: AsRef<Path>>(data: &[u8], out_path: P) -> Vec<i16> {
     all_samples
 }
 
     all_samples
 }
 

+/// Subscribes to M17 streams and attempts to play the decoded Codec2

 pub struct Codec2Adapter {
     state: Arc<Mutex<AdapterState>>,
 pub struct Codec2Adapter {
     state: Arc<Mutex<AdapterState>>,

-    // TODO: make this configurable
-    output_card: String,
+    output_card: Option<String>,

 }
 
 impl Codec2Adapter {
 }
 
 impl Codec2Adapter {


@@ -54,11 +63,15 @@ impl Codec2Adapter {
                 out_buf: VecDeque::new(),
                 codec2: Codec2::new(Codec2Mode::MODE_3200),
                 end_tx: None,
                 out_buf: VecDeque::new(),
                 codec2: Codec2::new(Codec2Mode::MODE_3200),
                 end_tx: None,

+                resampler: None,

             })),
             })),

-            // TODO: this doesn't work on rpi. Use default_output_device() by default
-            output_card: "default".to_owned(),
+            output_card: None,

         }
     }
         }
     }

+
+    pub fn set_output_card<S: Into<String>>(&mut self, card_name: S) {
+        self.output_card = Some(card_name.into());
+    }

 }
 
 impl Default for Codec2Adapter {
 }
 
 impl Default for Codec2Adapter {


@@ -73,29 +86,44 @@ struct AdapterState {
     out_buf: VecDeque<i16>,
     codec2: Codec2,
     end_tx: Option<Sender<()>>,
     out_buf: VecDeque<i16>,
     codec2: Codec2,
     end_tx: Option<Sender<()>>,

+    resampler: Option<SincFixedIn<f32>>,

 }
 
 impl StreamAdapter for Codec2Adapter {
 }
 
 impl StreamAdapter for Codec2Adapter {

-    fn adapter_registered(&self, _id: usize, handle: TxHandle) {
+    fn start(&self, handle: TxHandle) -> Result<(), AdapterError> {

         self.state.lock().unwrap().tx = Some(handle);
 
         let (end_tx, end_rx) = channel();
         self.state.lock().unwrap().tx = Some(handle);
 
         let (end_tx, end_rx) = channel();

+        let (setup_tx, setup_rx) = channel();

         let state = self.state.clone();
         let output_card = self.output_card.clone();
         let state = self.state.clone();
         let output_card = self.output_card.clone();

-        std::thread::spawn(move || stream_thread(end_rx, state, output_card));
+        std::thread::spawn(move || stream_thread(end_rx, setup_tx, state, output_card));

         self.state.lock().unwrap().end_tx = Some(end_tx);
         self.state.lock().unwrap().end_tx = Some(end_tx);

+        // Propagate any errors arising in the thread
+        let sample_rate = setup_rx.recv()??;
+        debug!("selected codec2 output sample rate {sample_rate}");
+        if sample_rate != 8000 {
+            let params = SincInterpolationParameters {
+                sinc_len: 256,
+                f_cutoff: 0.95,
+                oversampling_factor: 256,
+                interpolation: rubato::SincInterpolationType::Cubic,
+                window: rubato::WindowFunction::BlackmanHarris2,
+            };
+            // TODO: fix unwrap
+            self.state.lock().unwrap().resampler =
+                Some(SincFixedIn::new(sample_rate as f64 / 8000f64, 1.0, params, 160, 1).unwrap());
+        }
+        Ok(())

     }
 
     }
 

-    fn adapter_removed(&self) {
+    fn close(&self) -> Result<(), AdapterError> {

         let mut state = self.state.lock().unwrap();
         state.tx = None;
         state.end_tx = None;
         let mut state = self.state.lock().unwrap();
         state.tx = None;
         state.end_tx = None;

+        Ok(())

     }
 
     }
 

-    fn tnc_started(&self) {}
-
-    fn tnc_closed(&self) {}
-

     fn stream_began(&self, _link_setup: LinkSetup) {
         // for now we will assume:
         // - unencrypted
     fn stream_began(&self, _link_setup: LinkSetup) {
         // for now we will assume:
         // - unencrypted


@@ -108,12 +136,21 @@ impl StreamAdapter for Codec2Adapter {
     fn stream_data(&self, _frame_number: u16, _is_final: bool, data: Arc<[u8; 16]>) {
         let mut state = self.state.lock().unwrap();
         for encoded in data.chunks(8) {
     fn stream_data(&self, _frame_number: u16, _is_final: bool, data: Arc<[u8; 16]>) {
         let mut state = self.state.lock().unwrap();
         for encoded in data.chunks(8) {

-            if state.out_buf.len() < 1024 {
+            if state.out_buf.len() < 8192 {

                 let mut samples = [i16::EQUILIBRIUM; 160]; // while assuming 3200
                 state.codec2.decode(&mut samples, encoded);
                 let mut samples = [i16::EQUILIBRIUM; 160]; // while assuming 3200
                 state.codec2.decode(&mut samples, encoded);

-                // TODO: maybe get rid of VecDeque so we can decode directly into ring buffer?
-                for s in samples {
-                    state.out_buf.push_back(s);
+                if let Some(resampler) = state.resampler.as_mut() {
+                    let samples_f: Vec<f32> =
+                        samples.iter().map(|s| *s as f32 / 16384.0f32).collect();
+                    let res = resampler.process(&vec![samples_f], None).unwrap();
+                    for s in &res[0] {
+                        state.out_buf.push_back((s * 16383.0f32) as i16);
+                    }
+                } else {
+                    // TODO: maybe get rid of VecDeque so we can decode directly into ring buffer?
+                    for s in samples {
+                        state.out_buf.push_back(s);
+                    }

                 }
             } else {
                 debug!("out_buf overflow");
                 }
             } else {
                 debug!("out_buf overflow");


@@ -122,49 +159,118 @@ impl StreamAdapter for Codec2Adapter {
     }
 }
 
     }
 }
 

-fn output_cb(data: &mut [i16], state: &Mutex<AdapterState>) {
+fn output_cb(data: &mut [i16], state: &Mutex<AdapterState>, channels: u16) {

     let mut state = state.lock().unwrap();
     let mut state = state.lock().unwrap();

-    for d in data {
-        *d = state.out_buf.pop_front().unwrap_or(i16::EQUILIBRIUM);
+    for d in data.chunks_mut(channels as usize) {
+        d.fill(state.out_buf.pop_front().unwrap_or(i16::EQUILIBRIUM));

     }
 }
 
 /// Create and manage the stream from a dedicated thread since it's `!Send`
     }
 }
 
 /// Create and manage the stream from a dedicated thread since it's `!Send`

-fn stream_thread(end: Receiver<()>, state: Arc<Mutex<AdapterState>>, output_card: String) {
+fn stream_thread(
+    end: Receiver<()>,
+    setup_tx: Sender<Result<u32, AdapterError>>,
+    state: Arc<Mutex<AdapterState>>,
+    output_card: Option<String>,
+) {

     let host = cpal::default_host();
     let host = cpal::default_host();

-    let device = host
-        .output_devices()
-        .unwrap()
-        .find(|d| d.name().unwrap() == output_card)
-        .unwrap();
-    let mut configs = device.supported_output_configs().unwrap();
-    // TODO: channels == 1 doesn't work on a Raspberry Pi
-    // make this configurable and support interleaving LRLR stereo samples if using 2 channels
-    let config = configs
-        .find(|c| c.channels() == 1 && c.sample_format() == SampleFormat::I16)
-        .unwrap()
-        .with_sample_rate(SampleRate(8000));
-    let stream = device
-        .build_output_stream(
-            &config.into(),
-            move |data: &mut [i16], _info: &cpal::OutputCallbackInfo| {
-                output_cb(data, &state);
-            },
-            |e| {
-                // trigger end_tx here? always more edge cases
-                debug!("error occurred in codec2 playback: {e:?}");
-            },
-            None,
-        )
-        .unwrap();
-    stream.play().unwrap();
+    let device = if let Some(output_card) = output_card {
+        // TODO: more error handling for unwraps
+        match host
+            .output_devices()
+            .unwrap()
+            .find(|d| d.name().unwrap() == output_card)
+        {
+            Some(d) => d,
+            None => {
+                let _ = setup_tx.send(Err(M17Codec2Error::CardUnavailable(output_card).into()));
+                return;
+            }
+        }
+    } else {
+        match host.default_output_device() {
+            Some(d) => d,
+            None => {
+                let _ = setup_tx.send(Err(M17Codec2Error::DefaultCardUnavailable.into()));
+                return;
+            }
+        }
+    };
+    let card_name = device.name().unwrap();
+    let mut configs = match device.supported_output_configs() {
+        Ok(c) => c,
+        Err(e) => {
+            let _ = setup_tx.send(Err(
+                M17Codec2Error::OutputConfigsUnavailable(card_name, e).into()
+            ));
+            return;
+        }
+    };
+    let config = match configs.find(|c| {
+        (c.channels() == 1 || c.channels() == 2) && c.sample_format() == SampleFormat::I16
+    }) {
+        Some(c) => c,
+        None => {
+            let _ = setup_tx.send(Err(
+                M17Codec2Error::SupportedOutputUnavailable(card_name).into()
+            ));
+            return;
+        }
+    };
+
+    let target_sample_rate =
+        if config.min_sample_rate().0 <= 8000 && config.max_sample_rate().0 >= 8000 {
+            8000
+        } else {
+            config.max_sample_rate().0
+        };
+    let channels = config.channels();
+
+    let config = config.with_sample_rate(SampleRate(target_sample_rate));
+    let stream = match device.build_output_stream(
+        &config.into(),
+        move |data: &mut [i16], _info: &cpal::OutputCallbackInfo| {
+            output_cb(data, &state, channels);
+        },
+        |e| {
+            // trigger end_tx here? always more edge cases
+            debug!("error occurred in codec2 playback: {e:?}");
+        },
+        None,
+    ) {
+        Ok(s) => s,
+        Err(e) => {
+            let _ = setup_tx.send(Err(
+                M17Codec2Error::OutputStreamBuildError(card_name, e).into()
+            ));
+            return;
+        }
+    };
+    match stream.play() {
+        Ok(()) => (),
+        Err(e) => {
+            let _ = setup_tx.send(Err(
+                M17Codec2Error::OutputStreamPlayError(card_name, e).into()
+            ));
+            return;
+        }
+    }
+    let _ = setup_tx.send(Ok(target_sample_rate));

     let _ = end.recv();
     // it seems concrete impls of Stream have a Drop implementation that will handle termination
 }
 
     let _ = end.recv();
     // it seems concrete impls of Stream have a Drop implementation that will handle termination
 }
 

+/// Transmits a wave file as an M17 stream

 pub struct WavePlayer;
 
 impl WavePlayer {
 pub struct WavePlayer;
 
 impl WavePlayer {

+    /// Plays a wave file (blocking).
+    ///
+    /// * `path`: wave file to transmit, must be 8 kHz mono and 16-bit LE
+    /// * `tx`: a `TxHandle` obtained from an `M17App`
+    /// * `source`: address of transmission source
+    /// * `destination`: address of transmission destination
+    /// * `channel_access_number`: from 0 to 15, usually 0

     pub fn play(
         path: PathBuf,
         tx: TxHandle,
     pub fn play(
         path: PathBuf,
         tx: TxHandle,


@@ -221,3 +327,24 @@ impl WavePlayer {
         }
     }
 }
         }
     }
 }

+
+#[derive(Debug, Error)]
+pub enum M17Codec2Error {
+    #[error("selected card '{0}' does not exist or is in use")]
+    CardUnavailable(String),
+
+    #[error("default output card is unavailable")]
+    DefaultCardUnavailable,
+
+    #[error("selected card '{0}' failed to list available output configs: '{1}'")]
+    OutputConfigsUnavailable(String, #[source] cpal::SupportedStreamConfigsError),
+
+    #[error("selected card '{0}' did not offer a compatible output config type, either due to hardware limitations or because it is currently in use")]
+    SupportedOutputUnavailable(String),
+
+    #[error("selected card '{0}' was unable to build an output stream: '{1}'")]
+    OutputStreamBuildError(String, #[source] cpal::BuildStreamError),
+
+    #[error("selected card '{0}' was unable to play an output stream: '{1}'")]
+    OutputStreamPlayError(String, #[source] cpal::PlayStreamError),
+}