""" devices/serial_device.py Serial / UART device module — reads data from a serial port. Parses CSV-format lines: "ch0,ch1,ch2,...\\n" """ import random import time from typing import Any, Dict from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QFormLayout, QGroupBox, QComboBox, QSpinBox, QLineEdit, QPushButton, QLabel ) from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus SER_COLORS = ["#7fff6e", "#4cc9f0", "#f72585", "#00d4ff"] 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, simulate: bool = True): channels = [ ChannelConfig( channel_id=f"s{i}", name=f"Serial {i}", unit="", min_value=0.0, max_value=1023.0, color=SER_COLORS[i % len(SER_COLORS)] ) for i in range(num_channels) ] info = DeviceInfo( device_id=device_id, name="Serial / UART", device_type=self.DEVICE_TYPE, description=f"Serial port {port} @ {baud_rate} baud", icon=self.ICON, channels=channels ) super().__init__(info) self.port = port self.baud_rate = baud_rate self.simulate = simulate self._serial = None # Replace with serial.Serial() for real hardware self._t0 = 0.0 def connect(self) -> bool: if self.simulate: self._t0 = time.time() self.status = DeviceStatus.SIMULATED return True try: import serial self._serial = serial.Serial(self.port, self.baud_rate, timeout=0.1) self.status = DeviceStatus.CONNECTED return True except Exception as e: print(f"[SerialDevice] Connect failed: {e}") self.status = DeviceStatus.ERROR return False def disconnect(self) -> None: if self._serial: try: self._serial.close() except Exception: pass self.status = DeviceStatus.DISCONNECTED def read_channels(self) -> Dict[str, float]: if self.simulate: return self._simulate_read() if not self._serial or not self._serial.is_open: return {} try: line = self._serial.readline().decode("utf-8").strip() if not line: return {} parts = line.split(",") return { ch.channel_id: float(parts[i]) for i, ch in enumerate(self.info.channels) if i < len(parts) } except Exception: return {} def _simulate_read(self) -> Dict[str, float]: t = time.time() - self._t0 import math return { ch.channel_id: round(512 + 400 * math.sin(2 * 3.14159 * (0.2 + i * 0.15) * t) + random.gauss(0, 5), 1) for i, ch in enumerate(self.info.channels) } def write_channel(self, channel_id: str, value: Any) -> bool: if self._serial and self._serial.is_open: try: cmd = f"{channel_id}:{value}\n" self._serial.write(cmd.encode()) return True except Exception: return False return False def get_config_widget(self) -> QWidget: w = QWidget() layout = QVBoxLayout(w) grp = QGroupBox("Serial Port Settings") form = QFormLayout(grp) self._port_edit = QLineEdit(self.port) 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.baud_rate)) form.addRow("Baud Rate:", self._baud_cb) parity_cb = QComboBox() parity_cb.addItems(["None", "Even", "Odd"]) form.addRow("Parity:", parity_cb) bits_cb = QComboBox() bits_cb.addItems(["8", "7"]) form.addRow("Data Bits:", bits_cb) apply_btn = QPushButton("Apply & Reconnect") form.addRow(apply_btn) layout.addWidget(grp) layout.addStretch() return w