summaryrefslogtreecommitdiff
path: root/daq_system/devices/serial_device.py
diff options
context:
space:
mode:
Diffstat (limited to 'daq_system/devices/serial_device.py')
-rw-r--r--daq_system/devices/serial_device.py136
1 files changed, 0 insertions, 136 deletions
diff --git a/daq_system/devices/serial_device.py b/daq_system/devices/serial_device.py
deleted file mode 100644
index 842a401..0000000
--- a/daq_system/devices/serial_device.py
+++ /dev/null
@@ -1,136 +0,0 @@
-"""
-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