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