feat:add slave

src-tauri/src/serial_core/serial.rs

@@ -1,32 +1,197 @@
 use crate::serial_core::codec::Codec;
-use crate::serial_core::frame::{FrameHandler, TestFrame};
-use crate::serial_core::model::HudChartState;
+use crate::serial_core::codecs::tactile_a::TactileACodec;
+use crate::serial_core::frame::{FrameHandler, TactileAFrame, TestFrame};
+use crate::serial_core::model::{HudChartState, HudPacket};
+use crate::serial_core::record::Recording;
 use anyhow::Result;
 use tauri::{AppHandle, Emitter};
-use tokio::io::AsyncReadExt;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::time::{self, Duration, MissedTickBehavior};
 use tokio_serial::SerialStream;
 use tokio_util::sync::CancellationToken;
+use std::future::pending;
 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>(
+pub enum PollMode<F> {
+    Disable,
+    Enabled(Box<dyn PollRequester<F>>)
+}
+
+pub trait SerialFrame: Clone + Send + 'static {
+    fn dts_ms(&self) -> u64;
+
+    fn to_hud_packet(
+        &self,
+        chart_state: &mut HudChartState,
+        display_values: Option<&[i32]>,
+    ) -> Option<HudPacket>;
+}
+
+impl SerialFrame for TestFrame {
+    fn dts_ms(&self) -> u64 {
+        self.dts_ms
+    }
+
+    fn to_hud_packet(
+        &self,
+        chart_state: &mut HudChartState,
+        display_values: Option<&[i32]>,
+    ) -> Option<HudPacket> {
+        Some(chart_state.apply_frame(self, display_values))
+    }
+}
+
+impl SerialFrame for TactileAFrame {
+    fn dts_ms(&self) -> u64 {
+        match self {
+            TactileAFrame::Req(_) => 0,
+            TactileAFrame::Rep(rep) => rep.dts_ms,
+        }
+    }
+
+    fn to_hud_packet(
+        &self,
+        _chart_state: &mut HudChartState,
+        _display_values: Option<&[i32]>,
+    ) -> Option<HudPacket> {
+        None
+    }
+}
+
+pub trait PollRequester<F>: Send {
+    fn poll_interval(&self) -> Option<Duration> {
+        None
+    }
+
+    fn should_request(&mut self) -> bool {
+        true
+    }
+
+    fn next_request(&mut self) -> Result<Option<F>> {
+        Ok(None)
+    }
+
+    fn on_rx_frame(&mut self, _frame: &F) {}
+}
+
+#[derive(Default)]
+pub struct NoopPollRequester;
+
+impl<F> PollRequester<F> for NoopPollRequester {}
+
+pub struct TactileAPollRequester {
+    period: Duration,
+    cols: usize,
+    rows: usize,
+    awaiting_reply: bool,
+    last_request_at: Option<Instant>,
+    reply_timeout: Duration,
+}
+
+impl TactileAPollRequester {
+    pub fn new(period: Duration, cols: usize, rows: usize, reply_timeout: Duration) -> Self {
+        Self {
+            period,
+            cols,
+            rows,
+            awaiting_reply: false,
+            last_request_at: None,
+            reply_timeout,
+        }
+    }
+}
+
+impl PollRequester<TactileAFrame> for TactileAPollRequester {
+    fn poll_interval(&self) -> Option<Duration> {
+        Some(self.period)
+    }
+
+    fn should_request(&mut self) -> bool {
+        if !self.awaiting_reply {
+            return true;
+        }
+        let timed_out = self
+            .last_request_at
+            .map(|t| t.elapsed() >= self.reply_timeout)
+            .unwrap_or(false);
+
+        if timed_out {
+            self.awaiting_reply = false;
+            self.last_request_at = None;
+            return true;
+        }
+
+        false
+    }
+
+    fn next_request(&mut self) -> Result<Option<TactileAFrame>> {
+        let req = TactileACodec::build_req_frame(self.cols, self.rows)?;
+        self.awaiting_reply = true;
+        self.last_request_at = Some(Instant::now());
+        Ok(Some(req))
+    }
+
+    fn on_rx_frame(&mut self, frame: &TactileAFrame) {
+        if matches!(frame, TactileAFrame::Rep(_)) {
+            self.awaiting_reply = false;
+            self.last_request_at = None
+        }
+    }
+}
+
+pub async fn run_serial<C, H, T, F>(
     app: AppHandle,
     mut port: SerialStream,
     mut codec: C,
     mut handler: H,
     session_started_at: Instant,
-    recording: Arc<Mutex<TestRecording>>,
+    recording: Arc<Mutex<Recording<F>>>,
     cancel: CancellationToken,
 ) -> Result<()>
 where
-    C: Codec<TestFrame> + Send + 'static,
-    H: FrameHandler<TestFrame, T> + Send + 'static,
+    F: SerialFrame,
+    C: Codec<F> + Send + 'static,
+    H: FrameHandler<F, T> + Send + 'static,
+    T: Into<i32>
+{
+    run_serial_with_poll(
+        app, port, codec, handler, session_started_at, recording, cancel, PollMode::Disable
+    ).await
+}
+
+pub async fn run_serial_with_poll<C, H, T, F>(
+    app: AppHandle,
+    mut port: SerialStream,
+    mut codec: C,
+    mut handler: H,
+    session_started_at: Instant,
+    recording: Arc<Mutex<Recording<F>>>,
+    cancel: CancellationToken,
+    poll_mode: PollMode<F>
+) -> Result<()>
+where
+    F: SerialFrame,
+    C: Codec<F> + Send + 'static,
+    H: FrameHandler<F, T> + Send + 'static,
     T: Into<i32>,
 {
+    let mut requester = match poll_mode {
+        PollMode::Disable => None,
+        PollMode::Enabled(r) => Some(r),
+    };
+
+    let mut poll_interval = requester
+    .as_ref()
+    .and_then(|r| r.poll_interval())
+    .map(|d| {
+        let mut it = time::interval(d);
+        it.set_missed_tick_behavior(MissedTickBehavior::Skip);
+        it
+    });
+
     let mut chart_state = HudChartState::new();
     let mut buffer = [0u8; 1024];
     let mut prune_interval = time::interval(Duration::from_millis(450));
@@ -35,6 +200,23 @@ where
     loop {
         tokio::select! {
             _ = cancel.cancelled() => break,
+            _ = async {
+                match poll_interval.as_mut() {
+                    Some(it) => {
+                        it.tick().await;
+                    }
+                    None => pending::<()>().await,
+                }
+            } => {
+                if let Some(r) = requester.as_mut() {
+                    if r.should_request() {
+                        if let Some(req) = r.next_request()? {
+                            let bytes = codec.encode(&req)?;
+                            port.write_all(&bytes).await?;
+                        }
+                    }
+                }
+            }
             _ = prune_interval.tick() => {
                 if let Some(packet) = chart_state.prune_stale() {
                     app.emit("hud_stream", packet)?;
@@ -48,6 +230,10 @@ where
 
                 let frames = codec.decode(&buffer[..n], session_started_at)?;
                 for frame in frames {
+                    if let Some(r) = requester.as_mut() {
+                        r.on_rx_frame(&frame);
+                    }
+
                     let decode_res = handler
                         .on_frame(&frame)
                         .await?
@@ -55,7 +241,7 @@ where
 
                     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 },
+                        timing: FrameTiming { pts_ms: None, dts_ms: frame.dts_ms() },
                         frame: frame.clone(),
                     });
 
@@ -69,12 +255,12 @@ where
                         None
                     };
 
-                    let packet = chart_state.apply_frame(&frame, display_values.as_deref());
-                    app.emit("hud_stream", packet)?;
+                    if let Some(packet) = frame.to_hud_packet(&mut chart_state, display_values.as_deref()) {
+                        app.emit("hud_stream", packet)?;
+                    }
                 }
             }
         }
     }
-
     Ok(())
 }