1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
|
"""
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._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),
)
|
