diff options
Diffstat (limited to 'devices/serial_device.py')
| -rw-r--r-- | devices/serial_device.py | 187 |
1 files changed, 187 insertions, 0 deletions
diff --git a/devices/serial_device.py b/devices/serial_device.py new file mode 100644 index 0000000..6bb7bf1 --- /dev/null +++ b/devices/serial_device.py @@ -0,0 +1,187 @@ +""" +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_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._build() + + def _build(self): + root = QVBoxLayout(self) + root.setContentsMargins(0, 0, 0, 0) + + grp = QGroupBox("Port Settings") + form = QFormLayout(grp) + + self.port_edit = QLineEdit(self.device._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.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_chk = QComboBox() + self.sim_chk.addItems(["Simulate", "Real Hardware"]) + self.sim_chk.setCurrentIndex(0 if self.device.simulate else 1) + form.addRow("Mode:", self.sim_chk) + + scan_btn = QPushButton("Scan Ports") + scan_btn.clicked.connect(self._scan) + form.addRow(scan_btn) + self.port_lbl = QLabel("") + form.addRow(self.port_lbl) + + root.addWidget(grp) + + apply_btn = QPushButton("Apply & Reconnect") + apply_btn.setObjectName("applyButton") + apply_btn.clicked.connect(self._apply) + root.addWidget(apply_btn) + root.addStretch() + + def _scan(self): + ports = ArduinoLayer.list_ports() + self.port_lbl.setText(", ".join(ports) if ports else "None found") + + def _apply(self): + self.device.reconfigure( + port=self.port_edit.text(), + baud=int(self.baud_cb.currentText()), + fmt=self.fmt_cb.currentText(), + simulate=(self.sim_chk.currentIndex() == 0), + ) |