summaryrefslogtreecommitdiff
path: root/devices/analog_input.py
diff options
context:
space:
mode:
authorChristian Kolset <christian.kolset@gmail.com>2026-04-13 15:22:56 -0600
committerChristian Kolset <christian.kolset@gmail.com>2026-04-13 15:22:56 -0600
commit2ed9d37da7b27d25173535550fb92702225ac14e (patch)
treed6ff360a656e577629d3cefdaa1ae02af7255864 /devices/analog_input.py
parent8d6acf3a8ea4b37f86b321dbf430be5be01b1267 (diff)
Removed QtPy5 snippet and fixed directories
Diffstat (limited to 'devices/analog_input.py')
-rw-r--r--devices/analog_input.py276
1 files changed, 276 insertions, 0 deletions
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()),
+ )