nordic-dev.net/hydrophonitor
nordic-dev.net/hydrophonitor
2
1
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Improve code clarity and save configuration for when writer is reset

Browse Source
This commit is contained in:
Julius Koskela 2022-09-22 17:25:10 +03:00
parent 3524e86bd1
commit fe14cef7b8
2 changed files with 162 additions and 132 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

74
audio-logger/src/main.rs
View File

@ -10,7 +10,7 @@ use recorder::{batch_recording, contiguous_recording};
use cli::*;
use std::path::Path;
use print_configs::*;
use std::sync::{Arc, Mutex, Condvar, atomic::{AtomicBool, Ordering}};
use std::sync::{Condvar, atomic::{AtomicBool, Ordering}};
const DEFAULT_SAMPLE_RATE: u32 = 44100;
const DEFAULT_CHANNEL_COUNT: u16 = 1;
@ -19,59 +19,6 @@ const ALLOWED_SAMPLE_RATES: &[u32] = &[44100, 48000, 88200, 96000, 176400, 19200
const MAX_CHANNEL_COUNT: u16 = 2;
const MIN_BUFFER_SIZE: usize = 64;
const MAX_BUFFER_SIZE: usize = 8192;
const FMT_TIME: &str = "%Y-%m-%d-%H:%M:%S.%f";
type StreamInterrupt = Arc<(Mutex<bool>, Condvar)>;
type BatchInterrupt= Arc<AtomicBool>;
/// # Interrupts Handling
///
/// The `Recorder` struct has two interrupt mechanisms:
///
/// 1. `stream_interrupt` is used to interrupt the stream when the user presses `ctrl+c`.
/// 2. `batch_interrupt` is used to interrupt the batch recording when the user presses `ctrl+c`.
#[derive(Clone)]
pub struct InterruptHandles {
batch_interrupt: BatchInterrupt,
stream_interrupt: StreamInterrupt,
}
impl InterruptHandles {
pub fn new() -> Result<Self, anyhow::Error> {
let stream_interrupt = Arc::new((Mutex::new(false), Condvar::new()));
let stream_interrupt_cloned = stream_interrupt.clone();
let batch_interrupt = Arc::new(AtomicBool::new(false));
let batch_interrupt_cloned = batch_interrupt.clone();
ctrlc::set_handler(move || {
// Set batch interrupt to true
batch_interrupt_cloned.store(true, Ordering::SeqCst);
// Release the stream
let &(ref lock, ref cvar) = &*stream_interrupt_cloned;
let mut started = lock.lock().unwrap();
*started = true;
cvar.notify_one();
})?;
Ok(Self {
batch_interrupt,
stream_interrupt,
})
}
pub fn stream_wait(&self) {
let &(ref lock, ref cvar) = &*self.stream_interrupt;
let mut started = lock.lock().unwrap();
while !*started {
started = cvar.wait(started).unwrap();
}
}
pub fn batch_is_running(&self) -> bool {
!self.batch_interrupt.load(Ordering::SeqCst)
}
}
fn main() -> Result<(), anyhow::Error> {
let args = Args::parse();
@ -81,14 +28,27 @@ fn main() -> Result<(), anyhow::Error> {
return Ok(());
}
let interrupt_handles = InterruptHandles::new()?;
let mut recorder = recorder::Recorder::init(
args.name.clone(),
match args.output.clone() {
Some(path) => path,
None => Path::new("./").to_path_buf(),
},
match args.host {
Hosts::Alsa => cpal::HostId::Alsa,
Hosts::Jack => cpal::HostId::Jack,
},
args.sample_rate.unwrap_or(DEFAULT_SAMPLE_RATE),
args.channels.unwrap_or(DEFAULT_CHANNEL_COUNT),
args.buffer_size.unwrap_or(DEFAULT_BUFFER_SIZE),
)?;
match args.batch_recording {
Some(secs) => {
batch_recording(&args, secs, interrupt_handles)?;
batch_recording(&mut recorder, secs)?;
},
None => {
contiguous_recording(&args, interrupt_handles)?;
contiguous_recording(&mut recorder)?;
}
}

220
audio-logger/src/recorder.rs
View File

@ -12,21 +12,42 @@ type WriteHandle = Arc<Mutex<Option<hound::WavWriter<BufWriter<File>>>>>;
pub struct Recorder {
writer: WriteHandle,
interrupt: InterruptHandles,
interrupt_handles: InterruptHandles,
default_config: SupportedStreamConfig,
user_config: StreamConfig,
device: Device,
filename: String,
spec: hound::WavSpec,
name: String,
path: PathBuf,
current_file: String,
}
/// # Stream User Config
fn get_host(host: HostId) -> Result<Host, anyhow::Error> {
Ok(cpal::host_from_id(cpal::available_hosts()
.into_iter()
.find(|id| *id == host)
.ok_or(anyhow!("Requested host device not found"))?
)?)
}
fn get_device(host: Host) -> Result<Device, anyhow::Error> {
Ok(host.default_input_device()
.ok_or(anyhow!("No input device available. Try running `jackd -R -d alsa -d hw:0`",
))?)
}
fn get_default_config(device: &Device) -> Result<SupportedStreamConfig, anyhow::Error> {
Ok(device.default_input_config()?)
}
/// # Get User Config
///
/// Overrides certain fields of the default stream config with the user's config.
///
/// sample_rate: The user's sample rate if it is supported by the device, otherwise the default sample rate.
/// channels: The user's number of channels if it is supported by the device, otherwise the default number of channels.
/// buffer_size: The user's buffer size if it is supported by the device, otherwise the default buffer size.
fn stream_user_config(sample_rate: u32, channels: u16, buffer_size: u32) -> Result<StreamConfig, anyhow::Error> {
fn get_user_config(sample_rate: u32, channels: u16, buffer_size: u32) -> Result<StreamConfig, anyhow::Error> {
if !ALLOWED_SAMPLE_RATES.contains(&sample_rate) {
return Err(anyhow!(
"Sample rate {} is not supported. Allowed sample rates: {:?}",
@ -56,6 +77,34 @@ fn stream_user_config(sample_rate: u32, channels: u16, buffer_size: u32) -> Resu
})
}
fn get_spec(default_config: &SupportedStreamConfig, user_config: &StreamConfig) -> Result<hound::WavSpec, anyhow::Error> {
Ok(hound::WavSpec {
channels: user_config.channels,
sample_rate: user_config.sample_rate.0,
bits_per_sample: (default_config.sample_format().sample_size() * 8) as u16,
sample_format: match default_config.sample_format() {
cpal::SampleFormat::U16 => hound::SampleFormat::Int,
cpal::SampleFormat::I16 => hound::SampleFormat::Int,
cpal::SampleFormat::F32 => hound::SampleFormat::Float,
},
})
}
fn get_filename(name: &str, path: &PathBuf) -> Result<String, anyhow::Error> {
let now: DateTime<Local> = Local::now();
let filename = format!(
"{}-{}-{}-{}-{}:{}:{}.wav",
name,
now.year(),
now.month(),
now.day(),
now.hour(),
now.minute(),
now.second(),
);
Ok(path.join(filename).to_str().unwrap().to_string())
}
/// # Recorder
///
/// The `Recorder` struct is used to record audio.
@ -69,53 +118,50 @@ impl Recorder {
sample_rate: u32,
channels: u16,
buffer_size: u32,
interrupt: InterruptHandles,
) -> Result<Self, anyhow::Error> {
// Select requested host
let host = cpal::host_from_id(cpal::available_hosts()
.into_iter()
.find(|id| *id == host)
.ok_or(anyhow!("Requested host device not found"))?
)?;
// Create interrupt handles to be used by the stream or batch loop.
let interrupt_handles = InterruptHandles::new()?;
// Select requested host.
let host = get_host(host)?;
// Set up the input device and stream with the default input config.
let device = host.default_input_device()
.ok_or(anyhow!("No input device available. Try running `jackd -R -d alsa -d hw:0`",
))?;
let device = get_device(host)?;
let default_config = device.default_input_config()?;
let user_config = stream_user_config(sample_rate, channels, buffer_size)?;
// Get default config for the device.
let default_config = get_default_config(&device)?;
let spec = hound::WavSpec {
channels: user_config.channels as _,
sample_rate: user_config.sample_rate.0 as _,
bits_per_sample: (default_config.sample_format().sample_size() * 8) as _,
sample_format: match default_config.sample_format() {
cpal::SampleFormat::U16 => hound::SampleFormat::Int,
cpal::SampleFormat::I16 => hound::SampleFormat::Int,
cpal::SampleFormat::F32 => hound::SampleFormat::Float,
},
};
// Override certain fields of the default stream config with the user's config.
let user_config = get_user_config(sample_rate, channels, buffer_size)?;
// The WAV file we're recording to.
let ts: String = Utc::now().format(FMT_TIME).to_string();
let filename: String = path.to_str().unwrap().to_owned() + &name + "-" + &ts + ".wav";
// Get the hound WAV spec for the user's config.
let spec = get_spec(&default_config, &user_config)?;
Ok(Self {
writer: Arc::new(Mutex::new(Some(hound::WavWriter::create(filename.clone(), spec)?))),
interrupt,
writer: Arc::new(Mutex::new(None)),
interrupt_handles,
default_config,
user_config,
device,
filename,
spec,
name,
path,
current_file: "".to_string(),
})
}
fn init_writer(&mut self) -> Result<(), anyhow::Error> {
let filename = get_filename(&self.name, &self.path)?;
self.current_file = filename.clone();
*self.writer.lock().unwrap() = Some(hound::WavWriter::create(filename, self.spec)?);
Ok(())
}
fn create_stream(&self) -> Result<Stream, anyhow::Error> {
let writer = self.writer.clone();
let config = self.user_config.clone();
let err_fn = |err| { eprintln!("an error occurred on stream: {}", err); };
let err_fn = |err| { eprintln!("An error occurred on stream: {}", err); };
let stream = match self.default_config.sample_format() {
cpal::SampleFormat::F32 => self.device.build_input_stream(
@ -137,21 +183,23 @@ impl Recorder {
Ok(stream)
}
pub fn record(&self) -> Result<(), anyhow::Error> {
pub fn record(&mut self) -> Result<(), anyhow::Error> {
self.init_writer()?;
let stream = self.create_stream()?;
stream.play()?;
println!("REC: {}", self.filename);
self.interrupt.stream_wait();
println!("REC: {}", self.current_file);
self.interrupt_handles.stream_wait();
drop(stream);
self.writer.lock().unwrap().take().unwrap().finalize()?;
println!("STOP: {}", self.filename);
println!("STOP: {}", self.current_file);
Ok(())
}
pub fn record_secs(&self, secs: u64) -> Result<(), anyhow::Error> {
pub fn record_secs(&mut self, secs: u64) -> Result<(), anyhow::Error> {
self.init_writer()?;
let stream = self.create_stream()?;
stream.play()?;
println!("REC: {}", self.filename);
println!("REC: {}", self.current_file);
let now = std::time::Instant::now();
loop {
std::thread::sleep(std::time::Duration::from_millis(500));
@ -161,7 +209,7 @@ impl Recorder {
}
drop(stream);
self.writer.lock().unwrap().take().unwrap().finalize()?;
println!("STOP: {}", self.filename);
println!("STOP: {}", self.current_file);
Ok(())
}
}
@ -181,44 +229,66 @@ where
}
}
pub fn batch_recording(args: &Args, secs: u64, interrupt_handles: InterruptHandles) -> Result<(), anyhow::Error> {
while interrupt_handles.batch_is_running() {
let recorder = recorder::Recorder::init(
args.name.clone(),
match args.output.clone() {
Some(path) => path,
None => Path::new("./").to_path_buf(),
},
match args.host {
Hosts::Alsa => cpal::HostId::Alsa,
Hosts::Jack => cpal::HostId::Jack,
},
args.sample_rate.unwrap_or(DEFAULT_SAMPLE_RATE),
args.channels.unwrap_or(DEFAULT_CHANNEL_COUNT),
args.buffer_size.unwrap_or(DEFAULT_BUFFER_SIZE),
interrupt_handles.clone(),
)?;
recorder.record_secs(secs)?;
pub fn batch_recording(rec: &mut Recorder, secs: u64) -> Result<(), anyhow::Error> {
while rec.interrupt_handles.batch_is_running() {
rec.record_secs(secs)?;
}
Ok(())
}
pub fn contiguous_recording(args: &Args, interrupt_handles: InterruptHandles) -> Result<(), anyhow::Error> {
let recorder = recorder::Recorder::init(
args.name.clone(),
match args.output.clone() {
Some(path) => path,
None => Path::new("./").to_path_buf(),
},
match args.host {
Hosts::Alsa => cpal::HostId::Alsa,
Hosts::Jack => cpal::HostId::Jack,
},
args.sample_rate.unwrap_or(DEFAULT_SAMPLE_RATE),
args.channels.unwrap_or(DEFAULT_CHANNEL_COUNT),
args.buffer_size.unwrap_or(DEFAULT_BUFFER_SIZE),
interrupt_handles.clone(),
)?;
recorder.record()?;
pub fn contiguous_recording(rec: &mut Recorder) -> Result<(), anyhow::Error> {
rec.record()?;
Ok(())
}
type StreamInterrupt = Arc<(Mutex<bool>, Condvar)>;
type BatchInterrupt= Arc<AtomicBool>;
/// # Interrupts Handling
///
/// The `Recorder` struct has two interrupt mechanisms:
///
/// 1. `stream_interrupt` is used to interrupt the stream when the user presses `ctrl+c`.
/// 2. `batch_interrupt` is used to interrupt the batch recording when the user presses `ctrl+c`.
#[derive(Clone)]
pub struct InterruptHandles {
batch_interrupt: BatchInterrupt,
stream_interrupt: StreamInterrupt,
}
impl InterruptHandles {
pub fn new() -> Result<Self, anyhow::Error> {
let stream_interrupt = Arc::new((Mutex::new(false), Condvar::new()));
let stream_interrupt_cloned = stream_interrupt.clone();
let batch_interrupt = Arc::new(AtomicBool::new(false));
let batch_interrupt_cloned = batch_interrupt.clone();
ctrlc::set_handler(move || {
// Set batch interrupt to true
batch_interrupt_cloned.store(true, Ordering::SeqCst);
// Release the stream
let &(ref lock, ref cvar) = &*stream_interrupt_cloned;
let mut started = lock.lock().unwrap();
*started = true;
cvar.notify_one();
})?;
Ok(Self {
batch_interrupt,
stream_interrupt,
})
}
pub fn stream_wait(&self) {
let &(ref lock, ref cvar) = &*self.stream_interrupt;
let mut started = lock.lock().unwrap();
while !*started {
started = cvar.wait(started).unwrap();
}
}
pub fn batch_is_running(&self) -> bool {
!self.batch_interrupt.load(Ordering::SeqCst)
}
}