//! Recording and CSV export for pressure sensor data. //! //! Provides two recording modes (Full and Snapshot), state management //! (Idle / Recording / Paused), and CSV import/export for replay. use anyhow::{Context, Result}; use std::fs::File; use std::io::{BufRead, BufReader, BufWriter, Write}; use std::path::Path; use std::sync::{Arc, Mutex}; use std::time::Instant; // ── Public types ──────────────────────────────────────────────────────── /// A single frame of pressure data captured at a point in time. #[derive(Debug, Clone)] pub struct Frame { /// Pressure values, one per channel (same order every frame). pub pressures: Vec, /// Milliseconds elapsed since the recording started. pub timestamp_ms: u64, } /// Whether the recording captures everything from connect (Full) or /// only between explicit start/stop calls (Snapshot). #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum RecordingMode { Full, Snapshot, } /// Transient state of a recording. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum RecordingState { Idle, Recording, Paused, } // ── Inner (unlocked) recorder ─────────────────────────────────────────── struct RecorderInner { mode: RecordingMode, state: RecordingState, frames: Vec, start: Option, /// Accumulated wall-clock ms while paused (subtracted from elapsed). paused_duration_ms: u64, /// Instant when we entered Paused (None when not paused). pause_start: Option, /// Number of channels in the first frame (used for CSV header). channel_count: Option, } impl RecorderInner { fn new(mode: RecordingMode) -> Self { Self { mode, state: RecordingState::Idle, frames: Vec::new(), start: None, paused_duration_ms: 0, pause_start: None, channel_count: None, } } fn elapsed_ms(&self) -> u64 { let Some(start) = self.start else { return 0 }; let raw = start.elapsed().as_millis() as u64; let paused = if let Some(ps) = self.pause_start { self.paused_duration_ms + ps.elapsed().as_millis() as u64 } else { self.paused_duration_ms }; raw.saturating_sub(paused) } fn push_frame(&mut self, pressures: Vec) { if self.channel_count.is_none() { self.channel_count = Some(pressures.len()); } let ts = self.elapsed_ms(); self.frames.push(Frame { pressures, timestamp_ms: ts, }); } } // ── Thread-safe wrapper ───────────────────────────────────────────────── /// Thread-safe recorder. Clone the `Arc` to share across threads. #[derive(Clone)] pub struct Recorder { inner: Arc>, } impl Recorder { /// Create a recorder in the given mode (starts in `Idle` state). pub fn new(mode: RecordingMode) -> Self { Self { inner: Arc::new(Mutex::new(RecorderInner::new(mode))), } } // ── Lifecycle ─────────────────────────────────────────────────── /// Start a **Full** recording (records every frame pushed from now on). pub fn start_full_recording(&self) -> Result<()> { let mut r = self.inner.lock().unwrap(); anyhow::ensure!( r.state == RecordingState::Idle, "can only start from Idle state" ); r.state = RecordingState::Recording; r.start = Some(Instant::now()); r.paused_duration_ms = 0; r.pause_start = None; r.frames.clear(); r.channel_count = None; Ok(()) } /// Start a **Snapshot** recording window. pub fn start_snapshot_recording(&self) -> Result<()> { let mut r = self.inner.lock().unwrap(); anyhow::ensure!( r.state == RecordingState::Idle, "can only start from Idle state" ); r.state = RecordingState::Recording; r.start = Some(Instant::now()); r.paused_duration_ms = 0; r.pause_start = None; r.frames.clear(); r.channel_count = None; Ok(()) } /// Stop recording (transitions to `Idle`). pub fn stop_recording(&self) -> Result<()> { let mut r = self.inner.lock().unwrap(); anyhow::ensure!( r.state == RecordingState::Recording || r.state == RecordingState::Paused, "nothing to stop" ); if r.state == RecordingState::Paused { if let Some(ps) = r.pause_start.take() { r.paused_duration_ms += ps.elapsed().as_millis() as u64; } } r.state = RecordingState::Idle; Ok(()) } /// Pause an active recording. pub fn pause_recording(&self) -> Result<()> { let mut r = self.inner.lock().unwrap(); anyhow::ensure!(r.state == RecordingState::Recording, "not recording"); r.pause_start = Some(Instant::now()); r.state = RecordingState::Paused; Ok(()) } /// Resume a paused recording. pub fn resume_recording(&self) -> Result<()> { let mut r = self.inner.lock().unwrap(); anyhow::ensure!(r.state == RecordingState::Paused, "not paused"); if let Some(ps) = r.pause_start.take() { r.paused_duration_ms += ps.elapsed().as_millis() as u64; } r.state = RecordingState::Recording; Ok(()) } // ── Data ──────────────────────────────────────────────────────── /// Feed one frame of pressure data. Ignored when not recording. pub fn add_frame(&self, pressures: &[u32]) { let mut r = self.inner.lock().unwrap(); if r.state == RecordingState::Recording { r.push_frame(pressures.to_vec()); } } /// Number of recorded frames. pub fn frame_count(&self) -> usize { self.inner.lock().unwrap().frames.len() } /// Whether the recorder is currently in the `Recording` state. pub fn is_recording(&self) -> bool { self.inner.lock().unwrap().state == RecordingState::Recording } /// Current recording state. pub fn state(&self) -> RecordingState { self.inner.lock().unwrap().state } /// Milliseconds elapsed (excludes paused time). pub fn duration_ms(&self) -> u64 { self.inner.lock().unwrap().elapsed_ms() } /// Snapshot of all recorded frames. pub fn recorded_frames(&self) -> Vec { self.inner.lock().unwrap().frames.clone() } // ── CSV export / import ───────────────────────────────────────── /// Export recorded frames to CSV. /// /// Header: `channel1,channel2,...,channelN,timestamp_ms` pub fn export_csv>(&self, path: P) -> Result<()> { let r = self.inner.lock().unwrap(); anyhow::ensure!(!r.frames.is_empty(), "no frames to export"); let file = File::create(path.as_ref()) .with_context(|| format!("creating {}", path.as_ref().display()))?; let mut w = BufWriter::new(file); let n = r.channel_count.unwrap_or(r.frames[0].pressures.len()); // Header for i in 0..n { write!(w, "channel{}", i + 1)?; if i < n - 1 { write!(w, ",")?; } } writeln!(w, ",timestamp_ms")?; // Rows for frame in &r.frames { for (i, val) in frame.pressures_iter().enumerate() { write!(w, "{}", val)?; if i < n - 1 { write!(w, ",")?; } } writeln!(w, ",{}", frame.timestamp_ms)?; } w.flush()?; Ok(()) } /// Import frames from a CSV file (same format as `export_csv`). /// /// The recorder must be in `Idle` state. After import it stays `Idle` /// so you can inspect / re-export; call `start_*` to continue recording. pub fn import_csv>(&self, path: P) -> Result<()> { let mut r = self.inner.lock().unwrap(); anyhow::ensure!(r.state == RecordingState::Idle, "must be Idle to import"); let file = File::open(path.as_ref()) .with_context(|| format!("opening {}", path.as_ref().display()))?; let reader = BufReader::new(file); let mut lines = reader.lines(); // Parse header to learn channel count let header = lines .next() .context("empty CSV file")? .context("reading header")?; let cols: Vec<&str> = header.split(',').map(str::trim).collect(); anyhow::ensure!(cols.len() >= 2, "need at least 1 channel + timestamp_ms"); let n_channels = cols.len() - 1; // last column is timestamp_ms r.frames.clear(); r.channel_count = Some(n_channels); let mut first_ts: Option = None; for (lineno, line) in lines.enumerate() { let line = line.with_context(|| format!("reading line {}", lineno + 2))?; let line = line.trim(); if line.is_empty() { continue; } let fields: Vec<&str> = line.split(',').map(str::trim).collect(); anyhow::ensure!( fields.len() == cols.len(), "line {}: expected {} columns, got {}", lineno + 2, cols.len(), fields.len() ); let mut pressures = Vec::with_capacity(n_channels); for f in &fields[..n_channels] { pressures.push( f.parse::().with_context(|| { format!("line {}: bad channel value '{}'", lineno + 2, f) })?, ); } let raw_ts = fields[n_channels] .parse::() .with_context(|| format!("line {}: bad timestamp", lineno + 2))?; // Normalise timestamps so the first frame starts at 0 let ts = if let Some(first) = first_ts { raw_ts.saturating_sub(first) } else { first_ts = Some(raw_ts); 0 }; r.frames.push(Frame { pressures, timestamp_ms: ts, }); } // Set the start instant so duration_ms() reports the imported span if !r.frames.is_empty() { if let Some(last) = r.frames.last() { // Pretend the recording happened `last.timestamp_ms` ago // so that elapsed_ms() would return that value. // We store a "fake" start by noting the offset. r.start = Some(Instant::now() - std::time::Duration::from_millis(last.timestamp_ms)); r.paused_duration_ms = 0; } } Ok(()) } } // ── Convenience constructors ──────────────────────────────────────────── impl Recorder { /// Shorthand: new Full recorder. pub fn full() -> Self { Self::new(RecordingMode::Full) } /// Shorthand: new Snapshot recorder. pub fn snapshot() -> Self { Self::new(RecordingMode::Snapshot) } } // ── Helper extension on Frame ─────────────────────────────────────────── impl Frame { fn pressures_iter(&self) -> impl Iterator + '_ { self.pressures.iter().copied() } } // ── Tests ─────────────────────────────────────────────────────────────── #[cfg(test)] mod tests { use super::*; use std::thread; #[test] fn full_recording_lifecycle() { let rec = Recorder::full(); assert_eq!(rec.state(), RecordingState::Idle); assert!(!rec.is_recording()); assert_eq!(rec.frame_count(), 0); rec.start_full_recording().unwrap(); assert!(rec.is_recording()); rec.add_frame(&[10, 20, 30]); rec.add_frame(&[40, 50, 60]); assert_eq!(rec.frame_count(), 2); rec.stop_recording().unwrap(); assert_eq!(rec.state(), RecordingState::Idle); assert!(!rec.is_recording()); assert!(rec.duration_ms() < 500); } #[test] fn snapshot_pause_resume() { let rec = Recorder::snapshot(); rec.start_snapshot_recording().unwrap(); rec.add_frame(&[1, 2]); rec.pause_recording().unwrap(); assert_eq!(rec.state(), RecordingState::Paused); rec.add_frame(&[9, 9]); // should be ignored assert_eq!(rec.frame_count(), 1); rec.resume_recording().unwrap(); rec.add_frame(&[3, 4]); assert_eq!(rec.frame_count(), 2); rec.stop_recording().unwrap(); } #[test] fn csv_round_trip() { let rec = Recorder::full(); rec.start_full_recording().unwrap(); rec.add_frame(&[100, 200, 300]); rec.add_frame(&[101, 201, 301]); rec.stop_recording().unwrap(); let dir = std::env::temp_dir().join("eskin_recording_test"); std::fs::create_dir_all(&dir).unwrap(); let csv_path = dir.join("test_roundtrip.csv"); rec.export_csv(&csv_path).unwrap(); let rec2 = Recorder::snapshot(); rec2.import_csv(&csv_path).unwrap(); assert_eq!(rec2.frame_count(), 2); let frames = rec2.recorded_frames(); assert_eq!(frames[0].pressures, vec![100, 200, 300]); assert_eq!(frames[1].pressures, vec![101, 201, 301]); assert_eq!(frames[0].timestamp_ms, 0); let _ = std::fs::remove_dir_all(&dir); } #[test] fn thread_safety() { let rec = Recorder::full(); let rec2 = rec.clone(); rec.start_full_recording().unwrap(); let h = thread::spawn(move || { for i in 0..100u32 { rec2.add_frame(&[i, i + 1, i + 2]); } }); h.join().unwrap(); rec.stop_recording().unwrap(); assert_eq!(rec.frame_count(), 100); } }