first commit

src-tauri/src/serial_core/serial.rs

@@ -0,0 +1,80 @@
+use crate::serial_core::codec::Codec;
+use crate::serial_core::frame::{FrameHandler, TestFrame};
+use crate::serial_core::model::HudChartState;
+use anyhow::Result;
+use tauri::{AppHandle, Emitter};
+use tokio::io::AsyncReadExt;
+use tokio::time::{self, Duration, MissedTickBehavior};
+use tokio_serial::SerialStream;
+use tokio_util::sync::CancellationToken;
+use std::sync::{Arc, Mutex};
+use std::time::Instant;
+use log::info;
+use crate::serial_core::record::{FrameTiming, RecordedFrame};
+use crate::serial_core::TestRecording;
+
+pub async fn run_serial<C, H, T>(
+    app: AppHandle,
+    mut port: SerialStream,
+    mut codec: C,
+    mut handler: H,
+    session_started_at: Instant,
+    recording: Arc<Mutex<TestRecording>>,
+    cancel: CancellationToken,
+) -> Result<()>
+where
+    C: Codec<TestFrame> + Send + 'static,
+    H: FrameHandler<TestFrame, T> + Send + 'static,
+    T: Into<i32>,
+{
+    let mut chart_state = HudChartState::new();
+    let mut buffer = [0u8; 1024];
+    let mut prune_interval = time::interval(Duration::from_millis(450));
+    prune_interval.set_missed_tick_behavior(MissedTickBehavior::Delay);
+
+    loop {
+        tokio::select! {
+            _ = cancel.cancelled() => break,
+            _ = prune_interval.tick() => {
+                if let Some(packet) = chart_state.prune_stale() {
+                    app.emit("hud_stream", packet)?;
+                }
+            }
+            read_result = port.read(&mut buffer) => {
+                let n = read_result?;
+                if n == 0 {
+                    continue;
+                }
+
+                let frames = codec.decode(&buffer[..n], session_started_at)?;
+                for frame in frames {
+                    let decode_res = handler
+                        .on_frame(&frame)
+                        .await?
+                        .map(|vals| vals.into_iter().map(Into::into).collect::<Vec<i32>>());
+
+                    let mut record = recording.lock().map_err(|_| anyhow::anyhow!("recording state poisoned"))?;
+                    record.push(RecordedFrame{
+                        timing: FrameTiming { pts_ms: None, dts_ms: frame.dts_ms },
+                        frame: frame.clone(),
+                    });
+
+                    let display_values = if let Some(vals) = decode_res.as_ref() {
+                        let summary = vals.iter().copied().sum::<i32>();
+                        info!("dot value summary: {}", summary);
+                        chart_state.record_summary(summary as f32);
+                        chart_state.record_pressure_matrix(vals.as_slice());
+                        Some(vec![summary])
+                    } else {
+                        None
+                    };
+
+                    let packet = chart_state.apply_frame(&frame, display_values.as_deref());
+                    app.emit("hud_stream", packet)?;
+                }
+            }
+        }
+    }
+
+    Ok(())
+}