From f3327d56d48d63b21ffc03a55fa5e7db753d1707 Mon Sep 17 00:00:00 2001 From: Christian Kolset Date: Sat, 19 Sep 2026 17:38:07 -0600 Subject: Redesign Signal/Channel naming and fix pin-remap breaking references Arduino analog Signal ids used to be the pin name itself, so fixing a wrong pin silently changed the id and orphaned any plot trace, control, or filter pipeline pointing at it. Analog Signal ids are now positional (ai0, ai1, ...) and decoupled from the underlying pin, matching how digital I/O already worked. More broadly, formalizes a Device -> Signal -> Channel -> Plot naming model (SignalConfig for physical device channels, Channel for derived/ computed ones) and adds core/reference_registry.py to propagate any Signal or Channel id rename through plot traces, controls, and filter pipelines instead of leaving them dangling. Signal ids are now editable in the Signals tab with the same propagation. .labui profile schema bumped to v2.0 with a transparent migration for old files, including remapping old pin-name Signal ids to the new positional ones. Removed six confirmed-dead files (two legacy device classes, four superseded top-level ui/ duplicates). --- devices/__init__.py | 4 +- devices/analog_input.py | 461 ---------------------------------------------- devices/arduino_device.py | 136 ++++++++------ devices/base_device.py | 22 ++- devices/digital_io.py | 334 --------------------------------- devices/nidaqmx_device.py | 44 ++--- devices/serial_device.py | 66 +++---- 7 files changed, 153 insertions(+), 914 deletions(-) delete mode 100644 devices/analog_input.py delete mode 100644 devices/digital_io.py (limited to 'devices') diff --git a/devices/__init__.py b/devices/__init__.py index 7cee4ac..d175390 100644 --- a/devices/__init__.py +++ b/devices/__init__.py @@ -1,4 +1,4 @@ -# devices/__init__.py from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus +# devices/__init__.py from devices.base_device import BaseDevice, SignalConfig, DeviceInfo, DeviceStatus from devices.device_registry import DeviceRegistry -__all__ = ["BaseDevice", "ChannelConfig", "DeviceInfo", "DeviceStatus", "DeviceRegistry"] +__all__ = ["BaseDevice", "SignalConfig", "DeviceInfo", "DeviceStatus", "DeviceRegistry"] diff --git a/devices/analog_input.py b/devices/analog_input.py deleted file mode 100644 index 8938bf2..0000000 --- a/devices/analog_input.py +++ /dev/null @@ -1,461 +0,0 @@ -""" -devices/analog_input.py - -Analog Input device. Backends: NI-DAQmx or Arduino. -""" - -from typing import Any, Dict, List - -from PyQt6.QtWidgets import ( - QWidget, QVBoxLayout, QHBoxLayout, QFormLayout, QGroupBox, - QComboBox, QDoubleSpinBox, QCheckBox, QLineEdit, QLabel, - QPushButton, QListWidget, QListWidgetItem, QFrame, QTextEdit, -) -from PyQt6.QtCore import Qt, QThread, pyqtSignal -from PyQt6.QtGui import QFont - -from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus -from api_layers.nidaqmx_layer import NidaqmxLayer -from api_layers.arduino_layer import ArduinoLayer - -_COLORS = [ - "#00d4ff", "#ff6b35", "#7fff6e", "#ffcc00", - "#c77dff", "#ff4d6d", "#4cc9f0", "#f72585", - "#38b000", "#e9c46a", "#a8dadc", "#e63946", - "#90e0ef", "#fb8500", "#b5e48c", "#d62828", -] - - -class AnalogInputDevice(BaseDevice): - DEVICE_TYPE = "analog_input" - ICON = "〜" - - def __init__( - self, - device_id: str = "ai_0", - num_channels: int = 4, - simulate: bool = True, - backend: str = "nidaqmx", - ni_device: str = "Dev1", - ni_min_v: float = -10.0, - ni_max_v: float = 10.0, - ard_port: str = "COM3", - ard_baud: int = 115200, - ): - self.backend = backend - self.simulate = simulate - - if backend == "arduino": - pins = ArduinoLayer.DEFAULT_ANALOG_PINS[:num_channels] - channels = [ - ChannelConfig(channel_id=p, name=p, unit="V", - min_value=0.0, max_value=5.0, - color=_COLORS[i % len(_COLORS)]) - for i, p in enumerate(pins) - ] - else: - ni_pins = [f"ai{i}" for i in range(num_channels)] - channels = [ - ChannelConfig(channel_id=p, name=p.upper(), unit="V", - min_value=ni_min_v, max_value=ni_max_v, - color=_COLORS[i % len(_COLORS)]) - for i, p in enumerate(ni_pins) - ] - - info = DeviceInfo( - device_id=device_id, - name=f"Analog Input ({backend.upper()})", - device_type=self.DEVICE_TYPE, - description=f"Multi-channel analog input via {backend}", - manufacturer="NI" if backend == "nidaqmx" else "Arduino", - icon=self.ICON, - channels=channels, - ) - super().__init__(info) - - self._ni_device = ni_device - self._ni_min_v = ni_min_v - self._ni_max_v = ni_max_v - self._ard_port = ard_port - self._ard_baud = ard_baud - self._num_channels = num_channels - self._last_error = "" - - self._layer = self._make_layer( - backend, simulate, ni_device, ni_min_v, ni_max_v, - ard_port, ard_baud, - [ch.channel_id for ch in channels], - ) - - def _make_layer(self, backend, simulate, - ni_device, ni_min_v, ni_max_v, - ard_port, ard_baud, channel_ids): - if backend == "arduino": - # Use shared layer so two devices on the same port don't conflict - from api_layers.port_registry import port_registry - return port_registry.get_layer( - port=ard_port, - baud=ard_baud, - simulate=simulate, - extra_pins=channel_ids, - ) - else: - return NidaqmxLayer( - device_name=ni_device, - channels=channel_ids, - min_val=ni_min_v, - max_val=ni_max_v, - simulate=simulate, - ) - - # ── BaseDevice ────────────────────────────────────────────────────────── - - def connect(self) -> bool: - self._last_error = "" - - # ArduinoLayer: use connect_if_needed so shared layers aren't - # opened twice when multiple devices share the same port. - # NidaqmxLayer: uses start() - if hasattr(self._layer, "connect"): - if self._layer.is_connected: - ok = True # already open — shared with another device - else: - ok = self._layer.connect() - else: - ok = self._layer.start() - - if not ok and hasattr(self._layer, "last_error"): - self._last_error = self._layer.last_error - - if self.simulate: - self.status = DeviceStatus.SIMULATED - else: - self.status = DeviceStatus.CONNECTED if ok else DeviceStatus.ERROR - - return ok - - def disconnect(self) -> None: - # For shared Arduino layers, release our ref count via port_registry. - # The layer stays open until the last device using it disconnects. - if self.backend == "arduino" and not self.simulate: - from api_layers.port_registry import port_registry - port_registry.release(self._ard_port, self._ard_baud) - else: - try: - if hasattr(self._layer, "stop"): - self._layer.stop() - elif hasattr(self._layer, "disconnect"): - self._layer.disconnect() - except Exception: - pass - self.status = DeviceStatus.DISCONNECTED - - def read_channels(self) -> Dict[str, float]: - return self._layer.read() - - def write_channel(self, channel_id: str, value: Any) -> bool: - """ - Route output commands through the Arduino layer. - - channel_id examples: - "D7" → digital write W:D7:1 or W:D7:0 - "D9" → PWM write P:D9: (if value is 0-255) - "HEAT" → named param C:HEAT:1.0 - """ - if self.backend != "arduino" or not hasattr(self._layer, "digital_write"): - return False - ch = channel_id.strip() - # Digital pin (D7, D13, etc.) - if ch.upper().startswith("D") and ch[1:].isdigit(): - return self._layer.digital_write(ch, int(bool(value))) - # Named parameter / setpoint - return self._layer.set_parameter(ch, value) - - def get_save_config(self) -> dict: - return { - "device_type": self.DEVICE_TYPE, - "device_id": self.info.device_id, - "name": self.info.name, - "num_channels": self._num_channels, - "simulate": self.simulate, - "backend": self.backend, - "ni_device": self._ni_device, - "ni_min_v": self._ni_min_v, - "ni_max_v": self._ni_max_v, - "ard_port": self._ard_port, - "ard_baud": self._ard_baud, - } - - def get_config_widget(self) -> QWidget: - return AnalogInputConfigWidget(self) - - def switch_backend(self, backend: str, simulate: bool, - ni_device: str, min_val: float, max_val: float, - port: str, baud: int) -> None: - was_running = self.status in (DeviceStatus.CONNECTED, DeviceStatus.SIMULATED) - if was_running: - self.disconnect() - - self.backend = backend - self.simulate = simulate - self._ni_device = ni_device - self._ni_min_v = min_val - self._ni_max_v = max_val - self._ard_port = port - self._ard_baud = baud - - self._layer = self._make_layer( - backend, simulate, ni_device, min_val, max_val, - port, baud, - [ch.channel_id for ch in self.info.channels], - ) - - if was_running: - self.connect() - - -# ── Port scanner thread ─────────────────────────────────────────────────────── - -class _PortScanThread(QThread): - done = pyqtSignal(list) - - def run(self): - ports = ArduinoLayer.list_ports() # returns [(device, desc), ...] - self.done.emit(ports) - - -# ── Config widget ───────────────────────────────────────────────────────────── - -class AnalogInputConfigWidget(QWidget): - def __init__(self, device: AnalogInputDevice): - super().__init__() - self.device = device - self._scanner = None - self._build() - - def _build(self): - root = QVBoxLayout(self) - root.setContentsMargins(10, 14, 10, 10) - root.setSpacing(12) - self.setMinimumWidth(460) - - # ── Backend / Simulate ─────────────────────────────────────────── - be_grp = QGroupBox("Backend Selection") - be_form = QFormLayout(be_grp) - be_form.setContentsMargins(10, 16, 10, 10) - - self.backend_cb = QComboBox() - self.backend_cb.addItems(["nidaqmx", "arduino"]) - self.backend_cb.setCurrentText(self.device.backend) - be_form.addRow("Backend:", self.backend_cb) - - self.sim_check = QCheckBox("Simulation Mode (no hardware)") - self.sim_check.setChecked(self.device.simulate) - from core.app_settings import is_developer_mode - self.sim_check.setVisible(is_developer_mode()) - be_form.addRow(self.sim_check) - root.addWidget(be_grp) - - # ── NI-DAQmx ──────────────────────────────────────────────────── - self.ni_grp = QGroupBox("NI-DAQmx Settings") - ni_form = QFormLayout(self.ni_grp) - ni_form.setContentsMargins(10, 16, 10, 10) - - self.ni_dev_edit = QLineEdit(self.device._ni_device) - ni_form.addRow("NI Device:", self.ni_dev_edit) - - self.ni_min_spin = QDoubleSpinBox() - self.ni_min_spin.setRange(-100, 0) - self.ni_min_spin.setValue(self.device._ni_min_v) - self.ni_min_spin.setSuffix(" V") - ni_form.addRow("Min Voltage:", self.ni_min_spin) - - self.ni_max_spin = QDoubleSpinBox() - self.ni_max_spin.setRange(0, 100) - self.ni_max_spin.setValue(self.device._ni_max_v) - self.ni_max_spin.setSuffix(" V") - ni_form.addRow("Max Voltage:", self.ni_max_spin) - - detect_btn = QPushButton("Detect NI Devices") - detect_btn.setObjectName("configButton") - detect_btn.clicked.connect(self._detect_ni) - ni_form.addRow(detect_btn) - self.ni_detect_lbl = QLabel("") - self.ni_detect_lbl.setObjectName("traceSource") - ni_form.addRow(self.ni_detect_lbl) - root.addWidget(self.ni_grp) - - # ── Arduino ────────────────────────────────────────────────────── - self.ard_grp = QGroupBox("Arduino Serial Settings") - ard_lay = QVBoxLayout(self.ard_grp) - ard_lay.setContentsMargins(10, 16, 10, 10) - ard_lay.setSpacing(8) - - ard_form = QFormLayout() - self.ard_port_edit = QLineEdit(self.device._ard_port) - self.ard_port_edit.setPlaceholderText("e.g. /dev/ttyUSB0 or COM3") - ard_form.addRow("Port:", self.ard_port_edit) - - self.ard_baud_cb = QComboBox() - self.ard_baud_cb.addItems(["9600", "19200", "57600", "115200", "230400"]) - self.ard_baud_cb.setCurrentText(str(self.device._ard_baud)) - ard_form.addRow("Baud Rate:", self.ard_baud_cb) - ard_lay.addLayout(ard_form) - - # Scan row - scan_row = QHBoxLayout() - self._scan_btn = QPushButton("🔍 Scan Ports") - self._scan_btn.setObjectName("addTraceBtn") - self._scan_btn.clicked.connect(self._scan_ports) - self._scan_lbl = QLabel("") - self._scan_lbl.setObjectName("traceSource") - scan_row.addWidget(self._scan_btn) - scan_row.addWidget(self._scan_lbl, 1) - ard_lay.addLayout(scan_row) - - # Port list - self._port_list = QListWidget() - self._port_list.setObjectName("portList") - self._port_list.setMaximumHeight(90) - self._port_list.itemClicked.connect(self._on_port_selected) - ard_lay.addWidget(self._port_list) - - hint = QLabel("↑ Click a port to fill the field above") - hint.setObjectName("traceSource") - ard_lay.addWidget(hint) - - # pyserial availability notice - if not ArduinoLayer.is_pyserial_available(): - warn = QLabel("⚠ pyserial not installed — run: pip install pyserial") - warn.setStyleSheet("color:#f59e0b; font-weight:600;") - ard_lay.addWidget(warn) - - root.addWidget(self.ard_grp) - - # ── Diagnostics ─────────────────────────────────────────────────── - self.diag_grp = QGroupBox("Connection Diagnostics") - diag_lay = QVBoxLayout(self.diag_grp) - diag_lay.setContentsMargins(10, 16, 10, 10) - - self._diag_box = QTextEdit() - self._diag_box.setObjectName("codeEditor") - self._diag_box.setReadOnly(True) - self._diag_box.setMaximumHeight(160) - self._diag_box.setPlaceholderText("Connection log will appear here…") - diag_lay.addWidget(self._diag_box) - - # Show any existing error - if self.device._last_error: - self._diag_box.setPlainText(f"Last error:\n{self.device._last_error}") - - root.addWidget(self.diag_grp) - - # ── Apply ───────────────────────────────────────────────────────── - apply_btn = QPushButton("Apply & Reconnect") - apply_btn.setObjectName("applyButton") - apply_btn.clicked.connect(self._apply) - root.addWidget(apply_btn) - root.addStretch() - - self._update_visibility() - self.backend_cb.currentTextChanged.connect(self._update_visibility) - self._refresh_diag() - - # ── Helpers ────────────────────────────────────────────────────────────── - - def _update_visibility(self): - be = self.backend_cb.currentText() - self.ni_grp.setVisible(be == "nidaqmx") - self.ard_grp.setVisible(be == "arduino") - - def _refresh_diag(self): - lines = [] - lines.append(f"Backend: {self.device.backend}") - lines.append(f"Simulate: {self.device.simulate}") - lines.append(f"Status: {self.device.status.value}") - if self.device.backend == "arduino": - lines.append(f"Port: {self.device._ard_port}") - lines.append(f"Baud: {self.device._ard_baud}") - lines.append(f"pyserial: {'available' if ArduinoLayer.is_pyserial_available() else 'NOT INSTALLED'}") - # Show last raw lines received from Arduino - if hasattr(self.device._layer, "raw_lines"): - raw = self.device._layer.raw_lines - if raw: - lines.append("\nLast lines from Arduino:") - for r in raw: - lines.append(f" {r}") - if self.device._last_error: - lines.append(f"\nError:\n{self.device._last_error}") - self._diag_box.setPlainText("\n".join(lines)) - - def _detect_ni(self): - devs = NidaqmxLayer.list_devices() - self.ni_detect_lbl.setText( - ", ".join(devs) if devs else "No NI devices found (nidaqmx package required)" - ) - - def _scan_ports(self): - self._scan_btn.setEnabled(False) - self._scan_lbl.setText("Scanning…") - self._port_list.clear() - self._scanner = _PortScanThread() - 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 ArduinoLayer.is_pyserial_available(): - self._scan_lbl.setText("pyserial not installed") - item = QListWidgetItem(" Install pyserial: pip install pyserial") - item.setFlags(item.flags() & ~Qt.ItemFlag.ItemIsSelectable) - self._port_list.addItem(item) - return - - if not ports: - self._scan_lbl.setText("No ports detected") - item = QListWidgetItem( - " No serial ports found.\n" - " • Check USB cable\n" - " • On Linux: sudo usermod -aG dialout $USER then re-login\n" - " • On Arch: sudo usermod -aG uucp $USER then re-login" - ) - item.setFlags(item.flags() & ~Qt.ItemFlag.ItemIsSelectable) - self._port_list.addItem(item) - else: - self._scan_lbl.setText(f"{len(ports)} port(s) found — 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) - - def _on_port_selected(self, item: QListWidgetItem): - port = item.data(Qt.ItemDataRole.UserRole) - if port: - self.ard_port_edit.setText(port) - - def _apply(self): - backend = self.backend_cb.currentText() - simulate = self.sim_check.isChecked() - port = self.ard_port_edit.text().strip() - baud = int(self.ard_baud_cb.currentText()) - - self._diag_box.setPlainText( - f"Connecting...\nBackend: {backend}\nSimulate: {simulate}\n" - + (f"Port: {port}\nBaud: {baud}" if backend == "arduino" else "") - ) - - self.device.switch_backend( - backend=backend, - simulate=simulate, - ni_device=self.ni_dev_edit.text().strip() or "Dev1", - min_val=self.ni_min_spin.value(), - max_val=self.ni_max_spin.value(), - port=port, - baud=baud, - ) - self._refresh_diag() diff --git a/devices/arduino_device.py b/devices/arduino_device.py index 1641eaa..5c954c8 100644 --- a/devices/arduino_device.py +++ b/devices/arduino_device.py @@ -15,7 +15,7 @@ from PyQt6.QtWidgets import ( ) from PyQt6.QtCore import Qt, QThread, pyqtSignal -from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus +from devices.base_device import BaseDevice, SignalConfig, DeviceInfo, DeviceStatus from api_layers.arduino_layer import ArduinoLayer _ANALOG_COLORS = [ @@ -91,26 +91,11 @@ class ArduinoDevice(BaseDevice): self._di_pins: List[str] = list(di_pins) self._do_pins: List[str] = list(do_pins) - channels = [] - for i, pin in enumerate(self._analog_pins): - channels.append(ChannelConfig( - channel_id=pin, name=pin, unit="V", - min_value=0.0, max_value=5.0, - color=_ANALOG_COLORS[i % len(_ANALOG_COLORS)], - )) - for i in range(len(self._di_pins)): - channels.append(ChannelConfig( - channel_id=f"di{i}", name=f"DI {i}", unit="", - min_value=0.0, max_value=1.0, - color=_DI_COLORS[i % len(_DI_COLORS)], - )) - for i in range(len(self._do_pins)): - channels.append(ChannelConfig( - channel_id=f"do{i}", name=f"DO {i}", unit="", - min_value=0.0, max_value=1.0, - writable=True, - color=_DO_COLORS[i % len(_DO_COLORS)], - )) + self._ai_map = {f"ai{i}": pin for i, pin in enumerate(self._analog_pins)} + self._di_map = {f"di{i}": pin for i, pin in enumerate(self._di_pins)} + self._do_map = {f"do{i}": pin for i, pin in enumerate(self._do_pins)} + + channels = self._build_channels() _desc = ArduinoLayer.describe_port(port) if not simulate else "" if _desc: @@ -121,22 +106,45 @@ class ArduinoDevice(BaseDevice): device_id=device_id, name=_name, device_type=self.DEVICE_TYPE, - description=(f"Arduino — {len(self._analog_pins)} analog, " - f"{len(self._di_pins)} DI, {len(self._do_pins)} DO"), + description=self._description(), manufacturer="Arduino", icon=self.ICON, - channels=channels, + signals=channels, ) super().__init__(info) self._output_state: Dict[str, int] = {f"do{i}": 0 for i in range(len(self._do_pins))} - self._di_state: Dict[str, int] = {f"di{i}": 0 for i in range(len(self._di_pins))} - - self._di_map = {f"di{i}": pin for i, pin in enumerate(self._di_pins)} - self._do_map = {f"do{i}": pin for i, pin in enumerate(self._do_pins)} + self._di_state: Dict[str, int] = {f"di{i}": 0 for i in range(len(self._di_pins))} self._layer = self._make_layer() + def _description(self) -> str: + return (f"Arduino — {len(self._analog_pins)} analog, " + f"{len(self._di_pins)} DI, {len(self._do_pins)} DO") + + def _build_channels(self) -> List[SignalConfig]: + channels = [] + for i, pin in enumerate(self._analog_pins): + channels.append(SignalConfig( + signal_id=f"ai{i}", name=pin, unit="V", + min_value=0.0, max_value=5.0, + color=_ANALOG_COLORS[i % len(_ANALOG_COLORS)], + )) + for i in range(len(self._di_pins)): + channels.append(SignalConfig( + signal_id=f"di{i}", name=f"DI {i}", unit="", + min_value=0.0, max_value=1.0, + color=_DI_COLORS[i % len(_DI_COLORS)], + )) + for i in range(len(self._do_pins)): + channels.append(SignalConfig( + signal_id=f"do{i}", name=f"DO {i}", unit="", + min_value=0.0, max_value=1.0, + writable=True, + color=_DO_COLORS[i % len(_DO_COLORS)], + )) + return channels + def _make_layer(self): from api_layers.port_registry import port_registry return port_registry.get_layer( @@ -177,10 +185,12 @@ class ArduinoDevice(BaseDevice): def read_channels(self) -> Dict[str, float]: raw = self._layer.read() result = {} - for ch in self.info.channels: - cid = ch.channel_id - if cid in raw: - result[cid] = raw[cid] + for ch in self.info.signals: + cid = ch.signal_id + if cid.startswith("ai"): + pin = self._ai_map.get(cid) + if pin and pin in raw: + result[cid] = raw[pin] elif cid.startswith("di"): pin = self._di_map.get(cid) if pin and pin in raw: @@ -244,15 +254,25 @@ class ArduinoDevice(BaseDevice): di_pins: List[str] = None, do_pins: List[str] = None, ) -> None: - """Update active pin mapping and send config commands if hardware-connected.""" + """Update active pin mapping, rebuild channels, and send config + commands if hardware-connected. Caller is responsible for emitting + a `device_reconfigured` signal so the acquisition engine, profile, + and UI pick up the new channel list (see ui/windows/config_window.py).""" if analog_pins is not None: self._analog_pins = list(analog_pins) + self._ai_map = {f"ai{i}": p for i, p in enumerate(analog_pins)} if di_pins is not None: self._di_pins = list(di_pins) self._di_map = {f"di{i}": p for i, p in enumerate(di_pins)} if do_pins is not None: self._do_pins = list(do_pins) self._do_map = {f"do{i}": p for i, p in enumerate(do_pins)} + + self.info.signals = self._build_channels() + self.info.description = self._description() + self._output_state = {f"do{i}": self._output_state.get(f"do{i}", 0) for i in range(len(self._do_pins))} + self._di_state = {f"di{i}": self._di_state.get(f"di{i}", 0) for i in range(len(self._di_pins))} + if self.status == DeviceStatus.CONNECTED: self._layer.configure_pins( di_pins=self._di_pins, @@ -329,27 +349,40 @@ class ArduinoConfigWidget(QWidget): # ── Pin Mapping ─────────────────────────────────────────────────── pin_grp = QGroupBox("Pin Mapping") - pin_form = QFormLayout(pin_grp) - pin_form.setContentsMargins(10, 16, 10, 10) + pin_outer = QVBoxLayout(pin_grp) + pin_outer.setContentsMargins(10, 16, 10, 10) + pin_outer.setSpacing(8) + + pin_cols = QHBoxLayout() + pin_cols.setSpacing(12) + analog_grp = QGroupBox("Analog Pins") + analog_form = QFormLayout(analog_grp) self.analog_pins_edit = QLineEdit(", ".join(p[1:] for p in self.device._analog_pins)) - self.analog_pins_edit.setPlaceholderText("e.g. 0, 1, 2, 3 (indices into A0–A5)") - pin_form.addRow("Analog Input Pins:", self.analog_pins_edit) + self.analog_pins_edit.setPlaceholderText("e.g. 0, 1, 2, 3") + analog_form.addRow("Input:", self.analog_pins_edit) + pin_cols.addWidget(analog_grp) + digital_grp = QGroupBox("Digital Pins") + digital_form = QFormLayout(digital_grp) self.di_pins_edit = QLineEdit(", ".join(p[1:] for p in self.device._di_pins)) self.di_pins_edit.setPlaceholderText("e.g. 2, 3, 8") - pin_form.addRow("Digital Input Pins:", self.di_pins_edit) - + digital_form.addRow("Input:", self.di_pins_edit) self.do_pins_edit = QLineEdit(", ".join(p[1:] for p in self.device._do_pins)) - self.do_pins_edit.setPlaceholderText("e.g. 5, 6, 7, 9") - pin_form.addRow("Digital Output Pins:", self.do_pins_edit) + self.do_pins_edit.setPlaceholderText("e.g. 5, 6, 7") + digital_form.addRow("Output:", self.do_pins_edit) + pin_cols.addWidget(digital_grp) + + pin_outer.addLayout(pin_cols) hint = QLabel("Changes take effect on Apply & Reconnect.\n" "Analog: enter A-pin indices (0=A0, 1=A1, …). " - "Digital: enter pin numbers.") + "Digital: enter pin numbers. " + "For analog-style output, drive a PWM-capable digital " + "output pin (e.g. D9) with a duty-cycle value.") hint.setObjectName("traceSource") hint.setWordWrap(True) - pin_form.addRow(hint) + pin_outer.addWidget(hint) root.addWidget(pin_grp) @@ -417,19 +450,18 @@ class ArduinoConfigWidget(QWidget): self.port_edit.setText(port) def _apply(self): - # Parse pin fields and update device before reconnect so connect() sends correct APIN/DPIN + # Parse pin fields and remap (rebuilds device.info.signals) before + # reconnect, so connect() sends correct APIN/DPIN and the acquisition + # engine picks up the new channel list once this dialog closes. analog = _parse_analog_pin_edit(self.analog_pins_edit.text()) di = _parse_pin_edit(self.di_pins_edit.text()) do = _parse_pin_edit(self.do_pins_edit.text()) - if analog: - self.device._analog_pins = analog - if di: - self.device._di_pins = di - self.device._di_map = {f"di{i}": p for i, p in enumerate(di)} - if do: - self.device._do_pins = do - self.device._do_map = {f"do{i}": p for i, p in enumerate(do)} + self.device.remap_pins( + analog_pins=analog or None, + di_pins=di or None, + do_pins=do or None, + ) self.device.switch_backend( simulate=self.sim_chk.isChecked(), port=self.port_edit.text().strip() or self.device._port, diff --git a/devices/base_device.py b/devices/base_device.py index 9803867..e3eed65 100644 --- a/devices/base_device.py +++ b/devices/base_device.py @@ -20,17 +20,19 @@ class DeviceStatus(Enum): @dataclass -class ChannelConfig: - """Configuration for a single I/O channel.""" - channel_id: str +class SignalConfig: + """Configuration for a single physical Signal — one I/O stream coming + off a Device (e.g. an Arduino analog pin, a digital line). Contrast + with a Channel (core/signal_processor.py), which takes one or more + Signals and applies math/filters to produce a new derived stream.""" + signal_id: str name: str unit: str = "" min_value: float = 0.0 max_value: float = 100.0 enabled: bool = True - is_output: bool = False color: str = "#00d4ff" - writable: bool = False # True if this channel accepts write_channel() calls + writable: bool = False # True if this signal accepts write_channel() calls extra: Dict[str, Any] = field(default_factory=dict) @@ -39,13 +41,13 @@ class DeviceInfo: """Metadata describing a device module.""" device_id: str name: str - device_type: str # "analog_input" | "digital_io" | "serial" | "temperature" + device_type: str # "arduino" | "serial" | "nidaqmx" | ... description: str = "" manufacturer: str = "" model: str = "" version: str = "1.0.0" icon: str = "⚙" - channels: List[ChannelConfig] = field(default_factory=list) + signals: List[SignalConfig] = field(default_factory=list) class BaseDevice(ABC): @@ -75,7 +77,7 @@ class BaseDevice(ABC): @abstractmethod def read_channels(self) -> Dict[str, float]: - """Return {channel_id: value} for all enabled channels.""" + """Return {signal_id: value} for all enabled signals.""" @abstractmethod def write_channel(self, channel_id: str, value: Any) -> bool: @@ -97,8 +99,8 @@ class BaseDevice(ABC): except Exception: pass - def get_channel(self, channel_id: str) -> Optional[ChannelConfig]: - return next((c for c in self.info.channels if c.channel_id == channel_id), None) + def get_signal(self, signal_id: str) -> Optional[SignalConfig]: + return next((s for s in self.info.signals if s.signal_id == signal_id), None) def __repr__(self): return f"<{self.__class__.__name__} id={self.info.device_id} status={self.status.value}>" diff --git a/devices/digital_io.py b/devices/digital_io.py deleted file mode 100644 index 7cda61f..0000000 --- a/devices/digital_io.py +++ /dev/null @@ -1,334 +0,0 @@ -""" -devices/digital_io.py - -Digital I/O device module. Backends: NI-DAQmx or Arduino. - -NI backend – uses nidaqmx digital line tasks (P0.0..P0.7) -Arduino – uses ArduinoLayer digital pin reads; writes via W:Dxx:val -""" - -import random -import time -from typing import Any, Dict - -from PyQt6.QtWidgets import ( - QWidget, QVBoxLayout, QHBoxLayout, QGroupBox, - QCheckBox, QPushButton, QLabel, QFormLayout, - QComboBox, QLineEdit, -) - -from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus - -_IN_COLORS = ["#00d4ff", "#4cc9f0", "#90e0ef", "#caf0f8", - "#0077b6", "#023e8a", "#48cae4", "#ade8f4"] -_OUT_COLORS = ["#ff6b35", "#ffcc00", "#f77f00", "#fcbf49", - "#d62828", "#e63946", "#fb8500", "#ffd166"] - - -class DigitalIODevice(BaseDevice): - DEVICE_TYPE = "digital_io" - ICON = "⬛" - - def __init__( - self, - device_id: str = "dio_0", - num_inputs: int = 8, - num_outputs: int = 8, - simulate: bool = True, - backend: str = "nidaqmx", # "nidaqmx" | "arduino" - ni_device: str = "Dev1", - ard_port: str = "COM3", - ard_baud: int = 115200, - ): - self.simulate = simulate - self.backend = backend - self._ni_device = ni_device - self._ard_port = ard_port - self._ard_baud = ard_baud - self._num_inputs = num_inputs - self._num_outputs = num_outputs - - channels = [] - for i in range(num_inputs): - channels.append(ChannelConfig( - channel_id=f"di{i}", name=f"DI {i}", unit="", - min_value=0.0, max_value=1.0, - color=_IN_COLORS[i % len(_IN_COLORS)], - )) - for i in range(num_outputs): - channels.append(ChannelConfig( - channel_id=f"do{i}", name=f"DO {i}", unit="", - min_value=0.0, max_value=1.0, - color=_OUT_COLORS[i % len(_OUT_COLORS)], - writable=True, - )) - - info = DeviceInfo( - device_id=device_id, - name=f"Digital I/O ({backend.upper()})", - device_type=self.DEVICE_TYPE, - description="Digital input/output module", - icon=self.ICON, - channels=channels, - ) - super().__init__(info) - - self._output_state: Dict[str, int] = { - f"do{i}": 0 for i in range(num_outputs) - } - self._sim_toggle: Dict[str, int] = {} - self._sim_state: Dict[str, int] = {} - - # Hardware task placeholders - self._ni_in_task = None - self._ni_out_task = None - self._ard_layer = None - - # ── BaseDevice ────────────────────────────────────────────────────── - - def connect(self) -> bool: - if self.simulate: - self.status = DeviceStatus.SIMULATED - return True - - if self.backend == "nidaqmx": - return self._ni_connect() - else: - return self._ard_connect() - - def _ni_connect(self) -> bool: - try: - import nidaqmx # type: ignore - from nidaqmx.constants import LineGrouping # type: ignore - n_in = sum(1 for c in self.info.channels if c.channel_id.startswith("di")) - n_out = sum(1 for c in self.info.channels if c.channel_id.startswith("do")) - - if n_in: - self._ni_in_task = nidaqmx.Task() - for i in range(n_in): - self._ni_in_task.di_channels.add_di_chan( - f"{self._ni_device}/port0/line{i}", - line_grouping=LineGrouping.CHAN_PER_LINE, - ) - self._ni_in_task.start() - - if n_out: - self._ni_out_task = nidaqmx.Task() - for i in range(n_out): - self._ni_out_task.do_channels.add_do_chan( - f"{self._ni_device}/port1/line{i}", - line_grouping=LineGrouping.CHAN_PER_LINE, - ) - self._ni_out_task.start() - - self.status = DeviceStatus.CONNECTED - return True - except Exception as e: - print(f"[DigitalIODevice] NI connect failed: {e}") - self.status = DeviceStatus.ERROR - return False - - def _ard_connect(self) -> bool: - from api_layers.port_registry import port_registry - # Get shared layer — won't open a second connection if AnalogInputDevice - # is already connected on the same port - self._ard_layer = port_registry.get_layer( - port=self._ard_port, - baud=self._ard_baud, - simulate=False, - ) - if self._ard_layer.is_connected: - # Port already open — just register ourselves and return OK - self.status = DeviceStatus.CONNECTED - return True - ok = self._ard_layer.connect() - self.status = DeviceStatus.CONNECTED if ok else DeviceStatus.ERROR - return ok - - def disconnect(self) -> None: - if self._ni_in_task: - try: self._ni_in_task.stop(); self._ni_in_task.close() - except Exception: pass - if self._ni_out_task: - try: self._ni_out_task.stop(); self._ni_out_task.close() - except Exception: pass - if self._ard_layer and self.backend == "arduino" and not self.simulate: - from api_layers.port_registry import port_registry - port_registry.release(self._ard_port, self._ard_baud) - self._ard_layer = None - self.status = DeviceStatus.DISCONNECTED - - def read_channels(self) -> Dict[str, float]: - if self.simulate: - return self._sim_read() - if self.backend == "nidaqmx": - return self._ni_read() - return self._ard_read() - - def _ni_read(self) -> Dict[str, float]: - result = {} - try: - if self._ni_in_task: - vals = self._ni_in_task.read() - for i, ch in enumerate(c for c in self.info.channels if c.channel_id.startswith("di")): - result[ch.channel_id] = float(vals[i] if isinstance(vals, list) else vals) - except Exception as e: - print(f"[DigitalIODevice] NI read failed: {e}") - for ch in (c for c in self.info.channels if c.channel_id.startswith("do")): - result[ch.channel_id] = float(self._output_state.get(ch.channel_id, 0)) - return result - - def _ard_read(self) -> Dict[str, float]: - if not self._ard_layer: - return {} - raw = self._ard_layer.read() - result = {} - for ch in self.info.channels: - if ch.channel_id in raw: - result[ch.channel_id] = raw[ch.channel_id] - elif ch.channel_id.startswith("do"): - result[ch.channel_id] = float(self._output_state.get(ch.channel_id, 0)) - return result - - def _sim_read(self) -> Dict[str, float]: - result = {} - for ch in self.info.channels: - if ch.channel_id.startswith("di"): - cnt = self._sim_toggle.get(ch.channel_id, 0) + 1 - if cnt >= random.randint(8, 40): - self._sim_state[ch.channel_id] = 1 - self._sim_state.get(ch.channel_id, 0) - cnt = 0 - self._sim_toggle[ch.channel_id] = cnt - result[ch.channel_id] = float(self._sim_state.get(ch.channel_id, 0)) - else: - result[ch.channel_id] = float(self._output_state.get(ch.channel_id, 0)) - return result - - def write_channel(self, channel_id: str, value: Any) -> bool: - self._output_state[channel_id] = int(bool(value)) - if not self.simulate: - if self.backend == "nidaqmx" and self._ni_out_task: - try: - out_chs = [c for c in self.info.channels if c.channel_id.startswith("do")] - idx = next((i for i, c in enumerate(out_chs) if c.channel_id == channel_id), None) - if idx is not None: - states = [self._output_state.get(c.channel_id, 0) for c in out_chs] - self._ni_out_task.write(states) - except Exception as e: - print(f"[DigitalIODevice] NI write failed: {e}") - elif self.backend == "arduino" and self._ard_layer: - # Map channel ID to Arduino pin name: - # do0→D5, do1→D6, do2→D7... (matches DIGITAL_OUT in firmware) - # Or the channel_id itself if it already looks like D5 - pin = channel_id if channel_id.upper().startswith("D") else f"D{5 + int(channel_id.replace('do',''))}" - self._ard_layer.digital_write(pin, int(bool(value))) - return True - - def pwm_channel(self, channel_id: str, duty_pct: float) -> bool: - """Send a PWM command to an Arduino output pin (0–100%).""" - if self.backend == "arduino" and self._ard_layer and not self.simulate: - pin = channel_id if channel_id.upper().startswith("D") else f"D{9 + int(channel_id.replace('do',''))}" - return self._ard_layer.pwm_write_pct(pin, duty_pct) - return False - - def set_parameter(self, name: str, value: Any) -> bool: - """Send a named parameter/setpoint to the Arduino.""" - if self.backend == "arduino" and self._ard_layer and not self.simulate: - return self._ard_layer.set_parameter(name, value) - return False - - def switch_backend(self, backend: str, simulate: bool, - ni_device: str, ard_port: str, ard_baud: int) -> None: - was_running = self.status in (DeviceStatus.CONNECTED, DeviceStatus.SIMULATED) - if was_running: - self.disconnect() - - self.backend = backend - self.simulate = simulate - self._ni_device = ni_device - self._ard_port = ard_port - self._ard_baud = ard_baud - self._ard_layer = None - self._ni_in_task = None - self._ni_out_task = None - - if was_running: - self.connect() - - def get_save_config(self) -> dict: - return { - "device_type": self.DEVICE_TYPE, - "device_id": self.info.device_id, - "name": self.info.name, - "num_inputs": self._num_inputs, - "num_outputs": self._num_outputs, - "simulate": self.simulate, - "backend": self.backend, - "ni_device": self._ni_device, - "ard_port": self._ard_port, - "ard_baud": self._ard_baud, - } - - def get_config_widget(self) -> QWidget: - return DigitalIOConfigWidget(self) - - -class DigitalIOConfigWidget(QWidget): - def __init__(self, device: DigitalIODevice): - super().__init__() - self.device = device - self._build() - - def _build(self): - root = QVBoxLayout(self) - root.setContentsMargins(8, 8, 8, 8) - self.setMinimumWidth(380) - - be_grp = QGroupBox("Backend") - be_form = QFormLayout(be_grp) - self.be_cb = QComboBox() - self.be_cb.addItems(["nidaqmx", "arduino"]) - self.be_cb.setCurrentText(self.device.backend) - be_form.addRow("Backend:", self.be_cb) - self.sim_chk = QCheckBox("Simulate") - self.sim_chk.setChecked(self.device.simulate) - from core.app_settings import is_developer_mode - self.sim_chk.setVisible(is_developer_mode()) - be_form.addRow(self.sim_chk) - - self.ni_dev_edit = QLineEdit(self.device._ni_device) - be_form.addRow("NI Device:", self.ni_dev_edit) - - self.ard_port_edit = QLineEdit(self.device._ard_port) - self.ard_port_edit.setPlaceholderText("e.g. /dev/ttyUSB0 or COM3") - be_form.addRow("Arduino Port:", self.ard_port_edit) - - self.ard_baud_cb = QComboBox() - self.ard_baud_cb.addItems(["9600", "19200", "57600", "115200", "230400"]) - self.ard_baud_cb.setCurrentText(str(self.device._ard_baud)) - be_form.addRow("Baud Rate:", self.ard_baud_cb) - - apply_btn = QPushButton("Apply & Reconnect") - apply_btn.setObjectName("applyButton") - apply_btn.clicked.connect(self._apply) - be_form.addRow(apply_btn) - - self.be_cb.currentTextChanged.connect(self._update_visibility) - root.addWidget(be_grp) - self._update_visibility(self.device.backend) - root.addStretch() - - def _update_visibility(self, backend: str = ""): - backend = backend or self.be_cb.currentText() - self.ni_dev_edit.setVisible(backend == "nidaqmx") - self.ard_port_edit.setVisible(backend == "arduino") - self.ard_baud_cb.setVisible(backend == "arduino") - - def _apply(self): - self.device.switch_backend( - backend=self.be_cb.currentText(), - simulate=self.sim_chk.isChecked(), - ni_device=self.ni_dev_edit.text().strip() or "Dev1", - ard_port=self.ard_port_edit.text().strip(), - ard_baud=int(self.ard_baud_cb.currentText()), - ) diff --git a/devices/nidaqmx_device.py b/devices/nidaqmx_device.py index 9ebec0f..652b7e2 100644 --- a/devices/nidaqmx_device.py +++ b/devices/nidaqmx_device.py @@ -13,7 +13,7 @@ from PyQt6.QtWidgets import ( ) from PyQt6.QtCore import Qt, QThread, pyqtSignal -from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus +from devices.base_device import BaseDevice, SignalConfig, DeviceInfo, DeviceStatus from api_layers.nidaqmx_layer import NidaqmxLayer _ANALOG_COLORS = [ @@ -53,20 +53,20 @@ class NidaqmxDevice(BaseDevice): channels = [] for i in range(num_analog): - channels.append(ChannelConfig( - channel_id=f"ai{i}", name=f"AI{i}", unit="V", + channels.append(SignalConfig( + signal_id=f"ai{i}", name=f"AI{i}", unit="V", min_value=min_v, max_value=max_v, color=_ANALOG_COLORS[i % len(_ANALOG_COLORS)], )) for i in range(num_di): - channels.append(ChannelConfig( - channel_id=f"di{i}", name=f"DI {i}", unit="", + channels.append(SignalConfig( + signal_id=f"di{i}", name=f"DI {i}", unit="", min_value=0.0, max_value=1.0, color=_DI_COLORS[i % len(_DI_COLORS)], )) for i in range(num_do): - channels.append(ChannelConfig( - channel_id=f"do{i}", name=f"DO {i}", unit="", + channels.append(SignalConfig( + signal_id=f"do{i}", name=f"DO {i}", unit="", min_value=0.0, max_value=1.0, color=_DO_COLORS[i % len(_DO_COLORS)], writable=True, @@ -79,7 +79,7 @@ class NidaqmxDevice(BaseDevice): description=f"NI-DAQmx — {num_analog} AI, {num_di} DI, {num_do} DO", manufacturer="National Instruments", icon=self.ICON, - channels=channels, + signals=channels, ) super().__init__(info) @@ -99,8 +99,8 @@ class NidaqmxDevice(BaseDevice): ok = True try: if self._num_analog > 0: - ai_ids = [ch.channel_id for ch in self.info.channels - if ch.channel_id.startswith("ai")] + ai_ids = [ch.signal_id for ch in self.info.signals + if ch.signal_id.startswith("ai")] self._ai_layer = NidaqmxLayer( device_name=self._ni_device, channels=ai_ids, @@ -179,14 +179,14 @@ class NidaqmxDevice(BaseDevice): try: if self._ni_in_task: vals = self._ni_in_task.read() - di_chs = [c for c in self.info.channels if c.channel_id.startswith("di")] + di_chs = [c for c in self.info.signals if c.signal_id.startswith("di")] for i, ch in enumerate(di_chs): - result[ch.channel_id] = float(vals[i] if isinstance(vals, list) else vals) + result[ch.signal_id] = float(vals[i] if isinstance(vals, list) else vals) except Exception: pass - for ch in (c for c in self.info.channels if c.channel_id.startswith("do")): - result[ch.channel_id] = float(self._output_state.get(ch.channel_id, 0)) + for ch in (c for c in self.info.signals if c.signal_id.startswith("do")): + result[ch.signal_id] = float(self._output_state.get(ch.signal_id, 0)) return result @@ -194,12 +194,12 @@ class NidaqmxDevice(BaseDevice): import math, time t = time.time() result = {} - for i, ch in enumerate([c for c in self.info.channels if c.channel_id.startswith("ai")]): - result[ch.channel_id] = math.sin(t + i) * (self._max_v * 0.5) - for ch in (c for c in self.info.channels if c.channel_id.startswith("di")): - result[ch.channel_id] = 0.0 - for ch in (c for c in self.info.channels if c.channel_id.startswith("do")): - result[ch.channel_id] = float(self._output_state.get(ch.channel_id, 0)) + for i, ch in enumerate([c for c in self.info.signals if c.signal_id.startswith("ai")]): + result[ch.signal_id] = math.sin(t + i) * (self._max_v * 0.5) + for ch in (c for c in self.info.signals if c.signal_id.startswith("di")): + result[ch.signal_id] = 0.0 + for ch in (c for c in self.info.signals if c.signal_id.startswith("do")): + result[ch.signal_id] = float(self._output_state.get(ch.signal_id, 0)) return result def write_channel(self, channel_id: str, value: Any) -> bool: @@ -208,8 +208,8 @@ class NidaqmxDevice(BaseDevice): self._output_state[cid] = int(bool(value)) if not self.simulate and self._ni_out_task: try: - do_chs = [c for c in self.info.channels if c.channel_id.startswith("do")] - states = [self._output_state.get(c.channel_id, 0) for c in do_chs] + do_chs = [c for c in self.info.signals if c.signal_id.startswith("do")] + states = [self._output_state.get(c.signal_id, 0) for c in do_chs] self._ni_out_task.write(states) except Exception as e: print(f"[NidaqmxDevice] write failed: {e}") diff --git a/devices/serial_device.py b/devices/serial_device.py index d402f09..86d5e80 100644 --- a/devices/serial_device.py +++ b/devices/serial_device.py @@ -31,7 +31,7 @@ from PyQt6.QtWidgets import ( QStackedWidget, QTableWidget, QTableWidgetItem, QTextEdit, QVBoxLayout, QWidget, ) -from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus +from devices.base_device import BaseDevice, SignalConfig, DeviceInfo, DeviceStatus from api_layers.arduino_layer import ArduinoLayer _COLORS = ["#7fff6e", "#4cc9f0", "#f72585", "#00d4ff", @@ -145,7 +145,7 @@ class SerialDevice(BaseDevice): device_type=self.DEVICE_TYPE, description=self._description(), icon=self.ICON, - channels=channels, + signals=channels, ) super().__init__(info) self._layer = self._make_layer() @@ -175,21 +175,21 @@ class SerialDevice(BaseDevice): # 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") + force_ch = self.get_signal("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. + # Protocol layers already key by signal_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] + for i, ch in enumerate(self.info.signals): + if ch.signal_id in raw: + mapped[ch.signal_id] = raw[ch.signal_id] elif i < len(raw_vals): - mapped[ch.channel_id] = raw_vals[i] + mapped[ch.signal_id] = raw_vals[i] return mapped def write_channel(self, channel_id: str, value: Any) -> bool: @@ -217,9 +217,9 @@ class SerialDevice(BaseDevice): cfg["port_serial"] = ArduinoLayer.port_serial_number(self._port) 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], + "num_channels": len(self.info.signals), + "channel_names": [c.signal_id for c in self.info.signals], + "units": [c.unit for c in self.info.signals], }) elif self._fmt == "scpi": cfg["scpi_channels"] = self._scpi_channels @@ -274,7 +274,7 @@ class SerialDevice(BaseDevice): self._mb_channels = modbus_channels or self._mb_channels self._motors = motors or self._motors - self.info.channels = self._build_channel_configs() + self.info.signals = self._build_channel_configs() self.info.description = self._description() self._layer = self._make_layer() @@ -287,14 +287,14 @@ class SerialDevice(BaseDevice): label = _LABEL_FOR.get(self._fmt, self._fmt) return f"{self._port} @ {self._baud} [{label}]" - def _build_channel_configs(self) -> List[ChannelConfig]: + def _build_channel_configs(self) -> List[SignalConfig]: 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], + SignalConfig( + signal_id=names[i], name=names[i], unit=_units[i], min_value=0.0, max_value=1023.0, color=_COLORS[i % len(_COLORS)], ) @@ -302,8 +302,8 @@ class SerialDevice(BaseDevice): ] elif fmt == "scpi": return [ - ChannelConfig( - channel_id=sc["channel_id"], + SignalConfig( + signal_id=sc["channel_id"], name=sc.get("name", sc["channel_id"]), unit=sc.get("unit", ""), min_value=-1e9, max_value=1e9, @@ -314,8 +314,8 @@ class SerialDevice(BaseDevice): ] elif fmt == "modbus_rtu": return [ - ChannelConfig( - channel_id=mc["channel_id"], + SignalConfig( + signal_id=mc["channel_id"], name=mc.get("name", mc["channel_id"]), unit=mc.get("unit", ""), min_value=-1e9, max_value=1e9, @@ -328,15 +328,15 @@ class SerialDevice(BaseDevice): 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]), + SignalConfig("force", "Force", "N", -5000.0, 5000.0, color=_COLORS[0]), + SignalConfig("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), + SignalConfig("zero", "Zero Gauge", "", 0.0, 1.0, color=_COLORS[2], writable=True), + SignalConfig("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]), + SignalConfig("extension", "Extension", "", -50.0, 50.0, color=_COLORS[0]), ] elif fmt == "cml": _CMD_UNITS = {"TP": "counts", "TV": "counts/s", "TC": "%×10", "TS": "flags"} @@ -346,8 +346,8 @@ class SerialDevice(BaseDevice): 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}", + channels.append(SignalConfig( + signal_id=f"{mid}_{cmd}", name=f"{mid} {cmd}", unit=_CMD_UNITS.get(cmd, ""), min_value=-1e6, max_value=1e6, @@ -356,8 +356,8 @@ class SerialDevice(BaseDevice): 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}", + channels.append(SignalConfig( + signal_id=f"{mid}_{cmd}", name=f"{mid} {cmd}", unit=_WRITE_UNITS.get(cmd, ""), min_value=-1e6, max_value=1e6, @@ -367,8 +367,8 @@ class SerialDevice(BaseDevice): 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}", + channels.append(SignalConfig( + signal_id=f"{mid}_{cmd}", name=f"{mid} {label}", unit="", min_value=0.0, max_value=1.0, @@ -376,13 +376,13 @@ class SerialDevice(BaseDevice): writable=True, )) color_idx += 1 - return channels or [ChannelConfig("M1_TP", "M1 TP", "counts", color=_COLORS[0])] + return channels or [SignalConfig("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] + names = [ch.signal_id for ch in self.info.signals] return ArduinoLayer( port=self._port, baud=self._baud, analog_pins=names, simulate=self.simulate, @@ -632,7 +632,7 @@ class _GenericPanel(QWidget): 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) + self._ch_spin.setValue(len(device.info.signals) if device._fmt in _GENERIC_FORMATS else 4) lay.addRow("Signals:", self._ch_spin) # ── Diagnostics ─────────────────────────────────────────────────── -- cgit v1.2.3