instrument_core/scpi/
session.rs

1use super::framing::extract_response;
2use super::protocol::{max_write_attempts, normalize_command, parse_f64, SessionCapabilities};
3use crate::connect::ConnectOptions;
4use crate::diagnostics::{CommsEventKind, Diagnostics};
5use crate::error::{Error, Result};
6use crate::ieee4882::Ieee4882;
7use crate::transport::{DynTransport, Transport};
8use std::thread;
9use std::time::{Duration, Instant};
10
11/// SCPI session over a boxed transport.
12pub struct ScpiSession {
13    transport: DynTransport,
14    opts: ConnectOptions,
15    capabilities: SessionCapabilities,
16    read_buffer: Vec<u8>,
17    diagnostics: Option<Diagnostics>,
18    pending_command: Option<String>,
19}
20
21impl ScpiSession {
22    pub fn new(mut transport: DynTransport, opts: ConnectOptions) -> Result<Self> {
23        transport.configure(&opts)?;
24        let mut session = Self {
25            transport,
26            opts,
27            capabilities: SessionCapabilities::default(),
28            read_buffer: Vec::with_capacity(4096),
29            diagnostics: None,
30            pending_command: None,
31        };
32        if session.opts.reset_on_connect {
33            let _ = Ieee4882::new(&mut session).clear_status();
34            let _ = Ieee4882::new(&mut session).reset();
35        }
36        Ok(session)
37    }
38
39    pub fn with_diagnostics(mut self, diagnostics: Diagnostics) -> Self {
40        self.diagnostics = Some(diagnostics);
41        self
42    }
43
44    pub fn transport(&self) -> &dyn Transport {
45        self.transport.as_ref()
46    }
47
48    pub fn options(&self) -> &ConnectOptions {
49        &self.opts
50    }
51
52    /// Drains pending bytes from the transport read buffer.
53    pub fn flush(&mut self) -> Result<()> {
54        let short = Duration::from_millis(50);
55        self.transport.set_read_timeout(short)?;
56        let mut chunk = [0u8; 256];
57        loop {
58            match self.transport.read(&mut chunk) {
59                Ok(0) => break,
60                Ok(_) => continue,
61                Err(Error::Timeout) => break,
62                Err(e) => return Err(e),
63            }
64        }
65        self.read_buffer.clear();
66        Ok(())
67    }
68
69    /// Writes a command without expecting a response.
70    pub fn write(&mut self, command: &str) -> Result<()> {
71        self.write_with_retry(command, false)
72    }
73
74    /// Sends a query and returns the response string.
75    pub fn query(&mut self, command: &str) -> Result<String> {
76        self.query_with_timeout(command, self.effective_read_timeout())
77    }
78
79    pub fn query_with_timeout(&mut self, command: &str, timeout: Duration) -> Result<String> {
80        self.write_with_retry(command, true)?;
81        let bytes = self.read_response(timeout)?;
82        Ok(String::from_utf8_lossy(&bytes).trim().to_string())
83    }
84
85    fn write_with_retry(&mut self, command: &str, idempotent: bool) -> Result<()> {
86        let payload = normalize_command(command, &self.opts.terminator);
87        let data = payload.as_bytes();
88
89        let mut attempts = 0;
90        let max_attempts = max_write_attempts(idempotent, self.opts.retries);
91
92        loop {
93            attempts += 1;
94            let started = Instant::now();
95            self.pending_command = Some(command.to_string());
96            match self.transport.write(data) {
97                Ok(()) => {
98                    self.record_success(CommsEventKind::WriteOk, Some(command), attempts, started);
99                    return Ok(());
100                }
101                Err(Error::Timeout) if attempts < max_attempts => {
102                    self.record_failure(
103                        CommsEventKind::Timeout,
104                        Some(command),
105                        attempts,
106                        started,
107                        "write timeout",
108                    );
109                    if self.opts.reconnect_on_failure {
110                        let _ = self.transport.reconnect();
111                        self.record_reconnect();
112                    }
113                    thread::sleep(self.opts.retry_backoff);
114                }
115                Err(Error::Timeout) => {
116                    self.record_failure(
117                        CommsEventKind::Timeout,
118                        Some(command),
119                        attempts,
120                        started,
121                        "write timeout",
122                    );
123                    return Err(Error::Timeout);
124                }
125                Err(e) => {
126                    self.record_failure(
127                        CommsEventKind::WriteFailed,
128                        Some(command),
129                        attempts,
130                        started,
131                        &e.to_string(),
132                    );
133                    return Err(e);
134                }
135            }
136        }
137    }
138
139    fn read_response(&mut self, timeout: Duration) -> Result<Vec<u8>> {
140        self.transport.set_read_timeout(timeout)?;
141        self.read_buffer.clear();
142        let command = self.pending_command.clone();
143
144        let mut chunk = [0u8; 1024];
145        loop {
146            let started = Instant::now();
147            match self.transport.read(&mut chunk) {
148                Ok(0) => thread::sleep(Duration::from_millis(1)),
149                Ok(n) => {
150                    self.read_buffer.extend_from_slice(&chunk[..n]);
151                    if let Ok((payload, _)) =
152                        extract_response(&self.read_buffer, &self.opts.terminator)
153                    {
154                        self.record_success(CommsEventKind::ReadOk, command.as_deref(), 1, started);
155                        return Ok(payload);
156                    }
157                }
158                Err(Error::Timeout) => {
159                    if !self.read_buffer.is_empty() {
160                        if let Ok((payload, _)) =
161                            extract_response(&self.read_buffer, &self.opts.terminator)
162                        {
163                            self.record_success(
164                                CommsEventKind::ReadOk,
165                                command.as_deref(),
166                                1,
167                                started,
168                            );
169                            return Ok(payload);
170                        }
171                    }
172                    if self.opts.reconnect_on_failure {
173                        let _ = self.transport.reconnect();
174                        self.record_reconnect();
175                    }
176                    self.record_failure(
177                        CommsEventKind::Timeout,
178                        command.as_deref(),
179                        1,
180                        started,
181                        "read timeout",
182                    );
183                    return Err(Error::Timeout);
184                }
185                Err(e) => {
186                    self.record_failure(
187                        CommsEventKind::ReadFailed,
188                        command.as_deref(),
189                        1,
190                        started,
191                        &e.to_string(),
192                    );
193                    return Err(e);
194                }
195            }
196        }
197    }
198
199    fn record_success(
200        &self,
201        kind: CommsEventKind,
202        command: Option<&str>,
203        attempt: u32,
204        started: Instant,
205    ) {
206        if let Some(diag) = &self.diagnostics {
207            diag.record_success(kind, command, attempt, started.elapsed());
208        }
209    }
210
211    fn record_failure(
212        &self,
213        kind: CommsEventKind,
214        command: Option<&str>,
215        attempt: u32,
216        started: Instant,
217        detail: &str,
218    ) {
219        if let Some(diag) = &self.diagnostics {
220            diag.record_failure(kind, command, attempt, started.elapsed(), detail);
221        }
222    }
223
224    fn record_reconnect(&self) {
225        if let Some(diag) = &self.diagnostics {
226            diag.record_success(CommsEventKind::Reconnect, None, 1, Duration::ZERO);
227        }
228    }
229
230    fn effective_read_timeout(&self) -> Duration {
231        self.opts.per_op_timeout.unwrap_or(self.opts.read_timeout)
232    }
233
234    /// Probes and caches whether SYST:ERR? is supported.
235    pub fn probe_syst_err(&mut self) -> bool {
236        if let Some(v) = self.capabilities.syst_err {
237            return v;
238        }
239        let supported = self
240            .query_with_timeout("SYST:ERR?", Duration::from_millis(500))
241            .is_ok();
242        self.capabilities.syst_err = Some(supported);
243        supported
244    }
245
246    /// Probes and caches whether *OPC? is supported.
247    pub fn probe_opc(&mut self) -> bool {
248        if let Some(v) = self.capabilities.opc {
249            return v;
250        }
251        let supported = self
252            .query_with_timeout("*OPC?", Duration::from_millis(500))
253            .is_ok();
254        self.capabilities.opc = Some(supported);
255        supported
256    }
257
258    /// Drains the instrument error queue when supported.
259    pub fn check_errors(&mut self) -> Result<Vec<String>> {
260        if !self.probe_syst_err() {
261            return Ok(Vec::new());
262        }
263        let mut errors = Vec::new();
264        loop {
265            let resp = self.query("SYST:ERR?")?;
266            if resp.starts_with("0,") || resp.starts_with("+0,") {
267                break;
268            }
269            errors.push(resp);
270            if errors.len() > 50 {
271                break;
272            }
273        }
274        Ok(errors)
275    }
276
277    /// Parses a numeric SCPI response.
278    pub fn parse_f64(response: &str) -> Result<f64> {
279        parse_f64(response)
280    }
281}
instrument_core/scpi/session.rs

instrument_core/scpi/
session.rs
