code.octet-stream.net Git - m17rt/blob - m17app/src/soundmodem.rs

   1 use crate::tnc::{Tnc, TncError};
   2 use cpal::traits::DeviceTrait;
   3 use cpal::traits::HostTrait;
   4 use cpal::traits::StreamTrait;
   5 use cpal::{SampleFormat, SampleRate};
   6 use log::debug;
   7 use m17core::kiss::MAX_FRAME_LEN;
   8 use m17core::modem::{Demodulator, Modulator, ModulatorAction, SoftDemodulator, SoftModulator};
   9 use m17core::tnc::SoftTnc;
  10 use std::collections::VecDeque;
  11 use std::fs::File;
  12 use std::io::{self, ErrorKind, Read, Write};
  13 use std::path::PathBuf;
  14 use std::sync::mpsc::{channel, sync_channel, Receiver, Sender, SyncSender, TryRecvError};
  15 use std::sync::RwLock;
  16 use std::sync::{Arc, Mutex};
  17 use std::time::{Duration, Instant};
  18
  19 pub struct Soundmodem {
  20     event_tx: SyncSender<SoundmodemEvent>,
  21     kiss_out_rx: Arc<Mutex<Receiver<Arc<[u8]>>>>,
  22     partial_kiss_out: Arc<Mutex<Option<PartialKissOut>>>,
  23 }
  24
  25 impl Soundmodem {
  26     pub fn new<I: InputSource, O: OutputSink, P: Ptt>(input: I, output: O, ptt: P) -> Self {
  27         // must create TNC here
  28         let (event_tx, event_rx) = sync_channel(128);
  29         let (kiss_out_tx, kiss_out_rx) = sync_channel(128);
  30         spawn_soundmodem_worker(
  31             event_tx.clone(),
  32             event_rx,
  33             kiss_out_tx,
  34             Box::new(input),
  35             Box::new(output),
  36             Box::new(ptt),
  37         );
  38         Self {
  39             event_tx,
  40             kiss_out_rx: Arc::new(Mutex::new(kiss_out_rx)),
  41             partial_kiss_out: Arc::new(Mutex::new(None)),
  42         }
  43     }
  44 }
  45
  46 struct PartialKissOut {
  47     output: Arc<[u8]>,
  48     idx: usize,
  49 }
  50
  51 impl Read for Soundmodem {
  52     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
  53         {
  54             let mut partial_kiss_out = self.partial_kiss_out.lock().unwrap();
  55             if let Some(partial) = partial_kiss_out.as_mut() {
  56                 let remaining = partial.output.len() - partial.idx;
  57                 let to_write = remaining.min(buf.len());
  58                 buf[0..to_write]
  59                     .copy_from_slice(&partial.output[partial.idx..(partial.idx + to_write)]);
  60                 if to_write == remaining {
  61                     *partial_kiss_out = None;
  62                 } else {
  63                     partial.idx += to_write;
  64                 }
  65                 return Ok(to_write);
  66             }
  67         }
  68         let output = {
  69             let rx = self.kiss_out_rx.lock().unwrap();
  70             rx.recv()
  71                 .map_err(|s| io::Error::new(ErrorKind::Other, format!("{:?}", s)))?
  72         };
  73         let to_write = output.len().min(buf.len());
  74         buf[0..to_write].copy_from_slice(&output[0..to_write]);
  75         if to_write != output.len() {
  76             *self.partial_kiss_out.lock().unwrap() = Some(PartialKissOut {
  77                 output,
  78                 idx: to_write,
  79             })
  80         }
  81         Ok(to_write)
  82     }
  83 }
  84
  85 impl Write for Soundmodem {
  86     fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
  87         let _ = self.event_tx.try_send(SoundmodemEvent::Kiss(buf.into()));
  88         Ok(buf.len())
  89     }
  90
  91     fn flush(&mut self) -> std::io::Result<()> {
  92         Ok(())
  93     }
  94 }
  95
  96 impl Tnc for Soundmodem {
  97     fn try_clone(&mut self) -> Result<Self, TncError> {
  98         Ok(Self {
  99             event_tx: self.event_tx.clone(),
 100             kiss_out_rx: self.kiss_out_rx.clone(),
 101             partial_kiss_out: self.partial_kiss_out.clone(),
 102         })
 103     }
 104
 105     fn start(&mut self) -> Result<(), TncError> {
 106         let _ = self.event_tx.send(SoundmodemEvent::Start);
 107         Ok(())
 108     }
 109
 110     fn close(&mut self) -> Result<(), TncError> {
 111         let _ = self.event_tx.send(SoundmodemEvent::Close);
 112         Ok(())
 113     }
 114 }
 115
 116 pub enum SoundmodemEvent {
 117     Kiss(Arc<[u8]>),
 118     BasebandInput(Arc<[i16]>),
 119     Start,
 120     Close,
 121     DidReadFromOutputBuffer { len: usize, timestamp: Instant },
 122     OutputUnderrun,
 123 }
 124
 125 fn spawn_soundmodem_worker(
 126     event_tx: SyncSender<SoundmodemEvent>,
 127     event_rx: Receiver<SoundmodemEvent>,
 128     kiss_out_tx: SyncSender<Arc<[u8]>>,
 129     input: Box<dyn InputSource>,
 130     output: Box<dyn OutputSink>,
 131     mut ptt_driver: Box<dyn Ptt>,
 132 ) {
 133     std::thread::spawn(move || {
 134         // TODO: should be able to provide a custom Demodulator for a soundmodem
 135         let mut demodulator = SoftDemodulator::new();
 136         let mut modulator = SoftModulator::new();
 137         let mut tnc = SoftTnc::new();
 138         let mut buf = [0u8; MAX_FRAME_LEN];
 139         let out_buffer = Arc::new(RwLock::new(OutputBuffer::new()));
 140         let mut out_samples = [0i16; 1024];
 141         let start = Instant::now();
 142         let mut ptt = false;
 143         while let Ok(ev) = event_rx.recv() {
 144             // Update clock on TNC before we do anything
 145             let sample_time = (start.elapsed().as_nanos() / 48000) as u64;
 146             tnc.set_now(sample_time);
 147
 148             // Handle event
 149             match ev {
 150                 SoundmodemEvent::Kiss(k) => {
 151                     let _n = tnc.write_kiss(&k);
 152                     // TODO: what does it mean if we fail to write it all?
 153                     // Probably we have to read frames for tx first - revisit this during tx
 154                 }
 155                 SoundmodemEvent::BasebandInput(b) => {
 156                     for sample in &*b {
 157                         if let Some(frame) = demodulator.demod(*sample) {
 158                             tnc.handle_frame(frame);
 159                             loop {
 160                                 let n = tnc.read_kiss(&mut buf);
 161                                 if n > 0 {
 162                                     let _ = kiss_out_tx.try_send(buf[0..n].into());
 163                                 } else {
 164                                     break;
 165                                 }
 166                             }
 167                         }
 168                     }
 169                     tnc.set_data_carrier_detect(demodulator.data_carrier_detect());
 170                 }
 171                 SoundmodemEvent::Start => {
 172                     input.start(event_tx.clone());
 173                     output.start(event_tx.clone(), out_buffer.clone());
 174                 }
 175                 SoundmodemEvent::Close => {
 176                     ptt_driver.ptt_off();
 177                     break;
 178                 }
 179                 SoundmodemEvent::DidReadFromOutputBuffer { len, timestamp } => {
 180                     let (occupied, internal_latency) = {
 181                         let out_buffer = out_buffer.read().unwrap();
 182                         (out_buffer.samples.len(), out_buffer.latency)
 183                     };
 184                     let internal_latency = (internal_latency.as_secs_f32() * 48000.0) as usize;
 185                     let dynamic_latency =
 186                         len.saturating_sub((timestamp.elapsed().as_secs_f32() * 48000.0) as usize);
 187                     modulator.update_output_buffer(
 188                         occupied,
 189                         48000,
 190                         internal_latency + dynamic_latency,
 191                     );
 192                 }
 193                 SoundmodemEvent::OutputUnderrun => {
 194                     // TODO: cancel transmission, send empty data frame to host
 195                 }
 196             }
 197
 198             // Update PTT state
 199             let new_ptt = tnc.ptt();
 200             if new_ptt != ptt {
 201                 if new_ptt {
 202                     ptt_driver.ptt_on();
 203                 } else {
 204                     ptt_driver.ptt_off();
 205                 }
 206             }
 207             ptt = new_ptt;
 208
 209             // Let the modulator do what it wants
 210             while let Some(action) = modulator.run() {
 211                 match action {
 212                     ModulatorAction::SetIdle(idling) => {
 213                         out_buffer.write().unwrap().idling = idling;
 214                     }
 215                     ModulatorAction::GetNextFrame => {
 216                         modulator.provide_next_frame(tnc.read_tx_frame());
 217                     }
 218                     ModulatorAction::ReadOutput => loop {
 219                         let n = modulator.read_output_samples(&mut out_samples);
 220                         if n == 0 {
 221                             break;
 222                         }
 223                         let mut out_buffer = out_buffer.write().unwrap();
 224                         for s in &out_samples[0..n] {
 225                             out_buffer.samples.push_back(*s);
 226                         }
 227                     },
 228                     ModulatorAction::TransmissionWillEnd(in_samples) => {
 229                         tnc.set_tx_end_time(in_samples);
 230                     }
 231                 }
 232             }
 233         }
 234     });
 235 }
 236
 237 pub trait InputSource: Send + Sync + 'static {
 238     fn start(&self, samples: SyncSender<SoundmodemEvent>);
 239     fn close(&self);
 240 }
 241
 242 pub struct InputSoundcard {
 243     // TODO: allow for inversion both here and in output
 244     cpal_name: Option<String>,
 245     end_tx: Mutex<Option<Sender<()>>>,
 246 }
 247
 248 impl InputSoundcard {
 249     pub fn new() -> Self {
 250         Self {
 251             cpal_name: None,
 252             end_tx: Mutex::new(None),
 253         }
 254     }
 255
 256     pub fn new_with_card(card_name: String) -> Self {
 257         Self {
 258             cpal_name: Some(card_name),
 259             end_tx: Mutex::new(None),
 260         }
 261     }
 262 }
 263
 264 impl InputSource for InputSoundcard {
 265     fn start(&self, samples: SyncSender<SoundmodemEvent>) {
 266         let (end_tx, end_rx) = channel();
 267         let cpal_name = self.cpal_name.clone();
 268         std::thread::spawn(move || {
 269             let host = cpal::default_host();
 270             let device = if let Some(name) = cpal_name.as_deref() {
 271                 host.input_devices()
 272                     .unwrap()
 273                     .find(|d| d.name().unwrap() == name)
 274                     .unwrap()
 275             } else {
 276                 host.default_input_device().unwrap()
 277             };
 278             let mut configs = device.supported_input_configs().unwrap();
 279             let config = configs
 280                 .find(|c| c.channels() == 1 && c.sample_format() == SampleFormat::I16)
 281                 .unwrap()
 282                 .with_sample_rate(SampleRate(48000));
 283             let stream = device
 284                 .build_input_stream(
 285                     &config.into(),
 286                     move |data: &[i16], _info: &cpal::InputCallbackInfo| {
 287                         let out: Vec<i16> = data.iter().map(|s| *s).collect();
 288                         let _ = samples.try_send(SoundmodemEvent::BasebandInput(out.into()));
 289                     },
 290                     |e| {
 291                         // TODO: abort?
 292                         debug!("error occurred in soundcard input: {e:?}");
 293                     },
 294                     None,
 295                 )
 296                 .unwrap();
 297             stream.play().unwrap();
 298             let _ = end_rx.recv();
 299         });
 300         *self.end_tx.lock().unwrap() = Some(end_tx);
 301     }
 302
 303     fn close(&self) {
 304         let _ = self.end_tx.lock().unwrap().take();
 305     }
 306 }
 307
 308 pub struct InputRrcFile {
 309     path: PathBuf,
 310     end_tx: Mutex<Option<Sender<()>>>,
 311 }
 312
 313 impl InputRrcFile {
 314     pub fn new(path: PathBuf) -> Self {
 315         Self {
 316             path,
 317             end_tx: Mutex::new(None),
 318         }
 319     }
 320 }
 321
 322 impl InputSource for InputRrcFile {
 323     fn start(&self, samples: SyncSender<SoundmodemEvent>) {
 324         let (end_tx, end_rx) = channel();
 325         let path = self.path.clone();
 326         std::thread::spawn(move || {
 327             // TODO: error handling
 328             let mut file = File::open(path).unwrap();
 329             let mut baseband = vec![];
 330             file.read_to_end(&mut baseband).unwrap();
 331
 332             // assuming 48 kHz for now
 333             const TICK: Duration = Duration::from_millis(25);
 334             const SAMPLES_PER_TICK: usize = 1200;
 335
 336             let mut next_tick = Instant::now() + TICK;
 337             let mut buf = [0i16; SAMPLES_PER_TICK];
 338             let mut idx = 0;
 339
 340             for sample in baseband
 341                 .chunks(2)
 342                 .map(|pair| i16::from_le_bytes([pair[0], pair[1]]))
 343             {
 344                 buf[idx] = sample;
 345                 idx += 1;
 346                 if idx == SAMPLES_PER_TICK {
 347                     if let Err(e) = samples.try_send(SoundmodemEvent::BasebandInput(buf.into())) {
 348                         debug!("overflow feeding soundmodem: {e:?}");
 349                     }
 350                     next_tick = next_tick + TICK;
 351                     idx = 0;
 352                     std::thread::sleep(next_tick.duration_since(Instant::now()));
 353                 }
 354                 if end_rx.try_recv() != Err(TryRecvError::Empty) {
 355                     break;
 356                 }
 357             }
 358         });
 359         *self.end_tx.lock().unwrap() = Some(end_tx);
 360     }
 361
 362     fn close(&self) {
 363         let _ = self.end_tx.lock().unwrap().take();
 364     }
 365 }
 366
 367 pub struct NullInputSource {
 368     end_tx: Mutex<Option<Sender<()>>>,
 369 }
 370
 371 impl NullInputSource {
 372     pub fn new() -> Self {
 373         Self {
 374             end_tx: Mutex::new(None),
 375         }
 376     }
 377 }
 378
 379 impl InputSource for NullInputSource {
 380     fn start(&self, samples: SyncSender<SoundmodemEvent>) {
 381         let (end_tx, end_rx) = channel();
 382         std::thread::spawn(move || {
 383             // assuming 48 kHz for now
 384             const TICK: Duration = Duration::from_millis(25);
 385             const SAMPLES_PER_TICK: usize = 1200;
 386             let mut next_tick = Instant::now() + TICK;
 387
 388             loop {
 389                 std::thread::sleep(next_tick.duration_since(Instant::now()));
 390                 next_tick = next_tick + TICK;
 391                 if end_rx.try_recv() != Err(TryRecvError::Empty) {
 392                     break;
 393                 }
 394                 if let Err(e) = samples.try_send(SoundmodemEvent::BasebandInput(
 395                     [0i16; SAMPLES_PER_TICK].into(),
 396                 )) {
 397                     debug!("overflow feeding soundmodem: {e:?}");
 398                 }
 399             }
 400         });
 401         *self.end_tx.lock().unwrap() = Some(end_tx);
 402     }
 403
 404     fn close(&self) {
 405         let _ = self.end_tx.lock().unwrap().take();
 406     }
 407 }
 408
 409 pub struct OutputBuffer {
 410     idling: bool,
 411     // TODO: something more efficient
 412     samples: VecDeque<i16>,
 413     latency: Duration,
 414 }
 415
 416 impl OutputBuffer {
 417     pub fn new() -> Self {
 418         Self {
 419             idling: true,
 420             samples: VecDeque::new(),
 421             latency: Duration::ZERO,
 422         }
 423     }
 424 }
 425
 426 pub trait OutputSink: Send + Sync + 'static {
 427     fn start(&self, event_tx: SyncSender<SoundmodemEvent>, buffer: Arc<RwLock<OutputBuffer>>);
 428     fn close(&self);
 429 }
 430
 431 pub struct OutputRrcFile {
 432     path: PathBuf,
 433     end_tx: Mutex<Option<Sender<()>>>,
 434 }
 435
 436 impl OutputRrcFile {
 437     pub fn new(path: PathBuf) -> Self {
 438         Self {
 439             path,
 440             end_tx: Mutex::new(None),
 441         }
 442     }
 443 }
 444
 445 impl OutputSink for OutputRrcFile {
 446     fn start(&self, event_tx: SyncSender<SoundmodemEvent>, buffer: Arc<RwLock<OutputBuffer>>) {
 447         let (end_tx, end_rx) = channel();
 448         let path = self.path.clone();
 449         std::thread::spawn(move || {
 450             // TODO: error handling
 451             let mut file = File::create(path).unwrap();
 452
 453             // assuming 48 kHz for now
 454             const TICK: Duration = Duration::from_millis(25);
 455             const SAMPLES_PER_TICK: usize = 1200;
 456
 457             // flattened BE i16s for writing
 458             let mut buf = [0u8; SAMPLES_PER_TICK * 2];
 459             let mut next_tick = Instant::now() + TICK;
 460
 461             loop {
 462                 std::thread::sleep(next_tick.duration_since(Instant::now()));
 463                 next_tick = next_tick + TICK;
 464                 if end_rx.try_recv() != Err(TryRecvError::Empty) {
 465                     break;
 466                 }
 467                 // For now only write deliberately modulated (non-idling) samples
 468                 // Multiple transmissions will get smooshed together
 469                 let mut buf_used = 0;
 470
 471                 let mut buffer = buffer.write().unwrap();
 472                 for out in buf.chunks_mut(2) {
 473                     if let Some(s) = buffer.samples.pop_front() {
 474                         let be = s.to_le_bytes();
 475                         out.copy_from_slice(&[be[0], be[1]]);
 476                         buf_used += 2;
 477                     } else if !buffer.idling {
 478                         debug!("output rrc file had underrun");
 479                         let _ = event_tx.send(SoundmodemEvent::OutputUnderrun);
 480                         break;
 481                     }
 482                 }
 483                 if let Err(e) = file.write_all(&buf[0..buf_used]) {
 484                     debug!("failed to write to rrc file: {e:?}");
 485                     break;
 486                 }
 487                 let _ = event_tx.send(SoundmodemEvent::DidReadFromOutputBuffer {
 488                     len: buf_used / 2,
 489                     timestamp: Instant::now(),
 490                 });
 491             }
 492         });
 493         *self.end_tx.lock().unwrap() = Some(end_tx);
 494     }
 495
 496     fn close(&self) {
 497         let _ = self.end_tx.lock().unwrap().take();
 498     }
 499 }
 500
 501 pub struct NullOutputSink {
 502     end_tx: Mutex<Option<Sender<()>>>,
 503 }
 504
 505 impl NullOutputSink {
 506     pub fn new() -> Self {
 507         Self {
 508             end_tx: Mutex::new(None),
 509         }
 510     }
 511 }
 512
 513 impl OutputSink for NullOutputSink {
 514     fn start(&self, event_tx: SyncSender<SoundmodemEvent>, buffer: Arc<RwLock<OutputBuffer>>) {
 515         let (end_tx, end_rx) = channel();
 516         std::thread::spawn(move || {
 517             // assuming 48 kHz for now
 518             const TICK: Duration = Duration::from_millis(25);
 519             const SAMPLES_PER_TICK: usize = 1200;
 520             let mut next_tick = Instant::now() + TICK;
 521
 522             loop {
 523                 std::thread::sleep(next_tick.duration_since(Instant::now()));
 524                 next_tick = next_tick + TICK;
 525                 if end_rx.try_recv() != Err(TryRecvError::Empty) {
 526                     break;
 527                 }
 528
 529                 let mut buffer = buffer.write().unwrap();
 530                 let mut taken = 0;
 531                 for _ in 0..SAMPLES_PER_TICK {
 532                     if !buffer.samples.pop_front().is_some() {
 533                         if !buffer.idling {
 534                             debug!("null output had underrun");
 535                             let _ = event_tx.send(SoundmodemEvent::OutputUnderrun);
 536                             break;
 537                         }
 538                     } else {
 539                         taken += 1;
 540                     }
 541                 }
 542                 let _ = event_tx.send(SoundmodemEvent::DidReadFromOutputBuffer {
 543                     len: taken,
 544                     timestamp: Instant::now(),
 545                 });
 546             }
 547         });
 548         *self.end_tx.lock().unwrap() = Some(end_tx);
 549     }
 550
 551     fn close(&self) {
 552         let _ = self.end_tx.lock().unwrap().take();
 553     }
 554 }
 555
 556 pub struct OutputSoundcard {
 557     // TODO: allow for inversion both here and in output
 558     cpal_name: Option<String>,
 559     end_tx: Mutex<Option<Sender<()>>>,
 560 }
 561
 562 impl OutputSoundcard {
 563     pub fn new() -> Self {
 564         Self {
 565             cpal_name: None,
 566             end_tx: Mutex::new(None),
 567         }
 568     }
 569
 570     pub fn new_with_card(card_name: String) -> Self {
 571         Self {
 572             cpal_name: Some(card_name),
 573             end_tx: Mutex::new(None),
 574         }
 575     }
 576 }
 577
 578 impl OutputSink for OutputSoundcard {
 579     fn start(&self, event_tx: SyncSender<SoundmodemEvent>, buffer: Arc<RwLock<OutputBuffer>>) {
 580         let (end_tx, end_rx) = channel();
 581         let cpal_name = self.cpal_name.clone();
 582         std::thread::spawn(move || {
 583             let host = cpal::default_host();
 584             let device = if let Some(name) = cpal_name.as_deref() {
 585                 host.output_devices()
 586                     .unwrap()
 587                     .find(|d| d.name().unwrap() == name)
 588                     .unwrap()
 589             } else {
 590                 host.default_output_device().unwrap()
 591             };
 592             let mut configs = device.supported_output_configs().unwrap();
 593             // TODO: more error handling
 594             let config = configs
 595                 .find(|c| c.channels() == 1 && c.sample_format() == SampleFormat::I16)
 596                 .unwrap()
 597                 .with_sample_rate(SampleRate(48000));
 598             let stream = device
 599                 .build_output_stream(
 600                     &config.into(),
 601                     move |data: &mut [i16], info: &cpal::OutputCallbackInfo| {
 602                         let mut taken = 0;
 603                         let ts = info.timestamp();
 604                         let latency = ts
 605                             .playback
 606                             .duration_since(&ts.callback)
 607                             .unwrap_or(Duration::ZERO);
 608                         let mut buffer = buffer.write().unwrap();
 609                         buffer.latency = latency;
 610                         for out in data.iter_mut() {
 611                             if let Some(s) = buffer.samples.pop_front() {
 612                                 *out = s;
 613                                 taken += 1;
 614                             } else if buffer.idling {
 615                                 *out = 0;
 616                             } else {
 617                                 debug!("output soundcard had underrun");
 618                                 let _ = event_tx.send(SoundmodemEvent::OutputUnderrun);
 619                                 break;
 620                             }
 621                         }
 622                         //debug!("latency is {} ms, taken {taken}", latency.as_millis());
 623                         let _ = event_tx.send(SoundmodemEvent::DidReadFromOutputBuffer {
 624                             len: taken,
 625                             timestamp: Instant::now(),
 626                         });
 627                     },
 628                     |e| {
 629                         // TODO: abort?
 630                         debug!("error occurred in soundcard output: {e:?}");
 631                     },
 632                     None,
 633                 )
 634                 .unwrap();
 635             stream.play().unwrap();
 636             let _ = end_rx.recv();
 637         });
 638         *self.end_tx.lock().unwrap() = Some(end_tx);
 639     }
 640
 641     fn close(&self) {
 642         let _ = self.end_tx.lock().unwrap().take();
 643     }
 644 }
 645
 646 pub trait Ptt: Send + 'static {
 647     fn ptt_on(&mut self);
 648     fn ptt_off(&mut self);
 649 }
 650
 651 /// There is no PTT because this TNC will never make transmissions on a real radio.
 652 pub struct NullPtt;
 653
 654 impl NullPtt {
 655     pub fn new() -> Self {
 656         Self
 657     }
 658 }
 659
 660 impl Ptt for NullPtt {
 661     fn ptt_on(&mut self) {}
 662     fn ptt_off(&mut self) {}
 663 }