From 2ed9d37da7b27d25173535550fb92702225ac14e Mon Sep 17 00:00:00 2001 From: Christian Kolset Date: Mon, 13 Apr 2026 15:22:56 -0600 Subject: Removed QtPy5 snippet and fixed directories --- devices/analog_input.py | 276 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 276 insertions(+) create mode 100644 devices/analog_input.py (limited to 'devices/analog_input.py') diff --git a/devices/analog_input.py b/devices/analog_input.py new file mode 100644 index 0000000..5bb5d28 --- /dev/null +++ b/devices/analog_input.py @@ -0,0 +1,276 @@ +""" +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()), + ) -- cgit v1.2.3