1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
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 })
}
}