From: Thomas Karpiniec Date: Mon, 26 May 2025 01:14:02 +0000 (+1000) Subject: Split m17codec2 into modules X-Git-Url: https://code.octet-stream.net/m17rt/commitdiff_plain/0ae30668b7142e60de9efcc6e2e2724f9ec77962?ds=inline Split m17codec2 into modules --- diff --git a/m17codec2/src/error.rs b/m17codec2/src/error.rs new file mode 100644 index 0000000..ccf94b9 --- /dev/null +++ b/m17codec2/src/error.rs @@ -0,0 +1,22 @@ +use thiserror::Error; + +#[derive(Debug, Error)] +pub enum M17Codec2Error { + #[error("selected card '{0}' does not exist or is in use")] + CardUnavailable(String), + + #[error("default output card is unavailable")] + DefaultCardUnavailable, + + #[error("selected card '{0}' failed to list available output configs: '{1}'")] + OutputConfigsUnavailable(String, #[source] cpal::SupportedStreamConfigsError), + + #[error("selected card '{0}' did not offer a compatible output config type, either due to hardware limitations or because it is currently in use")] + SupportedOutputUnavailable(String), + + #[error("selected card '{0}' was unable to build an output stream: '{1}'")] + OutputStreamBuildError(String, #[source] cpal::BuildStreamError), + + #[error("selected card '{0}' was unable to play an output stream: '{1}'")] + OutputStreamPlayError(String, #[source] cpal::PlayStreamError), +} diff --git a/m17codec2/src/lib.rs b/m17codec2/src/lib.rs index 0c32a5b..4023133 100644 --- a/m17codec2/src/lib.rs +++ b/m17codec2/src/lib.rs @@ -1,350 +1,8 @@ #![doc = include_str!("../README.md")] -use codec2::{Codec2, Codec2Mode}; -use cpal::traits::DeviceTrait; -use cpal::traits::HostTrait; -use cpal::traits::StreamTrait; -use cpal::{Sample, SampleFormat, SampleRate}; -use log::debug; -use m17app::adapter::StreamAdapter; -use m17app::app::TxHandle; -use m17app::error::AdapterError; -use m17app::link_setup::LinkSetup; -use m17app::link_setup::M17Address; -use m17app::StreamFrame; -use rubato::Resampler; -use rubato::SincFixedIn; -use rubato::SincInterpolationParameters; -use std::collections::VecDeque; -use std::fs::File; -use std::io::Write; -use std::path::Path; -use std::path::PathBuf; -use std::sync::{ - mpsc::{channel, Receiver, Sender}, - Arc, Mutex, -}; -use std::time::Duration; -use std::time::Instant; -use thiserror::Error; - +pub mod error; +pub mod rx; pub mod soundcards; +pub mod tx; -/// Write one or more 8-byte chunks of 3200-bit Codec2 to a raw S16LE file -/// and return the samples. -pub fn decode_codec2>(data: &[u8], out_path: P) -> Vec { - let codec2 = Codec2::new(Codec2Mode::MODE_3200); - let var_name = codec2; - let mut codec = var_name; - let mut all_samples: Vec = vec![]; - for i in 0..(data.len() / 8) { - let mut samples = vec![0; codec.samples_per_frame()]; - codec.decode(&mut samples, &data[i * 8..((i + 1) * 8)]); - all_samples.append(&mut samples); - } - let mut speech_out = File::create(out_path).unwrap(); - for b in &all_samples { - speech_out.write_all(&b.to_le_bytes()).unwrap(); - } - all_samples -} - -/// Subscribes to M17 streams and attempts to play the decoded Codec2 -pub struct Codec2Adapter { - state: Arc>, - output_card: Option, -} - -impl Codec2Adapter { - pub fn new() -> Self { - Self { - state: Arc::new(Mutex::new(AdapterState { - tx: None, - out_buf: VecDeque::new(), - codec2: Codec2::new(Codec2Mode::MODE_3200), - end_tx: None, - resampler: None, - })), - output_card: None, - } - } - - pub fn set_output_card>(&mut self, card_name: S) { - self.output_card = Some(card_name.into()); - } -} - -impl Default for Codec2Adapter { - fn default() -> Self { - Self::new() - } -} - -struct AdapterState { - tx: Option, - /// Circular buffer of output samples for playback - out_buf: VecDeque, - codec2: Codec2, - end_tx: Option>, - resampler: Option>, -} - -impl StreamAdapter for Codec2Adapter { - fn start(&self, handle: TxHandle) -> Result<(), AdapterError> { - self.state.lock().unwrap().tx = Some(handle); - - let (end_tx, end_rx) = channel(); - let (setup_tx, setup_rx) = channel(); - let state = self.state.clone(); - let output_card = self.output_card.clone(); - std::thread::spawn(move || stream_thread(end_rx, setup_tx, state, output_card)); - self.state.lock().unwrap().end_tx = Some(end_tx); - // Propagate any errors arising in the thread - let sample_rate = setup_rx.recv()??; - debug!("selected codec2 output sample rate {sample_rate}"); - if sample_rate != 8000 { - let params = SincInterpolationParameters { - sinc_len: 256, - f_cutoff: 0.95, - oversampling_factor: 128, - interpolation: rubato::SincInterpolationType::Cubic, - window: rubato::WindowFunction::BlackmanHarris2, - }; - // TODO: fix unwrap - self.state.lock().unwrap().resampler = - Some(SincFixedIn::new(sample_rate as f64 / 8000f64, 1.0, params, 160, 1).unwrap()); - } - Ok(()) - } - - fn close(&self) -> Result<(), AdapterError> { - let mut state = self.state.lock().unwrap(); - state.tx = None; - state.end_tx = None; - Ok(()) - } - - fn stream_began(&self, _link_setup: LinkSetup) { - // for now we will assume: - // - unencrypted - // - data type is Voice (Codec2 3200), not Voice+Data - // TODO: is encryption handled here or in M17App, such that we get a decrypted stream? - // TODO: handle the Voice+Data combination with Codec2 1600 - self.state.lock().unwrap().codec2 = Codec2::new(Codec2Mode::MODE_3200); - } - - fn stream_data(&self, _frame_number: u16, _is_final: bool, data: Arc<[u8; 16]>) { - let mut state = self.state.lock().unwrap(); - for encoded in data.chunks(8) { - if state.out_buf.len() < 8192 { - let mut samples = [i16::EQUILIBRIUM; 160]; // while assuming 3200 - state.codec2.decode(&mut samples, encoded); - if let Some(resampler) = state.resampler.as_mut() { - let samples_f: Vec = - samples.iter().map(|s| *s as f32 / 16384.0f32).collect(); - let res = resampler.process(&vec![samples_f], None).unwrap(); - for s in &res[0] { - state.out_buf.push_back((s * 16383.0f32) as i16); - } - } else { - // TODO: maybe get rid of VecDeque so we can decode directly into ring buffer? - for s in samples { - state.out_buf.push_back(s); - } - } - } else { - debug!("out_buf overflow"); - } - } - } -} - -fn output_cb(data: &mut [i16], state: &Mutex, channels: u16) { - let mut state = state.lock().unwrap(); - for d in data.chunks_mut(channels as usize) { - d.fill(state.out_buf.pop_front().unwrap_or(i16::EQUILIBRIUM)); - } -} - -/// Create and manage the stream from a dedicated thread since it's `!Send` -fn stream_thread( - end: Receiver<()>, - setup_tx: Sender>, - state: Arc>, - output_card: Option, -) { - let host = cpal::default_host(); - let device = if let Some(output_card) = output_card { - // TODO: more error handling for unwraps - match host - .output_devices() - .unwrap() - .find(|d| d.name().unwrap() == output_card) - { - Some(d) => d, - None => { - let _ = setup_tx.send(Err(M17Codec2Error::CardUnavailable(output_card).into())); - return; - } - } - } else { - match host.default_output_device() { - Some(d) => d, - None => { - let _ = setup_tx.send(Err(M17Codec2Error::DefaultCardUnavailable.into())); - return; - } - } - }; - let card_name = device.name().unwrap(); - let mut configs = match device.supported_output_configs() { - Ok(c) => c, - Err(e) => { - let _ = setup_tx.send(Err( - M17Codec2Error::OutputConfigsUnavailable(card_name, e).into() - )); - return; - } - }; - let config = match configs.find(|c| { - (c.channels() == 1 || c.channels() == 2) && c.sample_format() == SampleFormat::I16 - }) { - Some(c) => c, - None => { - let _ = setup_tx.send(Err( - M17Codec2Error::SupportedOutputUnavailable(card_name).into() - )); - return; - } - }; - - let target_sample_rate = - if config.min_sample_rate().0 <= 8000 && config.max_sample_rate().0 >= 8000 { - 8000 - } else { - config.min_sample_rate().0 - }; - let channels = config.channels(); - - let config = config.with_sample_rate(SampleRate(target_sample_rate)); - let stream = match device.build_output_stream( - &config.into(), - move |data: &mut [i16], _info: &cpal::OutputCallbackInfo| { - output_cb(data, &state, channels); - }, - |e| { - // trigger end_tx here? always more edge cases - debug!("error occurred in codec2 playback: {e:?}"); - }, - None, - ) { - Ok(s) => s, - Err(e) => { - let _ = setup_tx.send(Err( - M17Codec2Error::OutputStreamBuildError(card_name, e).into() - )); - return; - } - }; - match stream.play() { - Ok(()) => (), - Err(e) => { - let _ = setup_tx.send(Err( - M17Codec2Error::OutputStreamPlayError(card_name, e).into() - )); - return; - } - } - let _ = setup_tx.send(Ok(target_sample_rate)); - let _ = end.recv(); - // it seems concrete impls of Stream have a Drop implementation that will handle termination -} - -/// Transmits a wave file as an M17 stream -pub struct WavePlayer; - -impl WavePlayer { - /// Plays a wave file (blocking). - /// - /// * `path`: wave file to transmit, must be 8 kHz mono and 16-bit LE - /// * `tx`: a `TxHandle` obtained from an `M17App` - /// * `source`: address of transmission source - /// * `destination`: address of transmission destination - /// * `channel_access_number`: from 0 to 15, usually 0 - pub fn play( - path: PathBuf, - tx: TxHandle, - source: &M17Address, - destination: &M17Address, - channel_access_number: u8, - ) { - let mut reader = hound::WavReader::open(path).unwrap(); - let mut samples = reader.samples::(); - - let mut codec = Codec2::new(Codec2Mode::MODE_3200); - let mut in_buf = [0i16; 160]; - let mut out_buf = [0u8; 16]; - let mut lsf_chunk: usize = 0; - const TICK: Duration = Duration::from_millis(40); - let mut next_tick = Instant::now() + TICK; - let mut frame_number = 0; - - let mut setup = LinkSetup::new_voice(source, destination); - setup.set_channel_access_number(channel_access_number); - tx.transmit_stream_start(&setup); - - loop { - let mut last_one = false; - for out in out_buf.chunks_mut(8) { - for i in in_buf.iter_mut() { - let sample = match samples.next() { - Some(Ok(sample)) => sample, - _ => { - last_one = true; - 0 - } - }; - *i = sample; - } - codec.encode(out, &in_buf); - } - tx.transmit_stream_next(&StreamFrame { - lich_idx: lsf_chunk as u8, - lich_part: setup.lich_part(lsf_chunk as u8), - frame_number, - end_of_stream: last_one, - stream_data: out_buf, - }); - frame_number += 1; - lsf_chunk = (lsf_chunk + 1) % 6; - - if last_one { - break; - } - - std::thread::sleep(next_tick.duration_since(Instant::now())); - next_tick += TICK; - } - } -} - -#[derive(Debug, Error)] -pub enum M17Codec2Error { - #[error("selected card '{0}' does not exist or is in use")] - CardUnavailable(String), - - #[error("default output card is unavailable")] - DefaultCardUnavailable, - - #[error("selected card '{0}' failed to list available output configs: '{1}'")] - OutputConfigsUnavailable(String, #[source] cpal::SupportedStreamConfigsError), - - #[error("selected card '{0}' did not offer a compatible output config type, either due to hardware limitations or because it is currently in use")] - SupportedOutputUnavailable(String), - - #[error("selected card '{0}' was unable to build an output stream: '{1}'")] - OutputStreamBuildError(String, #[source] cpal::BuildStreamError), - - #[error("selected card '{0}' was unable to play an output stream: '{1}'")] - OutputStreamPlayError(String, #[source] cpal::PlayStreamError), -} +pub use error::M17Codec2Error; diff --git a/m17codec2/src/rx.rs b/m17codec2/src/rx.rs new file mode 100644 index 0000000..c6d3cdc --- /dev/null +++ b/m17codec2/src/rx.rs @@ -0,0 +1,252 @@ +use crate::M17Codec2Error; +use codec2::{Codec2, Codec2Mode}; +use cpal::traits::DeviceTrait; +use cpal::traits::HostTrait; +use cpal::traits::StreamTrait; +use cpal::{Sample, SampleFormat, SampleRate}; +use log::debug; +use m17app::adapter::StreamAdapter; +use m17app::app::TxHandle; +use m17app::error::AdapterError; +use m17app::link_setup::LinkSetup; +use rubato::Resampler; +use rubato::SincFixedIn; +use rubato::SincInterpolationParameters; +use std::collections::VecDeque; +use std::fs::File; +use std::io::Write; +use std::path::Path; +use std::sync::{ + mpsc::{channel, Receiver, Sender}, + Arc, Mutex, +}; + +/// Write one or more 8-byte chunks of 3200-bit Codec2 to a raw S16LE file +/// and return the samples. +pub fn decode_codec2>(data: &[u8], out_path: P) -> Vec { + let codec2 = Codec2::new(Codec2Mode::MODE_3200); + let var_name = codec2; + let mut codec = var_name; + let mut all_samples: Vec = vec![]; + for i in 0..(data.len() / 8) { + let mut samples = vec![0; codec.samples_per_frame()]; + codec.decode(&mut samples, &data[i * 8..((i + 1) * 8)]); + all_samples.append(&mut samples); + } + let mut speech_out = File::create(out_path).unwrap(); + for b in &all_samples { + speech_out.write_all(&b.to_le_bytes()).unwrap(); + } + all_samples +} + +/// Subscribes to M17 streams and attempts to play the decoded Codec2 +pub struct Codec2RxAdapter { + state: Arc>, + output_card: Option, +} + +impl Codec2RxAdapter { + pub fn new() -> Self { + Self { + state: Arc::new(Mutex::new(AdapterState { + tx: None, + out_buf: VecDeque::new(), + codec2: Codec2::new(Codec2Mode::MODE_3200), + end_tx: None, + resampler: None, + })), + output_card: None, + } + } + + pub fn set_output_card>(&mut self, card_name: S) { + self.output_card = Some(card_name.into()); + } +} + +impl Default for Codec2RxAdapter { + fn default() -> Self { + Self::new() + } +} + +struct AdapterState { + tx: Option, + /// Circular buffer of output samples for playback + out_buf: VecDeque, + codec2: Codec2, + end_tx: Option>, + resampler: Option>, +} + +impl StreamAdapter for Codec2RxAdapter { + fn start(&self, handle: TxHandle) -> Result<(), AdapterError> { + self.state.lock().unwrap().tx = Some(handle); + + let (end_tx, end_rx) = channel(); + let (setup_tx, setup_rx) = channel(); + let state = self.state.clone(); + let output_card = self.output_card.clone(); + std::thread::spawn(move || stream_thread(end_rx, setup_tx, state, output_card)); + self.state.lock().unwrap().end_tx = Some(end_tx); + // Propagate any errors arising in the thread + let sample_rate = setup_rx.recv()??; + debug!("selected codec2 output sample rate {sample_rate}"); + if sample_rate != 8000 { + let params = SincInterpolationParameters { + sinc_len: 256, + f_cutoff: 0.95, + oversampling_factor: 128, + interpolation: rubato::SincInterpolationType::Cubic, + window: rubato::WindowFunction::BlackmanHarris2, + }; + // TODO: fix unwrap + self.state.lock().unwrap().resampler = + Some(SincFixedIn::new(sample_rate as f64 / 8000f64, 1.0, params, 160, 1).unwrap()); + } + Ok(()) + } + + fn close(&self) -> Result<(), AdapterError> { + let mut state = self.state.lock().unwrap(); + state.tx = None; + state.end_tx = None; + Ok(()) + } + + fn stream_began(&self, _link_setup: LinkSetup) { + // for now we will assume: + // - unencrypted + // - data type is Voice (Codec2 3200), not Voice+Data + // TODO: is encryption handled here or in M17App, such that we get a decrypted stream? + // TODO: handle the Voice+Data combination with Codec2 1600 + self.state.lock().unwrap().codec2 = Codec2::new(Codec2Mode::MODE_3200); + } + + fn stream_data(&self, _frame_number: u16, _is_final: bool, data: Arc<[u8; 16]>) { + let mut state = self.state.lock().unwrap(); + for encoded in data.chunks(8) { + if state.out_buf.len() < 8192 { + let mut samples = [i16::EQUILIBRIUM; 160]; // while assuming 3200 + state.codec2.decode(&mut samples, encoded); + if let Some(resampler) = state.resampler.as_mut() { + let samples_f: Vec = + samples.iter().map(|s| *s as f32 / 16384.0f32).collect(); + let res = resampler.process(&[samples_f], None).unwrap(); + for s in &res[0] { + state.out_buf.push_back((s * 16383.0f32) as i16); + } + } else { + // TODO: maybe get rid of VecDeque so we can decode directly into ring buffer? + for s in samples { + state.out_buf.push_back(s); + } + } + } else { + debug!("out_buf overflow"); + } + } + } +} + +fn output_cb(data: &mut [i16], state: &Mutex, channels: u16) { + let mut state = state.lock().unwrap(); + for d in data.chunks_mut(channels as usize) { + d.fill(state.out_buf.pop_front().unwrap_or(i16::EQUILIBRIUM)); + } +} + +/// Create and manage the stream from a dedicated thread since it's `!Send` +fn stream_thread( + end: Receiver<()>, + setup_tx: Sender>, + state: Arc>, + output_card: Option, +) { + let host = cpal::default_host(); + let device = if let Some(output_card) = output_card { + // TODO: more error handling for unwraps + match host + .output_devices() + .unwrap() + .find(|d| d.name().unwrap() == output_card) + { + Some(d) => d, + None => { + let _ = setup_tx.send(Err(M17Codec2Error::CardUnavailable(output_card).into())); + return; + } + } + } else { + match host.default_output_device() { + Some(d) => d, + None => { + let _ = setup_tx.send(Err(M17Codec2Error::DefaultCardUnavailable.into())); + return; + } + } + }; + let card_name = device.name().unwrap(); + let mut configs = match device.supported_output_configs() { + Ok(c) => c, + Err(e) => { + let _ = setup_tx.send(Err( + M17Codec2Error::OutputConfigsUnavailable(card_name, e).into() + )); + return; + } + }; + let config = match configs.find(|c| { + (c.channels() == 1 || c.channels() == 2) && c.sample_format() == SampleFormat::I16 + }) { + Some(c) => c, + None => { + let _ = setup_tx.send(Err( + M17Codec2Error::SupportedOutputUnavailable(card_name).into() + )); + return; + } + }; + + let target_sample_rate = + if config.min_sample_rate().0 <= 8000 && config.max_sample_rate().0 >= 8000 { + 8000 + } else { + config.min_sample_rate().0 + }; + let channels = config.channels(); + + let config = config.with_sample_rate(SampleRate(target_sample_rate)); + let stream = match device.build_output_stream( + &config.into(), + move |data: &mut [i16], _info: &cpal::OutputCallbackInfo| { + output_cb(data, &state, channels); + }, + |e| { + // trigger end_tx here? always more edge cases + debug!("error occurred in codec2 playback: {e:?}"); + }, + None, + ) { + Ok(s) => s, + Err(e) => { + let _ = setup_tx.send(Err( + M17Codec2Error::OutputStreamBuildError(card_name, e).into() + )); + return; + } + }; + match stream.play() { + Ok(()) => (), + Err(e) => { + let _ = setup_tx.send(Err( + M17Codec2Error::OutputStreamPlayError(card_name, e).into() + )); + return; + } + } + let _ = setup_tx.send(Ok(target_sample_rate)); + let _ = end.recv(); + // it seems concrete impls of Stream have a Drop implementation that will handle termination +} diff --git a/m17codec2/src/tx.rs b/m17codec2/src/tx.rs new file mode 100644 index 0000000..b824268 --- /dev/null +++ b/m17codec2/src/tx.rs @@ -0,0 +1,76 @@ +use codec2::{Codec2, Codec2Mode}; +use m17app::app::TxHandle; +use m17app::link_setup::LinkSetup; +use m17app::link_setup::M17Address; +use m17app::StreamFrame; +use std::path::PathBuf; +use std::time::Duration; +use std::time::Instant; + +/// Transmits a wave file as an M17 stream +pub struct WavePlayer; + +impl WavePlayer { + /// Plays a wave file (blocking). + /// + /// * `path`: wave file to transmit, must be 8 kHz mono and 16-bit LE + /// * `tx`: a `TxHandle` obtained from an `M17App` + /// * `source`: address of transmission source + /// * `destination`: address of transmission destination + /// * `channel_access_number`: from 0 to 15, usually 0 + pub fn play( + path: PathBuf, + tx: TxHandle, + source: &M17Address, + destination: &M17Address, + channel_access_number: u8, + ) { + let mut reader = hound::WavReader::open(path).unwrap(); + let mut samples = reader.samples::(); + + let mut codec = Codec2::new(Codec2Mode::MODE_3200); + let mut in_buf = [0i16; 160]; + let mut out_buf = [0u8; 16]; + let mut lsf_chunk: usize = 0; + const TICK: Duration = Duration::from_millis(40); + let mut next_tick = Instant::now() + TICK; + let mut frame_number = 0; + + let mut setup = LinkSetup::new_voice(source, destination); + setup.set_channel_access_number(channel_access_number); + tx.transmit_stream_start(&setup); + + loop { + let mut last_one = false; + for out in out_buf.chunks_mut(8) { + for i in in_buf.iter_mut() { + let sample = match samples.next() { + Some(Ok(sample)) => sample, + _ => { + last_one = true; + 0 + } + }; + *i = sample; + } + codec.encode(out, &in_buf); + } + tx.transmit_stream_next(&StreamFrame { + lich_idx: lsf_chunk as u8, + lich_part: setup.lich_part(lsf_chunk as u8), + frame_number, + end_of_stream: last_one, + stream_data: out_buf, + }); + frame_number += 1; + lsf_chunk = (lsf_chunk + 1) % 6; + + if last_one { + break; + } + + std::thread::sleep(next_tick.duration_since(Instant::now())); + next_tick += TICK; + } + } +}