csi_webclient/export/
mod.rs

1//! Local CSI stream export.
2//!
3//! Decodes raw WebSocket CSI frames (COBS-framed postcard records, identical to
4//! what `csi-webserver-rs` consumes) and writes them to a Parquet file on the
5//! client host. The output schema matches the server's `csi_dump_*.parquet`, so
6//! a locally-recorded file and a server-side dump are interchangeable.
7
8pub mod csi;
9pub mod parquet_sink;
10
11use csi::ChipVariant;
12use parquet_sink::{ParquetSink, ParquetSinkError};
13
14/// An active recording of one device's CSI stream to a Parquet file.
15///
16/// Decodes each incoming frame against the device's chip layout and appends it
17/// to the sink. Lives outside [`crate::state::AppState`] (which is `Clone`)
18/// because the underlying file writer is not cloneable.
19pub struct Recorder {
20    sink: ParquetSink,
21    chip: ChipVariant,
22    /// Frames successfully decoded and written.
23    pub frames_written: u64,
24    /// Frames that failed to decode (wire drift, truncation, wrong chip).
25    pub decode_errors: u64,
26}
27
28impl Recorder {
29    /// Open a Parquet file at `path` for a stream from the given `chip` string.
30    ///
31    /// Returns `Err` if the chip is unrecognized (no known wire layout) or the
32    /// file cannot be created.
33    pub fn start(path: &str, chip: &str) -> Result<Self, String> {
34        let variant = ChipVariant::from_chip_str(chip)
35            .ok_or_else(|| format!("unknown chip '{chip}'; cannot decode CSI frames"))?;
36        let sink = ParquetSink::open(path, chip).map_err(|e| e.to_string())?;
37        Ok(Self {
38            sink,
39            chip: variant,
40            frames_written: 0,
41            decode_errors: 0,
42        })
43    }
44
45    /// The output file path being written.
46    pub fn path(&self) -> &str {
47        self.sink.path()
48    }
49
50    /// Decode one raw WebSocket frame and append it, stamped with `host_rx_micros`
51    /// (UTC microseconds). Decode failures are counted, not propagated, so a
52    /// single malformed frame never aborts a recording.
53    pub fn record_frame(&mut self, bytes: &[u8], host_rx_micros: i64) -> Result<(), ParquetSinkError> {
54        match csi::decode(bytes, self.chip) {
55            Ok(decoded) => {
56                self.sink.push(decoded, host_rx_micros)?;
57                self.frames_written += 1;
58            }
59            Err(_) => {
60                self.decode_errors += 1;
61            }
62        }
63        Ok(())
64    }
65
66    /// Flush remaining rows and finalize the Parquet footer.
67    pub fn finish(mut self) -> Result<(), ParquetSinkError> {
68        self.sink.finish()
69    }
70}
csi_webclient/export/mod.rs

csi_webclient/export/
mod.rs
