diff options
| author | Christian Kolset <christian.kolset@gmail.com> | 2026-09-19 17:38:07 -0600 |
|---|---|---|
| committer | Christian Kolset <christian.kolset@gmail.com> | 2026-09-19 17:38:07 -0600 |
| commit | f3327d56d48d63b21ffc03a55fa5e7db753d1707 (patch) | |
| tree | e19d4c7846dd4a434f8a18f82290597731b4a12c /devices/analog_input.py | |
| parent | 00671a90bb51bfe485d98a928a456d55adbe1faf (diff) | |
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).
Diffstat (limited to 'devices/analog_input.py')
| -rw-r--r-- | devices/analog_input.py | 461 |
1 files changed, 0 insertions, 461 deletions
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() |
