""" 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._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_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) 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()