summaryrefslogtreecommitdiff
path: root/devices/serial_device.py
diff options
context:
space:
mode:
authorChristian Kolset <christian.kolset@gmail.com>2026-04-13 15:22:56 -0600
committerChristian Kolset <christian.kolset@gmail.com>2026-04-13 15:22:56 -0600
commit2ed9d37da7b27d25173535550fb92702225ac14e (patch)
treed6ff360a656e577629d3cefdaa1ae02af7255864 /devices/serial_device.py
parent8d6acf3a8ea4b37f86b321dbf430be5be01b1267 (diff)
Removed QtPy5 snippet and fixed directories
Diffstat (limited to 'devices/serial_device.py')
-rw-r--r--devices/serial_device.py187
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),
+ )