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
|
"""
api_layers/protocols/dscusb.py
DSCUSB displacement transducer (extensometer) ASCII protocol layer.
Command (sent with CR terminator):
!001:SYS? — request current reading ("001" = device address; hardcoded
for now — fine for one unit per COM port, which is the only
configuration seen so far)
Response: a plain ASCII number, no unit suffix — read up to CR with
Serial.read_until(b"\\r") rather than readline(), since pyserial's
readline() looks for LF by default and this device only terminates with
CR. Using readline() here is the failure mode that was actually observed:
port opens and shows CONNECTED fine, but zero lines are ever received,
because readline() never sees the LF it's waiting for. Same reasoning as
Mark-10 (api_layers/protocols/mark10.py) — also CR-only.
Reports the raw device reading, unit unconverted (no unit is reported by
the instrument itself). If you need e.g. inches→mm, apply a Scale/Offset
Channel operation downstream (Configuration > Channels) rather than
hardcoding a conversion here — keeps this layer honest about what the
device actually said. See docs/developer-new-protocols.md.
Channels exposed:
extension — current raw reading
"""
import math
import re
from typing import Dict
from api_layers.protocols.base_protocol import BaseProtocol
_QUERY = b"!001:SYS?\r"
_RESP_RE = re.compile(r"[+-]?\d+\.?\d*(?:[eE][+-]?\d+)?")
class DSCUSBLayer(BaseProtocol):
def __init__(
self,
port: str,
baud: int = 115200,
poll_interval: float = 0.05,
simulate: bool = True,
):
super().__init__(port, baud, poll_interval, simulate)
# ── Protocol ──────────────────────────────────────────────────────────
def _poll(self) -> Dict[str, float]:
try:
with self._io_lock:
self._ser.reset_input_buffer()
self._ser.write(_QUERY)
self._ser.flush()
resp = self._ser.read_until(b"\r").decode(errors="replace").strip()
return self._parse(resp)
except Exception:
return {}
def _parse(self, resp: str) -> Dict[str, float]:
m = _RESP_RE.search(resp)
if not m:
return {}
return {"extension": float(m.group())}
def _simulate(self, t: float) -> Dict[str, float]:
return {"extension": math.sin(t * 0.4) * 2.0 + 2.0}
def write(self, channel_id: str, value) -> bool:
return False # no write-capable channels on this device
|
