code.octet-stream.net Git - m17rt/blob - m17codec2/src/rx.rs

   1 use crate::M17Codec2Error;
   2 use codec2::{Codec2, Codec2Mode};
   3 use cpal::traits::DeviceTrait;
   4 use cpal::traits::HostTrait;
   5 use cpal::traits::StreamTrait;
   6 use cpal::{Sample, SampleFormat, SampleRate};
   7 use log::debug;
   8 use m17app::adapter::StreamAdapter;
   9 use m17app::app::TxHandle;
  10 use m17app::error::AdapterError;
  11 use m17app::link_setup::LinkSetup;
  12 use rubato::Resampler;
  13 use rubato::SincFixedIn;
  14 use rubato::SincInterpolationParameters;
  15 use std::collections::VecDeque;
  16 use std::fs::File;
  17 use std::io::Write;
  18 use std::path::Path;
  19 use std::sync::{
  20     mpsc::{channel, Receiver, Sender},
  21     Arc, Mutex,
  22 };
  23
  24 /// Write one or more 8-byte chunks of 3200-bit Codec2 to a raw S16LE file
  25 /// and return the samples.
  26 pub fn decode_codec2<P: AsRef<Path>>(data: &[u8], out_path: P) -> Vec<i16> {
  27     let codec2 = Codec2::new(Codec2Mode::MODE_3200);
  28     let var_name = codec2;
  29     let mut codec = var_name;
  30     let mut all_samples: Vec<i16> = vec![];
  31     for i in 0..(data.len() / 8) {
  32         let mut samples = vec![0; codec.samples_per_frame()];
  33         codec.decode(&mut samples, &data[i * 8..((i + 1) * 8)]);
  34         all_samples.append(&mut samples);
  35     }
  36     let mut speech_out = File::create(out_path).unwrap();
  37     for b in &all_samples {
  38         speech_out.write_all(&b.to_le_bytes()).unwrap();
  39     }
  40     all_samples
  41 }
  42
  43 /// Subscribes to M17 streams and attempts to play the decoded Codec2
  44 pub struct Codec2RxAdapter {
  45     state: Arc<Mutex<AdapterState>>,
  46     output_card: Option<String>,
  47 }
  48
  49 impl Codec2RxAdapter {
  50     pub fn new() -> Self {
  51         Self {
  52             state: Arc::new(Mutex::new(AdapterState {
  53                 out_buf: VecDeque::new(),
  54                 codec2: Codec2::new(Codec2Mode::MODE_3200),
  55                 end_tx: None,
  56                 resampler: None,
  57             })),
  58             output_card: None,
  59         }
  60     }
  61
  62     pub fn set_output_card<S: Into<String>>(&mut self, card_name: S) {
  63         self.output_card = Some(card_name.into());
  64     }
  65
  66     /// List sound cards supported for audio output.
  67     ///
  68     /// M17RT will handle any card with 1 or 2 channels and 16-bit output.
  69     pub fn supported_output_cards() -> Vec<String> {
  70         let mut out = vec![];
  71         let host = cpal::default_host();
  72         let Ok(output_devices) = host.output_devices() else {
  73             return out;
  74         };
  75         for d in output_devices {
  76             let Ok(mut configs) = d.supported_output_configs() else {
  77                 continue;
  78             };
  79             if configs.any(|config| {
  80                 (config.channels() == 1 || config.channels() == 2)
  81                     && config.sample_format() == SampleFormat::I16
  82             }) {
  83                 let Ok(name) = d.name() else {
  84                     continue;
  85                 };
  86                 out.push(name);
  87             }
  88         }
  89         out.sort();
  90         out
  91     }
  92 }
  93
  94 impl Default for Codec2RxAdapter {
  95     fn default() -> Self {
  96         Self::new()
  97     }
  98 }
  99
 100 struct AdapterState {
 101     /// Circular buffer of output samples for playback
 102     out_buf: VecDeque<i16>,
 103     codec2: Codec2,
 104     end_tx: Option<Sender<()>>,
 105     resampler: Option<SincFixedIn<f32>>,
 106 }
 107
 108 impl StreamAdapter for Codec2RxAdapter {
 109     fn start(&self, _handle: TxHandle) -> Result<(), AdapterError> {
 110         let (end_tx, end_rx) = channel();
 111         let (setup_tx, setup_rx) = channel();
 112         let state = self.state.clone();
 113         let output_card = self.output_card.clone();
 114         std::thread::spawn(move || stream_thread(end_rx, setup_tx, state, output_card));
 115         self.state.lock().unwrap().end_tx = Some(end_tx);
 116         // Propagate any errors arising in the thread
 117         let sample_rate = setup_rx.recv()??;
 118         debug!("selected codec2 speaker sample rate {sample_rate}");
 119         if sample_rate != 8000 {
 120             let params = SincInterpolationParameters {
 121                 sinc_len: 256,
 122                 f_cutoff: 0.95,
 123                 oversampling_factor: 128,
 124                 interpolation: rubato::SincInterpolationType::Cubic,
 125                 window: rubato::WindowFunction::BlackmanHarris2,
 126             };
 127             // TODO: fix unwrap
 128             self.state.lock().unwrap().resampler =
 129                 Some(SincFixedIn::new(sample_rate as f64 / 8000f64, 1.0, params, 160, 1).unwrap());
 130         }
 131         Ok(())
 132     }
 133
 134     fn close(&self) -> Result<(), AdapterError> {
 135         let mut state = self.state.lock().unwrap();
 136         state.end_tx = None;
 137         Ok(())
 138     }
 139
 140     fn stream_began(&self, _link_setup: LinkSetup) {
 141         // for now we will assume:
 142         // - unencrypted
 143         // - data type is Voice (Codec2 3200), not Voice+Data
 144         // TODO: is encryption handled here or in M17App, such that we get a decrypted stream?
 145         // TODO: handle the Voice+Data combination with Codec2 1600
 146         self.state.lock().unwrap().codec2 = Codec2::new(Codec2Mode::MODE_3200);
 147     }
 148
 149     fn stream_data(&self, _frame_number: u16, _is_final: bool, data: Arc<[u8; 16]>) {
 150         let mut state = self.state.lock().unwrap();
 151         for encoded in data.chunks(8) {
 152             if state.out_buf.len() < 8192 {
 153                 let mut samples = [i16::EQUILIBRIUM; 160]; // while assuming 3200
 154                 state.codec2.decode(&mut samples, encoded);
 155                 if let Some(resampler) = state.resampler.as_mut() {
 156                     let samples_f: Vec<f32> =
 157                         samples.iter().map(|s| *s as f32 / 16384.0f32).collect();
 158                     let res = resampler.process(&[samples_f], None).unwrap();
 159                     for s in &res[0] {
 160                         state.out_buf.push_back((s * 16383.0f32) as i16);
 161                     }
 162                 } else {
 163                     // TODO: maybe get rid of VecDeque so we can decode directly into ring buffer?
 164                     for s in samples {
 165                         state.out_buf.push_back(s);
 166                     }
 167                 }
 168             } else {
 169                 debug!("out_buf overflow");
 170             }
 171         }
 172     }
 173 }
 174
 175 fn output_cb(data: &mut [i16], state: &Mutex<AdapterState>, channels: u16) {
 176     let mut state = state.lock().unwrap();
 177     for d in data.chunks_mut(channels as usize) {
 178         d.fill(state.out_buf.pop_front().unwrap_or(i16::EQUILIBRIUM));
 179     }
 180 }
 181
 182 /// Create and manage the stream from a dedicated thread since it's `!Send`
 183 fn stream_thread(
 184     end: Receiver<()>,
 185     setup_tx: Sender<Result<u32, AdapterError>>,
 186     state: Arc<Mutex<AdapterState>>,
 187     output_card: Option<String>,
 188 ) {
 189     let host = cpal::default_host();
 190     let device = if let Some(output_card) = output_card {
 191         // TODO: more error handling for unwraps
 192         match host
 193             .output_devices()
 194             .unwrap()
 195             .find(|d| d.name().unwrap() == output_card)
 196         {
 197             Some(d) => d,
 198             None => {
 199                 let _ = setup_tx.send(Err(M17Codec2Error::CardUnavailable(output_card).into()));
 200                 return;
 201             }
 202         }
 203     } else {
 204         match host.default_output_device() {
 205             Some(d) => d,
 206             None => {
 207                 let _ = setup_tx.send(Err(M17Codec2Error::DefaultCardUnavailable.into()));
 208                 return;
 209             }
 210         }
 211     };
 212     let card_name = device.name().unwrap();
 213     let mut configs = match device.supported_output_configs() {
 214         Ok(c) => c,
 215         Err(e) => {
 216             let _ = setup_tx.send(Err(
 217                 M17Codec2Error::OutputConfigsUnavailable(card_name, e).into()
 218             ));
 219             return;
 220         }
 221     };
 222     let config = match configs.find(|c| {
 223         (c.channels() == 1 || c.channels() == 2) && c.sample_format() == SampleFormat::I16
 224     }) {
 225         Some(c) => c,
 226         None => {
 227             let _ = setup_tx.send(Err(
 228                 M17Codec2Error::SupportedOutputUnavailable(card_name).into()
 229             ));
 230             return;
 231         }
 232     };
 233
 234     let target_sample_rate =
 235         if config.min_sample_rate().0 <= 8000 && config.max_sample_rate().0 >= 8000 {
 236             8000
 237         } else {
 238             config.min_sample_rate().0
 239         };
 240     let channels = config.channels();
 241
 242     let config = config.with_sample_rate(SampleRate(target_sample_rate));
 243     let stream = match device.build_output_stream(
 244         &config.into(),
 245         move |data: &mut [i16], _info: &cpal::OutputCallbackInfo| {
 246             output_cb(data, &state, channels);
 247         },
 248         |e| {
 249             // trigger end_tx here? always more edge cases
 250             debug!("error occurred in codec2 playback: {e:?}");
 251         },
 252         None,
 253     ) {
 254         Ok(s) => s,
 255         Err(e) => {
 256             let _ = setup_tx.send(Err(
 257                 M17Codec2Error::OutputStreamBuildError(card_name, e).into()
 258             ));
 259             return;
 260         }
 261     };
 262     match stream.play() {
 263         Ok(()) => (),
 264         Err(e) => {
 265             let _ = setup_tx.send(Err(
 266                 M17Codec2Error::OutputStreamPlayError(card_name, e).into()
 267             ));
 268             return;
 269         }
 270     }
 271     let _ = setup_tx.send(Ok(target_sample_rate));
 272     let _ = end.recv();
 273     // it seems concrete impls of Stream have a Drop implementation that will handle termination
 274 }