use crate::recording::Recorder; use crate::serial_core::codec::Codec; use crate::serial_core::codecs::hand_gateway::{HandGatewayCodec, HandGatewayFrame}; use crate::serial_core::codecs::tactile_a::TactileACodec; use crate::serial_core::frame::TactileAFrame; use crossbeam_channel::{Receiver, Sender}; use std::io::{Read, Write}; use std::time::{Duration, Instant}; const POLL_INTERVAL_MS: u64 = 5; const DEFAULT_HAND_GATEWAY_NODE_SAMPLE_COUNTS: &[u16] = &[84, 84, 84, 84, 84, 70, 44]; #[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] pub struct SerialIoStats { pub rx_bytes: u64, pub tx_bytes: u64, pub rx_frames: u64, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum SerialProtocol { TactileA, HandGateway, } /// Runs the serial polling loop on the calling (background) thread. /// Sends decoded pressure matrix data (Vec) to the output channel. pub fn run_serial_loop( port: &mut dyn ReadWrite, rows: usize, cols: usize, protocol: SerialProtocol, cancel_rx: &Receiver<()>, sample_tx: &Sender>, stats_tx: Option<&Sender>, recorder: Option<&Recorder>, ) { match protocol { SerialProtocol::TactileA => { run_tactile_a_loop(port, rows, cols, cancel_rx, sample_tx, stats_tx, recorder) } SerialProtocol::HandGateway => { run_hand_gateway_loop(port, cancel_rx, sample_tx, stats_tx, recorder) } } } fn run_tactile_a_loop( port: &mut dyn ReadWrite, rows: usize, cols: usize, cancel_rx: &Receiver<()>, sample_tx: &Sender>, stats_tx: Option<&Sender>, recorder: Option<&Recorder>, ) { let session_started_at = Instant::now(); let mut codec = TactileACodec::new(cols, rows); let req_frame = match TactileACodec::build_req_frame(cols, rows) { Ok(frame) => frame, Err(err) => { eprintln!("[serial] request frame build error: {err}"); return; } }; let mut buffer = [0u8; 1024]; let poll_interval = Duration::from_millis(POLL_INTERVAL_MS); let mut io_stats = SerialIoStats::default(); loop { // Check cancel if cancel_rx.try_recv().is_ok() { break; } // Send poll request if let Ok(req_bytes) = codec.encode(&req_frame) { if port.write_all(&req_bytes).is_ok() { io_stats.tx_bytes += req_bytes.len() as u64; publish_stats(stats_tx, io_stats); } } // Read response with poll interval let deadline = Instant::now() + poll_interval; loop { if Instant::now() >= deadline { break; } match port.read(&mut buffer) { Ok(n) if n > 0 => { io_stats.rx_bytes += n as u64; publish_stats(stats_tx, io_stats); if let Ok(frames) = codec.decode(&buffer[..n], session_started_at) { for frame in frames { if let TactileAFrame::Rep(rep) = frame { if let Ok(vals) = TactileACodec::parse_data_frame(&rep.payload) { if let Some(recorder) = recorder { let pressures: Vec = vals.iter().map(|v| (*v).max(0) as u32).collect(); recorder.add_frame(&pressures); } let _ = sample_tx.try_send(vals); io_stats.rx_frames += 1; publish_stats(stats_tx, io_stats); } } } } } Ok(_) => { std::thread::sleep(Duration::from_millis(1)); } Err(ref e) if e.kind() == std::io::ErrorKind::TimedOut => { continue; } Err(e) => { eprintln!("[serial] read error: {e}"); return; } } } } } fn run_hand_gateway_loop( port: &mut dyn ReadWrite, cancel_rx: &Receiver<()>, sample_tx: &Sender>, stats_tx: Option<&Sender>, recorder: Option<&Recorder>, ) { let session_started_at = Instant::now(); let mut codec = HandGatewayCodec::new(DEFAULT_HAND_GATEWAY_NODE_SAMPLE_COUNTS); let mut buffer = [0u8; 1024]; let poll_interval = Duration::from_millis(POLL_INTERVAL_MS); let mut io_stats = SerialIoStats::default(); loop { if cancel_rx.try_recv().is_ok() { break; } let deadline = Instant::now() + poll_interval; loop { if Instant::now() >= deadline { break; } match port.read(&mut buffer) { Ok(n) if n > 0 => { io_stats.rx_bytes += n as u64; publish_stats(stats_tx, io_stats); if let Ok(frames) = codec.decode(&buffer[..n], session_started_at) { for frame in frames { let HandGatewayFrame::DataRep(rep) = frame; // println!( // "[hand-packet-raw] bytes={} timestamp_us={} config=0x{:08X} valid_config=0x{:08X} block_count={} payload_bytes={} raw={}", // rep.raw.len(), // rep.timestamp, // rep.config, // rep.valid_config, // rep.block_count, // rep.nodes // .iter() // .map(|node| node.payload.len()) // .sum::(), // format_hex_bytes(&rep.raw) // ); let mut vals = Vec::new(); let mut parse_ok = true; for node in &rep.nodes { match HandGatewayCodec::parse_node_payload(&node.payload) { Ok(node_values) => { vals.extend(node_values.into_iter().map(|raw| { let raw = raw as i32; if raw < 15 { 0 } else { raw } })); } Err(err) => { parse_ok = false; eprintln!("[hand-packet-values] parse error: {err}"); break; } } } if parse_ok { // println!("[hand-packet-values] samples={}", vals.len()); if let Some(recorder) = recorder { let pressures: Vec = vals.iter().map(|v| (*v).max(0) as u32).collect(); recorder.add_frame(&pressures); } let _ = sample_tx.try_send(vals); io_stats.rx_frames += 1; publish_stats(stats_tx, io_stats); } } } } Ok(_) => { std::thread::sleep(Duration::from_millis(1)); } Err(ref e) if e.kind() == std::io::ErrorKind::TimedOut => { continue; } Err(e) => { eprintln!("[serial] hand gateway read error: {e}"); return; } } } } } fn format_hex_bytes(bytes: &[u8]) -> String { bytes .iter() .map(|byte| format!("{byte:02X}")) .collect::>() .join(" ") } fn publish_stats(stats_tx: Option<&Sender>, stats: SerialIoStats) { if let Some(tx) = stats_tx { let _ = tx.try_send(stats); } } /// Trait abstracting read+write for the serial port pub trait ReadWrite: Send { fn read(&mut self, buf: &mut [u8]) -> std::io::Result; fn write_all(&mut self, buf: &[u8]) -> std::io::Result<()>; } /// Wrapper for serialport's Box pub struct SerialPortReadWrite { inner: Box, } impl SerialPortReadWrite { pub fn new(port: Box) -> Self { Self { inner: port } } } impl ReadWrite for SerialPortReadWrite { fn read(&mut self, buf: &mut [u8]) -> std::io::Result { self.inner.read(buf) } fn write_all(&mut self, buf: &[u8]) -> std::io::Result<()> { self.inner.write_all(buf) } }