nanonis_rs/

tcplogger_stream.rs

use crate::error::NanonisError;
use crate::signals::SignalFrame;
use crate::tcplog::TCPLogStatus;
use byteorder::{BigEndian, ReadBytesExt};
use std::io::{Cursor, Read};
use std::net::{SocketAddr, TcpStream};
use std::sync::mpsc;
use std::thread::{self, JoinHandle};
use std::time::Duration;

/// Simple TCP Logger Stream - connects to data stream only, no control
pub struct TCPLoggerStream {
    stream: TcpStream,
    buffer: Vec<u8>,
}

impl TCPLoggerStream {
    /// Connect to TCP Logger data stream only
    ///
    /// Creates a simple connection to the TCP data stream without any control operations.
    /// All control (start/stop/configure) should be handled externally.
    ///
    /// # Arguments
    /// * `addr` - Server address (e.g., "127.0.0.1")
    /// * `stream_port` - TCP Logger data stream port (typically 6590)
    ///
    /// # Returns
    /// Connected `TCPLoggerStream` ready to read data frames.
    pub fn new(addr: &str, stream_port: u16) -> Result<Self, NanonisError> {
        let socket_addr: SocketAddr = format!("{addr}:{stream_port}")
            .parse()
            .map_err(|_| NanonisError::Protocol(format!("Invalid address: {addr}")))?;

        let connect_timeout = Duration::from_secs(10);
        let stream = TcpStream::connect_timeout(&socket_addr, connect_timeout).map_err(|e| {
            NanonisError::from_io(
                e,
                format!("Failed to connect to TCP stream at {socket_addr}"),
            )
        })?;

        // Set read timeout for continuous reading
        stream
            .set_read_timeout(Some(Duration::from_secs(30)))
            .map_err(|e| NanonisError::Io {
                source: e,
                context: "Setting TCP stream read timeout".to_string(),
            })?;

        Ok(Self {
            stream,
            buffer: Vec::with_capacity(1024),
        })
    }

    /// Spawn background reader thread.
    ///
    /// Creates a named background thread that continuously reads TCP Logger
    /// data frames and sends them through a channel. Returns both the
    /// receiver and a `JoinHandle` so the caller can detect *why* the
    /// thread exited (clean shutdown vs. I/O error).
    ///
    /// The thread exits when:
    /// - The receiver is dropped (clean shutdown, returns `Ok(())`)
    /// - A `read_frame` call fails (returns `Err(NanonisError)`)
    pub fn spawn_background_reader(
        mut self,
    ) -> (
        mpsc::Receiver<SignalFrame>,
        JoinHandle<Result<(), NanonisError>>,
    ) {
        let (sender, receiver) = mpsc::channel();

        let handle = thread::Builder::new()
            .name("tcp-logger-reader".into())
            .spawn(move || {
                loop {
                    match self.read_frame() {
                        Ok(frame) => {
                            if sender.send(frame).is_err() {
                                return Ok(()); // receiver dropped, clean shutdown
                            }
                        }
                        Err(e) => return Err(e),
                    }
                }
            })
            .expect("failed to spawn tcp-logger-reader thread");

        (receiver, handle)
    }

    /// Read a single data frame from the stream.
    ///
    /// Automatically skips the initial metadata frame (counter == 0) that
    /// the Nanonis TCP logger sends when first started. This frame contains
    /// signal index information rather than measurement data and is not
    /// useful to callers.
    ///
    /// # Returns
    /// A `SignalFrame` containing the frame counter and channel data.
    ///
    /// # Frame Format
    /// Each frame is 18 bytes header + (num_channels * 4) bytes of f32 data.
    pub fn read_frame(&mut self) -> Result<SignalFrame, NanonisError> {
        loop {
            let frame = self.read_frame_raw()?;
            // The first frame after logger start has counter == 0 and carries
            // signal index metadata, not measurement data. Skip it.
            if frame.counter == 0 {
                continue;
            }
            return Ok(frame);
        }
    }

    /// Read a single raw frame from the stream without filtering.
    ///
    /// Unlike [`read_frame`], this returns every frame including the
    /// counter-0 metadata frame. Use this only if you need to inspect
    /// the signal index metadata.
    pub fn read_frame_raw(&mut self) -> Result<SignalFrame, NanonisError> {
        // First read header to determine frame size
        let header_size = 18;
        self.buffer.resize(header_size, 0);

        // Read header into buffer
        self.stream
            .read_exact(&mut self.buffer[..header_size])
            .map_err(|e| NanonisError::Io {
                source: e,
                context: "Reading TCP Logger frame header".to_string(),
            })?;

        // Parse header from buffer
        let mut cursor = Cursor::new(&self.buffer[..header_size]);
        let num_channels = cursor.read_u32::<BigEndian>()?;
        let _oversampling = cursor.read_f32::<BigEndian>()?;
        let counter = cursor.read_u64::<BigEndian>()?;
        let state_val = cursor.read_u16::<BigEndian>()?;
        let _state = TCPLogStatus::try_from(state_val as i32)?;

        // Calculate total frame size and read data portion
        let data_size = (num_channels as usize)
            .checked_mul(4) // 4 bytes per f32
            .ok_or_else(|| {
                NanonisError::Protocol(format!("Channel count overflow: {num_channels} channels"))
            })?;
        self.buffer.resize(data_size, 0);

        self.stream
            .read_exact(&mut self.buffer[..data_size])
            .map_err(|e| NanonisError::Io {
                source: e,
                context: "Reading TCP Logger frame data".to_string(),
            })?;

        // Parse data values from buffer
        let mut cursor = Cursor::new(&self.buffer[..data_size]);
        let mut data = Vec::with_capacity(num_channels as usize);
        for _ in 0..num_channels {
            data.push(cursor.read_f32::<BigEndian>()?);
        }

        Ok(SignalFrame { counter, data })
    }
}