diff options
Diffstat (limited to 'devices')
| -rw-r--r-- | devices/__init__.py | 4 | ||||
| -rw-r--r-- | devices/analog_input.py | 461 | ||||
| -rw-r--r-- | devices/arduino_device.py | 136 | ||||
| -rw-r--r-- | devices/base_device.py | 22 | ||||
| -rw-r--r-- | devices/digital_io.py | 334 | ||||
| -rw-r--r-- | devices/nidaqmx_device.py | 44 | ||||
| -rw-r--r-- | devices/serial_device.py | 66 |
7 files changed, 153 insertions, 914 deletions
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:<duty> (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 ─────────────────────────────────────────────────── |