eskin-player/src/connection.rs

use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};
use std::time::Duration;

use crossbeam_channel::{self, Receiver, Sender, TryRecvError};

use crate::recording::Recorder;
use crate::serial_core::serial::{
    SerialIoStats, SerialPortReadWrite, SerialProtocol, run_serial_loop,
};

/// 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>>,
    stats_rx: Receiver<SerialIoStats>,
    rows: u32,
    cols: u32,
}

/// 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>>>,
    stats: Arc<Mutex<SerialIoStats>>,
}

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)),
            stats: Arc::new(Mutex::new(SerialIoStats::default())),
        }
    }

    pub fn state(&self) -> ConnectionState {
        *self.state.lock().unwrap()
    }

    pub fn set_state(&self, new_state: ConnectionState) {
        *self.state.lock().unwrap() = new_state;
    }

    pub fn stats(&self) -> SerialIoStats {
        let session = self.session.lock().unwrap();
        if let Some(ref session) = *session {
            loop {
                match session.stats_rx.try_recv() {
                    Ok(stats) => *self.stats.lock().unwrap() = stats,
                    Err(TryRecvError::Empty) => break,
                    Err(TryRecvError::Disconnected) => break,
                }
            }
        }

        *self.stats.lock().unwrap()
    }

    /// Connect to the given serial port and start streaming in a background thread.
    pub fn connect(
        &self,
        port_name: &str,
        rows: u32,
        cols: u32,
        baud_rate: u32,
        protocol: SerialProtocol,
        recorder: Recorder,
    ) {
        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 (stats_tx, stats_rx) = crossbeam_channel::bounded::<SerialIoStats>(16);
        *self.stats.lock().unwrap() = SerialIoStats::default();

        let handle = thread::spawn(move || {
            let result = run_device_loop(
                &port,
                rows,
                cols,
                baud_rate,
                protocol,
                &state,
                &cancel_rx,
                &sample_tx,
                &stats_tx,
                &latest_sample,
                recorder,
            );
            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,
            stats_rx,
            rows,
            cols,
        });
    }

    /// 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;
        *self.stats.lock().unwrap() = SerialIoStats::default();
    }

    /// 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 matrix = vals.iter().map(|v| (*v).max(0) as u32).collect();
                        last = Some(PressureSample {
                            matrix,
                            rows: session.rows,
                            cols: session.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.
#[allow(clippy::too_many_arguments)]
fn run_device_loop(
    port_name: &str,
    rows: u32,
    cols: u32,
    baud_rate: u32,
    protocol: SerialProtocol,
    state: &Arc<Mutex<ConnectionState>>,
    cancel_rx: &Receiver<()>,
    sample_tx: &Sender<Vec<i32>>,
    stats_tx: &Sender<SerialIoStats>,
    latest_sample: &Arc<Mutex<Option<PressureSample>>>,
    recorder: Recorder,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let port = serialport::new(port_name, baud_rate)
        .timeout(Duration::from_millis(1))
        .open()?;

    *state.lock().unwrap() = ConnectionState::Connected;

    let mut rw = SerialPortReadWrite::new(port);
    *state.lock().unwrap() = ConnectionState::Streaming;

    run_serial_loop(
        &mut rw,
        rows as usize,
        cols as usize,
        protocol,
        cancel_rx,
        sample_tx,
        Some(stats_tx),
        Some(&recorder),
    );

    if let Ok(mut latest) = latest_sample.lock() {
        *latest = None;
    }
    Ok(())
}