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

   1 use codec2::{Codec2, Codec2Mode};
   2 use cpal::traits::DeviceTrait;
   3 use cpal::traits::HostTrait;
   4 use cpal::traits::StreamTrait;
   5 use cpal::SampleFormat;
   6 use cpal::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 m17app::link_setup::M17Address;
  13 use m17app::StreamFrame;
  14 use rubato::Resampler;
  15 use rubato::SincFixedOut;
  16 use rubato::SincInterpolationParameters;
  17 use std::path::PathBuf;
  18 use std::sync::mpsc;
  19 use std::sync::mpsc::channel;
  20 use std::sync::Arc;
  21 use std::sync::Mutex;
  22 use std::time::Duration;
  23 use std::time::Instant;
  24
  25 use crate::M17Codec2Error;
  26
  27 /// Transmits a wave file as an M17 stream
  28 pub struct WavePlayer;
  29
  30 impl WavePlayer {
  31     /// Plays a wave file (blocking).
  32     ///
  33     /// * `path`: wave file to transmit, must be 8 kHz mono and 16-bit LE
  34     /// * `tx`: a `TxHandle` obtained from an `M17App`
  35     /// * `source`: address of transmission source
  36     /// * `destination`: address of transmission destination
  37     /// * `channel_access_number`: from 0 to 15, usually 0
  38     pub fn play(
  39         path: PathBuf,
  40         tx: TxHandle,
  41         source: &M17Address,
  42         destination: &M17Address,
  43         channel_access_number: u8,
  44     ) {
  45         let mut reader = hound::WavReader::open(path).unwrap();
  46         let mut samples = reader.samples::<i16>();
  47
  48         let mut codec = Codec2::new(Codec2Mode::MODE_3200);
  49         let mut in_buf = [0i16; 160];
  50         let mut out_buf = [0u8; 16];
  51         let mut lsf_chunk: usize = 0;
  52         const TICK: Duration = Duration::from_millis(40);
  53         let mut next_tick = Instant::now() + TICK;
  54         let mut frame_number = 0;
  55
  56         let mut setup = LinkSetup::new_voice(source, destination);
  57         setup.set_channel_access_number(channel_access_number);
  58         tx.transmit_stream_start(&setup);
  59
  60         loop {
  61             let mut last_one = false;
  62             for out in out_buf.chunks_mut(8) {
  63                 for i in in_buf.iter_mut() {
  64                     let sample = match samples.next() {
  65                         Some(Ok(sample)) => sample,
  66                         _ => {
  67                             last_one = true;
  68                             0
  69                         }
  70                     };
  71                     *i = sample;
  72                 }
  73                 codec.encode(out, &in_buf);
  74             }
  75             tx.transmit_stream_next(&StreamFrame {
  76                 lich_idx: lsf_chunk as u8,
  77                 lich_part: setup.lich_part(lsf_chunk as u8),
  78                 frame_number,
  79                 end_of_stream: last_one,
  80                 stream_data: out_buf,
  81             });
  82             frame_number += 1;
  83             lsf_chunk = (lsf_chunk + 1) % 6;
  84
  85             if last_one {
  86                 break;
  87             }
  88
  89             std::thread::sleep(next_tick.duration_since(Instant::now()));
  90             next_tick += TICK;
  91         }
  92     }
  93 }
  94
  95 /// Control transmissions into a Codec2TxAdapter
  96 #[derive(Clone)]
  97 pub struct Ptt {
  98     tx: mpsc::Sender<Event>,
  99 }
 100
 101 impl Ptt {
 102     pub fn set_ptt(&self, ptt: bool) {
 103         let _ = self.tx.send(if ptt { Event::PttOn } else { Event::PttOff });
 104     }
 105 }
 106
 107 /// Use a microphone and local PTT to transmit Codec2 voice data into an M17 channel.
 108 pub struct Codec2TxAdapter {
 109     input_card: Option<String>,
 110     event_tx: mpsc::Sender<Event>,
 111     event_rx: Mutex<Option<mpsc::Receiver<Event>>>,
 112     source: M17Address,
 113     destination: M17Address,
 114 }
 115
 116 impl Codec2TxAdapter {
 117     pub fn new(source: M17Address, destination: M17Address) -> Self {
 118         let (event_tx, event_rx) = mpsc::channel();
 119         Self {
 120             input_card: None,
 121             event_tx,
 122             event_rx: Mutex::new(Some(event_rx)),
 123             source,
 124             destination,
 125         }
 126     }
 127
 128     pub fn set_input_card<S: Into<String>>(&mut self, card_name: S) {
 129         self.input_card = Some(card_name.into());
 130     }
 131
 132     pub fn ptt(&self) -> Ptt {
 133         Ptt {
 134             tx: self.event_tx.clone(),
 135         }
 136     }
 137
 138     /// List sound cards supported for audio input.
 139     ///
 140     /// M17RT will handle any card with 1 or 2 channels and 16-bit output.
 141     pub fn supported_input_cards() -> Vec<String> {
 142         let mut out = vec![];
 143         let host = cpal::default_host();
 144         let Ok(input_devices) = host.input_devices() else {
 145             return out;
 146         };
 147         for d in input_devices {
 148             let Ok(mut configs) = d.supported_input_configs() else {
 149                 continue;
 150             };
 151             if configs.any(|config| {
 152                 (config.channels() == 1 || config.channels() == 2)
 153                     && config.sample_format() == SampleFormat::I16
 154             }) {
 155                 let Ok(name) = d.name() else {
 156                     continue;
 157                 };
 158                 out.push(name);
 159             }
 160         }
 161         out.sort();
 162         out
 163     }
 164 }
 165
 166 enum Event {
 167     PttOn,
 168     PttOff,
 169     MicSamples(Arc<[i16]>),
 170     Close,
 171 }
 172
 173 impl StreamAdapter for Codec2TxAdapter {
 174     fn start(&self, handle: TxHandle) -> Result<(), AdapterError> {
 175         let Some(event_rx) = self.event_rx.lock().unwrap().take() else {
 176             return Err(M17Codec2Error::RepeatStart.into());
 177         };
 178         let event_tx = self.event_tx.clone();
 179         let (setup_tx, setup_rx) = channel();
 180         let input_card = self.input_card.clone();
 181         let from = self.source.clone();
 182         let to = self.destination.clone();
 183         std::thread::spawn(move || {
 184             stream_thread(event_tx, event_rx, setup_tx, input_card, handle, from, to)
 185         });
 186         let sample_rate = setup_rx.recv()??;
 187         debug!("selected codec2 microphone sample rate {sample_rate}");
 188
 189         Ok(())
 190     }
 191
 192     fn close(&self) -> Result<(), AdapterError> {
 193         let _ = self.event_tx.send(Event::Close);
 194         Ok(())
 195     }
 196
 197     fn stream_began(&self, _link_setup: LinkSetup) {
 198         // not interested in incoming transmissions
 199     }
 200
 201     fn stream_data(&self, _frame_number: u16, _is_final: bool, _data: Arc<[u8; 16]>) {
 202         // not interested in incoming transmissions
 203
 204         // the only reason this is an adapter at all is for future "transmission aborted" feedback
 205         // when that's implemented by m17app
 206     }
 207 }
 208
 209 fn stream_thread(
 210     event_tx: mpsc::Sender<Event>,
 211     event_rx: mpsc::Receiver<Event>,
 212     setup_tx: mpsc::Sender<Result<u32, AdapterError>>,
 213     input_card: Option<String>,
 214     handle: TxHandle,
 215     source: M17Address,
 216     destination: M17Address,
 217 ) {
 218     let host = cpal::default_host();
 219     let device = if let Some(input_card) = input_card {
 220         // TODO: more error handling for unwraps
 221         match host
 222             .input_devices()
 223             .unwrap()
 224             .find(|d| d.name().unwrap() == input_card)
 225         {
 226             Some(d) => d,
 227             None => {
 228                 let _ = setup_tx.send(Err(M17Codec2Error::CardUnavailable(input_card).into()));
 229                 return;
 230             }
 231         }
 232     } else {
 233         match host.default_input_device() {
 234             Some(d) => d,
 235             None => {
 236                 let _ = setup_tx.send(Err(M17Codec2Error::DefaultCardUnavailable.into()));
 237                 return;
 238             }
 239         }
 240     };
 241     let card_name = device.name().unwrap();
 242     let mut configs = match device.supported_input_configs() {
 243         Ok(c) => c,
 244         Err(e) => {
 245             let _ = setup_tx.send(Err(
 246                 M17Codec2Error::InputConfigsUnavailable(card_name, e).into()
 247             ));
 248             return;
 249         }
 250     };
 251     // TODO: rank these by most favourable, same for rx
 252     let config = match configs.find(|c| {
 253         (c.channels() == 1 || c.channels() == 2) && c.sample_format() == SampleFormat::I16
 254     }) {
 255         Some(c) => c,
 256         None => {
 257             let _ = setup_tx.send(Err(
 258                 M17Codec2Error::SupportedInputUnavailable(card_name).into()
 259             ));
 260             return;
 261         }
 262     };
 263
 264     let target_sample_rate =
 265         if config.min_sample_rate().0 <= 8000 && config.max_sample_rate().0 >= 8000 {
 266             8000
 267         } else {
 268             config.min_sample_rate().0
 269         };
 270     let channels = config.channels();
 271
 272     let mut acc: Box<dyn Accumulator> = if target_sample_rate != 8000 {
 273         Box::new(ResamplingAccumulator::new(target_sample_rate as f64))
 274     } else {
 275         Box::new(DirectAccumulator::new())
 276     };
 277
 278     let config = config.with_sample_rate(SampleRate(target_sample_rate));
 279     let stream = match device.build_input_stream(
 280         &config.into(),
 281         move |data: &[i16], _info: &cpal::InputCallbackInfo| {
 282             let mut samples = vec![];
 283             for d in data.chunks(channels as usize) {
 284                 // if we were given multi-channel input we'll pick the first channel
 285                 // TODO: configurable?
 286                 samples.push(d[0]);
 287             }
 288             let _ = event_tx.send(Event::MicSamples(samples.into()));
 289         },
 290         |e| {
 291             // abort here?
 292             debug!("error occurred in codec2 recording: {e:?}");
 293         },
 294         None,
 295     ) {
 296         Ok(s) => s,
 297         Err(e) => {
 298             let _ = setup_tx.send(Err(
 299                 M17Codec2Error::InputStreamBuildError(card_name, e).into()
 300             ));
 301             return;
 302         }
 303     };
 304
 305     let _ = setup_tx.send(Ok(target_sample_rate));
 306     let mut state = State::Idle;
 307     let mut codec2 = Codec2::new(Codec2Mode::MODE_3200);
 308     let link_setup = LinkSetup::new_voice(&source, &destination);
 309     let mut lich_idx = 0;
 310     let mut frame_number = 0;
 311
 312     // Now the main loop
 313     while let Ok(ev) = event_rx.recv() {
 314         match ev {
 315             Event::PttOn => {
 316                 match state {
 317                     State::Idle => {
 318                         match stream.play() {
 319                             Ok(()) => (),
 320                             Err(_e) => {
 321                                 // TODO: report M17Codec2Error::InputStreamPlayError(card_name, e).into()
 322                                 break;
 323                             }
 324                         }
 325                         acc.reset();
 326                         codec2 = Codec2::new(Codec2Mode::MODE_3200);
 327                         state = State::StartTransmitting;
 328                     }
 329                     State::StartTransmitting => {}
 330                     State::Transmitting => {}
 331                     State::Ending => state = State::EndingWithPttRestart,
 332                     State::EndingWithPttRestart => {}
 333                 }
 334             }
 335             Event::PttOff => match state {
 336                 State::Idle => {}
 337                 State::StartTransmitting => state = State::Idle,
 338                 State::Transmitting => state = State::Ending,
 339                 State::Ending => {}
 340                 State::EndingWithPttRestart => state = State::Ending,
 341             },
 342             Event::MicSamples(samples) => {
 343                 match state {
 344                     State::Idle => {}
 345                     State::StartTransmitting
 346                     | State::Transmitting
 347                     | State::Ending
 348                     | State::EndingWithPttRestart => {
 349                         acc.handle_samples(&samples);
 350                         while let Some(frame) = acc.try_next_frame() {
 351                             let mut stream_data = [0u8; 16];
 352                             codec2.encode(&mut stream_data[0..8], &frame[0..160]);
 353                             codec2.encode(&mut stream_data[8..16], &frame[160..320]);
 354
 355                             if state == State::StartTransmitting {
 356                                 handle.transmit_stream_start(&link_setup);
 357                                 lich_idx = 0;
 358                                 frame_number = 0;
 359                                 state = State::Transmitting;
 360                             }
 361
 362                             let end_of_stream = state != State::Transmitting;
 363                             handle.transmit_stream_next(&StreamFrame {
 364                                 lich_idx,
 365                                 lich_part: link_setup.lich_part(lich_idx),
 366                                 frame_number,
 367                                 end_of_stream,
 368                                 stream_data,
 369                             });
 370                             frame_number += 1;
 371                             lich_idx = (lich_idx + 1) % 6;
 372
 373                             if end_of_stream {
 374                                 break;
 375                             }
 376                         }
 377
 378                         if state == State::Ending {
 379                             // when finished sending final stream frame
 380                             let _ = stream.pause();
 381                             state = State::Idle;
 382                         }
 383
 384                         if state == State::EndingWithPttRestart {
 385                             acc.reset();
 386                             codec2 = Codec2::new(Codec2Mode::MODE_3200);
 387                             state = State::StartTransmitting;
 388                         }
 389                     }
 390                 }
 391             }
 392             Event::Close => {
 393                 // assume PTT etc. will clean up itself responsibly on close
 394                 break;
 395             }
 396         }
 397     }
 398 }
 399
 400 #[derive(Debug, PartialEq, Eq)]
 401 enum State {
 402     /// Waiting for PTT
 403     Idle,
 404     /// PTT engaged but we are collecting the first full frame of audio data before starting the stream
 405     StartTransmitting,
 406     /// Streaming voice frames
 407     Transmitting,
 408     /// PTT disengaged; we are collecting the next frame of audio to use as a final frame
 409     Ending,
 410     /// PTT was re-enaged while ending; we will send final frame then go back to StartTransmitting
 411     EndingWithPttRestart,
 412 }
 413
 414 fn resampler_params() -> SincInterpolationParameters {
 415     SincInterpolationParameters {
 416         sinc_len: 256,
 417         f_cutoff: 0.95,
 418         oversampling_factor: 128,
 419         interpolation: rubato::SincInterpolationType::Cubic,
 420         window: rubato::WindowFunction::BlackmanHarris2,
 421     }
 422 }
 423
 424 trait Accumulator {
 425     fn handle_samples(&mut self, samples: &[i16]);
 426     /// Return 320 samples, enough for two Codec2 frames
 427     fn try_next_frame(&mut self) -> Option<Vec<i16>>;
 428     fn reset(&mut self);
 429 }
 430
 431 struct DirectAccumulator {
 432     buffer: Vec<i16>,
 433 }
 434
 435 impl DirectAccumulator {
 436     fn new() -> Self {
 437         Self { buffer: Vec::new() }
 438     }
 439 }
 440
 441 impl Accumulator for DirectAccumulator {
 442     fn handle_samples(&mut self, samples: &[i16]) {
 443         self.buffer.extend_from_slice(samples);
 444     }
 445
 446     fn try_next_frame(&mut self) -> Option<Vec<i16>> {
 447         if self.buffer.len() >= 320 {
 448             let part = self.buffer.split_off(320);
 449             Some(std::mem::replace(&mut self.buffer, part))
 450         } else {
 451             None
 452         }
 453     }
 454
 455     fn reset(&mut self) {
 456         self.buffer.clear();
 457     }
 458 }
 459
 460 struct ResamplingAccumulator {
 461     input_rate: f64,
 462     buffer: Vec<i16>,
 463     resampler: SincFixedOut<f32>,
 464 }
 465
 466 impl ResamplingAccumulator {
 467     fn new(input_rate: f64) -> Self {
 468         Self {
 469             input_rate,
 470             buffer: Vec::new(),
 471             resampler: make_resampler(input_rate),
 472         }
 473     }
 474 }
 475
 476 impl Accumulator for ResamplingAccumulator {
 477     fn handle_samples(&mut self, samples: &[i16]) {
 478         self.buffer.extend_from_slice(samples);
 479     }
 480
 481     fn try_next_frame(&mut self) -> Option<Vec<i16>> {
 482         let required = self.resampler.input_frames_next();
 483         if self.buffer.len() >= required {
 484             let mut part = self.buffer.split_off(required);
 485             std::mem::swap(&mut self.buffer, &mut part);
 486             let samples_f: Vec<f32> = part.iter().map(|s| *s as f32 / 16384.0f32).collect();
 487             let out = self.resampler.process(&[samples_f], None).unwrap();
 488             Some(out[0].iter().map(|s| (*s * 16383.0f32) as i16).collect())
 489         } else {
 490             None
 491         }
 492     }
 493
 494     fn reset(&mut self) {
 495         self.buffer.clear();
 496         self.resampler = make_resampler(self.input_rate);
 497     }
 498 }
 499
 500 fn make_resampler(input_rate: f64) -> SincFixedOut<f32> {
 501     // want 320 samples at a time to create 2x Codec2 frames per M17 Voice frame
 502     SincFixedOut::new(8000f64 / input_rate, 1.0, resampler_params(), 320, 1).unwrap()
 503 }