""" devices/serial_device.py Generic Serial / UART device module. Uses ArduinoLayer for communication, but works with ANY instrument that sends newline-terminated data. Configurable parse formats: • "csv" – plain comma-separated values mapped to channels in order • "key:val" – "CH0:1.23,CH1:4.56" key-colon-value pairs • "json" – {"CH0":1.23,"CH1":4.56} Switch format in the config widget without restarting. """ import json import math import random import threading import time from typing import Any, Dict, List from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QFormLayout, QGroupBox, QComboBox, QLineEdit, QSpinBox, QLabel, QPushButton, ) from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus from api_layers.arduino_layer import ArduinoLayer _COLORS = ["#7fff6e", "#4cc9f0", "#f72585", "#00d4ff", "#ffcc00", "#c77dff", "#ff6b35", "#38b000"] class SerialDevice(BaseDevice): DEVICE_TYPE = "serial" ICON = "⇌" def __init__( self, device_id: str = "ser_0", port: str = "COM3", baud_rate: int = 115200, num_channels: int = 4, channel_names: List[str] = None, units: List[str] = None, parse_format: str = "key:val", # "csv" | "key:val" | "json" simulate: bool = True, ): self._port = port self._baud = baud_rate self._parse_format = parse_format self.simulate = simulate names = channel_names or [f"CH{i}" for i in range(num_channels)] _units = units or ["" for _ in range(num_channels)] channels = [ ChannelConfig( channel_id=names[i], name=names[i], unit=_units[i], min_value=0.0, max_value=1023.0, color=_COLORS[i % len(_COLORS)], ) for i in range(num_channels) ] info = DeviceInfo( device_id=device_id, name="Serial / UART", device_type=self.DEVICE_TYPE, description=f"{port} @ {baud_rate}", icon=self.ICON, channels=channels, ) super().__init__(info) self._layer = ArduinoLayer( port=port, baud=baud_rate, analog_pins=[ch.channel_id for ch in channels], simulate=simulate, ) self._t0 = 0.0 # ── BaseDevice ────────────────────────────────────────────────────── def connect(self) -> bool: self._t0 = time.time() ok = self._layer.connect() self.status = DeviceStatus.SIMULATED if self.simulate else ( DeviceStatus.CONNECTED if ok else DeviceStatus.ERROR ) return ok def disconnect(self) -> None: self._layer.disconnect() self.status = DeviceStatus.DISCONNECTED def read_channels(self) -> Dict[str, float]: raw = self._layer.read() # Map by order if keys don't match channel IDs if raw: mapped: Dict[str, float] = {} raw_vals = list(raw.values()) for i, ch in enumerate(self.info.channels): if ch.channel_id in raw: mapped[ch.channel_id] = raw[ch.channel_id] elif i < len(raw_vals): mapped[ch.channel_id] = raw_vals[i] return mapped return {} def write_channel(self, channel_id: str, value: Any) -> bool: return self._layer.write(channel_id, int(value)) def get_save_config(self) -> dict: return { "device_type": self.DEVICE_TYPE, "device_id": self.info.device_id, "port": self._port, "baud_rate": self._baud, "num_channels": len(self.info.channels), "channel_names": [c.channel_id for c in self.info.channels], "units": [c.unit for c in self.info.channels], "parse_format": self._parse_format, "simulate": self.simulate, } def get_config_widget(self) -> QWidget: return SerialConfigWidget(self) def reconfigure(self, port: str, baud: int, fmt: str, simulate: bool): was_on = self.status in (DeviceStatus.CONNECTED, DeviceStatus.SIMULATED) if was_on: self.disconnect() self._port = port self._baud = baud self._parse_format = fmt self.simulate = simulate self._layer = ArduinoLayer( port=port, baud=baud, analog_pins=[ch.channel_id for ch in self.info.channels], simulate=simulate, ) if was_on: self.connect() class SerialConfigWidget(QWidget): def __init__(self, device: SerialDevice): super().__init__() self.device = device self._scanner = None self._build() def _build(self): from PyQt6.QtWidgets import QListWidget, QListWidgetItem from PyQt6.QtCore import Qt as _Qt root = QVBoxLayout(self) root.setContentsMargins(8, 8, 8, 8) root.setSpacing(8) self.setMinimumWidth(420) # ── Port settings ───────────────────────────────────────────── grp = QGroupBox("Port Settings") form = QFormLayout(grp) self.port_edit = QLineEdit(self.device._port) self.port_edit.setPlaceholderText("e.g. COM3 or /dev/ttyUSB0") form.addRow("Port:", self.port_edit) self.baud_cb = QComboBox() self.baud_cb.addItems(["9600", "19200", "38400", "57600", "115200", "230400", "460800"]) self.baud_cb.setCurrentText(str(self.device._baud)) form.addRow("Baud Rate:", self.baud_cb) self.fmt_cb = QComboBox() self.fmt_cb.addItems(["key:val", "csv", "json"]) self.fmt_cb.setCurrentText(self.device._parse_format) form.addRow("Parse Format:", self.fmt_cb) self.sim_cb = QComboBox() self.sim_cb.addItems(["Simulate", "Real Hardware"]) self.sim_cb.setCurrentIndex(0 if self.device.simulate else 1) form.addRow("Mode:", self.sim_cb) root.addWidget(grp) # ── Port scanner ────────────────────────────────────────────── scan_grp = QGroupBox("Available Serial Ports") scan_lay = QVBoxLayout(scan_grp) scan_lay.setSpacing(4) scan_top = QHBoxLayout() self._scan_btn = QPushButton("🔍 Scan Ports") self._scan_btn.setObjectName("addTraceBtn") self._scan_btn.clicked.connect(self._scan) self._scan_lbl = QLabel("Click Scan to detect ports") self._scan_lbl.setObjectName("traceSource") scan_top.addWidget(self._scan_btn) scan_top.addWidget(self._scan_lbl, 1) scan_lay.addLayout(scan_top) self._port_list = QListWidget() self._port_list.setObjectName("portList") self._port_list.setMaximumHeight(110) self._port_list.itemClicked.connect( lambda item: self.port_edit.setText( item.data(_Qt.ItemDataRole.UserRole) or "" ) if item.data(_Qt.ItemDataRole.UserRole) else None ) hint = QLabel("↑ Click a port above to select it") hint.setObjectName("traceSource") scan_lay.addWidget(self._port_list) scan_lay.addWidget(hint) root.addWidget(scan_grp) # ── Apply ───────────────────────────────────────────────────── apply_btn = QPushButton("Apply & Reconnect") apply_btn.setObjectName("applyButton") apply_btn.clicked.connect(self._apply) root.addWidget(apply_btn) root.addStretch() def _scan(self): self._scan_btn.setEnabled(False) self._scan_lbl.setText("Scanning…") self._port_list.clear() from PyQt6.QtCore import QThread, pyqtSignal as _sig class _Scanner(QThread): done = _sig(list) def run(self): self.done.emit(ArduinoLayer.list_ports()) self._scanner = _Scanner() self._scanner.done.connect(self._on_scan_done) self._scanner.start() def _on_scan_done(self, ports): from PyQt6.QtWidgets import QListWidgetItem 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 — check USB / permissions") 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 _apply(self): self.device.reconfigure( port=self.port_edit.text().strip() or "COM3", baud=int(self.baud_cb.currentText()), fmt=self.fmt_cb.currentText(), simulate=(self.sim_cb.currentIndex() == 0), )