diff options
Diffstat (limited to 'devices/arduino_device.py')
| -rw-r--r-- | devices/arduino_device.py | 423 |
1 files changed, 423 insertions, 0 deletions
diff --git a/devices/arduino_device.py b/devices/arduino_device.py new file mode 100644 index 0000000..7c1d21b --- /dev/null +++ b/devices/arduino_device.py @@ -0,0 +1,423 @@ +""" +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, + 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() |