From 67cfa0a514c7de4605ed7360e15a81aa781e510e Mon Sep 17 00:00:00 2001 From: Christian Kolset Date: Fri, 24 Apr 2026 11:51:26 -0600 Subject: Added difference serial protocols --- devices/serial_device.py | 866 ++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 739 insertions(+), 127 deletions(-) (limited to 'devices/serial_device.py') diff --git a/devices/serial_device.py b/devices/serial_device.py index 6f54cb5..78f74a2 100644 --- a/devices/serial_device.py +++ b/devices/serial_device.py @@ -1,27 +1,34 @@ """ devices/serial_device.py -Generic Serial / UART device module. - -Uses ArduinoLayer for communication, but works with ANY instrument -that sends newline-terminated data. Configurable parse formats: - • "csv" – plain comma-separated values mapped to channels in order - • "key:val" – "CH0:1.23,CH1:4.56" key-colon-value pairs - • "json" – {"CH0":1.23,"CH1":4.56} - -Switch format in the config widget without restarting. +Generic Serial / UART device. + +Parse formats (select in config widget): + Generic stream (uses ArduinoLayer): + key:val — "CH0:1.23,CH1:4.56" + csv — "1.23,4.56" (positional) + json — {"CH0":1.23,"CH1":4.56} + + Instrument protocols (uses api_layers/protocols/): + scpi — SCPI query/response (configurable per-channel queries) + modbus_rtu — Modbus RTU binary (configurable register map) + mark10 — ASCII Mark-10 (Series 5 tensile tester) + cml — ASCII CML (CoolMuscle servo motors) """ -import json +from __future__ import annotations + import math -import random import threading import time -from typing import Any, Dict, List +from typing import Any, Dict, List, Optional +from PyQt6.QtCore import Qt, QThread, pyqtSignal from PyQt6.QtWidgets import ( - QWidget, QVBoxLayout, QFormLayout, QGroupBox, - QComboBox, QLineEdit, QSpinBox, QLabel, QPushButton, + QCheckBox, QComboBox, QDoubleSpinBox, QFormLayout, QGroupBox, + QHBoxLayout, QHeaderView, QLabel, QLineEdit, QListWidget, + QListWidgetItem, QPushButton, QScrollArea, QSpinBox, + QStackedWidget, QTableWidget, QTableWidgetItem, QVBoxLayout, QWidget, ) from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus @@ -30,6 +37,34 @@ from api_layers.arduino_layer import ArduinoLayer _COLORS = ["#7fff6e", "#4cc9f0", "#f72585", "#00d4ff", "#ffcc00", "#c77dff", "#ff6b35", "#38b000"] +# Human-readable label → internal format key +_FORMAT_LABELS: Dict[str, str] = { + "Key:Value (Arduino)": "key:val", + "CSV": "csv", + "JSON": "json", + "SCPI": "scpi", + "Modbus RTU": "modbus_rtu", + "ASCII Mark-10": "mark10", + "ASCII CML": "cml", +} +_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"} + +# Index into the config-widget stacked panel +_STACK_IDX = { + "key:val": 0, + "csv": 0, + "json": 0, + "scpi": 1, + "modbus_rtu": 2, + "mark10": 3, + "cml": 4, +} + + +# ── SerialDevice ────────────────────────────────────────────────────────────── class SerialDevice(BaseDevice): DEVICE_TYPE = "serial" @@ -37,51 +72,62 @@ class SerialDevice(BaseDevice): def __init__( self, - device_id: str = "ser_0", - port: str = "COM3", - baud_rate: int = 115200, - num_channels: int = 4, - channel_names: List[str] = None, - units: List[str] = None, - parse_format: str = "key:val", # "csv" | "key:val" | "json" - simulate: bool = True, + device_id: str = "ser_0", + port: str = "COM3", + baud_rate: int = 115200, + parse_format: str = "key:val", + simulate: bool = True, + # ── Generic stream options ────────────────────────────────────── + num_channels: int = 4, + channel_names: List[str] = None, + units: List[str] = None, + # ── SCPI ─────────────────────────────────────────────────────── + scpi_channels: List[Dict] = None, + # ── Modbus RTU ───────────────────────────────────────────────── + slave_addr: int = 1, + parity: str = "N", + stopbits: int = 1, + modbus_channels: List[Dict] = None, + # ── CML ──────────────────────────────────────────────────────── + motors: List[Dict] = None, ): self._port = port self._baud = baud_rate - self._parse_format = parse_format + self._fmt = parse_format self.simulate = simulate - names = channel_names or [f"CH{i}" for i in range(num_channels)] - _units = units or ["" for _ in range(num_channels)] + # Generic + self._num_channels = num_channels + self._chan_names = channel_names + self._units = units - channels = [ - ChannelConfig( - channel_id=names[i], name=names[i], unit=_units[i], - min_value=0.0, max_value=1023.0, - color=_COLORS[i % len(_COLORS)], - ) - for i in range(num_channels) - ] + # SCPI + self._scpi_channels: List[Dict] = scpi_channels or _default_scpi() + + # Modbus + self._slave_addr = slave_addr + self._parity = parity + self._stopbits = stopbits + self._mb_channels: List[Dict] = modbus_channels or _default_modbus() + # CML + self._motors: List[Dict] = motors or _default_motors() + + channels = self._build_channel_configs() info = DeviceInfo( - device_id=device_id, name="Serial / UART", + device_id=device_id, + name="Serial / UART", device_type=self.DEVICE_TYPE, - description=f"{port} @ {baud_rate}", - icon=self.ICON, channels=channels, + description=self._description(), + icon=self.ICON, + channels=channels, ) super().__init__(info) + self._layer = self._make_layer() - self._layer = ArduinoLayer( - port=port, baud=baud_rate, - analog_pins=[ch.channel_id for ch in channels], - simulate=simulate, - ) - self._t0 = 0.0 - - # ── BaseDevice ────────────────────────────────────────────────────── + # ── BaseDevice ──────────────────────────────────────────────────────── def connect(self) -> bool: - self._t0 = time.time() ok = self._layer.connect() self.status = DeviceStatus.SIMULATED if self.simulate else ( DeviceStatus.CONNECTED if ok else DeviceStatus.ERROR @@ -94,53 +140,224 @@ class SerialDevice(BaseDevice): def read_channels(self) -> Dict[str, float]: raw = self._layer.read() - # Map by order if keys don't match channel IDs - if raw: - mapped: Dict[str, float] = {} - raw_vals = list(raw.values()) - for i, ch in enumerate(self.info.channels): - if ch.channel_id in raw: - mapped[ch.channel_id] = raw[ch.channel_id] - elif i < len(raw_vals): - mapped[ch.channel_id] = raw_vals[i] - return mapped - return {} + if not raw: + return {} + # Protocol layers already use channel_id keys — pass through. + # Generic (ArduinoLayer) may use arbitrary names — remap by position. + mapped: Dict[str, float] = {} + raw_vals = list(raw.values()) + for i, ch in enumerate(self.info.channels): + if ch.channel_id in raw: + mapped[ch.channel_id] = raw[ch.channel_id] + elif i < len(raw_vals): + mapped[ch.channel_id] = raw_vals[i] + return mapped def write_channel(self, channel_id: str, value: Any) -> bool: return self._layer.write(channel_id, int(value)) - def get_save_config(self) -> dict: - return { - "device_type": self.DEVICE_TYPE, - "device_id": self.info.device_id, - "port": self._port, - "baud_rate": self._baud, - "num_channels": len(self.info.channels), - "channel_names": [c.channel_id for c in self.info.channels], - "units": [c.unit for c in self.info.channels], - "parse_format": self._parse_format, - "simulate": self.simulate, - } - def get_config_widget(self) -> QWidget: return SerialConfigWidget(self) - def reconfigure(self, port: str, baud: int, fmt: str, simulate: bool): + def get_save_config(self) -> dict: + cfg: Dict[str, Any] = { + "device_type": self.DEVICE_TYPE, + "device_id": self.info.device_id, + "port": self._port, + "baud_rate": self._baud, + "parse_format": self._fmt, + "simulate": self.simulate, + } + if self._fmt in _GENERIC_FORMATS: + cfg.update({ + "num_channels": len(self.info.channels), + "channel_names": [c.channel_id for c in self.info.channels], + "units": [c.unit for c in self.info.channels], + }) + elif self._fmt == "scpi": + cfg["scpi_channels"] = self._scpi_channels + elif self._fmt == "modbus_rtu": + cfg.update({ + "slave_addr": self._slave_addr, + "parity": self._parity, + "stopbits": self._stopbits, + "modbus_channels": self._mb_channels, + }) + elif self._fmt == "mark10": + pass # no extra config + elif self._fmt == "cml": + cfg["motors"] = self._motors + return cfg + + # ── Reconfigure ─────────────────────────────────────────────────────── + + def reconfigure( + self, + port: str, + baud: int, + fmt: str, + simulate: bool, + num_channels: int = 4, + channel_names: List[str] = None, + units: List[str] = None, + scpi_channels: List[Dict] = None, + slave_addr: int = 1, + parity: str = "N", + stopbits: int = 1, + modbus_channels: List[Dict] = None, + motors: List[Dict] = None, + ): was_on = self.status in (DeviceStatus.CONNECTED, DeviceStatus.SIMULATED) if was_on: self.disconnect() + self._port = port self._baud = baud - self._parse_format = fmt + self._fmt = fmt self.simulate = simulate - self._layer = ArduinoLayer( - port=port, baud=baud, - analog_pins=[ch.channel_id for ch in self.info.channels], - simulate=simulate, - ) + self._num_channels = num_channels + self._chan_names = channel_names + self._units = units + self._scpi_channels = scpi_channels or self._scpi_channels + self._slave_addr = slave_addr + self._parity = parity + self._stopbits = stopbits + self._mb_channels = modbus_channels or self._mb_channels + self._motors = motors or self._motors + + self.info.channels = self._build_channel_configs() + self.info.description = self._description() + self._layer = self._make_layer() + if was_on: self.connect() + # ── Internal ────────────────────────────────────────────────────────── + + def _description(self) -> str: + label = _LABEL_FOR.get(self._fmt, self._fmt) + return f"{self._port} @ {self._baud} [{label}]" + + def _build_channel_configs(self) -> List[ChannelConfig]: + fmt = self._fmt + if fmt in _GENERIC_FORMATS: + names = self._chan_names or [f"CH{i}" for i in range(self._num_channels)] + _units = self._units or [""] * len(names) + return [ + ChannelConfig( + channel_id=names[i], name=names[i], unit=_units[i], + min_value=0.0, max_value=1023.0, + color=_COLORS[i % len(_COLORS)], + ) + for i in range(len(names)) + ] + elif fmt == "scpi": + return [ + ChannelConfig( + channel_id=sc["channel_id"], + name=sc.get("name", sc["channel_id"]), + unit=sc.get("unit", ""), + min_value=-1e9, max_value=1e9, + color=_COLORS[i % len(_COLORS)], + ) + for i, sc in enumerate(self._scpi_channels) + ] + elif fmt == "modbus_rtu": + return [ + ChannelConfig( + channel_id=mc["channel_id"], + name=mc.get("name", mc["channel_id"]), + unit=mc.get("unit", ""), + min_value=-1e9, max_value=1e9, + color=_COLORS[i % len(_COLORS)], + ) + for i, mc in enumerate(self._mb_channels) + ] + elif fmt == "mark10": + from api_layers.protocols.mark10 import UNITS as _MARK10_UNITS + return [ + ChannelConfig("force", "Force", "N", -5000.0, 5000.0, color=_COLORS[0]), + ChannelConfig("unit_code", "Unit Code", "", 0.0, float(len(_MARK10_UNITS) - 1), color=_COLORS[1]), + ] + elif fmt == "cml": + _CMD_UNITS = {"TP": "counts", "TV": "counts/s", "TC": "%×10", "TS": "flags"} + channels = [] + color_idx = 0 + for motor in self._motors: + for cmd in motor.get("read_cmds", ["TP", "TV", "TC"]): + mid = motor.get("motor_id", "M1") + channels.append(ChannelConfig( + channel_id=f"{mid}_{cmd}", + name=f"{mid} {cmd}", + unit=_CMD_UNITS.get(cmd, ""), + min_value=-1e6, max_value=1e6, + color=_COLORS[color_idx % len(_COLORS)], + )) + color_idx += 1 + return channels or [ChannelConfig("M1_TP", "M1 TP", "counts", color=_COLORS[0])] + return [] + + def _make_layer(self): + fmt = self._fmt + if fmt in _GENERIC_FORMATS: + names = [ch.channel_id for ch in self.info.channels] + return ArduinoLayer( + port=self._port, baud=self._baud, + analog_pins=names, simulate=self.simulate, + ) + elif fmt == "scpi": + from api_layers.protocols.scpi import SCPILayer, SCPIChannel + chs = [ + SCPIChannel( + channel_id=sc["channel_id"], + name=sc.get("name", sc["channel_id"]), + query=sc.get("query", ""), + unit=sc.get("unit", ""), + scale=sc.get("scale", 1.0), + write_cmd=sc.get("write_cmd", ""), + ) + for sc in self._scpi_channels + ] + return SCPILayer(self._port, self._baud, chs, simulate=self.simulate) + elif fmt == "modbus_rtu": + from api_layers.protocols.modbus_rtu import ModbusRTULayer, ModbusChannel + chs = [ + ModbusChannel( + channel_id=mc["channel_id"], + name=mc.get("name", mc["channel_id"]), + register=mc.get("register", 0), + function_code=mc.get("function_code", 0x03), + data_type=mc.get("data_type", "uint16"), + scale=mc.get("scale", 1.0), + offset=mc.get("offset", 0.0), + unit=mc.get("unit", ""), + ) + for mc in self._mb_channels + ] + return ModbusRTULayer( + self._port, self._baud, + slave_addr=self._slave_addr, channels=chs, + parity=self._parity, stopbits=self._stopbits, + simulate=self.simulate, + ) + elif fmt == "mark10": + from api_layers.protocols.mark10 import Mark10Layer + return Mark10Layer(self._port, self._baud, simulate=self.simulate) + elif fmt == "cml": + from api_layers.protocols.cml import CMLLayer, CMLMotor + cml_motors = [ + CMLMotor( + motor_id=m.get("motor_id", "M1"), + address=m.get("address", 1), + read_cmds=m.get("read_cmds", ["TP", "TV", "TC"]), + ) + for m in self._motors + ] + return CMLLayer(self._port, self._baud, cml_motors, simulate=self.simulate) + raise ValueError(f"Unknown parse_format: {fmt!r}") + + +# ── Config widget ───────────────────────────────────────────────────────────── class SerialConfigWidget(QWidget): def __init__(self, device: SerialDevice): @@ -150,44 +367,54 @@ class SerialConfigWidget(QWidget): self._build() def _build(self): - from PyQt6.QtWidgets import QListWidget, QListWidgetItem - from PyQt6.QtCore import Qt as _Qt - root = QVBoxLayout(self) root.setContentsMargins(8, 8, 8, 8) root.setSpacing(8) - self.setMinimumWidth(420) - - # ── Port settings ───────────────────────────────────────────── - grp = QGroupBox("Port Settings") - form = QFormLayout(grp) - - self.port_edit = QLineEdit(self.device._port) - self.port_edit.setPlaceholderText("e.g. COM3 or /dev/ttyUSB0") - form.addRow("Port:", self.port_edit) - - self.baud_cb = QComboBox() - self.baud_cb.addItems(["9600", "19200", "38400", "57600", "115200", "230400", "460800"]) - self.baud_cb.setCurrentText(str(self.device._baud)) - form.addRow("Baud Rate:", self.baud_cb) - - self.fmt_cb = QComboBox() - self.fmt_cb.addItems(["key:val", "csv", "json"]) - self.fmt_cb.setCurrentText(self.device._parse_format) - form.addRow("Parse Format:", self.fmt_cb) - - self.sim_cb = QComboBox() - self.sim_cb.addItems(["Simulate", "Real Hardware"]) - self.sim_cb.setCurrentIndex(0 if self.device.simulate else 1) - form.addRow("Mode:", self.sim_cb) - - root.addWidget(grp) - - # ── Port scanner ────────────────────────────────────────────── + self.setMinimumWidth(480) + + # ── Common connection settings ───────────────────────────────────── + conn_grp = QGroupBox("Port Settings") + conn_form = QFormLayout(conn_grp) + + self._port_edit = QLineEdit(self.device._port) + self._port_edit.setPlaceholderText("e.g. COM3 or /dev/ttyUSB0") + conn_form.addRow("Port:", self._port_edit) + + self._baud_cb = QComboBox() + self._baud_cb.addItems(["1200", "2400", "4800", "9600", "19200", "38400", + "57600", "115200", "230400", "460800"]) + self._baud_cb.setCurrentText(str(self.device._baud)) + conn_form.addRow("Baud Rate:", self._baud_cb) + + self._fmt_cb = QComboBox() + self._fmt_cb.addItems(list(_FORMAT_LABELS.keys())) + self._fmt_cb.setCurrentText(_LABEL_FOR.get(self.device._fmt, "Key:Value (Arduino)")) + self._fmt_cb.currentIndexChanged.connect(self._on_fmt_changed) + conn_form.addRow("Protocol / Format:", self._fmt_cb) + + self._sim_cb = QComboBox() + self._sim_cb.addItems(["Simulate", "Real Hardware"]) + self._sim_cb.setCurrentIndex(0 if self.device.simulate else 1) + conn_form.addRow("Mode:", self._sim_cb) + + root.addWidget(conn_grp) + + # ── Protocol-specific settings (stacked) ────────────────────────── + self._proto_stack = QStackedWidget() + self._generic_panel = _GenericPanel(device=self.device) + self._scpi_panel = _SCPIPanel(device=self.device) + self._modbus_panel = _ModbusPanel(device=self.device) + self._mark10_panel = _Mark10Panel() + self._cml_panel = _CMLPanel(device=self.device) + for panel in (self._generic_panel, self._scpi_panel, + self._modbus_panel, self._mark10_panel, self._cml_panel): + self._proto_stack.addWidget(panel) + self._proto_stack.setCurrentIndex(_STACK_IDX.get(self.device._fmt, 0)) + root.addWidget(self._proto_stack) + + # ── Port scanner ─────────────────────────────────────────────────── scan_grp = QGroupBox("Available Serial Ports") scan_lay = QVBoxLayout(scan_grp) - scan_lay.setSpacing(4) - scan_top = QHBoxLayout() self._scan_btn = QPushButton("🔍 Scan Ports") self._scan_btn.setObjectName("addTraceBtn") @@ -197,14 +424,12 @@ class SerialConfigWidget(QWidget): scan_top.addWidget(self._scan_btn) scan_top.addWidget(self._scan_lbl, 1) scan_lay.addLayout(scan_top) - self._port_list = QListWidget() self._port_list.setObjectName("portList") - self._port_list.setMaximumHeight(110) + self._port_list.setMaximumHeight(100) self._port_list.itemClicked.connect( - lambda item: self.port_edit.setText( - item.data(_Qt.ItemDataRole.UserRole) or "" - ) if item.data(_Qt.ItemDataRole.UserRole) else None + lambda item: self._port_edit.setText(item.data(Qt.ItemDataRole.UserRole) or "") + if item.data(Qt.ItemDataRole.UserRole) else None ) hint = QLabel("↑ Click a port above to select it") hint.setObjectName("traceSource") @@ -212,33 +437,55 @@ class SerialConfigWidget(QWidget): scan_lay.addWidget(hint) root.addWidget(scan_grp) - # ── Apply ───────────────────────────────────────────────────── + # ── Apply ────────────────────────────────────────────────────────── apply_btn = QPushButton("Apply & Reconnect") apply_btn.setObjectName("applyButton") apply_btn.clicked.connect(self._apply) root.addWidget(apply_btn) root.addStretch() + def _on_fmt_changed(self, _): + fmt = _FORMAT_LABELS.get(self._fmt_cb.currentText(), "key:val") + self._proto_stack.setCurrentIndex(_STACK_IDX.get(fmt, 0)) + + def _current_fmt(self) -> str: + return _FORMAT_LABELS.get(self._fmt_cb.currentText(), "key:val") + + def _apply(self): + fmt = self._current_fmt() + kwargs: Dict[str, Any] = dict( + port=self._port_edit.text().strip() or "COM3", + baud=int(self._baud_cb.currentText()), + fmt=fmt, + simulate=(self._sim_cb.currentIndex() == 0), + ) + if fmt in _GENERIC_FORMATS: + kwargs.update(self._generic_panel.collect()) + elif fmt == "scpi": + kwargs["scpi_channels"] = self._scpi_panel.collect() + elif fmt == "modbus_rtu": + kwargs.update(self._modbus_panel.collect()) + elif fmt == "cml": + kwargs["motors"] = self._cml_panel.collect() + self.device.reconfigure(**kwargs) + def _scan(self): self._scan_btn.setEnabled(False) self._scan_lbl.setText("Scanning…") self._port_list.clear() - from PyQt6.QtCore import QThread, pyqtSignal as _sig class _Scanner(QThread): - done = _sig(list) - def run(self): - self.done.emit(ArduinoLayer.list_ports()) + done = pyqtSignal(list) + def run(self): self.done.emit(ArduinoLayer.list_ports()) self._scanner = _Scanner() self._scanner.done.connect(self._on_scan_done) self._scanner.start() def _on_scan_done(self, ports): - from PyQt6.QtWidgets import QListWidgetItem self._scan_btn.setEnabled(True) self._port_list.clear() if not ports: self._scan_lbl.setText("No ports found") - item = QListWidgetItem(" No serial ports detected — check USB / permissions") + item = QListWidgetItem(" No serial ports detected") item.setFlags(item.flags() & ~Qt.ItemFlag.ItemIsSelectable) self._port_list.addItem(item) else: @@ -251,10 +498,375 @@ class SerialConfigWidget(QWidget): item.setData(Qt.ItemDataRole.UserRole, device) self._port_list.addItem(item) - def _apply(self): - self.device.reconfigure( - port=self.port_edit.text().strip() or "COM3", - baud=int(self.baud_cb.currentText()), - fmt=self.fmt_cb.currentText(), - simulate=(self.sim_cb.currentIndex() == 0), + +# ── Protocol-specific panels ────────────────────────────────────────────────── + +class _GenericPanel(QWidget): + """Settings for key:val / csv / json formats.""" + + def __init__(self, device: SerialDevice): + super().__init__() + self.device = device + lay = QFormLayout(self) + lay.setContentsMargins(0, 4, 0, 4) + + self._ch_spin = QSpinBox() + self._ch_spin.setRange(1, 32) + self._ch_spin.setValue(len(device.info.channels) if device._fmt in _GENERIC_FORMATS else 4) + lay.addRow("Channels:", self._ch_spin) + + def collect(self) -> dict: + return {"num_channels": self._ch_spin.value()} + + +class _SCPIPanel(QWidget): + """Per-channel query table for SCPI format.""" + + def __init__(self, device: SerialDevice): + super().__init__() + self.device = device + lay = QVBoxLayout(self) + lay.setContentsMargins(0, 4, 0, 4) + lay.setSpacing(6) + + grp = QGroupBox("Channel Queries") + grp_lay = QVBoxLayout(grp) + + self._table = QTableWidget() + self._table.setColumnCount(5) + self._table.setHorizontalHeaderLabels(["ID", "Name", "Query", "Unit", "Scale"]) + self._table.horizontalHeader().setSectionResizeMode(2, QHeaderView.ResizeMode.Stretch) + self._table.setMinimumHeight(150) + grp_lay.addWidget(self._table) + + btn_row = QHBoxLayout() + add_btn = QPushButton("+ Add") + add_btn.setObjectName("addTraceBtn") + add_btn.clicked.connect(self._add_row) + del_btn = QPushButton("Remove") + del_btn.setObjectName("removeTraceBtn") + del_btn.clicked.connect(self._del_row) + + idn_btn = QPushButton("Query *IDN?") + idn_btn.setObjectName("addTraceBtn") + idn_btn.clicked.connect(self._query_idn) + self._idn_lbl = QLabel("—") + self._idn_lbl.setObjectName("traceSource") + + btn_row.addWidget(add_btn) + btn_row.addWidget(del_btn) + btn_row.addSpacing(12) + btn_row.addWidget(idn_btn) + btn_row.addWidget(self._idn_lbl, 1) + grp_lay.addLayout(btn_row) + lay.addWidget(grp) + + for sc in device._scpi_channels: + self._add_row(sc) + + def _add_row(self, sc: dict = None): + row = self._table.rowCount() + self._table.insertRow(row) + n = row + self._table.setItem(row, 0, QTableWidgetItem(sc["channel_id"] if sc else f"CH{n}")) + self._table.setItem(row, 1, QTableWidgetItem(sc.get("name", f"CH{n}") if sc else f"CH{n}")) + self._table.setItem(row, 2, QTableWidgetItem(sc.get("query", "MEAS:VOLT?") if sc else "MEAS:VOLT?")) + self._table.setItem(row, 3, QTableWidgetItem(sc.get("unit", "") if sc else "")) + self._table.setItem(row, 4, QTableWidgetItem(str(sc.get("scale", 1.0)) if sc else "1.0")) + + def _del_row(self): + row = self._table.currentRow() + if row >= 0: + self._table.removeRow(row) + + def collect(self) -> List[Dict]: + result = [] + for row in range(self._table.rowCount()): + def cell(c): return (self._table.item(row, c) or QTableWidgetItem("")).text().strip() + try: + scale = float(cell(4)) + except ValueError: + scale = 1.0 + result.append({ + "channel_id": cell(0) or f"CH{row}", + "name": cell(1) or f"CH{row}", + "query": cell(2), + "unit": cell(3), + "scale": scale, + "write_cmd": "", + }) + return result + + def _query_idn(self): + layer = self.device._layer + idn = getattr(layer, "query_idn", lambda: "—")() + self._idn_lbl.setText(idn or "—") + + +_FC_OPTIONS = {"Holding (FC03)": 0x03, "Input (FC04)": 0x04} +_TYPE_OPTIONS = ["uint16", "int16", "float32", "int32"] + + +class _ModbusPanel(QWidget): + """Slave address + register map table for Modbus RTU.""" + + def __init__(self, device: SerialDevice): + super().__init__() + self.device = device + lay = QVBoxLayout(self) + lay.setContentsMargins(0, 4, 0, 4) + lay.setSpacing(6) + + bus_form = QFormLayout() + self._slave_spin = QSpinBox() + self._slave_spin.setRange(1, 247) + self._slave_spin.setValue(device._slave_addr) + bus_form.addRow("Slave Address:", self._slave_spin) + + self._parity_cb = QComboBox() + self._parity_cb.addItems(["N — None", "E — Even", "O — Odd"]) + self._parity_cb.setCurrentIndex({"N": 0, "E": 1, "O": 2}.get(device._parity, 0)) + bus_form.addRow("Parity:", self._parity_cb) + + self._stop_cb = QComboBox() + self._stop_cb.addItems(["1", "2"]) + self._stop_cb.setCurrentText(str(device._stopbits)) + bus_form.addRow("Stop Bits:", self._stop_cb) + lay.addLayout(bus_form) + + grp = QGroupBox("Register Map") + grp_lay = QVBoxLayout(grp) + + self._table = QTableWidget() + self._table.setColumnCount(7) + self._table.setHorizontalHeaderLabels( + ["ID", "Name", "Register", "FC", "Type", "Scale", "Unit"] ) + self._table.horizontalHeader().setSectionResizeMode(1, QHeaderView.ResizeMode.Stretch) + self._table.setMinimumHeight(140) + grp_lay.addWidget(self._table) + + btn_row = QHBoxLayout() + add_btn = QPushButton("+ Add") + add_btn.setObjectName("addTraceBtn") + add_btn.clicked.connect(self._add_row) + del_btn = QPushButton("Remove") + del_btn.setObjectName("removeTraceBtn") + del_btn.clicked.connect(self._del_row) + btn_row.addWidget(add_btn) + btn_row.addWidget(del_btn) + btn_row.addStretch() + grp_lay.addLayout(btn_row) + lay.addWidget(grp) + + for mc in device._mb_channels: + self._add_row(mc) + + def _add_row(self, mc: dict = None): + row = self._table.rowCount() + self._table.insertRow(row) + self._table.setItem(row, 0, QTableWidgetItem(mc["channel_id"] if mc else f"R{row}")) + self._table.setItem(row, 1, QTableWidgetItem(mc.get("name", f"R{row}") if mc else f"R{row}")) + self._table.setItem(row, 2, QTableWidgetItem(str(mc.get("register", row * 2)) if mc else str(row * 2))) + + fc_cb = QComboBox() + fc_cb.addItems(list(_FC_OPTIONS.keys())) + if mc: + for label, code in _FC_OPTIONS.items(): + if code == mc.get("function_code", 0x03): + fc_cb.setCurrentText(label) + self._table.setCellWidget(row, 3, fc_cb) + + type_cb = QComboBox() + type_cb.addItems(_TYPE_OPTIONS) + if mc: + type_cb.setCurrentText(mc.get("data_type", "uint16")) + self._table.setCellWidget(row, 4, type_cb) + + self._table.setItem(row, 5, QTableWidgetItem(str(mc.get("scale", 1.0)) if mc else "1.0")) + self._table.setItem(row, 6, QTableWidgetItem(mc.get("unit", "") if mc else "")) + + def _del_row(self): + row = self._table.currentRow() + if row >= 0: + self._table.removeRow(row) + + def collect(self) -> dict: + parity_map = {"N — None": "N", "E — Even": "E", "O — Odd": "O"} + channels = [] + for row in range(self._table.rowCount()): + def cell(c): return (self._table.item(row, c) or QTableWidgetItem("")).text().strip() + def wtext(c): + w = self._table.cellWidget(row, c) + return w.currentText() if w else "" + try: + reg = int(cell(2)) + except ValueError: + reg = 0 + try: + scale = float(cell(5)) + except ValueError: + scale = 1.0 + channels.append({ + "channel_id": cell(0) or f"R{row}", + "name": cell(1) or f"R{row}", + "register": reg, + "function_code": _FC_OPTIONS.get(wtext(3), 0x03), + "data_type": wtext(4) or "uint16", + "scale": scale, + "offset": 0.0, + "unit": cell(6), + }) + return { + "slave_addr": self._slave_spin.value(), + "parity": parity_map.get(self._parity_cb.currentText(), "N"), + "stopbits": int(self._stop_cb.currentText()), + "modbus_channels": channels, + } + + +class _Mark10Panel(QWidget): + """Info panel for Mark-10 — no extra user config needed.""" + + def __init__(self): + super().__init__() + lay = QVBoxLayout(self) + lay.setContentsMargins(0, 4, 0, 4) + grp = QGroupBox("Mark-10 Series 5 Protocol") + grp_lay = QVBoxLayout(grp) + lbl = QLabel( + "Polls '?' command, parses '+X.XXX kgF' response.\n\n" + "Fixed channels:\n" + " force — current reading in active unit\n" + " unit_code — 0=lb 1=kgF 2=N 3=ozF\n\n" + "Serial: 8N1, no flow control.\n" + "Typical baud: 115200 (Series 5), 9600 (Series 3/4)." + ) + lbl.setObjectName("traceSource") + lbl.setWordWrap(True) + grp_lay.addWidget(lbl) + lay.addWidget(grp) + lay.addStretch() + + +class _CMLPanel(QWidget): + """Motor list for CoolMuscle CML.""" + + _READ_CMDS = ["TP", "TV", "TC", "TS"] + + def __init__(self, device: SerialDevice): + super().__init__() + self.device = device + self._rows: List[_MotorRow] = [] + lay = QVBoxLayout(self) + lay.setContentsMargins(0, 4, 0, 4) + lay.setSpacing(6) + + grp = QGroupBox("Motors (RS-232: addr=0 / RS-485: addr 1-31)") + grp_lay = QVBoxLayout(grp) + + scroll = QScrollArea() + scroll.setWidgetResizable(True) + scroll.setMaximumHeight(200) + container = QWidget() + self._container_lay = QVBoxLayout(container) + self._container_lay.setSpacing(2) + self._container_lay.addStretch() + scroll.setWidget(container) + grp_lay.addWidget(scroll) + + btn_row = QHBoxLayout() + add_btn = QPushButton("+ Add Motor") + add_btn.setObjectName("addTraceBtn") + add_btn.clicked.connect(self._add_row) + del_btn = QPushButton("Remove Last") + del_btn.setObjectName("removeTraceBtn") + del_btn.clicked.connect(self._del_last) + btn_row.addWidget(add_btn) + btn_row.addWidget(del_btn) + btn_row.addStretch() + grp_lay.addLayout(btn_row) + lay.addWidget(grp) + + for m in device._motors: + self._add_row(m) + + def _add_row(self, m: dict = None): + n = len(self._rows) + row = _MotorRow(m or {"motor_id": f"M{n+1}", "address": n+1, "read_cmds": ["TP", "TV", "TC"]}) + self._rows.append(row) + stretch = self._container_lay.takeAt(self._container_lay.count() - 1) + self._container_lay.addWidget(row) + self._container_lay.addStretch() + + def _del_last(self): + if not self._rows: + return + row = self._rows.pop() + self._container_lay.removeWidget(row) + row.deleteLater() + + def collect(self) -> List[Dict]: + return [r.to_dict() for r in self._rows] + + +class _MotorRow(QWidget): + _READ_CMDS = ["TP", "TV", "TC", "TS"] + + def __init__(self, m: dict): + super().__init__() + lay = QHBoxLayout(self) + lay.setContentsMargins(0, 1, 0, 1) + lay.setSpacing(6) + + self._id_edit = QLineEdit(m.get("motor_id", "M1")) + self._id_edit.setMaximumWidth(52) + lay.addWidget(QLabel("ID:")) + lay.addWidget(self._id_edit) + + self._addr_spin = QSpinBox() + self._addr_spin.setRange(0, 31) + self._addr_spin.setValue(m.get("address", 1)) + self._addr_spin.setToolTip("0 = RS-232 (no prefix), 1-31 = RS-485") + lay.addWidget(QLabel("Addr:")) + lay.addWidget(self._addr_spin) + + lay.addWidget(QLabel("Read:")) + self._checks: Dict[str, QCheckBox] = {} + for cmd in self._READ_CMDS: + cb = QCheckBox(cmd) + cb.setChecked(cmd in m.get("read_cmds", ["TP", "TV", "TC"])) + lay.addWidget(cb) + self._checks[cmd] = cb + lay.addStretch() + + def to_dict(self) -> dict: + return { + "motor_id": self._id_edit.text().strip() or "M1", + "address": self._addr_spin.value(), + "read_cmds": [cmd for cmd, cb in self._checks.items() if cb.isChecked()], + } + + +# ── Default protocol configs ─────────────────────────────────────────────────── + +def _default_scpi() -> List[Dict]: + return [ + {"channel_id": "CH1", "name": "Channel 1", "query": "MEAS:VOLT?", + "unit": "V", "scale": 1.0, "write_cmd": ""}, + {"channel_id": "CH2", "name": "Channel 2", "query": "MEAS:CURR?", + "unit": "A", "scale": 1.0, "write_cmd": ""}, + ] + + +def _default_modbus() -> List[Dict]: + return [ + {"channel_id": "R0", "name": "Register 0", "register": 0, + "function_code": 0x03, "data_type": "uint16", "scale": 1.0, "offset": 0.0, "unit": ""}, + {"channel_id": "R1", "name": "Register 1", "register": 2, + "function_code": 0x03, "data_type": "uint16", "scale": 1.0, "offset": 0.0, "unit": ""}, + ] + + +def _default_motors() -> List[Dict]: + return [{"motor_id": "M1", "address": 1, "read_cmds": ["TP", "TV", "TC"]}] -- cgit v1.2.3