""" ui/add_device_dialog.py Add Device dialog — user selects physical device type (Arduino / NI-DAQ / Serial). Backend is determined by device type; Arduino and NI-DAQ combine analog input and digital I/O into one physical device entry. """ import os from PyQt6.QtWidgets import ( QDialog, QVBoxLayout, QHBoxLayout, QFormLayout, QComboBox, QLineEdit, QSpinBox, QDoubleSpinBox, QCheckBox, QPushButton, QLabel, QMessageBox, QStackedWidget, QWidget, QGroupBox, QListWidget, QListWidgetItem, QFrame, QSizePolicy, ) from PyQt6.QtCore import Qt, QThread, pyqtSignal from PyQt6.QtGui import QFont from devices.device_registry import DeviceRegistry from devices.arduino_device import ArduinoDevice from devices.nidaqmx_device import NidaqmxDevice from devices.serial_device import SerialDevice, _FORMAT_LABELS # ── Background scan threads ─────────────────────────────────────────────────── class PortScanThread(QThread): ports_found = pyqtSignal(list) def run(self): from api_layers.arduino_layer import ArduinoLayer self.ports_found.emit(ArduinoLayer.list_ports()) class NIScanThread(QThread): devices_found = pyqtSignal(list) def run(self): try: import nidaqmx # type: ignore devs = [(d.name, d.product_type) for d in nidaqmx.system.System().devices] except Exception: devs = [] self.devices_found.emit(devs) # ── Reusable scanner widgets ────────────────────────────────────────────────── class PortScanGroup(QGroupBox): """Scan-and-click widget that fills a target QLineEdit with the chosen port.""" def __init__(self, target_edit: QLineEdit): super().__init__("Available Serial Ports") self._target = target_edit self._scanner = None lay = QVBoxLayout(self) lay.setSpacing(4) top = QHBoxLayout() self._scan_btn = QPushButton("🔍 Scan Ports") self._scan_btn.setObjectName("addTraceBtn") self._scan_btn.clicked.connect(self._scan) self._status = QLabel("Click Scan to detect ports") self._status.setObjectName("traceSource") top.addWidget(self._scan_btn) top.addWidget(self._status, 1) lay.addLayout(top) self._list = QListWidget() self._list.setMaximumHeight(100) self._list.setObjectName("portList") self._list.itemClicked.connect(self._on_select) self._list.setToolTip("Click a port to select it") lay.addWidget(self._list) hint = QLabel("↑ Click a port above to fill the Port field") hint.setObjectName("traceSource") lay.addWidget(hint) def _scan(self): self._scan_btn.setEnabled(False) self._status.setText("Scanning…") self._list.clear() self._scanner = PortScanThread() self._scanner.ports_found.connect(self._on_found) self._scanner.start() def _on_found(self, ports): self._scan_btn.setEnabled(True) self._list.clear() if not ports: self._status.setText("No ports found") item = QListWidgetItem(" No serial ports detected") item.setFlags(item.flags() & ~Qt.ItemFlag.ItemIsSelectable) self._list.addItem(item) else: self._status.setText(f"{len(ports)} port(s) found") for device, desc in ports: label = f" {device}" if desc and desc != device: label += f" — {desc}" item = QListWidgetItem(label) item.setData(Qt.ItemDataRole.UserRole, device) self._list.addItem(item) def _on_select(self, item: QListWidgetItem): port = item.data(Qt.ItemDataRole.UserRole) if port: self._target.setText(port) class NIScanGroup(QGroupBox): """Scan-and-click widget that fills a target QLineEdit with the chosen NI device.""" def __init__(self, target_edit: QLineEdit): super().__init__("Available NI Devices") self._target = target_edit self._scanner = None lay = QVBoxLayout(self) lay.setSpacing(4) top = QHBoxLayout() self._scan_btn = QPushButton("🔍 Scan NI Devices") self._scan_btn.setObjectName("addTraceBtn") self._scan_btn.clicked.connect(self._scan) self._status = QLabel("Click Scan to detect NI devices") self._status.setObjectName("traceSource") top.addWidget(self._scan_btn) top.addWidget(self._status, 1) lay.addLayout(top) self._list = QListWidget() self._list.setMaximumHeight(100) self._list.setObjectName("portList") self._list.itemClicked.connect(self._on_select) self._list.setToolTip("Click a device to select it") lay.addWidget(self._list) hint = QLabel("↑ Click a device above to fill the NI Device field") hint.setObjectName("traceSource") lay.addWidget(hint) def _scan(self): self._scan_btn.setEnabled(False) self._status.setText("Scanning…") self._list.clear() self._scanner = NIScanThread() self._scanner.devices_found.connect(self._on_found) self._scanner.start() def _on_found(self, devices): self._scan_btn.setEnabled(True) self._list.clear() if not devices: self._status.setText("No NI devices found") item = QListWidgetItem(" No NI devices detected") item.setFlags(item.flags() & ~Qt.ItemFlag.ItemIsSelectable) self._list.addItem(item) else: self._status.setText(f"{len(devices)} device(s) found") for name, product in devices: label = f" {name}" if product: label += f" — {product}" item = QListWidgetItem(label) item.setData(Qt.ItemDataRole.UserRole, name) self._list.addItem(item) def _on_select(self, item: QListWidgetItem): name = item.data(Qt.ItemDataRole.UserRole) if name: self._target.setText(name) # ── Per-type config panels ──────────────────────────────────────────────────── class ArduinoPanel(QWidget): """Config fields for ArduinoDevice (analog + digital I/O combined).""" def __init__(self): super().__init__() lay = QFormLayout(self) lay.setContentsMargins(0, 4, 0, 4) self.port_edit = QLineEdit("") self.port_edit.setPlaceholderText("e.g. COM3 or /dev/ttyUSB0") lay.addRow("Port:", self.port_edit) self.baud_cb = QComboBox() self.baud_cb.addItems(["9600", "57600", "115200", "230400"]) self.baud_cb.setCurrentText("115200") lay.addRow("Baud Rate:", self.baud_cb) self.analog_pins_edit = QLineEdit("0, 1, 2, 3") self.analog_pins_edit.setPlaceholderText("e.g. 0, 1, 2, 3 (indices into A0–A5)") lay.addRow("Analog Input Pins:", self.analog_pins_edit) self.di_pins_edit = QLineEdit("2, 3") self.di_pins_edit.setPlaceholderText("e.g. 2, 3, 8") lay.addRow("Digital Input Pins:", self.di_pins_edit) self.do_pins_edit = QLineEdit("5, 6, 7, 9") self.do_pins_edit.setPlaceholderText("e.g. 5, 6, 7, 9") lay.addRow("Digital Output Pins:", self.do_pins_edit) self.sim_chk = QCheckBox("Simulation mode") self.sim_chk.setChecked(False) lay.addRow(self.sim_chk) lay.addRow(PortScanGroup(self.port_edit)) def build_device(self, device_id: str) -> ArduinoDevice: from devices.arduino_device import _parse_pin_edit, _parse_analog_pin_edit analog = _parse_analog_pin_edit(self.analog_pins_edit.text()) or None di = _parse_pin_edit(self.di_pins_edit.text()) or None do = _parse_pin_edit(self.do_pins_edit.text()) or None return ArduinoDevice( device_id=device_id, analog_pins=analog, di_pins=di, do_pins=do, simulate=self.sim_chk.isChecked(), port=self.port_edit.text().strip() or "COM3", baud=int(self.baud_cb.currentText()), ) class NidaqmxPanel(QWidget): """Config fields for NidaqmxDevice (analog + digital I/O combined).""" def __init__(self): super().__init__() lay = QFormLayout(self) lay.setContentsMargins(0, 4, 0, 4) self.ni_device_edit = QLineEdit("Dev1") lay.addRow("NI Device:", self.ni_device_edit) self.analog_spin = QSpinBox() self.analog_spin.setRange(0, 16) self.analog_spin.setValue(4) lay.addRow("Analog Inputs:", self.analog_spin) self.min_v_spin = QDoubleSpinBox() self.min_v_spin.setRange(-100.0, 0.0) self.min_v_spin.setValue(-10.0) self.min_v_spin.setSuffix(" V") lay.addRow("Min Voltage:", self.min_v_spin) self.max_v_spin = QDoubleSpinBox() self.max_v_spin.setRange(0.0, 100.0) self.max_v_spin.setValue(10.0) self.max_v_spin.setSuffix(" V") lay.addRow("Max Voltage:", self.max_v_spin) self.di_spin = QSpinBox() self.di_spin.setRange(0, 32) self.di_spin.setValue(2) lay.addRow("Digital Inputs:", self.di_spin) self.do_spin = QSpinBox() self.do_spin.setRange(0, 32) self.do_spin.setValue(4) lay.addRow("Digital Outputs:", self.do_spin) self.sim_chk = QCheckBox("Simulation mode") self.sim_chk.setChecked(False) lay.addRow(self.sim_chk) lay.addRow(NIScanGroup(self.ni_device_edit)) def build_device(self, device_id: str) -> NidaqmxDevice: return NidaqmxDevice( device_id=device_id, num_analog=self.analog_spin.value(), min_v=self.min_v_spin.value(), max_v=self.max_v_spin.value(), num_di=self.di_spin.value(), num_do=self.do_spin.value(), simulate=self.sim_chk.isChecked(), ni_device=self.ni_device_edit.text().strip() or "Dev1", ) class SerialPanel(QWidget): """Config fields for SerialDevice — port, baud, protocol/format, simulate.""" def __init__(self): super().__init__() lay = QVBoxLayout(self) lay.setContentsMargins(0, 4, 0, 4) lay.setSpacing(8) form = QFormLayout() self.port_edit = QLineEdit("") self.port_edit.setPlaceholderText("e.g. COM3 or /dev/ttyUSB0") form.addRow("Port:", self.port_edit) self.baud_cb = QComboBox() self.baud_cb.addItems(["1200", "2400", "4800", "9600", "19200", "38400", "57600", "115200", "230400", "460800"]) self.baud_cb.setCurrentText("115200") form.addRow("Baud Rate:", self.baud_cb) self.fmt_cb = QComboBox() self.fmt_cb.addItems(list(_FORMAT_LABELS.keys())) form.addRow("Protocol / Format:", self.fmt_cb) self.sim_chk = QCheckBox("Simulation mode") self.sim_chk.setChecked(False) form.addRow(self.sim_chk) note = QLabel("Protocol-specific settings available in device config dialog.") note.setObjectName("traceSource") note.setWordWrap(True) lay.addLayout(form) lay.addWidget(note) lay.addWidget(PortScanGroup(self.port_edit)) def build_device(self, device_id: str) -> SerialDevice: fmt = _FORMAT_LABELS.get(self.fmt_cb.currentText(), "key:val") return SerialDevice( device_id=device_id, port=self.port_edit.text().strip() or "COM3", baud_rate=int(self.baud_cb.currentText()), parse_format=fmt, simulate=self.sim_chk.isChecked(), ) # ── Main dialog ─────────────────────────────────────────────────────────────── _PANELS = { "Arduino": (ArduinoPanel, "ard"), "NI-DAQ": (NidaqmxPanel, "ni"), "Serial / UART":(SerialPanel, "ser"), } class AddDeviceDialog(QDialog): def __init__(self, registry: DeviceRegistry, parent=None): super().__init__(parent) self.registry = registry self.created_device = None self.setWindowTitle("Add Device") self.setMinimumSize(480, 480) self.resize(500, 620) self._build() def _build(self): root = QVBoxLayout(self) root.setSpacing(10) hdr = QLabel("Add New Device") hdr.setObjectName("devWindowTitle") root.addWidget(hdr) div = QFrame() div.setFrameShape(QFrame.Shape.HLine) div.setObjectName("devWindowDivider") root.addWidget(div) type_row = QFormLayout() self._type_cb = QComboBox() self._type_cb.addItems(list(_PANELS.keys())) self._type_cb.currentIndexChanged.connect(self._on_type_changed) type_row.addRow("Device Type:", self._type_cb) self._id_edit = QLineEdit() self._id_edit.setPlaceholderText("Leave blank for auto") type_row.addRow("Device ID:", self._id_edit) root.addLayout(type_row) div2 = QFrame() div2.setFrameShape(QFrame.Shape.HLine) div2.setObjectName("devWindowDivider") root.addWidget(div2) self._stack = QStackedWidget() self._panels = {} for name, (cls, _prefix) in _PANELS.items(): panel = cls() self._panels[name] = panel self._stack.addWidget(panel) root.addWidget(self._stack, 1) div3 = QFrame() div3.setFrameShape(QFrame.Shape.HLine) div3.setObjectName("devWindowDivider") root.addWidget(div3) btn_row = QHBoxLayout() btn_row.addStretch() cancel = QPushButton("Cancel") cancel.clicked.connect(self.reject) add = QPushButton("Add Device") add.setObjectName("applyButton") add.setDefault(True) add.clicked.connect(self._on_add) btn_row.addWidget(cancel) btn_row.addWidget(add) root.addLayout(btn_row) self._on_type_changed(0) def _on_type_changed(self, idx: int): self._stack.setCurrentIndex(idx) type_name = self._type_cb.currentText() _, prefix = _PANELS.get(type_name, (None, "dev")) existing = {d.info.device_id for d in self.registry.all_instances()} for i in range(100): candidate = f"{prefix}_{i}" if candidate not in existing: self._id_edit.setPlaceholderText(f"e.g. {candidate} (auto)") break def _on_add(self): type_name = self._type_cb.currentText() panel = self._panels[type_name] dev_id = self._id_edit.text().strip() if not dev_id: _, prefix = _PANELS.get(type_name, (None, "dev")) existing = {d.info.device_id for d in self.registry.all_instances()} for i in range(100): candidate = f"{prefix}_{i}" if candidate not in existing: dev_id = candidate break if self.registry.get_instance(dev_id): QMessageBox.warning(self, "Duplicate ID", f"A device with ID '{dev_id}' already exists.\n" f"Choose a different ID.") return try: dev = panel.build_device(dev_id) self.created_device = dev self.accept() except Exception as e: QMessageBox.critical(self, "Error creating device", str(e))