From b7b3a84f37cc94494e17ea2098f72d6317090c53 Mon Sep 17 00:00:00 2001 From: Christian Kolset Date: Thu, 13 Aug 2026 10:28:25 -0600 Subject: Added DSCUSB Protocol. --- api_layers/protocols/dscusb.py | 73 ++++++++++++++++++++++++++++++++++++ devices/serial_device.py | 85 ++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 155 insertions(+), 3 deletions(-) create mode 100644 api_layers/protocols/dscusb.py diff --git a/api_layers/protocols/dscusb.py b/api_layers/protocols/dscusb.py new file mode 100644 index 0000000..5601338 --- /dev/null +++ b/api_layers/protocols/dscusb.py @@ -0,0 +1,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 diff --git a/devices/serial_device.py b/devices/serial_device.py index 61c3a64..cef34e8 100644 --- a/devices/serial_device.py +++ b/devices/serial_device.py @@ -28,7 +28,7 @@ from PyQt6.QtWidgets import ( QCheckBox, QComboBox, QDoubleSpinBox, QFormLayout, QGroupBox, QHBoxLayout, QHeaderView, QLabel, QLineEdit, QListWidget, QListWidgetItem, QPushButton, QScrollArea, QSizePolicy, QSpinBox, - QStackedWidget, QTableWidget, QTableWidgetItem, QVBoxLayout, QWidget, + QStackedWidget, QTableWidget, QTableWidgetItem, QTextEdit, QVBoxLayout, QWidget, ) from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus @@ -46,11 +46,12 @@ _FORMAT_LABELS: Dict[str, str] = { "Modbus RTU": "modbus_rtu", "ASCII Mark-10": "mark10", "ASCII CML": "cml", + "ASCII DSCUSB": "dscusb", } _LABEL_FOR: Dict[str, str] = {v: k for k, v in _FORMAT_LABELS.items()} _GENERIC_FORMATS = {"key:val", "csv", "json"} -_PROTOCOL_FORMATS = {"scpi", "modbus_rtu", "mark10", "cml"} +_PROTOCOL_FORMATS = {"scpi", "modbus_rtu", "mark10", "cml", "dscusb"} # parse_format → auto-name label — fallback only, used when the OS can't # describe the port (simulation mode, or the device isn't enumerable) @@ -59,6 +60,7 @@ _AUTO_NAME_LABEL: Dict[str, str] = { "scpi": "SCPI Serial Device", "modbus_rtu": "Modbus Serial Device", "cml": "CML Serial Device", + "dscusb": "DSCUSB Extensometer", } @@ -82,6 +84,7 @@ _STACK_IDX = { "modbus_rtu": 2, "mark10": 3, "cml": 4, + "dscusb": 5, } @@ -116,6 +119,7 @@ class SerialDevice(BaseDevice): self._baud = baud_rate self._fmt = parse_format self.simulate = simulate + self._last_error = "" # Generic self._num_channels = num_channels @@ -149,7 +153,10 @@ class SerialDevice(BaseDevice): # ── BaseDevice ──────────────────────────────────────────────────────── def connect(self) -> bool: + self._last_error = "" ok = self._layer.connect() + if not ok and hasattr(self._layer, "last_error"): + self._last_error = self._layer.last_error self.status = DeviceStatus.SIMULATED if self.simulate else ( DeviceStatus.CONNECTED if ok else DeviceStatus.ERROR ) @@ -220,6 +227,8 @@ class SerialDevice(BaseDevice): }) elif self._fmt == "mark10": pass # no extra config + elif self._fmt == "dscusb": + pass # no extra config elif self._fmt == "cml": cfg["motors"] = self._motors return cfg @@ -320,6 +329,10 @@ class SerialDevice(BaseDevice): ChannelConfig("zero", "Zero Gauge", "", 0.0, 1.0, color=_COLORS[2], writable=True), ChannelConfig("cycle_units", "Cycle Units", "", 0.0, 1.0, color=_COLORS[3], writable=True), ] + elif fmt == "dscusb": + return [ + ChannelConfig("extension", "Extension", "", -50.0, 50.0, color=_COLORS[0]), + ] elif fmt == "cml": _CMD_UNITS = {"TP": "counts", "TV": "counts/s", "TC": "%×10", "TS": "flags"} _WRITE_UNITS = {"VS": "counts/s", "MA": "counts"} @@ -407,6 +420,9 @@ class SerialDevice(BaseDevice): elif fmt == "mark10": from api_layers.protocols.mark10 import Mark10Layer return Mark10Layer(self._port, self._baud, simulate=self.simulate) + elif fmt == "dscusb": + from api_layers.protocols.dscusb import DSCUSBLayer + return DSCUSBLayer(self._port, self._baud, simulate=self.simulate) elif fmt == "cml": from api_layers.protocols.cml import CMLLayer, CMLMotor cml_motors = [ @@ -496,8 +512,10 @@ class SerialConfigWidget(QWidget): self._modbus_panel = _ModbusPanel(device=self.device) self._mark10_panel = _Mark10Panel() self._cml_panel = _CMLPanel(device=self.device) + self._dscusb_panel = _DSCUSBPanel() for panel in (self._generic_panel, self._scpi_panel, - self._modbus_panel, self._mark10_panel, self._cml_panel): + self._modbus_panel, self._mark10_panel, self._cml_panel, + self._dscusb_panel): self._proto_stack.addWidget(panel) self._proto_stack.setCurrentIndex(_STACK_IDX.get(self.device._fmt, 0)) sp = self._proto_stack.sizePolicy() @@ -612,6 +630,42 @@ class _GenericPanel(QWidget): self._ch_spin.setValue(len(device.info.channels) if device._fmt in _GENERIC_FORMATS else 4) lay.addRow("Signals:", self._ch_spin) + # ── Diagnostics ─────────────────────────────────────────────────── + # Connected but no channels showing values? This is where to look — + # shows exactly what's arriving on the wire, so you can see whether + # it matches the expected key:val/csv/json shape at all. + diag_grp = QGroupBox("Connection Diagnostics") + diag_lay = QVBoxLayout(diag_grp) + diag_lay.setContentsMargins(10, 12, 10, 10) + self._diag_box = QTextEdit() + self._diag_box.setObjectName("codeEditor") + self._diag_box.setReadOnly(True) + self._diag_box.setMaximumHeight(120) + diag_lay.addWidget(self._diag_box) + refresh_btn = QPushButton("🔍 Refresh") + refresh_btn.setToolTip("Show the last few raw lines received on this port") + refresh_btn.clicked.connect(self._refresh_diag) + diag_lay.addWidget(refresh_btn) + lay.addRow(diag_grp) + self._refresh_diag() + + def _refresh_diag(self): + lines = [f"Status: {self.device.status.value}"] + layer = getattr(self.device, "_layer", None) + raw = getattr(layer, "raw_lines", None) if layer is not None else None + if raw: + lines.append("Last lines received:") + for r in raw: + lines.append(f" {r!r}") + else: + lines.append( + "(no lines received yet — check the port is receiving data " + "at all, and that baud rate matches the device)" + ) + if getattr(self.device, "_last_error", ""): + lines.append(f"\nError:\n{self.device._last_error}") + self._diag_box.setPlainText("\n".join(lines)) + def collect(self) -> dict: return {"num_channels": self._ch_spin.value()} @@ -846,6 +900,31 @@ class _Mark10Panel(QWidget): lay.addStretch() +class _DSCUSBPanel(QWidget): + """Info panel for DSCUSB — no extra user config needed.""" + + def __init__(self): + super().__init__() + lay = QVBoxLayout(self) + lay.setContentsMargins(0, 4, 0, 4) + grp = QGroupBox("DSCUSB Protocol") + grp_lay = QVBoxLayout(grp) + lbl = QLabel( + "Sends '!001:SYS?\\r', parses a plain numeric response.\n\n" + "Fixed channel:\n" + " extension — raw reading, unit as reported by the device\n" + " (add a Scale/Offset Channel downstream for\n" + " unit conversion, e.g. ×25.4 for in→mm)\n\n" + "Response is CR-terminated only (no LF) — this protocol reads\n" + "with read_until(CR), not the generic reader's readline()." + ) + lbl.setObjectName("traceSource") + lbl.setWordWrap(True) + grp_lay.addWidget(lbl) + lay.addWidget(grp) + lay.addStretch() + + class _CMLPanel(QWidget): """Motor list for CoolMuscle CML.""" -- cgit v1.2.3