""" devices/arduino_device.py Combined Arduino physical device — analog inputs + digital I/O on one board. Uses port_registry so the single ArduinoLayer is shared between analog reads and digital writes without opening the serial port twice. """ from typing import Any, Dict, List, Optional from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QFormLayout, QGroupBox, QComboBox, QCheckBox, QLineEdit, QLabel, QPushButton, QListWidget, QListWidgetItem, QTextEdit, ) from PyQt6.QtCore import Qt, QThread, pyqtSignal from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus from api_layers.arduino_layer import ArduinoLayer _ANALOG_COLORS = [ "#00d4ff", "#ff6b35", "#7fff6e", "#ffcc00", "#c77dff", "#ff4d6d", ] _DI_COLORS = ["#4cc9f0", "#90e0ef", "#caf0f8", "#0077b6", "#023e8a", "#48cae4", "#ade8f4", "#00b4d8"] _DO_COLORS = ["#ff6b35", "#ffcc00", "#f77f00", "#fcbf49", "#d62828", "#e63946", "#fb8500", "#ffd166"] def _parse_pin_edit(text: str) -> List[str]: """Parse "2,3,8" or "D2,D3,D8" → ["D2","D3","D8"]. Empty string → [].""" pins = [] for tok in text.split(","): tok = tok.strip().upper() if not tok: continue if not tok.startswith("D"): tok = "D" + tok pins.append(tok) return pins def _parse_analog_pin_edit(text: str) -> List[str]: """Parse "A0,A1,A3" or "0,1,3" → ["A0","A1","A3"]. Empty string → [].""" pins = [] for tok in text.split(","): tok = tok.strip().upper() if not tok: continue if not tok.startswith("A"): tok = "A" + tok pins.append(tok) return pins class ArduinoDevice(BaseDevice): DEVICE_TYPE = "arduino" ICON = "⚡" def __init__( self, device_id: str = "ard_0", analog_pins: Optional[List[str]] = None, di_pins: Optional[List[str]] = None, do_pins: Optional[List[str]] = None, # backward compat — accepted when new list params not provided num_analog: Optional[int] = None, num_di: Optional[int] = None, num_do: Optional[int] = None, simulate: bool = True, port: str = "COM3", baud: int = 115200, ): self.simulate = simulate self.backend = "arduino" self._port = port self._baud = baud self._last_error = "" # Resolve each pin list — new list param wins, else fall back to count if analog_pins is None: n = num_analog if num_analog is not None else 4 analog_pins = ArduinoLayer.DEFAULT_ANALOG_PINS[:n] if di_pins is None: n = num_di if num_di is not None else 2 di_pins = ArduinoLayer.DEFAULT_DI_PINS[:n] if do_pins is None: n = num_do if num_do is not None else 4 do_pins = ArduinoLayer.DEFAULT_DO_PINS[:n] self._analog_pins: List[str] = list(analog_pins) self._di_pins: List[str] = list(di_pins) self._do_pins: List[str] = list(do_pins) channels = [] for i, pin in enumerate(self._analog_pins): channels.append(ChannelConfig( channel_id=pin, name=pin, unit="V", min_value=0.0, max_value=5.0, color=_ANALOG_COLORS[i % len(_ANALOG_COLORS)], )) for i in range(len(self._di_pins)): channels.append(ChannelConfig( channel_id=f"di{i}", name=f"DI {i}", unit="", min_value=0.0, max_value=1.0, color=_DI_COLORS[i % len(_DI_COLORS)], )) for i in range(len(self._do_pins)): channels.append(ChannelConfig( channel_id=f"do{i}", name=f"DO {i}", unit="", min_value=0.0, max_value=1.0, writable=True, color=_DO_COLORS[i % len(_DO_COLORS)], )) info = DeviceInfo( device_id=device_id, name="Arduino", device_type=self.DEVICE_TYPE, description=(f"Arduino — {len(self._analog_pins)} analog, " f"{len(self._di_pins)} DI, {len(self._do_pins)} DO"), manufacturer="Arduino", icon=self.ICON, channels=channels, ) super().__init__(info) self._output_state: Dict[str, int] = {f"do{i}": 0 for i in range(len(self._do_pins))} self._di_state: Dict[str, int] = {f"di{i}": 0 for i in range(len(self._di_pins))} self._di_map = {f"di{i}": pin for i, pin in enumerate(self._di_pins)} self._do_map = {f"do{i}": pin for i, pin in enumerate(self._do_pins)} self._layer = self._make_layer() def _make_layer(self): from api_layers.port_registry import port_registry return port_registry.get_layer( port=self._port, baud=self._baud, simulate=self.simulate, extra_pins=self._analog_pins, ) # ── BaseDevice ──────────────────────────────────────────────────────────── def connect(self) -> bool: self._last_error = "" layer = self._layer if layer.is_connected: ok = True else: ok = layer.connect() if not ok and hasattr(layer, "last_error"): self._last_error = layer.last_error if ok and not self.simulate: layer.configure_pins( di_pins=self._di_pins, do_pins=self._do_pins, analog_pins=self._analog_pins, ) self.status = DeviceStatus.SIMULATED if self.simulate else ( DeviceStatus.CONNECTED if ok else DeviceStatus.ERROR ) return ok def disconnect(self) -> None: if not self.simulate: from api_layers.port_registry import port_registry port_registry.release(self._port, self._baud) self.status = DeviceStatus.DISCONNECTED def read_channels(self) -> Dict[str, float]: raw = self._layer.read() result = {} for ch in self.info.channels: cid = ch.channel_id if cid in raw: result[cid] = raw[cid] elif cid.startswith("di"): pin = self._di_map.get(cid) if pin and pin in raw: self._di_state[cid] = int(raw[pin]) result[cid] = float(self._di_state.get(cid, 0)) elif cid.startswith("do"): result[cid] = float(self._output_state.get(cid, 0)) return result def write_channel(self, channel_id: str, value: Any) -> bool: cid = channel_id.strip() if cid.startswith("do"): self._output_state[cid] = int(bool(value)) if not self.simulate and hasattr(self._layer, "digital_write"): pin = self._do_map.get(cid) if pin: return self._layer.digital_write(pin, int(bool(value))) return True if cid.upper().startswith("D") and cid[1:].isdigit(): return self._layer.digital_write(cid, int(bool(value))) if hasattr(self._layer, "digital_write") else False if hasattr(self._layer, "set_parameter"): return self._layer.set_parameter(cid, value) return False def get_save_config(self) -> dict: return { "device_type": self.DEVICE_TYPE, "device_id": self.info.device_id, "analog_pins": self._analog_pins, "di_pins": self._di_pins, "do_pins": self._do_pins, "simulate": self.simulate, "port": self._port, "baud": self._baud, } def get_config_widget(self) -> QWidget: return ArduinoConfigWidget(self) def switch_backend(self, simulate: bool, port: str, baud: int) -> None: was_running = self.status in (DeviceStatus.CONNECTED, DeviceStatus.SIMULATED) if was_running: self.disconnect() self.simulate = simulate self._port = port self._baud = baud self._layer = self._make_layer() if was_running: self.connect() def remap_pins( self, analog_pins: List[str] = None, di_pins: List[str] = None, do_pins: List[str] = None, ) -> None: """Update active pin mapping and send config commands if hardware-connected.""" if analog_pins is not None: self._analog_pins = list(analog_pins) if di_pins is not None: self._di_pins = list(di_pins) self._di_map = {f"di{i}": p for i, p in enumerate(di_pins)} if do_pins is not None: self._do_pins = list(do_pins) self._do_map = {f"do{i}": p for i, p in enumerate(do_pins)} if self.status == DeviceStatus.CONNECTED: self._layer.configure_pins( di_pins=self._di_pins, do_pins=self._do_pins, analog_pins=self._analog_pins, ) # ── Port scanner ────────────────────────────────────────────────────────────── class _PortScanThread(QThread): done = pyqtSignal(list) def run(self): self.done.emit(ArduinoLayer.list_ports()) # ── Config widget ───────────────────────────────────────────────────────────── class ArduinoConfigWidget(QWidget): def __init__(self, device: ArduinoDevice): 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(420) # ── Serial settings ────────────────────────────────────────────── ser_grp = QGroupBox("Serial Connection") ser_form = QFormLayout(ser_grp) ser_form.setContentsMargins(10, 16, 10, 10) self.port_edit = QLineEdit(self.device._port) self.port_edit.setPlaceholderText("e.g. /dev/ttyUSB0 or COM3") ser_form.addRow("Port:", self.port_edit) self.baud_cb = QComboBox() self.baud_cb.addItems(["9600", "19200", "57600", "115200", "230400"]) self.baud_cb.setCurrentText(str(self.device._baud)) ser_form.addRow("Baud Rate:", self.baud_cb) self.sim_chk = QCheckBox("Simulation Mode (no hardware)") self.sim_chk.setChecked(self.device.simulate) ser_form.addRow(self.sim_chk) 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) ser_form.addRow(scan_row) self._port_list = QListWidget() self._port_list.setObjectName("portList") self._port_list.setMaximumHeight(90) self._port_list.itemClicked.connect(self._on_port_selected) ser_form.addRow(self._port_list) apply_btn = QPushButton("Apply & Reconnect") apply_btn.setObjectName("applyButton") apply_btn.clicked.connect(self._apply) ser_form.addRow(apply_btn) root.addWidget(ser_grp) # ── Pin Mapping ─────────────────────────────────────────────────── pin_grp = QGroupBox("Pin Mapping") pin_form = QFormLayout(pin_grp) pin_form.setContentsMargins(10, 16, 10, 10) self.analog_pins_edit = QLineEdit(", ".join(p[1:] for p in self.device._analog_pins)) self.analog_pins_edit.setPlaceholderText("e.g. 0, 1, 2, 3 (indices into A0–A5)") pin_form.addRow("Analog Input Pins:", self.analog_pins_edit) self.di_pins_edit = QLineEdit(", ".join(p[1:] for p in self.device._di_pins)) self.di_pins_edit.setPlaceholderText("e.g. 2, 3, 8") pin_form.addRow("Digital Input Pins:", self.di_pins_edit) self.do_pins_edit = QLineEdit(", ".join(p[1:] for p in self.device._do_pins)) self.do_pins_edit.setPlaceholderText("e.g. 5, 6, 7, 9") pin_form.addRow("Digital Output Pins:", self.do_pins_edit) hint = QLabel("Changes take effect on Apply & Reconnect.\n" "Analog: enter A-pin indices (0=A0, 1=A1, …). " "Digital: enter pin numbers.") hint.setObjectName("traceSource") hint.setWordWrap(True) pin_form.addRow(hint) root.addWidget(pin_grp) # ── Diagnostics ─────────────────────────────────────────────────── diag_grp = QGroupBox("Connection Diagnostics") diag_lay = QVBoxLayout(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(120) diag_lay.addWidget(self._diag_box) root.addWidget(diag_grp) root.addStretch() self._refresh_diag() def _refresh_diag(self): lines = [ f"Port: {self.device._port}", f"Baud: {self.device._baud}", f"Simulate: {self.device.simulate}", f"Status: {self.device.status.value}", f"Analog pins: {', '.join(self.device._analog_pins) or '(none)'}", f"DI pins: {', '.join(self.device._di_pins) or '(none)'}", f"DO pins: {', '.join(self.device._do_pins) or '(none)'}", f"pyserial: {'available' if ArduinoLayer.is_pyserial_available() else 'NOT INSTALLED'}", ] if hasattr(self.device._layer, "raw_lines") and self.device._layer.raw_lines: lines.append("\nLast lines from Arduino:") for r in self.device._layer.raw_lines: 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 _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 ports: self._scan_lbl.setText("No ports found") item = QListWidgetItem(" No serial ports detected") item.setFlags(item.flags() & ~Qt.ItemFlag.ItemIsSelectable) self._port_list.addItem(item) else: self._scan_lbl.setText(f"{len(ports)} port(s) — 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.port_edit.setText(port) def _apply(self): # Parse pin fields and update device before reconnect so connect() sends correct APIN/DPIN analog = _parse_analog_pin_edit(self.analog_pins_edit.text()) di = _parse_pin_edit(self.di_pins_edit.text()) do = _parse_pin_edit(self.do_pins_edit.text()) if analog: self.device._analog_pins = analog if di: self.device._di_pins = di self.device._di_map = {f"di{i}": p for i, p in enumerate(di)} if do: self.device._do_pins = do self.device._do_map = {f"do{i}": p for i, p in enumerate(do)} self.device.switch_backend( simulate=self.sim_chk.isChecked(), port=self.port_edit.text().strip() or self.device._port, baud=int(self.baud_cb.currentText()), ) self._refresh_diag()