""" devices/serial_device.py 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) """ from __future__ import annotations import math import threading import time from typing import Any, Dict, List, Optional from PyQt6.QtCore import Qt, QThread, pyqtSignal from PyQt6.QtWidgets import ( QCheckBox, QComboBox, QDoubleSpinBox, QFormLayout, QGroupBox, QHBoxLayout, QHeaderView, QLabel, QLineEdit, QListWidget, QListWidgetItem, QPushButton, QScrollArea, QSizePolicy, QSpinBox, QStackedWidget, QTableWidget, QTableWidgetItem, QVBoxLayout, QWidget, ) from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus 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" ICON = "⇌" def __init__( self, 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._fmt = parse_format self.simulate = simulate # Generic self._num_channels = num_channels self._chan_names = channel_names self._units = units # 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_type=self.DEVICE_TYPE, description=self._description(), icon=self.ICON, channels=channels, ) super().__init__(info) self._layer = self._make_layer() # ── BaseDevice ──────────────────────────────────────────────────────── def connect(self) -> bool: ok = self._layer.connect() self.status = DeviceStatus.SIMULATED if self.simulate else ( DeviceStatus.CONNECTED if ok else DeviceStatus.ERROR ) return ok def disconnect(self) -> None: self._layer.disconnect() self.status = DeviceStatus.DISCONNECTED def read_channels(self) -> Dict[str, float]: raw = self._layer.read() if not raw: return {} if self._fmt == "mark10" and hasattr(self._layer, "current_unit"): # Gauge can be switched between lb/kgF/N/ozF on the device itself — # keep the channel's unit in sync so new plots/pickers/CSV log # headers pick up the currently-selected unit instead of a fixed # default. Does not relabel the axis of an already-open plot pane. force_ch = self.get_channel("force") if force_ch is not None: force_ch.unit = self._layer.current_unit if self._fmt not in _GENERIC_FORMATS: # Protocol layers already key by channel_id — pass through untouched. # (Write-only channels, e.g. CML "M1_VS", never appear in raw — correctly dropped.) return raw # 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: if self.status not in (DeviceStatus.CONNECTED, DeviceStatus.SIMULATED): return False return self._layer.write(channel_id, int(value)) def get_config_widget(self) -> QWidget: return SerialConfigWidget(self) def get_save_config(self) -> dict: cfg: Dict[str, Any] = { "device_type": self.DEVICE_TYPE, "device_id": self.info.device_id, "name": self.info.name, "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._fmt = fmt self.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"sig{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)], writable=bool(sc.get("write_cmd")), ) 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)], # Holding registers (FC03) accept writes (FC06); input registers (FC04) don't. writable=(mc.get("function_code", 0x03) == 0x03), ) 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]), # Write-only action channels (no reading — for control buttons) 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 == "cml": _CMD_UNITS = {"TP": "counts", "TV": "counts/s", "TC": "%×10", "TS": "flags"} _WRITE_UNITS = {"VS": "counts/s", "MA": "counts"} channels = [] color_idx = 0 for motor in self._motors: mid = motor.get("motor_id", "M1") for cmd in motor.get("read_cmds", ["TP", "TV", "TC"]): 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 # Write-only setpoint channels (no reading — for control widgets) for cmd in ("VS", "MA"): channels.append(ChannelConfig( channel_id=f"{mid}_{cmd}", name=f"{mid} {cmd}", unit=_WRITE_UNITS.get(cmd, ""), min_value=-1e6, max_value=1e6, color=_COLORS[color_idx % len(_COLORS)], writable=True, )) color_idx += 1 # Write-only action channels — motor must be enabled (ME) before VS/MA take effect for cmd, label in (("ME", "Enable"), ("MD", "Disable"), ("ST", "Stop")): channels.append(ChannelConfig( channel_id=f"{mid}_{cmd}", name=f"{mid} {label}", unit="", min_value=0.0, max_value=1.0, color=_COLORS[color_idx % len(_COLORS)], writable=True, )) 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 _AdaptiveStack(QStackedWidget): """QStackedWidget sized to its current panel only — no gap from hidden panels.""" def _current_height(self) -> int: w = self.currentWidget() if not w: return 0 lyt = w.layout() if lyt: m = lyt.contentsMargins() return lyt.sizeHint().height() + m.top() + m.bottom() return w.sizeHint().height() def sizeHint(self): from PyQt6.QtCore import QSize h = self._current_height() w = self.currentWidget() sw = w.sizeHint().width() if w else super().sizeHint().width() return QSize(sw, h) if h > 0 else super().sizeHint() def minimumSizeHint(self): return self.sizeHint() class SerialConfigWidget(QWidget): def __init__(self, device: SerialDevice): super().__init__() self.device = device self._scanner = None self._build() def _build(self): root = QVBoxLayout(self) root.setContentsMargins(8, 8, 8, 8) root.setSpacing(8) 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) from core.app_settings import is_developer_mode mode_lbl = QLabel("Mode:") dev_mode = is_developer_mode() mode_lbl.setVisible(dev_mode); self._sim_cb.setVisible(dev_mode) conn_form.addRow(mode_lbl, self._sim_cb) root.addWidget(conn_grp) # ── Protocol-specific settings (stacked) ────────────────────────── self._proto_stack = _AdaptiveStack() 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)) sp = self._proto_stack.sizePolicy() sp.setVerticalPolicy(QSizePolicy.Policy.Fixed) self._proto_stack.setSizePolicy(sp) self._proto_stack.currentChanged.connect(self._fix_stack_height) root.addWidget(self._proto_stack) # ── Port scanner ─────────────────────────────────────────────────── scan_grp = QGroupBox("Available Serial Ports") scan_lay = QVBoxLayout(scan_grp) scan_top = QHBoxLayout() self._scan_btn = QPushButton("🔍 Scan Ports") self._scan_btn.setObjectName("addTraceBtn") self._scan_btn.clicked.connect(self._scan) self._scan_lbl = QLabel("Click Scan to detect ports") self._scan_lbl.setObjectName("traceSource") 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(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 ) hint = QLabel("↑ Click a port above to select it") hint.setObjectName("traceSource") scan_lay.addWidget(self._port_list) scan_lay.addWidget(hint) root.addWidget(scan_grp) root.addStretch() # ── Apply ────────────────────────────────────────────────────────── apply_btn = QPushButton("Apply & Reconnect") apply_btn.setObjectName("applyButton") apply_btn.clicked.connect(self._apply) root.addWidget(apply_btn) def _fix_stack_height(self, _: int): self._proto_stack.updateGeometry() 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() class _Scanner(QThread): 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): 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") item.setFlags(item.flags() & ~Qt.ItemFlag.ItemIsSelectable) self._port_list.addItem(item) else: self._scan_lbl.setText(f"{len(ports)} port(s) — click to select") for device, desc in ports: label = f" {device}" if desc and desc.strip() and desc.strip() != device: label += f" — {desc}" item = QListWidgetItem(label) item.setData(Qt.ItemDataRole.UserRole, device) self._port_list.addItem(item) # ── 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("Signals:", 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("Signal 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"sig{n}")) self._table.setItem(row, 1, QTableWidgetItem(sc.get("name", f"sig{n}") if sc else f"sig{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"sig{row}", "name": cell(1) or f"sig{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 (motor ID 1-31, always sent explicitly)") 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(1, 31) self._addr_spin.setValue(max(1, m.get("address", 1))) self._addr_spin.setToolTip("CM1-C motor ID — always sent explicitly as \".\"") 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"]}]