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

   1 use std::io::{self, ErrorKind, Read, Write};
   2 use std::sync::RwLock;
   3 use std::collections::VecDeque;
   4 use crate::tnc::{Tnc, TncError};
   5 use cpal::traits::DeviceTrait;
   6 use cpal::traits::HostTrait;
   7 use cpal::traits::StreamTrait;
   8 use cpal::{SampleFormat, SampleRate};
   9 use log::debug;
  10 use m17core::kiss::MAX_FRAME_LEN;
  11 use m17core::modem::{Demodulator, Modulator, ModulatorAction, SoftDemodulator, SoftModulator};
  12 use m17core::tnc::SoftTnc;
  13 use std::fs::File;
  14 use std::path::PathBuf;
  15 use std::sync::mpsc::{channel, sync_channel, Receiver, Sender, SyncSender, TryRecvError};
  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(event_tx.clone(), event_rx, kiss_out_tx, Box::new(input), Box::new(output));
  31         Self {
  32             event_tx,
  33             kiss_out_rx: Arc::new(Mutex::new(kiss_out_rx)),
  34             partial_kiss_out: Arc::new(Mutex::new(None)),
  35         }
  36     }
  37 }
  38
  39 struct PartialKissOut {
  40     output: Arc<[u8]>,
  41     idx: usize,
  42 }
  43
  44 impl Read for Soundmodem {
  45     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
  46         {
  47             let mut partial_kiss_out = self.partial_kiss_out.lock().unwrap();
  48             if let Some(partial) = partial_kiss_out.as_mut() {
  49                 let remaining = partial.output.len() - partial.idx;
  50                 let to_write = remaining.min(buf.len());
  51                 buf[0..to_write]
  52                     .copy_from_slice(&partial.output[partial.idx..(partial.idx + to_write)]);
  53                 if to_write == remaining {
  54                     *partial_kiss_out = None;
  55                 } else {
  56                     partial.idx += to_write;
  57                 }
  58                 return Ok(to_write);
  59             }
  60         }
  61         let output = {
  62             let rx = self.kiss_out_rx.lock().unwrap();
  63             rx.recv()
  64                 .map_err(|s| io::Error::new(ErrorKind::Other, format!("{:?}", s)))?
  65         };
  66         let to_write = output.len().min(buf.len());
  67         buf[0..to_write].copy_from_slice(&output[0..to_write]);
  68         if to_write != output.len() {
  69             *self.partial_kiss_out.lock().unwrap() = Some(PartialKissOut {
  70                 output,
  71                 idx: to_write,
  72             })
  73         }
  74         Ok(to_write)
  75     }
  76 }
  77
  78 impl Write for Soundmodem {
  79     fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
  80         let _ = self.event_tx.try_send(SoundmodemEvent::Kiss(buf.into()));
  81         Ok(buf.len())
  82     }
  83
  84     fn flush(&mut self) -> std::io::Result<()> {
  85         Ok(())
  86     }
  87 }
  88
  89 impl Tnc for Soundmodem {
  90     fn try_clone(&mut self) -> Result<Self, TncError> {
  91         Ok(Self {
  92             event_tx: self.event_tx.clone(),
  93             kiss_out_rx: self.kiss_out_rx.clone(),
  94             partial_kiss_out: self.partial_kiss_out.clone(),
  95         })
  96     }
  97
  98     fn start(&mut self) -> Result<(), TncError> {
  99         let _ = self.event_tx.send(SoundmodemEvent::Start);
 100         Ok(())
 101     }
 102
 103     fn close(&mut self) -> Result<(), TncError> {
 104         let _ = self.event_tx.send(SoundmodemEvent::Close);
 105         Ok(())
 106     }
 107 }
 108
 109 pub enum SoundmodemEvent {
 110     Kiss(Arc<[u8]>),
 111     BasebandInput(Arc<[i16]>),
 112     Start,
 113     Close,
 114     DidReadFromOutputBuffer {
 115         len: usize,
 116         timestamp: Instant,
 117     },
 118     OutputUnderrun,
 119 }
 120
 121 fn spawn_soundmodem_worker(
 122     event_tx: SyncSender<SoundmodemEvent>,
 123     event_rx: Receiver<SoundmodemEvent>,
 124     kiss_out_tx: SyncSender<Arc<[u8]>>,
 125     input: Box<dyn InputSource>,
 126     output: Box<dyn OutputSink>,
 127 ) {
 128     std::thread::spawn(move || {
 129         // TODO: should be able to provide a custom Demodulator for a soundmodem
 130         let mut demodulator = SoftDemodulator::new();
 131         let mut modulator = SoftModulator::new();
 132         let mut tnc = SoftTnc::new();
 133         let mut buf = [0u8; MAX_FRAME_LEN];
 134         let out_buffer = Arc::new(RwLock::new(OutputBuffer::new()));
 135         let mut out_samples = [0i16; 1024];
 136         let start = Instant::now();
 137         while let Ok(ev) = event_rx.recv() {
 138             // Update clock on TNC before we do anything
 139             let sample_time = (start.elapsed().as_nanos() / 48000) as u64;
 140             tnc.set_now(sample_time);
 141
 142             // Handle event
 143             match ev {
 144                 SoundmodemEvent::Kiss(k) => {
 145                     let _n = tnc.write_kiss(&k);
 146                     // TODO: what does it mean if we fail to write it all?
 147                     // Probably we have to read frames for tx first - revisit this during tx
 148                 }
 149                 SoundmodemEvent::BasebandInput(b) => {
 150                     for sample in &*b {
 151                         if let Some(frame) = demodulator.demod(*sample) {
 152                             tnc.handle_frame(frame);
 153                             loop {
 154                                 let n = tnc.read_kiss(&mut buf);
 155                                 if n > 0 {
 156                                     let _ = kiss_out_tx.try_send(buf[0..n].into());
 157                                 } else {
 158                                     break;
 159                                 }
 160                             }
 161                         }
 162                     }
 163                     tnc.set_data_carrier_detect(demodulator.data_carrier_detect());
 164                 }
 165                 SoundmodemEvent::Start => {
 166                     input.start(event_tx.clone());
 167                     output.start(event_tx.clone(), out_buffer.clone());
 168                 },
 169                 SoundmodemEvent::Close => break,
 170                 SoundmodemEvent::DidReadFromOutputBuffer { len, timestamp } => {
 171                     let (occupied, internal_latency) = {
 172                         let out_buffer = out_buffer.read().unwrap();
 173                         (out_buffer.samples.len(), out_buffer.latency)
 174                     };
 175                     let internal_latency = (internal_latency.as_secs_f32() * 48000.0) as usize;
 176                     let dynamic_latency = len.saturating_sub((timestamp.elapsed().as_secs_f32() * 48000.0) as usize);
 177                     modulator.update_output_buffer(occupied, 48000, internal_latency + dynamic_latency);
 178                 },
 179                 SoundmodemEvent::OutputUnderrun => {
 180                     // TODO: cancel transmission, send empty data frame to host
 181                 }
 182             }
 183
 184             // Let the modulator do what it wants
 185             while let Some(action) = modulator.run() {
 186                 match action {
 187                     ModulatorAction::SetIdle(idling) => {
 188                         out_buffer.write().unwrap().idling = idling;
 189                     },
 190                     ModulatorAction::GetNextFrame => {
 191                         modulator.provide_next_frame(tnc.read_tx_frame());
 192                     },
 193                     ModulatorAction::ReadOutput => {
 194                         loop {
 195                             let n = modulator.read_output_samples(&mut out_samples);
 196                             if n == 0 {
 197                                 break;
 198                             }
 199                             let mut out_buffer = out_buffer.write().unwrap();
 200                             for s in &out_samples[0..n] {
 201                                 out_buffer.samples.push_back(*s);
 202                             }
 203                         }
 204
 205                     },
 206                     ModulatorAction::TransmissionWillEnd(in_samples) => {
 207                         tnc.set_tx_end_time(in_samples);
 208                     },
 209                 }
 210             }
 211         }
 212     });
 213 }
 214
 215 pub trait InputSource: Send + Sync + 'static {
 216     fn start(&self, samples: SyncSender<SoundmodemEvent>);
 217     fn close(&self);
 218 }
 219
 220 pub struct InputSoundcard {
 221     // TODO: allow for inversion both here and in output
 222     cpal_name: Option<String>,
 223     end_tx: Mutex<Option<Sender<()>>>,
 224 }
 225
 226 impl InputSoundcard {
 227     pub fn new() -> Self {
 228         Self {
 229             cpal_name: None,
 230             end_tx: Mutex::new(None),
 231         }
 232     }
 233
 234     pub fn new_with_card(card_name: String) -> Self {
 235         Self {
 236             cpal_name: Some(card_name),
 237             end_tx: Mutex::new(None),
 238         }
 239     }
 240 }
 241
 242 impl InputSource for InputSoundcard {
 243     fn start(&self, samples: SyncSender<SoundmodemEvent>) {
 244         let (end_tx, end_rx) = channel();
 245         let cpal_name = self.cpal_name.clone();
 246         std::thread::spawn(move || {
 247             let host = cpal::default_host();
 248             let device = if let Some(name) = cpal_name.as_deref() {
 249                 host.input_devices()
 250                     .unwrap()
 251                     .find(|d| d.name().unwrap() == name)
 252                     .unwrap()
 253             } else {
 254                 host.default_input_device().unwrap()
 255             };
 256             let mut configs = device.supported_input_configs().unwrap();
 257             let config = configs
 258                 .find(|c| c.channels() == 1 && c.sample_format() == SampleFormat::I16)
 259                 .unwrap()
 260                 .with_sample_rate(SampleRate(48000));
 261             let stream = device
 262                 .build_input_stream(
 263                     &config.into(),
 264                     move |data: &[i16], _info: &cpal::InputCallbackInfo| {
 265                         debug!("input has given us {} samples", data.len());
 266                         let out: Vec<i16> = data.iter().map(|s| *s).collect();
 267                         let _ = samples.try_send(SoundmodemEvent::BasebandInput(out.into()));
 268                     },
 269                     |e| {
 270                         // TODO: abort?
 271                         debug!("error occurred in soundcard input: {e:?}");
 272                     },
 273                     None,
 274                 )
 275                 .unwrap();
 276             stream.play().unwrap();
 277             let _ = end_rx.recv();
 278         });
 279         *self.end_tx.lock().unwrap() = Some(end_tx);
 280     }
 281
 282     fn close(&self) {
 283         let _ = self.end_tx.lock().unwrap().take();
 284     }
 285 }
 286
 287 pub struct InputRrcFile {
 288     path: PathBuf,
 289     end_tx: Mutex<Option<Sender<()>>>,
 290 }
 291
 292 impl InputRrcFile {
 293     pub fn new(path: PathBuf) -> Self {
 294         Self {
 295             path,
 296             end_tx: Mutex::new(None),
 297         }
 298     }
 299 }
 300
 301 impl InputSource for InputRrcFile {
 302     fn start(&self, samples: SyncSender<SoundmodemEvent>) {
 303         let (end_tx, end_rx) = channel();
 304         let path = self.path.clone();
 305         std::thread::spawn(move || {
 306             // TODO: error handling
 307             let mut file = File::open(path).unwrap();
 308             let mut baseband = vec![];
 309             file.read_to_end(&mut baseband).unwrap();
 310
 311             // assuming 48 kHz for now
 312             const TICK: Duration = Duration::from_millis(25);
 313             const SAMPLES_PER_TICK: usize = 1200;
 314
 315             let mut next_tick = Instant::now() + TICK;
 316             let mut buf = [0i16; SAMPLES_PER_TICK];
 317             let mut idx = 0;
 318
 319             for sample in baseband
 320                 .chunks(2)
 321                 .map(|pair| i16::from_le_bytes([pair[0], pair[1]]))
 322             {
 323                 buf[idx] = sample;
 324                 idx += 1;
 325                 if idx == SAMPLES_PER_TICK {
 326                     if let Err(e) = samples.try_send(SoundmodemEvent::BasebandInput(buf.into())) {
 327                         debug!("overflow feeding soundmodem: {e:?}");
 328                     }
 329                     next_tick = next_tick + TICK;
 330                     idx = 0;
 331                     std::thread::sleep(next_tick.duration_since(Instant::now()));
 332                 }
 333                 if end_rx.try_recv() != Err(TryRecvError::Empty) {
 334                     break;
 335                 }
 336             }
 337         });
 338         *self.end_tx.lock().unwrap() = Some(end_tx);
 339     }
 340
 341     fn close(&self) {
 342         let _ = self.end_tx.lock().unwrap().take();
 343     }
 344 }
 345
 346 pub struct OutputBuffer {
 347     idling: bool,
 348     // TODO: something more efficient
 349     samples: VecDeque<i16>,
 350     latency: Duration,
 351 }
 352
 353 impl OutputBuffer {
 354     pub fn new() -> Self {
 355         Self {
 356             idling: true,
 357             samples: VecDeque::new(),
 358             latency: Duration::ZERO,
 359         }
 360     }
 361 }
 362
 363 pub trait OutputSink: Send + Sync + 'static {
 364     fn start(&self, event_tx: SyncSender<SoundmodemEvent>, buffer: Arc<RwLock<OutputBuffer>>);
 365     fn close(&self);
 366 }
 367
 368 pub struct OutputRrcFile {
 369     path: PathBuf,
 370     end_tx: Mutex<Option<Sender<()>>>,
 371 }
 372
 373 impl OutputRrcFile {
 374     pub fn new(path: PathBuf) -> Self {
 375         Self {
 376             path,
 377             end_tx: Mutex::new(None),
 378         }
 379     }
 380 }
 381
 382 impl OutputSink for OutputRrcFile {
 383     fn start(&self, event_tx: SyncSender<SoundmodemEvent>, buffer: Arc<RwLock<OutputBuffer>>) {
 384         let (end_tx, end_rx) = channel();
 385         let path = self.path.clone();
 386         std::thread::spawn(move || {
 387             // TODO: error handling
 388             let mut file = File::create(path).unwrap();
 389
 390             // assuming 48 kHz for now
 391             const TICK: Duration = Duration::from_millis(25);
 392             const SAMPLES_PER_TICK: usize = 1200;
 393
 394             // flattened BE i16s for writing
 395             let mut buf = [0u8; SAMPLES_PER_TICK * 2];
 396             let mut next_tick = Instant::now() + TICK;
 397
 398             loop {
 399                 std::thread::sleep(next_tick.duration_since(Instant::now()));
 400                 next_tick = next_tick + TICK;
 401                 if end_rx.try_recv() != Err(TryRecvError::Empty) {
 402                     break;
 403                 }
 404
 405                 let mut buffer = buffer.write().unwrap();
 406                 for out in buf.chunks_mut(2) {
 407                     if let Some(s) = buffer.samples.pop_front() {
 408                         let be = s.to_be_bytes();
 409                         out.copy_from_slice(&[be[0], be[1]]);
 410                     } else if buffer.idling {
 411                         out.copy_from_slice(&[0, 0]);
 412                     } else {
 413                         debug!("output rrc file had underrun");
 414                         let _ = event_tx.send(SoundmodemEvent::OutputUnderrun);
 415                         break;
 416                     }
 417                 }
 418                 if let Err(e) = file.write_all(&buf) {
 419                     debug!("failed to write to rrc file: {e:?}");
 420                     break;
 421                 }
 422             }
 423
 424         });
 425         *self.end_tx.lock().unwrap() = Some(end_tx);
 426     }
 427
 428     fn close(&self) {
 429         let _ = self.end_tx.lock().unwrap().take();
 430     }
 431 }