From 7cbff539df1bd9a622aafdd38316b915d79d5a02 Mon Sep 17 00:00:00 2001 From: Victoria Fischer Date: Mon, 22 Jun 2026 10:23:24 +0200 Subject: [PATCH] src: drop a lot of unwraps() --- src/artifacts/archive.rs | 1 - src/artifacts/bandcamp.rs | 6 +- src/artifacts/beets.rs | 52 +++++++------- src/audio.rs | 144 ++++++++++++++++++++++++++------------ src/main.rs | 4 +- src/transcription.rs | 14 ++-- src/tts.rs | 4 +- src/ui.rs | 4 +- 8 files changed, 142 insertions(+), 87 deletions(-) diff --git a/src/artifacts/archive.rs b/src/artifacts/archive.rs index 8cfd4f1..96c1a5c 100644 --- a/src/artifacts/archive.rs +++ b/src/artifacts/archive.rs @@ -12,7 +12,6 @@ pub struct ArtifactRef<'a> { archive: &'a Archive } - pub struct ArtifactRefMut<'a> { id: Uuid, archive: &'a mut Archive diff --git a/src/artifacts/bandcamp.rs b/src/artifacts/bandcamp.rs index d27b87a..a034b41 100644 --- a/src/artifacts/bandcamp.rs +++ b/src/artifacts/bandcamp.rs @@ -32,7 +32,7 @@ pub struct BandcampSource; impl DataSource for BandcampSource { type Args = BandcampQueryArgs; - type Error = (); + type Error = bandcamp::Error; async fn synchronize(&self, _artifact: &mut Artifact) -> Result, Self::Error> { todo!() @@ -47,11 +47,11 @@ impl DataSource for BandcampSource { match result { SearchResultItem::Artist(data) => { // TODO: The artist and album detailed fetchers should also be separate args - let result = bandcamp::fetch_artist(data.artist_id).await.unwrap().into(); + let result = bandcamp::fetch_artist(data.artist_id).await?.into(); json_results.push(result); }, SearchResultItem::Album(data) => { - let result = bandcamp::fetch_album(data.band_id, data.album_id).await.unwrap().into(); + let result = bandcamp::fetch_album(data.band_id, data.album_id).await?.into(); json_results.push(result); }, SearchResultItem::Track(data) => { diff --git a/src/artifacts/beets.rs b/src/artifacts/beets.rs index be91b91..4d4319c 100644 --- a/src/artifacts/beets.rs +++ b/src/artifacts/beets.rs @@ -56,10 +56,30 @@ impl Into for BeetsTrack { } } +#[derive(Debug)] +#[allow(unused)] +pub enum BeetsError { + Command(std::io::Error), + Json(serde_json::Error), + EmptyQuery +} + +impl From for BeetsError { + fn from(value: std::io::Error) -> Self { + Self::Command(value) + } +} + +impl From for BeetsError { + fn from(value: serde_json::Error) -> Self { + Self::Json(value) + } +} + pub struct BeetsDB; impl BeetsDB { - async fn query_multi(&self, args: &BeatsQueryMultiArgs) -> Result, ()> { + async fn query_multi(&self, args: &BeatsQueryMultiArgs) -> Result, BeetsError> { let mut ret = vec![]; for arg in &args.args { for artifact in self.query_single(arg).await.unwrap_or_default() { @@ -69,7 +89,7 @@ impl BeetsDB { Ok(ret) } - async fn query_single(&self, args: &BeatsQueryArgs) -> Result, ()> { + async fn query_single(&self, args: &BeatsQueryArgs) -> Result, BeetsError> { let mut beets_cmd = Command::new("beet"); beets_cmd.args(["export", "-f", "json", "-i", "title,label,year,genres,album,artist,mb_trackid"]); let mut valid = false; @@ -100,29 +120,20 @@ impl BeetsDB { if !valid { log::warn!("Tried to execute an empty beets query"); - return Err(()) + return Err(BeetsError::EmptyQuery) } log::debug!("Executing beets: {:?}", beets_cmd); - if let Ok(output) = beets_cmd.stdout(Stdio::piped()).stderr(Stdio::null()).spawn().unwrap().wait_with_output().await { - match serde_json::from_str::>(str::from_utf8(&output.stdout).unwrap()) { - Ok(track) => Ok(track.into_iter().map(|t| { t.into()}).collect()), - Err(err) => { - log::error!("Failed to decode beets json: {:?}", err); - Err(()) - } - } - } else { - log::error!("Unable to execute query!"); - Err(()) - } + let output = beets_cmd.stdout(Stdio::piped()).stderr(Stdio::null()).spawn()?.wait_with_output().await?; + let track = serde_json::from_str::>(str::from_utf8(&output.stdout).unwrap())?; + Ok(track.into_iter().map(|t| { t.into()}).collect()) } } impl DataSource for BeetsDB { type Args = BeatsQueryMultiArgs; - type Error = (); + type Error = BeetsError; async fn synchronize(&self, artifact: &mut Artifact) -> Result, Self::Error> { match artifact.contents { @@ -134,7 +145,7 @@ impl DataSource for BeetsDB { ..Default::default() }; - let results = self.query(&BeatsQueryMultiArgs { args: vec![args] }).await.unwrap(); + let results = self.query(&BeatsQueryMultiArgs { args: vec![args] }).await?; if let Some(first) = results.first() { artifact.merge(first.clone()); @@ -150,13 +161,6 @@ impl DataSource for BeetsDB { } fn query(&self, args: &Self::Args) -> impl Future, Self::Error>> { - /*let mut ret = vec![]; - for arg in args.0 { - for artifact in self.query_single(&arg).await.unwrap_or_default() { - ret.push(artifact); - } - } - futures::ready!(Ok(ret))*/ self.query_multi(args) } diff --git a/src/audio.rs b/src/audio.rs index 84fd212..1b889e3 100644 --- a/src/audio.rs +++ b/src/audio.rs @@ -5,6 +5,41 @@ use oximedia_metering::vu_meter::VuMeter; use serde::{Deserialize, Serialize}; use tokio::sync::*; +use crate::events::AudioRecordRequest; + +#[derive(Debug)] +#[allow(unused)] +pub enum AudioError { + Jack(jack::Error), + AudioBufferSend(mpsc::error::SendError>), + AudioBufferRecv(mpsc::error::TryRecvError), + AudioRequestSend(watch::error::SendError) +} + +impl From for AudioError { + fn from(value: jack::Error) -> Self { + Self::Jack(value) + } +} + +impl From>> for AudioError { + fn from(value: mpsc::error::SendError>) -> Self { + Self::AudioBufferSend(value) + } +} + +impl From for AudioError { + fn from(value: mpsc::error::TryRecvError) -> Self { + Self::AudioBufferRecv(value) + } +} + +impl From> for AudioError { + fn from(value: watch::error::SendError) -> Self { + Self::AudioRequestSend(value) + } +} + #[derive(Debug)] pub struct JackClientRef { killswitch: Option> @@ -12,7 +47,7 @@ pub struct JackClientRef { impl Drop for JackClientRef { fn drop(&mut self) { - self.killswitch.take().unwrap().send(()).unwrap(); + self.killswitch.take().expect("Killswitch was already dropped!").send(()).expect("Cannot fire Jack killswitch"); } } @@ -23,9 +58,9 @@ pub struct AudioInputControl { } impl AudioInputControl { - pub async fn next(&mut self) -> f64 { - self.volume_src.changed().await.unwrap(); - *self.volume_src.borrow_and_update() + pub async fn next(&mut self) -> Result { + self.volume_src.changed().await?; + Ok(*self.volume_src.borrow_and_update()) } } @@ -76,28 +111,28 @@ struct AudioSource { } impl AudioSource { - fn new(client: &jack::Client, name: &str) -> (Self, AudioInStream) { + fn new(client: &jack::Client, name: &str) -> Result<(Self, AudioInStream), AudioError> { let (sample_sink, receiver) = mpsc::channel(32); - let port = client.register_port(name, AudioIn::default()).unwrap(); - (AudioSource { + let port = client.register_port(name, AudioIn::default())?; + Ok((AudioSource { port, sample_sink, meter: VuMeter::new(client.sample_rate().into(), 1, None) }, AudioInStream { sample_rate: client.sample_rate(), src: receiver - }) + })) } - fn process(&mut self, scope: &ProcessScope) -> Option { - if self.port.connected_count().unwrap() > 0 { + fn process(&mut self, scope: &ProcessScope) -> Result, AudioError> { + if self.port.connected_count()? > 0 { let buf: Vec<_> = self.port.as_slice(scope).iter().copied().collect(); self.meter.process_interleaved(&buf); - self.sample_sink.blocking_send(buf).unwrap(); + self.sample_sink.blocking_send(buf)?; - self.meter.channel_vu(0) + Ok(self.meter.channel_vu(0)) } else { - None + Ok(None) } } } @@ -110,25 +145,24 @@ struct AudioSink { } impl AudioSink { - fn new(client: &jack::Client, name: &str) -> (Self, AudioOutStream) { + fn new(client: &jack::Client, name: &str) -> Result<(Self, AudioOutStream), AudioError> { let (sender, sample_src) = mpsc::channel(32); - let port = client.register_port(name, AudioOut::default()).unwrap(); - (AudioSink { + let port = client.register_port(name, AudioOut::default())?; + Ok((AudioSink { output_buf: Vec::with_capacity(1024), port, sample_src }, AudioOutStream { sample_rate: client.sample_rate(), sink: sender, - }) + })) } - fn process(&mut self, scope: &ProcessScope) { - if let Ok(mut next_outbuf) = self.sample_src.try_recv() { - self.output_buf.append(&mut next_outbuf); - } + fn process(&mut self, scope: &ProcessScope) -> Result<(), AudioError> { + let mut next_outbuf = self.sample_src.try_recv()?; + self.output_buf.append(&mut next_outbuf); - if self.port.connected_count().unwrap() > 0 && !self.output_buf.is_empty() { + if self.port.connected_count()? > 0 && !self.output_buf.is_empty() { let outbuf = self.port.as_mut_slice(scope); let mut next_segment: Vec = self.output_buf.drain(0..(outbuf.len()).min(self.output_buf.len())).collect(); let underrun = outbuf.len() - next_segment.len(); @@ -140,6 +174,8 @@ impl AudioSink { outbuf.copy_from_slice(&next_segment); } + + Ok(()) } } @@ -151,7 +187,13 @@ struct AudioConfig { impl AudioConfig { pub fn load() -> Self { if let Ok(contents) = std::fs::read_to_string("audio.json") { - serde_json::from_str(contents.as_str()).unwrap() + match serde_json::from_str(contents.as_str()) { + Err(err) => { + log::error!("Failed to load audio.json: {:?}", err); + Default::default() + }, + Ok(ret) => ret + } } else { Default::default() } @@ -186,10 +228,7 @@ impl NotificationHandler for Notify { }).next(); if let Some((role, Ok(target_port))) = port_match { - if !self.config.connections.contains_key(role) { - self.config.connections.insert(*role, Default::default()); - } - let cfg_slot = self.config.connections.get_mut(role).unwrap(); + let cfg_slot = self.config.connections.entry(*role).or_insert_with(|| Default::default()); if are_connected { log::info!("{} connected to {}", role, target_port); @@ -200,7 +239,9 @@ impl NotificationHandler for Notify { } let save_data = serde_json::to_string_pretty(&self.config).unwrap(); - std::fs::write("audio.json", save_data).unwrap(); + if let Err(err) = std::fs::write("audio.json", save_data) { + log::error!("Failed to write audio.json: {:?}", err); + } } } } @@ -213,11 +254,11 @@ pub async fn start_audio_input() -> (AudioInputControl, AudioInStream, AudioOutS let (volume_sink, volume_src) = watch::channel(0.); - let (client, _status) = jack::Client::new("Eva-Cohost", ClientOptions::default() | ClientOptions::SESSION_ID).unwrap(); + let (client, _status) = jack::Client::new("Eva-Cohost", ClientOptions::default() | ClientOptions::SESSION_ID).expect("Could not create JACK client!"); - let (mut tts_sink, tts_stream) = AudioSink::new(&client, "tts-out"); - let (mut sfx_sink, sfx_stream) = AudioSink::new(&client, "sfx-out"); - let (mut mic_src, mic_stream) = AudioSource::new(&client, "microphone-in"); + let (mut tts_sink, tts_stream) = AudioSink::new(&client, "tts-out").expect("Could not create TTS sink"); + let (mut sfx_sink, sfx_stream) = AudioSink::new(&client, "sfx-out").expect("Could not create SFX sink"); + let (mut mic_src, mic_stream) = AudioSource::new(&client, "microphone-in").expect("Could not create microphone source"); let notifier = Notify { config, @@ -248,30 +289,41 @@ pub async fn start_audio_input() -> (AudioInputControl, AudioInStream, AudioOutS } let handler = jack::contrib::ClosureProcessHandler::new(move |_client, scope| { - if let Some(next_vu) = mic_src.process(scope) { - volume_sink.send_if_modified(|v| { - let next_vu = (next_vu * 100.0).round() / 100.0; - if *v != next_vu { - *v = next_vu; - true - } else { - false - } - }); + match mic_src.process(scope) { + Ok(Some(next_vu)) => { + volume_sink.send_if_modified(|v| { + let next_vu = (next_vu * 100.0).round() / 100.0; + if *v != next_vu { + *v = next_vu; + true + } else { + false + } + }); + }, + Ok(None) => (), + Err(err) => { + log::error!("Error while processing mic source: {:?}", err); + return jack::Control::Quit + } } for sink in [&mut tts_sink, &mut sfx_sink] { - sink.process(scope); + if let Err(err) = sink.process(scope) { + log::error!("Error while processing {:?} audio sink: {:?}", sink, err); + } } jack::Control::Continue }); tokio::spawn(async move { - let async_client = client.activate_async(notifier, handler).unwrap(); + let async_client = client.activate_async(notifier, handler).expect("Unable to start jack client!"); - exit_rx.await.unwrap(); + if let Err(err) = exit_rx.await { + log::warn!("Premature killswitch triggered: {:?}", err); + } - async_client.deactivate().unwrap(); + async_client.deactivate().expect("Unable to stop the jack client"); }); (AudioInputControl { diff --git a/src/main.rs b/src/main.rs index 8009d17..46262aa 100644 --- a/src/main.rs +++ b/src/main.rs @@ -61,9 +61,9 @@ pub struct SaveData { } impl SaveData { - fn save(&self) { + fn save(&self) -> Result<(), std::io::Error> { let save_data = serde_json::to_string_pretty(self).unwrap(); - std::fs::write("save.json", save_data).unwrap(); + std::fs::write("save.json", save_data) } } diff --git a/src/transcription.rs b/src/transcription.rs index 6d48508..85dbb17 100644 --- a/src/transcription.rs +++ b/src/transcription.rs @@ -4,7 +4,7 @@ use async_openai::{Client, config::OpenAIConfig, types::{InputSource, audio::{Au use tempfile::SpooledData; use tokio::sync::{mpsc, watch}; -use crate::{audio::AudioInStream, events::AudioRecordRequest}; +use crate::{audio::{AudioError, AudioInStream}, events::AudioRecordRequest}; #[derive(Debug)] pub struct TranscriptionControl { @@ -13,16 +13,16 @@ pub struct TranscriptionControl { } impl TranscriptionControl { - pub fn start(&mut self) { - self.record_state_sink.send(AudioRecordRequest::Start).unwrap(); + pub fn start(&mut self) -> Result<(), AudioError> { + Ok(self.record_state_sink.send(AudioRecordRequest::Start)?) } - pub fn stop(&mut self) { - self.record_state_sink.send(AudioRecordRequest::Finish).unwrap(); + pub fn stop(&mut self) -> Result<(), AudioError> { + Ok(self.record_state_sink.send(AudioRecordRequest::Finish)?) } - pub async fn next(&mut self) -> String { - self.transcription_result_src.recv().await.unwrap() + pub async fn next(&mut self) -> Option { + self.transcription_result_src.recv().await } } diff --git a/src/tts.rs b/src/tts.rs index e6d67db..7729042 100644 --- a/src/tts.rs +++ b/src/tts.rs @@ -8,8 +8,8 @@ pub struct TtsControl { } impl TtsControl { - pub async fn speak(&self, text: String) { - self.request_sink.send(text).await.unwrap(); + pub async fn speak(&self, text: String) -> Result<(), tokio::sync::mpsc::error::SendError> { + self.request_sink.send(text).await } } diff --git a/src/ui.rs b/src/ui.rs index dc6877d..764e77b 100644 --- a/src/ui.rs +++ b/src/ui.rs @@ -299,10 +299,10 @@ impl Ui { } }, next_volume = self.audio.next() => { - self.audio_level = next_volume + self.audio_level = next_volume.expect("Audio volume event stream has died") }, transcription_result = self.transcription.next() => { - self.predictions.insert(PredictionAction::ConversationAppend(ConversationEntry::Spoken(Speaker::User, transcription_result))).await; + self.predictions.insert(PredictionAction::ConversationAppend(ConversationEntry::Spoken(Speaker::User, transcription_result.expect("Transcription stream has died")))).await; }, } }