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 std::sync::Arc;
12
13use csi::ChipVariant;
14use parquet_sink::{ParquetSink, ParquetSinkError};
15
16use crate::profile::ClientProfile;
17
18/// An active recording of one device's CSI stream to a Parquet file.
19///
20/// Decodes each incoming frame against the device's chip layout and appends it
21/// to the sink. Lives outside [`crate::state::AppState`] (which is `Clone`)
22/// because the underlying file writer is not cloneable.
23pub struct Recorder {
24    sink: ParquetSink,
25    chip: ChipVariant,
26    /// Frames successfully decoded and written.
27    pub frames_written: u64,
28    /// Frames that failed to decode (wire drift, truncation, wrong chip).
29    pub decode_errors: u64,
30}
31
32impl Recorder {
33    /// Open a Parquet file at `path` for a stream from the given `chip` string.
34    ///
35    /// `profile` labels the `data_format` column from the numeric
36    /// `cur_bb_format` where it can (see [`ClientProfile::label_format`]).
37    ///
38    /// Returns `Err` if the chip is unrecognized (no known wire layout) or the
39    /// file cannot be created.
40    pub fn start(
41        path: &str,
42        chip: &str,
43        profile: Arc<dyn ClientProfile>,
44    ) -> Result<Self, String> {
45        let variant = ChipVariant::from_chip_str(chip)
46            .ok_or_else(|| format!("unknown chip '{chip}'; cannot decode CSI frames"))?;
47        let sink = ParquetSink::open(path, chip, profile).map_err(|e| e.to_string())?;
48        Ok(Self {
49            sink,
50            chip: variant,
51            frames_written: 0,
52            decode_errors: 0,
53        })
54    }
55
56    /// The output file path being written.
57    pub fn path(&self) -> &str {
58        self.sink.path()
59    }
60
61    /// Decode one raw WebSocket frame and append it, stamped with `host_rx_micros`
62    /// (UTC microseconds). Decode failures are counted, not propagated, so a
63    /// single malformed frame never aborts a recording.
64    pub fn record_frame(&mut self, bytes: &[u8], host_rx_micros: i64) -> Result<(), ParquetSinkError> {
65        match csi::decode(bytes, self.chip) {
66            Ok(decoded) => {
67                self.sink.push(decoded, host_rx_micros)?;
68                self.frames_written += 1;
69            }
70            Err(_) => {
71                self.decode_errors += 1;
72            }
73        }
74        Ok(())
75    }
76
77    /// Flush remaining rows and finalize the Parquet footer.
78    pub fn finish(mut self) -> Result<(), ParquetSinkError> {
79        self.sink.finish()
80    }
81}
csi_webclient/export/mod.rs

csi_webclient/export/
mod.rs
