Wire live serial data into matrix renderer

src/connection.rs

@@ -0,0 +1,197 @@
+use std::sync::{Arc, Mutex};
+use std::thread::{self, JoinHandle};
+use std::time::Duration;
+
+use crossbeam_channel::{self, Receiver, Sender, TryRecvError};
+
+use crate::serial_core::serial::{run_serial_loop, SerialPortReadWrite};
+
+/// Connection state visible to the UI.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ConnectionState {
+    Disconnected,
+    Connecting,
+    Connected,
+    Streaming,
+    Error,
+}
+
+/// A pressure sample forwarded to the render layer.
+pub struct PressureSample {
+    pub matrix: Vec<u32>,
+    pub rows: u32,
+    pub cols: u32,
+}
+
+struct Session {
+    cancel_tx: Sender<()>,
+    handle: JoinHandle<()>,
+    sample_rx: Receiver<Vec<i32>>,
+}
+
+/// Thread-safe connection manager that the UI and renderer can share.
+pub struct ConnectionManager {
+    state: Arc<Mutex<ConnectionState>>,
+    session: Arc<Mutex<Option<Session>>>,
+    latest_sample: Arc<Mutex<Option<PressureSample>>>,
+}
+
+impl ConnectionManager {
+    pub fn new() -> Self {
+        Self {
+            state: Arc::new(Mutex::new(ConnectionState::Disconnected)),
+            session: Arc::new(Mutex::new(None)),
+            latest_sample: Arc::new(Mutex::new(None)),
+        }
+    }
+
+    pub fn state(&self) -> ConnectionState {
+        *self.state.lock().unwrap()
+    }
+
+    pub fn set_state(&self, new_state: ConnectionState) {
+        *self.state.lock().unwrap() = new_state;
+    }
+
+    /// Connect to the given serial port and start streaming in a background thread.
+    pub fn connect(&self, port_name: &str, rows: u32, cols: u32) {
+        self.disconnect();
+        self.set_state(ConnectionState::Connecting);
+
+        let port = port_name.to_owned();
+        let state = Arc::clone(&self.state);
+        let session_guard = Arc::clone(&self.session);
+        let latest_sample = Arc::clone(&self.latest_sample);
+        let (cancel_tx, cancel_rx) = crossbeam_channel::bounded::<()>(1);
+        let (sample_tx, sample_rx) = crossbeam_channel::bounded::<Vec<i32>>(16);
+
+        let handle = thread::spawn(move || {
+            let result = run_device_loop(
+                &port,
+                rows,
+                cols,
+                &state,
+                &cancel_rx,
+                &sample_tx,
+                &latest_sample,
+            );
+            if let Err(e) = result {
+                eprintln!("[connection] device loop error: {e}");
+                *state.lock().unwrap() = ConnectionState::Error;
+            }
+        });
+
+        *session_guard.lock().unwrap() = Some(Session {
+            cancel_tx,
+            handle,
+            sample_rx,
+        });
+    }
+
+    /// Disconnect from the device, stopping the background thread.
+    pub fn disconnect(&self) {
+        let session = {
+            let mut guard = self.session.lock().unwrap();
+            guard.take()
+        };
+
+        if let Some(session) = session {
+            let _ = session.cancel_tx.send(());
+            let _ = session.handle.join();
+        }
+
+        self.set_state(ConnectionState::Disconnected);
+        *self.latest_sample.lock().unwrap() = None;
+    }
+
+    /// Drain pending samples (non-blocking) and return the last one.
+    pub fn take_latest_sample(&self) -> Option<PressureSample> {
+        let session = self.session.lock().unwrap();
+        if let Some(ref session) = *session {
+            let mut last = None;
+            loop {
+                match session.sample_rx.try_recv() {
+                    Ok(vals) => {
+                        let rows = 12u32;
+                        let cols = 7u32;
+                        let matrix = vals.iter().map(|v| (*v).max(0) as u32).collect();
+                        last = Some(PressureSample {
+                            matrix,
+                            rows,
+                            cols,
+                        });
+                    }
+                    Err(TryRecvError::Empty) => break,
+                    Err(TryRecvError::Disconnected) => break,
+                }
+            }
+            if last.is_some() {
+                return last;
+            }
+        }
+        None
+    }
+}
+
+impl Default for ConnectionManager {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+/// The blocking device loop that runs on a background thread.
+fn run_device_loop(
+    port_name: &str,
+    rows: u32,
+    cols: u32,
+    state: &Arc<Mutex<ConnectionState>>,
+    cancel_rx: &Receiver<()>,
+    sample_tx: &Sender<Vec<i32>>,
+    latest_sample: &Arc<Mutex<Option<PressureSample>>>,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    let port = serialport::new(port_name, 921_600)
+        .timeout(Duration::from_millis(100))
+        .open()?;
+
+    *state.lock().unwrap() = ConnectionState::Connected;
+
+    let mut rw = SerialPortReadWrite::new(port);
+    *state.lock().unwrap() = ConnectionState::Streaming;
+
+    // We need to also forward samples to latest_sample
+    let (inner_tx, inner_rx) = crossbeam_channel::bounded::<Vec<i32>>(16);
+    let latest = Arc::clone(latest_sample);
+    let outer_tx = sample_tx.clone();
+
+    // Bridge thread: reads from inner channel, forwards to both sample_tx and latest_sample
+    let bridge_cancel = cancel_rx.clone();
+    let bridge_handle = thread::spawn(move || {
+        loop {
+            if bridge_cancel.try_recv().is_ok() {
+                break;
+            }
+            match inner_rx.try_recv() {
+                Ok(vals) => {
+                    // Store latest
+                    let matrix = vals.iter().map(|v| (*v).max(0) as u32).collect();
+                    *latest.lock().unwrap() = Some(PressureSample {
+                        matrix,
+                        rows,
+                        cols,
+                    });
+                    // Forward
+                    let _ = outer_tx.try_send(vals);
+                }
+                Err(TryRecvError::Empty) => {
+                    std::thread::sleep(Duration::from_millis(1));
+                }
+                Err(TryRecvError::Disconnected) => break,
+            }
+        }
+    });
+
+    run_serial_loop(&mut rw, rows as usize, cols as usize, cancel_rx, &inner_tx);
+
+    let _ = bridge_handle.join();
+    Ok(())
+}