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

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