diff --git a/src/app/mod.rs b/src/app/mod.rs index 339a632..4c9e10b 100644 --- a/src/app/mod.rs +++ b/src/app/mod.rs @@ -63,6 +63,8 @@ pub enum AppMessage { ToggleEchoCancellation(bool), CustomSoundPathChanged(Sound, String), ToggleMicTest(bool), + /// Start/stop recording the call; the core confirms via Recording{Started,Stopped}. + ToggleRecording, } fn core_subscription() -> impl iced::futures::Stream { @@ -103,6 +105,10 @@ pub struct AppState { locally_muted: HashSet, /// When we joined the current room, for the in-room call-duration timer. call_started: Option, + /// Whether a local call recording is in progress (confirmed by the core). + recording: bool, + /// When the current recording started, for the header REC timer. + recording_started: Option, /// Raw mic level (normalized RMS, `0.0..=1.0`) for the settings meter. mic_level: f32, /// Whether the standalone (off-call) mic test stream is running. @@ -174,6 +180,8 @@ impl Default for AppState { audio_levels: HashMap::new(), locally_muted: HashSet::new(), call_started: None, + recording: false, + recording_started: None, mic_level: 0.0, mic_test_active: false, connecting: HashSet::new(), @@ -310,6 +318,8 @@ fn update(state: &mut AppState, message: AppMessage) -> Task { state.audio_levels.clear(); state.locally_muted.clear(); state.call_started = None; + state.recording = false; + state.recording_started = None; state.connecting.clear(); state.ever_connected.clear(); state.status_message = "Ready to connect".to_string(); @@ -362,6 +372,16 @@ fn update(state: &mut AppState, message: AppMessage) -> Task { UiEvent::MicLevel(level) => { state.mic_level = level; } + UiEvent::RecordingStarted { path } => { + state.recording = true; + state.recording_started = Some(std::time::Instant::now()); + state.status_message = format!("Recording → {path}"); + } + UiEvent::RecordingStopped { path } => { + state.recording = false; + state.recording_started = None; + state.status_message = format!("Saved recording → {path}"); + } UiEvent::Error(err) => { state.status_message = format!("Error: {}", err); } @@ -454,6 +474,11 @@ fn update(state: &mut AppState, message: AppMessage) -> Task { Sound::ReconnectFailed => state.config.custom_sound_reconnect_failed = path_opt, } } + AppMessage::ToggleRecording => { + // Optimistic intent; the core flips `recording` for real via the + // Recording{Started,Stopped} events (so a failed start won't lie). + let _ = state.controller.send(CoreCommand::SetRecording(!state.recording)); + } AppMessage::ToggleMicTest(enabled) => { state.mic_test_active = enabled; if !enabled { @@ -897,6 +922,18 @@ fn view(state: &AppState) -> Element<'_, AppMessage> { text(format!("⏱ {}", format_duration(call_secs))) .size(14) .color(color_subtext), + if state.recording { + let rec_secs = state.recording_started.map(|t| t.elapsed().as_secs()).unwrap_or(0); + container( + text(format!("● REC {}", format_duration(rec_secs))) + .size(13) + .color(color_red) + ) + .style(c_style(color_crust, color_red, 6.0)) + .padding(6) + } else { + container(text("")).padding(0) + }, horizontal_space(), text(format!("My ID: {}", &state.self_id[..8])) .size(14) @@ -1091,6 +1128,23 @@ fn view(state: &AppState) -> Element<'_, AppMessage> { } else { column![] }, + vertical_space(20.0), + { + let (rec_label, rec_bg, rec_hover, rec_fg) = if state.recording { + ("⏹ Stop Recording", color_red, color_maroon, color_crust) + } else { + ("⏺ Record Call", color_surface, color_blue, color_text) + }; + button( + text(rec_label) + .size(16) + .align_x(iced::alignment::Horizontal::Center) + ) + .on_press(AppMessage::ToggleRecording) + .style(b_style(rec_bg, rec_hover, rec_fg, 8.0)) + .padding(14) + .width(iced::Length::Fill) + }, vertical_space(30.0), button( text("Leave Room") diff --git a/src/audio/mod.rs b/src/audio/mod.rs index 5074cf2..4e2a47d 100644 --- a/src/audio/mod.rs +++ b/src/audio/mod.rs @@ -56,3 +56,4 @@ pub mod gate; pub mod limiter; pub mod pipewire_impl; pub mod pw_cli; +pub mod recorder; diff --git a/src/audio/recorder.rs b/src/audio/recorder.rs new file mode 100644 index 0000000..bde1cec --- /dev/null +++ b/src/audio/recorder.rs @@ -0,0 +1,242 @@ +//! Local call recording to a mono 16-bit PCM WAV file. +//! +//! Records the **full call as you experienced it**: the mixed incoming audio +//! (everyone you hear) summed with your own transmitted mic, into a single mono +//! WAV. Writing is driven by the playout mixer (one [`Recorder::write_frame`] +//! per produced 20ms frame, paced by the hardware clock); your mic arrives +//! separately from the capture thread via [`Recorder::push_mic`] and is buffered +//! in a small FIFO so the two independently-clocked streams stay roughly aligned. +//! Minor clock drift just slowly grows/shrinks that FIFO (capped, so the lag +//! between your voice and the recording is bounded) — harmless for a voice +//! recording, no realtime crackle concern. +//! +//! No external crates: the WAV writer emits the 44-byte canonical header itself +//! and patches the two size fields on [`Recorder::finalize`]. + +use std::collections::VecDeque; +use std::fs::File; +use std::io::{self, Seek, SeekFrom, Write}; +use std::path::{Path, PathBuf}; + +/// Capture sample rate (mono, 48kHz, matching the rest of the audio path). +const SAMPLE_RATE: u32 = 48_000; +const BITS_PER_SAMPLE: u16 = 16; +const CHANNELS: u16 = 1; + +/// Cap on buffered mic samples (~200ms). Bounds how far recording lag can drift +/// if the capture clock runs persistently faster than playout — past this we drop +/// the oldest mic audio rather than let the offset grow without limit. +const MAX_MIC_FIFO: usize = SAMPLE_RATE as usize / 5; + +/// A minimal canonical PCM WAV writer (mono S16LE). Writes a placeholder header +/// up front, streams sample data, then patches the RIFF + data chunk sizes on +/// [`WavWriter::finalize`]. +pub struct WavWriter { + file: File, + /// Bytes of PCM data written so far (for the size fields). + data_bytes: u32, +} + +impl WavWriter { + /// Create the file and write the 44-byte header with zeroed size fields. + pub fn new(path: &Path) -> io::Result { + let mut file = File::create(path)?; + file.write_all(&Self::header(0))?; + Ok(Self { file, data_bytes: 0 }) + } + + /// The 44-byte canonical WAV/PCM header for the given data length in bytes. + fn header(data_bytes: u32) -> [u8; 44] { + let byte_rate = SAMPLE_RATE * CHANNELS as u32 * (BITS_PER_SAMPLE as u32 / 8); + let block_align = CHANNELS * (BITS_PER_SAMPLE / 8); + let mut h = [0u8; 44]; + h[0..4].copy_from_slice(b"RIFF"); + h[4..8].copy_from_slice(&(36 + data_bytes).to_le_bytes()); + h[8..12].copy_from_slice(b"WAVE"); + h[12..16].copy_from_slice(b"fmt "); + h[16..20].copy_from_slice(&16u32.to_le_bytes()); // fmt chunk size + h[20..22].copy_from_slice(&1u16.to_le_bytes()); // PCM + h[22..24].copy_from_slice(&CHANNELS.to_le_bytes()); + h[24..28].copy_from_slice(&SAMPLE_RATE.to_le_bytes()); + h[28..32].copy_from_slice(&byte_rate.to_le_bytes()); + h[32..34].copy_from_slice(&block_align.to_le_bytes()); + h[34..36].copy_from_slice(&BITS_PER_SAMPLE.to_le_bytes()); + h[36..40].copy_from_slice(b"data"); + h[40..44].copy_from_slice(&data_bytes.to_le_bytes()); + h + } + + /// Append PCM samples to the data chunk. + pub fn write_samples(&mut self, samples: &[i16]) -> io::Result<()> { + let mut buf = Vec::with_capacity(samples.len() * 2); + for &s in samples { + buf.extend_from_slice(&s.to_le_bytes()); + } + self.file.write_all(&buf)?; + self.data_bytes += (samples.len() * 2) as u32; + Ok(()) + } + + /// Patch the RIFF + data size fields and flush. Consumes the writer. + pub fn finalize(mut self) -> io::Result<()> { + self.file.seek(SeekFrom::Start(4))?; + self.file.write_all(&(36 + self.data_bytes).to_le_bytes())?; + self.file.seek(SeekFrom::Start(40))?; + self.file.write_all(&self.data_bytes.to_le_bytes())?; + self.file.flush()?; + Ok(()) + } +} + +/// A live call recorder: a [`WavWriter`] plus a small mic FIFO that aligns your +/// transmitted mic with the playout mixer's incoming-mix frames. +pub struct Recorder { + writer: WavWriter, + /// Your transmitted mic samples, awaiting alignment with the next mix frame. + mic_fifo: VecDeque, + path: PathBuf, +} + +impl Recorder { + /// Create a recording at `dir/.wav`. The directory is assumed to + /// exist (the caller creates it). + pub fn create(dir: &Path, now_unix_secs: u64) -> io::Result { + let path = dir.join(timestamp_filename(now_unix_secs)); + let writer = WavWriter::new(&path)?; + Ok(Self { + writer, + mic_fifo: VecDeque::new(), + path, + }) + } + + /// The path being written. + pub fn path(&self) -> &Path { + &self.path + } + + /// Buffer a frame of your transmitted mic audio. Bounded: if the FIFO exceeds + /// [`MAX_MIC_FIFO`] (capture outrunning playout), the oldest samples are + /// dropped so the recording's mic offset can't grow without limit. + pub fn push_mic(&mut self, frame: &[i16]) { + self.mic_fifo.extend(frame.iter().copied()); + let overflow = self.mic_fifo.len().saturating_sub(MAX_MIC_FIFO); + if overflow > 0 { + self.mic_fifo.drain(..overflow); + } + } + + /// Write one recording frame: the incoming mix summed (saturating) with the + /// next aligned slice of buffered mic. Mic samples beyond what's buffered are + /// treated as silence (you weren't transmitting), so quiet stretches record + /// the incoming mix alone. + pub fn write_frame(&mut self, mixed: &[i16]) -> io::Result<()> { + let mut out = Vec::with_capacity(mixed.len()); + for &m in mixed { + let mic = self.mic_fifo.pop_front().unwrap_or(0); + let sum = (m as i32 + mic as i32).clamp(i16::MIN as i32, i16::MAX as i32); + out.push(sum as i16); + } + self.writer.write_samples(&out) + } + + /// Finish the file, patching its size fields. Consumes the recorder. + pub fn finalize(self) -> io::Result<()> { + self.writer.finalize() + } +} + +/// Civil date (year, month, day) from a count of days since the Unix epoch. +/// Howard Hinnant's `civil_from_days`; valid across the whole practical range. +fn civil_from_days(z: i64) -> (i64, u32, u32) { + let z = z + 719_468; + let era = (if z >= 0 { z } else { z - 146_096 }) / 146_097; + let doe = z - era * 146_097; // [0, 146096] + let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146_096) / 365; // [0, 399] + let y = yoe + era * 400; + let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); // [0, 365] + let mp = (5 * doy + 2) / 153; // [0, 11] + let d = (doy - (153 * mp + 2) / 5 + 1) as u32; // [1, 31] + let m = if mp < 10 { mp + 3 } else { mp - 9 } as u32; // [1, 12] + (if m <= 2 { y + 1 } else { y }, m, d) +} + +/// A sortable, human-readable recording filename from a Unix timestamp (UTC): +/// `peerspeak-YYYY-MM-DD_HHMMSS.wav`. +pub fn timestamp_filename(unix_secs: u64) -> String { + let days = (unix_secs / 86_400) as i64; + let rem = unix_secs % 86_400; + let (y, m, d) = civil_from_days(days); + let (h, mi, s) = (rem / 3600, (rem % 3600) / 60, rem % 60); + format!("peerspeak-{y:04}-{m:02}-{d:02}_{h:02}{mi:02}{s:02}.wav") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn timestamp_filename_is_utc_and_padded() { + // 1_700_000_000 = 2023-11-14 22:13:20 UTC. + assert_eq!( + timestamp_filename(1_700_000_000), + "peerspeak-2023-11-14_221320.wav" + ); + // Epoch. + assert_eq!(timestamp_filename(0), "peerspeak-1970-01-01_000000.wav"); + } + + #[test] + fn wav_header_round_trips_sizes() { + let dir = std::env::temp_dir(); + let path = dir.join(format!("peerspeak-test-{}.wav", std::process::id())); + let mut w = WavWriter::new(&path).unwrap(); + // 100 samples = 200 data bytes. + w.write_samples(&vec![1234i16; 100]).unwrap(); + w.finalize().unwrap(); + + let bytes = std::fs::read(&path).unwrap(); + assert_eq!(&bytes[0..4], b"RIFF"); + assert_eq!(&bytes[8..12], b"WAVE"); + assert_eq!(&bytes[36..40], b"data"); + // RIFF size = 36 + data, data = 200. + let riff = u32::from_le_bytes([bytes[4], bytes[5], bytes[6], bytes[7]]); + let data = u32::from_le_bytes([bytes[40], bytes[41], bytes[42], bytes[43]]); + assert_eq!(data, 200); + assert_eq!(riff, 236); + // File is header + data. + assert_eq!(bytes.len(), 44 + 200); + let _ = std::fs::remove_file(&path); + } + + #[test] + fn mic_is_summed_with_mix_when_present() { + let dir = std::env::temp_dir(); + let mut r = Recorder { + writer: WavWriter::new(&dir.join(format!("ps-sum-{}.wav", std::process::id()))).unwrap(), + mic_fifo: VecDeque::new(), + path: PathBuf::new(), + }; + r.push_mic(&[1000, 2000, 3000]); + // write_frame pops mic per-sample and sums; we can't read the file mid-stream, + // so assert the FIFO drains exactly by frame length. + r.write_frame(&[10, 20]).unwrap(); + assert_eq!(r.mic_fifo.len(), 1, "two samples consumed, one mic left"); + r.write_frame(&[0, 0]).unwrap(); + assert_eq!(r.mic_fifo.len(), 0, "remaining mic sample consumed; rest is silence"); + let _ = r.finalize(); + } + + #[test] + fn mic_fifo_is_capped() { + let dir = std::env::temp_dir(); + let mut r = Recorder { + writer: WavWriter::new(&dir.join(format!("ps-cap-{}.wav", std::process::id()))).unwrap(), + mic_fifo: VecDeque::new(), + path: PathBuf::new(), + }; + r.push_mic(&vec![5i16; MAX_MIC_FIFO * 2]); + assert_eq!(r.mic_fifo.len(), MAX_MIC_FIFO, "FIFO is bounded to the cap"); + let _ = r.finalize(); + } +} diff --git a/src/core/messages.rs b/src/core/messages.rs index 3c89cf6..8510871 100644 --- a/src/core/messages.rs +++ b/src/core/messages.rs @@ -26,6 +26,9 @@ pub enum CoreCommand { /// Set the relay/discovery posture. Takes effect on the next room join, /// since the endpoint is (re)built then. SetNetworkMode(NetworkMode), + /// Start/stop recording the call to a local WAV (your mic + the incoming + /// mix). No-op start if already recording / not in a call. + SetRecording(bool), } #[derive(Debug, Clone)] @@ -44,5 +47,9 @@ pub enum UiEvent { /// Raw (pre-gate, pre-mute) normalized RMS of the local mic, `0.0..=1.0`, /// for the settings level meter. Throttled to ~10/sec. MicLevel(f32), + /// Call recording started; carries the absolute WAV path being written. + RecordingStarted { path: String }, + /// Call recording stopped; carries the finished WAV path. + RecordingStopped { path: String }, Error(String), } diff --git a/src/core/mod.rs b/src/core/mod.rs index 014ce3b..1fc8328 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -358,6 +358,26 @@ impl ActiveSession { } } +/// Finalize and clear the active recording, if any, emitting `RecordingStopped`. +/// No-op when not recording. Called on stop, room leave, and room switch so a +/// recording is always closed cleanly (its WAV size fields patched). +async fn stop_recording( + recorder: &Arc>>, + is_recording: &Arc, + ui_tx: &mpsc::Sender, +) { + is_recording.store(false, Ordering::Relaxed); + let rec = recorder.lock().unwrap().take(); + if let Some(rec) = rec { + let path = rec.path().to_string_lossy().to_string(); + if let Err(e) = rec.finalize() { + crate::log_msg(&format!("Failed to finalize recording: {e}")); + } + crate::log_msg(&format!("Recording saved: {path}")); + let _ = ui_tx.send(UiEvent::RecordingStopped { path }).await; + } +} + async fn run_core_loop( mut cmd_rx: mpsc::Receiver, ui_tx: mpsc::Sender, @@ -375,6 +395,13 @@ async fn run_core_loop( // App-internal capture/playback gains (f32 bits), live-read by the audio loops. let input_gain = Arc::new(std::sync::atomic::AtomicU32::new(1.0f32.to_bits())); let output_gain = Arc::new(std::sync::atomic::AtomicU32::new(1.0f32.to_bits())); + // Call recording: an optional live recorder (mic FIFO + WAV writer), shared + // by the capture thread (pushes mic) and the mixer task (writes mix frames). + // `is_recording` is a fast-path gate so the audio loops only take the lock + // while a recording is actually running. + let recorder: Arc>> = + Arc::new(std::sync::Mutex::new(None)); + let is_recording = Arc::new(AtomicBool::new(false)); let peer_volumes = Arc::new(Mutex::new(HashMap::::new())); // Peers locally muted by us: decoded for level metering but not mixed. let locally_muted = Arc::new(Mutex::new(HashSet::::new())); @@ -390,6 +417,10 @@ async fn run_core_loop( CoreCommand::Join { name, ticket, input_device, output_device, echo_cancellation } => { current_name = name.clone(); + // Finalize any recording before tearing down the old session — its + // capture/mixer feeders are about to stop. + stop_recording(&recorder, &is_recording, &ui_tx).await; + // Clean up any existing session if let Some(session) = active_session.take() { crate::log_msg("Shutting down existing active session"); @@ -552,6 +583,8 @@ async fn run_core_loop( let input_gain_clone = input_gain.clone(); let transport_clone = transport.clone(); let ui_tx_capture = ui_tx.clone(); + let recorder_capture = recorder.clone(); + let is_recording_capture = is_recording.clone(); let capture_thread = std::thread::spawn(move || { use opus::{Channels, Application}; @@ -598,6 +631,15 @@ async fn run_core_loop( continue; } + // Record what we transmit (post-gain, post-gate, post-mute): + // this is exactly the mic audio peers receive from us. The + // mixer task pairs it with the incoming mix. + if is_recording_capture.load(Ordering::Relaxed) + && let Some(rec) = recorder_capture.lock().unwrap().as_mut() + { + rec.push_mic(&pcm); + } + if let Ok(encoded) = encoder.encode(&pcm) { // Frame on the wire: [seq: u32 LE][opus payload]. let mut packet = Vec::with_capacity(4 + encoded.len()); @@ -662,6 +704,8 @@ async fn run_core_loop( let output_gain_mixer = output_gain.clone(); let ui_tx_mixer = ui_tx.clone(); let ring_fill_mixer = ring_fill.clone(); + let recorder_mixer = recorder.clone(); + let is_recording_mixer = is_recording.clone(); let mixer_task = tokio::spawn(async move { // Mix-bus soft limiter: rides loud multi-peer moments down to // the ceiling instead of hard-clipping. State carries across @@ -728,6 +772,16 @@ async fn run_core_loop( let out_gain = f32::from_bits(output_gain_mixer.load(Ordering::Relaxed)); let mixed = limiter.process(&mixed_sum, out_gain); + // Record the true call audio (incoming mix + our mic), + // independent of local deafen — deafen only silences our + // own monitor, not what the call actually carried. + if is_recording_mixer.load(Ordering::Relaxed) + && let Some(rec) = recorder_mixer.lock().unwrap().as_mut() + && let Err(e) = rec.write_frame(&mixed) + { + crate::log_msg(&format!("Recording write failed: {e}")); + } + let frame_to_send = if is_deafened_clone.load(Ordering::Relaxed) { vec![0i16; FRAME_SAMPLES] } else { @@ -861,6 +915,8 @@ async fn run_core_loop( } CoreCommand::Leave => { + // Finalize any recording first, while the audio feeders are alive. + stop_recording(&recorder, &is_recording, &ui_tx).await; if let Some(session) = active_session.take() { session.shutdown(audio_backend.clone()).await; let _ = ui_tx.send(UiEvent::RoomLeft).await; @@ -952,6 +1008,56 @@ async fn run_core_loop( CoreCommand::SetNetworkMode(mode) => { network_mode = mode; } + + CoreCommand::SetRecording(enabled) => { + if enabled { + // Only record while in a call, and not already recording. + if active_session.is_none() { + let _ = ui_tx + .send(UiEvent::Error("Join a call before recording".into())) + .await; + } else if !is_recording.load(Ordering::Relaxed) { + match dirs::home_dir() { + Some(home) => { + let dir = home.join("peerspeak-recordings"); + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); + let started = std::fs::create_dir_all(&dir) + .map_err(|e| e.to_string()) + .and_then(|_| { + crate::audio::recorder::Recorder::create(&dir, now) + .map_err(|e| e.to_string()) + }); + match started { + Ok(rec) => { + let path = rec.path().to_string_lossy().to_string(); + *recorder.lock().unwrap() = Some(rec); + is_recording.store(true, Ordering::Relaxed); + crate::log_msg(&format!("Recording started: {path}")); + let _ = ui_tx + .send(UiEvent::RecordingStarted { path }) + .await; + } + Err(e) => { + let _ = ui_tx + .send(UiEvent::Error(format!("Recording failed: {e}"))) + .await; + } + } + } + None => { + let _ = ui_tx + .send(UiEvent::Error("No home directory for recordings".into())) + .await; + } + } + } + } else { + stop_recording(&recorder, &is_recording, &ui_tx).await; + } + } } }