eskin-player/src/serial_core/serial.rs

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,
}

#[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<i32>) 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<Vec<i32>>,
    stats_tx: Option<&Sender<SerialIoStats>>,
    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<Vec<i32>>,
    stats_tx: Option<&Sender<SerialIoStats>>,
    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<u32> =
                                            vals.iter().map(|v| (*v).max(0) as u32).collect();
                                        recorder.add_frame(&pressures);
                                    }
                                    let _ = sample_tx.try_send(vals);
                                }
                            }
                        }
                    }
                }
                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<Vec<i32>>,
    stats_tx: Option<&Sender<SerialIoStats>>,
    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::<usize>(),
                            //     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<u32> =
                                        vals.iter().map(|v| (*v).max(0) as u32).collect();
                                    recorder.add_frame(&pressures);
                                }
                                let _ = sample_tx.try_send(vals);
                            }
                        }
                    }
                }
                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::<Vec<_>>()
        .join(" ")
}

fn publish_stats(stats_tx: Option<&Sender<SerialIoStats>>, 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<usize>;
    fn write_all(&mut self, buf: &[u8]) -> std::io::Result<()>;
}

/// Wrapper for serialport's Box<dyn SerialPort>
pub struct SerialPortReadWrite {
    inner: Box<dyn serialport::SerialPort>,
}

impl SerialPortReadWrite {
    pub fn new(port: Box<dyn serialport::SerialPort>) -> Self {
        Self { inner: port }
    }
}

impl ReadWrite for SerialPortReadWrite {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        self.inner.read(buf)
    }

    fn write_all(&mut self, buf: &[u8]) -> std::io::Result<()> {
        self.inner.write_all(buf)
    }
}