summaryrefslogtreecommitdiff
path: root/daq_system/devices
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 /daq_system/devices
parent8d6acf3a8ea4b37f86b321dbf430be5be01b1267 (diff)
Removed QtPy5 snippet and fixed directories
Diffstat (limited to 'daq_system/devices')
-rw-r--r--daq_system/devices/__init__.py1
-rw-r--r--daq_system/devices/__pycache__/__init__.cpython-312.pycbin164 -> 0 bytes
-rw-r--r--daq_system/devices/__pycache__/analog_input.cpython-312.pycbin9034 -> 0 bytes
-rw-r--r--daq_system/devices/__pycache__/base_device.cpython-312.pycbin6014 -> 0 bytes
-rw-r--r--daq_system/devices/__pycache__/device_registry.cpython-312.pycbin5219 -> 0 bytes
-rw-r--r--daq_system/devices/__pycache__/digital_io.cpython-312.pycbin6027 -> 0 bytes
-rw-r--r--daq_system/devices/__pycache__/serial_device.cpython-312.pycbin7405 -> 0 bytes
-rw-r--r--daq_system/devices/__pycache__/temperature.cpython-312.pycbin6707 -> 0 bytes
-rw-r--r--daq_system/devices/analog_input.py184
-rw-r--r--daq_system/devices/base_device.py127
-rw-r--r--daq_system/devices/device_registry.py87
-rw-r--r--daq_system/devices/digital_io.py95
-rw-r--r--daq_system/devices/serial_device.py136
-rw-r--r--daq_system/devices/temperature.py115
14 files changed, 0 insertions, 745 deletions
diff --git a/daq_system/devices/__init__.py b/daq_system/devices/__init__.py
deleted file mode 100644
index 9c98181..0000000
--- a/daq_system/devices/__init__.py
+++ /dev/null
@@ -1 +0,0 @@
-# devices/__init__.py
diff --git a/daq_system/devices/__pycache__/__init__.cpython-312.pyc b/daq_system/devices/__pycache__/__init__.cpython-312.pyc
deleted file mode 100644
index af251ba..0000000
--- a/daq_system/devices/__pycache__/__init__.cpython-312.pyc
+++ /dev/null
Binary files differ
diff --git a/daq_system/devices/__pycache__/analog_input.cpython-312.pyc b/daq_system/devices/__pycache__/analog_input.cpython-312.pyc
deleted file mode 100644
index efe5752..0000000
--- a/daq_system/devices/__pycache__/analog_input.cpython-312.pyc
+++ /dev/null
Binary files differ
diff --git a/daq_system/devices/__pycache__/base_device.cpython-312.pyc b/daq_system/devices/__pycache__/base_device.cpython-312.pyc
deleted file mode 100644
index 4f08c29..0000000
--- a/daq_system/devices/__pycache__/base_device.cpython-312.pyc
+++ /dev/null
Binary files differ
diff --git a/daq_system/devices/__pycache__/device_registry.cpython-312.pyc b/daq_system/devices/__pycache__/device_registry.cpython-312.pyc
deleted file mode 100644
index 8d5ca55..0000000
--- a/daq_system/devices/__pycache__/device_registry.cpython-312.pyc
+++ /dev/null
Binary files differ
diff --git a/daq_system/devices/__pycache__/digital_io.cpython-312.pyc b/daq_system/devices/__pycache__/digital_io.cpython-312.pyc
deleted file mode 100644
index 9ca7bc7..0000000
--- a/daq_system/devices/__pycache__/digital_io.cpython-312.pyc
+++ /dev/null
Binary files differ
diff --git a/daq_system/devices/__pycache__/serial_device.cpython-312.pyc b/daq_system/devices/__pycache__/serial_device.cpython-312.pyc
deleted file mode 100644
index f4a2190..0000000
--- a/daq_system/devices/__pycache__/serial_device.cpython-312.pyc
+++ /dev/null
Binary files differ
diff --git a/daq_system/devices/__pycache__/temperature.cpython-312.pyc b/daq_system/devices/__pycache__/temperature.cpython-312.pyc
deleted file mode 100644
index ab2a2b9..0000000
--- a/daq_system/devices/__pycache__/temperature.cpython-312.pyc
+++ /dev/null
Binary files differ
diff --git a/daq_system/devices/analog_input.py b/daq_system/devices/analog_input.py
deleted file mode 100644
index 8aaf9db..0000000
--- a/daq_system/devices/analog_input.py
+++ /dev/null
@@ -1,184 +0,0 @@
-"""
-devices/analog_input.py
-
-Analog Input device module — reads voltage/current channels.
-Includes a simulation mode (no hardware required) for development/demo.
-"""
-
-import math
-import random
-import time
-from typing import Any, Dict
-
-from PyQt6.QtWidgets import (
- QWidget, QVBoxLayout, QHBoxLayout, QLabel, QComboBox,
- QDoubleSpinBox, QGroupBox, QCheckBox, QSpinBox, QFormLayout
-)
-from PyQt6.QtCore import Qt
-
-from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus
-
-
-CHANNEL_COLORS = ["#00d4ff", "#ff6b35", "#7fff6e", "#ffcc00", "#c77dff", "#ff4d6d", "#4cc9f0", "#f72585"]
-
-
-class AnalogInputDevice(BaseDevice):
- """
- Analog voltage/current input module.
-
- Supports up to 16 channels. In simulation mode, generates
- realistic waveforms (sine, ramp, noise) for each channel.
- Real hardware: override read_channels() with your SDK calls.
- """
-
- DEVICE_TYPE = "analog_input"
- ICON = "〜"
-
- def __init__(self, device_id: str = "ai_0", num_channels: int = 4,
- simulate: bool = True, sample_rate_hz: float = 10.0):
- channels = [
- ChannelConfig(
- channel_id=f"ch{i}",
- name=f"AI {i}",
- unit="V",
- min_value=-10.0,
- max_value=10.0,
- alarm_low=-8.0,
- alarm_high=8.0,
- color=CHANNEL_COLORS[i % len(CHANNEL_COLORS)],
- )
- for i in range(num_channels)
- ]
- info = DeviceInfo(
- device_id=device_id,
- name="Analog Input",
- device_type=self.DEVICE_TYPE,
- description="Multi-channel analog voltage/current input",
- manufacturer="Generic",
- model="AI-16",
- icon=self.ICON,
- channels=channels,
- )
- super().__init__(info)
- self.simulate = simulate
- self.sample_rate_hz = sample_rate_hz
- self._start_time = 0.0
- # Sim parameters per channel
- self._sim_params = [
- {"freq": 0.5 + i * 0.3, "amp": 5.0, "offset": 0.0, "noise": 0.05, "mode": "sine"}
- for i in range(num_channels)
- ]
-
- # ------------------------------------------------------------------ #
- # BaseDevice interface #
- # ------------------------------------------------------------------ #
-
- def connect(self) -> bool:
- if self.simulate:
- self._start_time = time.time()
- self.status = DeviceStatus.SIMULATED
- return True
- # TODO: Replace with real hardware SDK init
- # e.g. import nidaqmx; self._task = nidaqmx.Task(); ...
- self.status = DeviceStatus.ERROR
- return False
-
- def disconnect(self) -> None:
- self.status = DeviceStatus.DISCONNECTED
-
- def read_channels(self) -> Dict[str, float]:
- if self.simulate:
- return self._simulate_read()
- # TODO: Replace with real hardware read
- return {}
-
- def write_channel(self, channel_id: str, value: Any) -> bool:
- # Analog inputs don't support write — subclass for AO
- return False
-
- def get_config_widget(self) -> QWidget:
- return AnalogInputConfigWidget(self)
-
- # ------------------------------------------------------------------ #
- # Simulation #
- # ------------------------------------------------------------------ #
-
- def _simulate_read(self) -> Dict[str, float]:
- t = time.time() - self._start_time
- result = {}
- for i, ch in enumerate(self.info.channels):
- if not ch.enabled:
- continue
- p = self._sim_params[i]
- if p["mode"] == "sine":
- val = p["amp"] * math.sin(2 * math.pi * p["freq"] * t) + p["offset"]
- elif p["mode"] == "ramp":
- period = 1.0 / max(p["freq"], 0.01)
- val = p["amp"] * ((t % period) / period) * 2 - p["amp"] + p["offset"]
- elif p["mode"] == "square":
- val = p["amp"] * math.copysign(1, math.sin(2 * math.pi * p["freq"] * t)) + p["offset"]
- else:
- val = p["offset"]
- val += random.gauss(0, p["noise"] * p["amp"])
- val = max(ch.min_value, min(ch.max_value, val))
- result[ch.channel_id] = round(val, 4)
- return result
-
-
-# ------------------------------------------------------------------ #
-# Config Widget #
-# ------------------------------------------------------------------ #
-
-class AnalogInputConfigWidget(QWidget):
- def __init__(self, device: AnalogInputDevice):
- super().__init__()
- self.device = device
- self._build_ui()
-
- def _build_ui(self):
- layout = QVBoxLayout(self)
- layout.setContentsMargins(0, 0, 0, 0)
-
- # Global settings
- global_group = QGroupBox("Device Settings")
- form = QFormLayout(global_group)
-
- self.sim_check = QCheckBox("Simulation Mode")
- self.sim_check.setChecked(self.device.simulate)
- form.addRow(self.sim_check)
-
- self.rate_spin = QDoubleSpinBox()
- self.rate_spin.setRange(0.1, 1000.0)
- self.rate_spin.setValue(self.device.sample_rate_hz)
- self.rate_spin.setSuffix(" Hz")
- form.addRow("Sample Rate:", self.rate_spin)
-
- layout.addWidget(global_group)
-
- # Per-channel
- ch_group = QGroupBox("Channel Configuration")
- ch_layout = QVBoxLayout(ch_group)
-
- for i, ch in enumerate(self.device.info.channels):
- row = QHBoxLayout()
- en = QCheckBox(ch.name)
- en.setChecked(ch.enabled)
- row.addWidget(en)
-
- mode_cb = QComboBox()
- mode_cb.addItems(["sine", "ramp", "square", "dc"])
- mode_cb.setCurrentText(self.device._sim_params[i]["mode"])
- row.addWidget(QLabel("Mode:"))
- row.addWidget(mode_cb)
-
- freq_sp = QDoubleSpinBox()
- freq_sp.setRange(0.01, 100.0)
- freq_sp.setValue(self.device._sim_params[i]["freq"])
- freq_sp.setSuffix(" Hz")
- row.addWidget(QLabel("Freq:"))
- row.addWidget(freq_sp)
-
- ch_layout.addLayout(row)
-
- layout.addWidget(ch_group)
- layout.addStretch()
diff --git a/daq_system/devices/base_device.py b/daq_system/devices/base_device.py
deleted file mode 100644
index 4898a7a..0000000
--- a/daq_system/devices/base_device.py
+++ /dev/null
@@ -1,127 +0,0 @@
-"""
-devices/base_device.py
-
-Abstract base class for all DAQ I/O modules.
-Every device plugin must subclass BaseDevice and implement the required methods.
-"""
-
-from abc import ABC, abstractmethod
-from dataclasses import dataclass, field
-from typing import Any, Dict, List, Optional
-from enum import Enum
-
-
-class DeviceStatus(Enum):
- DISCONNECTED = "disconnected"
- CONNECTING = "connecting"
- CONNECTED = "connected"
- ERROR = "error"
- SIMULATED = "simulated"
-
-
-@dataclass
-class ChannelConfig:
- """Configuration for a single I/O channel."""
- channel_id: str
- name: str
- unit: str = ""
- min_value: float = 0.0
- max_value: float = 100.0
- alarm_low: Optional[float] = None
- alarm_high: Optional[float] = None
- enabled: bool = True
- color: str = "#00d4ff" # For plotting
- extra: Dict[str, Any] = field(default_factory=dict)
-
-
-@dataclass
-class DeviceInfo:
- """Metadata describing a device module."""
- device_id: str
- name: str
- device_type: str # e.g. "analog_input", "digital_io", "serial", "temperature"
- description: str = ""
- manufacturer: str = ""
- model: str = ""
- version: str = "1.0.0"
- icon: str = "⚙" # Unicode icon for UI display
- channels: List[ChannelConfig] = field(default_factory=list)
-
-
-class BaseDevice(ABC):
- """
- Abstract base class for all DAQ device plugins.
-
- To create a new device module:
- 1. Subclass BaseDevice
- 2. Implement all @abstractmethod methods
- 3. Place the file in the devices/ directory
- 4. The DeviceRegistry will auto-discover it
- """
-
- def __init__(self, device_info: DeviceInfo):
- self.info = device_info
- self.status = DeviceStatus.DISCONNECTED
- self._callbacks: List[callable] = []
-
- # ------------------------------------------------------------------ #
- # Abstract interface — every device must implement these #
- # ------------------------------------------------------------------ #
-
- @abstractmethod
- def connect(self) -> bool:
- """
- Open connection to the physical device.
- Returns True on success, False on failure.
- Sets self.status appropriately.
- """
-
- @abstractmethod
- def disconnect(self) -> None:
- """Close the connection and release resources."""
-
- @abstractmethod
- def read_channels(self) -> Dict[str, float]:
- """
- Read current values from all enabled channels.
- Returns dict mapping channel_id -> float value.
- Called repeatedly by the acquisition loop.
- """
-
- @abstractmethod
- def write_channel(self, channel_id: str, value: Any) -> bool:
- """
- Write a value to an output channel (if supported).
- Returns True on success.
- """
-
- @abstractmethod
- def get_config_widget(self):
- """
- Return a QWidget with device-specific configuration controls.
- This widget is embedded in the Device Config panel.
- """
-
- # ------------------------------------------------------------------ #
- # Shared helpers #
- # ------------------------------------------------------------------ #
-
- def add_data_callback(self, callback: callable) -> None:
- """Register a callback: callback(device_id, channel_id, value, timestamp)"""
- self._callbacks.append(callback)
-
- def _emit(self, channel_id: str, value: float, timestamp: float) -> None:
- for cb in self._callbacks:
- try:
- cb(self.info.device_id, channel_id, value, timestamp)
- except Exception:
- pass
-
- def get_channel(self, channel_id: str) -> Optional[ChannelConfig]:
- for ch in self.info.channels:
- if ch.channel_id == channel_id:
- return ch
- return None
-
- def __repr__(self):
- return f"<{self.__class__.__name__} id={self.info.device_id} status={self.status.value}>"
diff --git a/daq_system/devices/device_registry.py b/daq_system/devices/device_registry.py
deleted file mode 100644
index e407bc3..0000000
--- a/daq_system/devices/device_registry.py
+++ /dev/null
@@ -1,87 +0,0 @@
-"""
-devices/device_registry.py
-
-Discovers, instantiates, and manages all device modules.
-Add new devices by dropping a .py file into the devices/ directory.
-"""
-
-import importlib
-import inspect
-import pkgutil
-from pathlib import Path
-from typing import Dict, List, Optional, Type
-
-from devices.base_device import BaseDevice, DeviceInfo
-
-
-class DeviceRegistry:
- """Central registry for all DAQ device modules."""
-
- def __init__(self):
- self._device_classes: Dict[str, Type[BaseDevice]] = {}
- self._instances: Dict[str, BaseDevice] = {}
- self._auto_discover()
-
- # ------------------------------------------------------------------ #
- # Discovery #
- # ------------------------------------------------------------------ #
-
- def _auto_discover(self):
- """Scan the devices/ package for BaseDevice subclasses."""
- devices_path = Path(__file__).parent
- package = "devices"
-
- for _, module_name, _ in pkgutil.iter_modules([str(devices_path)]):
- if module_name.startswith("_") or module_name in ("base_device", "device_registry"):
- continue
- try:
- module = importlib.import_module(f"{package}.{module_name}")
- for name, obj in inspect.getmembers(module, inspect.isclass):
- if issubclass(obj, BaseDevice) and obj is not BaseDevice:
- self._device_classes[name] = obj
- except Exception as e:
- print(f"[DeviceRegistry] Failed to load {module_name}: {e}")
-
- def register_class(self, cls: Type[BaseDevice]) -> None:
- """Manually register a device class (for testing / runtime plugins)."""
- self._device_classes[cls.__name__] = cls
-
- # ------------------------------------------------------------------ #
- # Instance management #
- # ------------------------------------------------------------------ #
-
- def create_device(self, class_name: str, device_id: str, **kwargs) -> Optional[BaseDevice]:
- """Instantiate a device by class name with a unique device_id."""
- cls = self._device_classes.get(class_name)
- if cls is None:
- raise ValueError(f"Unknown device class: {class_name}")
- instance = cls(device_id=device_id, **kwargs)
- self._instances[device_id] = instance
- return instance
-
- def add_instance(self, device: BaseDevice) -> None:
- """Register a pre-built device instance."""
- self._instances[device.info.device_id] = device
-
- def remove_instance(self, device_id: str) -> None:
- dev = self._instances.pop(device_id, None)
- if dev:
- try:
- dev.disconnect()
- except Exception:
- pass
-
- def get_instance(self, device_id: str) -> Optional[BaseDevice]:
- return self._instances.get(device_id)
-
- def all_instances(self) -> List[BaseDevice]:
- return list(self._instances.values())
-
- def available_classes(self) -> List[str]:
- return list(self._device_classes.keys())
-
- def get_class(self, class_name: str) -> Optional[Type[BaseDevice]]:
- return self._device_classes.get(class_name)
-
- def __len__(self):
- return len(self._instances)
diff --git a/daq_system/devices/digital_io.py b/daq_system/devices/digital_io.py
deleted file mode 100644
index b78b5cf..0000000
--- a/daq_system/devices/digital_io.py
+++ /dev/null
@@ -1,95 +0,0 @@
-"""
-devices/digital_io.py
-
-Digital I/O device module — reads and writes binary channels.
-"""
-
-import random
-import time
-from typing import Any, Dict
-
-from PyQt6.QtWidgets import QWidget, QVBoxLayout, QHBoxLayout, QCheckBox, QGroupBox, QPushButton, QLabel
-from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus
-
-
-class DigitalIODevice(BaseDevice):
- DEVICE_TYPE = "digital_io"
- ICON = "⬛"
-
- def __init__(self, device_id: str = "dio_0", num_inputs: int = 8,
- num_outputs: int = 8, simulate: bool = True):
- channels = []
- for i in range(num_inputs):
- channels.append(ChannelConfig(
- channel_id=f"di{i}", name=f"DI {i}", unit="",
- min_value=0.0, max_value=1.0,
- color="#00d4ff" if i % 2 == 0 else "#4cc9f0"
- ))
- for i in range(num_outputs):
- channels.append(ChannelConfig(
- channel_id=f"do{i}", name=f"DO {i}", unit="",
- min_value=0.0, max_value=1.0,
- color="#ff6b35" if i % 2 == 0 else "#ffcc00"
- ))
- info = DeviceInfo(
- device_id=device_id, name="Digital I/O",
- device_type=self.DEVICE_TYPE,
- description="Digital input/output module",
- icon=self.ICON, channels=channels
- )
- super().__init__(info)
- self.simulate = simulate
- self._output_state: Dict[str, int] = {f"do{i}": 0 for i in range(num_outputs)}
- self._toggle_counters = [0] * num_inputs
-
- def connect(self) -> bool:
- self.status = DeviceStatus.SIMULATED if self.simulate else DeviceStatus.ERROR
- return self.simulate
-
- def disconnect(self) -> None:
- self.status = DeviceStatus.DISCONNECTED
-
- def read_channels(self) -> Dict[str, float]:
- result = {}
- # Simulate toggling inputs randomly
- for i, ch in enumerate(self.info.channels):
- if ch.channel_id.startswith("di"):
- self._toggle_counters[i] += 1
- if self._toggle_counters[i] > random.randint(5, 30):
- self._toggle_counters[i] = 0
- result[ch.channel_id] = float(random.randint(0, 1))
- else:
- result[ch.channel_id] = result.get(ch.channel_id, 0.0)
- elif ch.channel_id.startswith("do"):
- result[ch.channel_id] = float(self._output_state.get(ch.channel_id, 0))
- return result
-
- def write_channel(self, channel_id: str, value: Any) -> bool:
- if channel_id in self._output_state:
- self._output_state[channel_id] = int(bool(value))
- return True
- return False
-
- def get_config_widget(self) -> QWidget:
- w = QWidget()
- layout = QVBoxLayout(w)
- grp = QGroupBox("Output Controls")
- grp_layout = QVBoxLayout(grp)
- for k in self._output_state:
- row = QHBoxLayout()
- lbl = QLabel(k.upper())
- btn = QPushButton("OFF")
- btn.setCheckable(True)
- btn.setChecked(bool(self._output_state[k]))
- btn.setText("ON" if self._output_state[k] else "OFF")
- channel_id = k
- def on_toggle(checked, cid=channel_id, b=btn):
- self.write_channel(cid, checked)
- b.setText("ON" if checked else "OFF")
- btn.toggled.connect(on_toggle)
- row.addWidget(lbl)
- row.addWidget(btn)
- grp_layout.addLayout(row)
- layout.addWidget(grp)
- layout.addStretch()
- return w
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
diff --git a/daq_system/devices/temperature.py b/daq_system/devices/temperature.py
deleted file mode 100644
index 5427cf6..0000000
--- a/daq_system/devices/temperature.py
+++ /dev/null
@@ -1,115 +0,0 @@
-"""
-devices/temperature.py
-
-Temperature sensor module — thermocouple / RTD / thermistor inputs.
-"""
-
-import math
-import random
-import time
-from typing import Any, Dict
-
-from PyQt6.QtWidgets import (
- QWidget, QVBoxLayout, QFormLayout, QGroupBox,
- QComboBox, QDoubleSpinBox, QLabel, QCheckBox
-)
-from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus
-
-TEMP_COLORS = ["#ff6b35", "#ffcc00", "#c77dff", "#ff4d6d"]
-
-
-class TemperatureDevice(BaseDevice):
- DEVICE_TYPE = "temperature"
- ICON = "🌡"
-
- def __init__(self, device_id: str = "temp_0", num_channels: int = 4,
- simulate: bool = True, sensor_type: str = "thermocouple"):
- channels = [
- ChannelConfig(
- channel_id=f"tc{i}", name=f"TC {i}",
- unit="°C", min_value=-200.0, max_value=1200.0,
- alarm_low=0.0, alarm_high=100.0,
- color=TEMP_COLORS[i % len(TEMP_COLORS)]
- )
- for i in range(num_channels)
- ]
- info = DeviceInfo(
- device_id=device_id, name="Temperature",
- device_type=self.DEVICE_TYPE,
- description=f"{sensor_type.title()} temperature input",
- icon=self.ICON, channels=channels
- )
- super().__init__(info)
- self.simulate = simulate
- self.sensor_type = sensor_type
- self._start = time.time()
- # Simulate slow thermal drift
- self._targets = [20.0 + i * 5 for i in range(num_channels)]
- self._currents = [20.0 + i * 5 for i in range(num_channels)]
-
- def connect(self) -> bool:
- self._start = time.time()
- self.status = DeviceStatus.SIMULATED if self.simulate else DeviceStatus.ERROR
- return self.simulate
-
- def disconnect(self) -> None:
- self.status = DeviceStatus.DISCONNECTED
-
- def read_channels(self) -> Dict[str, float]:
- result = {}
- for i, ch in enumerate(self.info.channels):
- if not ch.enabled:
- continue
- # Slow drift toward target with noise
- diff = self._targets[i] - self._currents[i]
- self._currents[i] += diff * 0.05 + random.gauss(0, 0.02)
- # Occasionally shift target
- if random.random() < 0.01:
- self._targets[i] += random.gauss(0, 2.0)
- self._targets[i] = max(10.0, min(200.0, self._targets[i]))
- result[ch.channel_id] = round(self._currents[i], 2)
- return result
-
- def write_channel(self, channel_id: str, value: Any) -> bool:
- return False # Read-only
-
- def get_config_widget(self) -> QWidget:
- w = QWidget()
- layout = QVBoxLayout(w)
- grp = QGroupBox("Sensor Configuration")
- form = QFormLayout(grp)
-
- sensor_cb = QComboBox()
- sensor_cb.addItems(["thermocouple", "rtd", "thermistor", "ic_sensor"])
- sensor_cb.setCurrentText(self.sensor_type)
- form.addRow("Sensor Type:", sensor_cb)
-
- tc_type = QComboBox()
- tc_type.addItems(["K", "J", "T", "E", "N", "R", "S", "B"])
- form.addRow("TC Type:", tc_type)
-
- unit_cb = QComboBox()
- unit_cb.addItems(["°C", "°F", "K"])
- form.addRow("Units:", unit_cb)
-
- layout.addWidget(grp)
-
- # Alarm config per channel
- alarm_grp = QGroupBox("Alarm Setpoints")
- alarm_layout = QVBoxLayout(alarm_grp)
- for ch in self.info.channels:
- row_layout = QFormLayout()
- lo = QDoubleSpinBox()
- lo.setRange(-200, 1200)
- lo.setValue(ch.alarm_low or 0.0)
- lo.setSuffix(" °C")
- hi = QDoubleSpinBox()
- hi.setRange(-200, 1200)
- hi.setValue(ch.alarm_high or 100.0)
- hi.setSuffix(" °C")
- row_layout.addRow(f"{ch.name} Low:", lo)
- row_layout.addRow(f"{ch.name} High:", hi)
- alarm_layout.addLayout(row_layout)
- layout.addWidget(alarm_grp)
- layout.addStretch()
- return w