From 8d6acf3a8ea4b37f86b321dbf430be5be01b1267 Mon Sep 17 00:00:00 2001
From: Christian Kolset <christian.kolset@gmail.com>
Date: Mon, 13 Apr 2026 14:36:55 -0600
Subject: init

---
 daq_system/devices/serial_device.py | 136 ++++++++++++++++++++++++++++++++++++
 1 file changed, 136 insertions(+)
 create mode 100644 daq_system/devices/serial_device.py

(limited to 'daq_system/devices/serial_device.py')

diff --git a/daq_system/devices/serial_device.py b/daq_system/devices/serial_device.py
new file mode 100644
index 0000000..842a401
--- /dev/null
+++ b/daq_system/devices/serial_device.py
@@ -0,0 +1,136 @@
+"""
+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
-- 
cgit v1.2.3