""" devices/analog_input.py Analog Input device module. Supports two interchangeable backends: • backend="nidaqmx" → NidaqmxLayer (NI hardware or sim) • backend="arduino" → ArduinoLayer (Arduino hardware or sim) The device is backend-agnostic at the acquisition layer — swap the backend without changing any other code. """ from typing import Any, Dict, List from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QFormLayout, QGroupBox, QComboBox, QDoubleSpinBox, QSpinBox, QCheckBox, QLineEdit, QLabel, QPushButton, QHBoxLayout, ) from PyQt6.QtCore import Qt from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus from api_layers.nidaqmx_layer import NidaqmxLayer from api_layers.arduino_layer import ArduinoLayer # Distinct colors for up to 16 channels _COLORS = [ "#00d4ff", "#ff6b35", "#7fff6e", "#ffcc00", "#c77dff", "#ff4d6d", "#4cc9f0", "#f72585", "#38b000", "#e9c46a", "#a8dadc", "#e63946", "#90e0ef", "#fb8500", "#b5e48c", "#d62828", ] class AnalogInputDevice(BaseDevice): """Multi-channel analog input. Backend: NI-DAQmx or Arduino.""" DEVICE_TYPE = "analog_input" ICON = "〜" def __init__( self, device_id: str = "ai_0", num_channels: int = 4, simulate: bool = True, backend: str = "nidaqmx", # "nidaqmx" | "arduino" # NI-specific ni_device: str = "Dev1", ni_min_v: float = -10.0, ni_max_v: float = 10.0, # Arduino-specific ard_port: str = "COM3", ard_baud: int = 115200, ): self.backend = backend self.simulate = simulate # Build channel list 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, alarm_low=None, alarm_high=4.8, 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, alarm_low=None, alarm_high=ni_max_v * 0.9, 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) # Instantiate the backend layer if backend == "arduino": self._layer = ArduinoLayer( port=ard_port, baud=ard_baud, analog_pins=[ch.channel_id for ch in channels], simulate=simulate, ) else: self._layer = NidaqmxLayer( device_name=ni_device, channels=[ch.channel_id for ch in channels], min_val=ni_min_v, max_val=ni_max_v, simulate=simulate, ) # Store config for the config widget 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 # ── BaseDevice interface ──────────────────────────────────────────── def connect(self) -> bool: ok = self._layer.start() if hasattr(self._layer, "start") else self._layer.connect() self.status = DeviceStatus.SIMULATED if self.simulate else ( DeviceStatus.CONNECTED if ok else DeviceStatus.ERROR ) return ok def disconnect(self) -> None: if hasattr(self._layer, "stop"): self._layer.stop() else: self._layer.disconnect() 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: return False # AI is read-only def get_config_widget(self) -> QWidget: return AnalogInputConfigWidget(self) def switch_backend(self, backend: str, **kwargs) -> None: """Hot-swap the API layer without re-creating the device object.""" was_running = self.status in (DeviceStatus.CONNECTED, DeviceStatus.SIMULATED) if was_running: self.disconnect() self.backend = backend self.simulate = kwargs.get("simulate", self.simulate) if backend == "arduino": self._layer = ArduinoLayer( port=kwargs.get("port", self._ard_port), baud=kwargs.get("baud", self._ard_baud), analog_pins=[ch.channel_id for ch in self.info.channels], simulate=self.simulate, ) else: self._layer = NidaqmxLayer( device_name=kwargs.get("ni_device", self._ni_device), channels=[ch.channel_id for ch in self.info.channels], min_val=kwargs.get("min_val", self._ni_min_v), max_val=kwargs.get("max_val", self._ni_max_v), simulate=self.simulate, ) if was_running: self.connect() # ── Config Widget ──────────────────────────────────────────────────────────── class AnalogInputConfigWidget(QWidget): def __init__(self, device: AnalogInputDevice): super().__init__() self.device = device self._build() def _build(self): root = QVBoxLayout(self) root.setContentsMargins(0, 0, 0, 0) # ── Backend selector ────────────────────────────────────────── be_grp = QGroupBox("API Backend") be_form = QFormLayout(be_grp) 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") self.sim_check.setChecked(self.device.simulate) be_form.addRow(self.sim_check) root.addWidget(be_grp) # ── NI settings ─────────────────────────────────────────────── self.ni_grp = QGroupBox("NI-DAQmx Settings") ni_form = QFormLayout(self.ni_grp) self.ni_dev_edit = QLineEdit(self.device._ni_device) ni_form.addRow("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 V:", 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 V:", self.ni_max_spin) # Detect button detect_btn = QPushButton("Detect NI Devices") detect_btn.clicked.connect(self._detect_ni) ni_form.addRow(detect_btn) self.ni_detect_lbl = QLabel("") ni_form.addRow(self.ni_detect_lbl) root.addWidget(self.ni_grp) # ── Arduino settings ────────────────────────────────────────── self.ard_grp = QGroupBox("Arduino Settings") ard_form = QFormLayout(self.ard_grp) self.ard_port_edit = QLineEdit(self.device._ard_port) ard_form.addRow("Port:", self.ard_port_edit) self.ard_baud_cb = QComboBox() self.ard_baud_cb.addItems(["9600", "57600", "115200", "230400"]) self.ard_baud_cb.setCurrentText(str(self.device._ard_baud)) ard_form.addRow("Baud Rate:", self.ard_baud_cb) scan_btn = QPushButton("Scan Serial Ports") scan_btn.clicked.connect(self._scan_ports) ard_form.addRow(scan_btn) self.ard_port_lbl = QLabel("") ard_form.addRow(self.ard_port_lbl) root.addWidget(self.ard_grp) # ── Apply button ────────────────────────────────────────────── 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) def _update_visibility(self): be = self.backend_cb.currentText() self.ni_grp.setVisible(be == "nidaqmx") self.ard_grp.setVisible(be == "arduino") def _detect_ni(self): from api_layers.nidaqmx_layer import NidaqmxLayer devs = NidaqmxLayer.list_devices() self.ni_detect_lbl.setText(", ".join(devs) if devs else "None detected") def _scan_ports(self): from api_layers.arduino_layer import ArduinoLayer ports = ArduinoLayer.list_ports() self.ard_port_lbl.setText(", ".join(ports) if ports else "None found") def _apply(self): self.device.switch_backend( backend=self.backend_cb.currentText(), simulate=self.sim_check.isChecked(), ni_device=self.ni_dev_edit.text(), min_val=self.ni_min_spin.value(), max_val=self.ni_max_spin.value(), port=self.ard_port_edit.text(), baud=int(self.ard_baud_cb.currentText()), )