From 2ed9d37da7b27d25173535550fb92702225ac14e Mon Sep 17 00:00:00 2001 From: Christian Kolset Date: Mon, 13 Apr 2026 15:22:56 -0600 Subject: Removed QtPy5 snippet and fixed directories --- daq_system/devices/__init__.py | 1 - .../devices/__pycache__/__init__.cpython-312.pyc | Bin 164 -> 0 bytes .../__pycache__/analog_input.cpython-312.pyc | Bin 9034 -> 0 bytes .../__pycache__/base_device.cpython-312.pyc | Bin 6014 -> 0 bytes .../__pycache__/device_registry.cpython-312.pyc | Bin 5219 -> 0 bytes .../devices/__pycache__/digital_io.cpython-312.pyc | Bin 6027 -> 0 bytes .../__pycache__/serial_device.cpython-312.pyc | Bin 7405 -> 0 bytes .../__pycache__/temperature.cpython-312.pyc | Bin 6707 -> 0 bytes daq_system/devices/analog_input.py | 184 --------------------- daq_system/devices/base_device.py | 127 -------------- daq_system/devices/device_registry.py | 87 ---------- daq_system/devices/digital_io.py | 95 ----------- daq_system/devices/serial_device.py | 136 --------------- daq_system/devices/temperature.py | 115 ------------- 14 files changed, 745 deletions(-) delete mode 100644 daq_system/devices/__init__.py delete mode 100644 daq_system/devices/__pycache__/__init__.cpython-312.pyc delete mode 100644 daq_system/devices/__pycache__/analog_input.cpython-312.pyc delete mode 100644 daq_system/devices/__pycache__/base_device.cpython-312.pyc delete mode 100644 daq_system/devices/__pycache__/device_registry.cpython-312.pyc delete mode 100644 daq_system/devices/__pycache__/digital_io.cpython-312.pyc delete mode 100644 daq_system/devices/__pycache__/serial_device.cpython-312.pyc delete mode 100644 daq_system/devices/__pycache__/temperature.cpython-312.pyc delete mode 100644 daq_system/devices/analog_input.py delete mode 100644 daq_system/devices/base_device.py delete mode 100644 daq_system/devices/device_registry.py delete mode 100644 daq_system/devices/digital_io.py delete mode 100644 daq_system/devices/serial_device.py delete mode 100644 daq_system/devices/temperature.py (limited to 'daq_system/devices') 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 Binary files a/daq_system/devices/__pycache__/__init__.cpython-312.pyc and /dev/null 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 Binary files a/daq_system/devices/__pycache__/analog_input.cpython-312.pyc and /dev/null 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 Binary files a/daq_system/devices/__pycache__/base_device.cpython-312.pyc and /dev/null 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 Binary files a/daq_system/devices/__pycache__/device_registry.cpython-312.pyc and /dev/null 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 Binary files a/daq_system/devices/__pycache__/digital_io.cpython-312.pyc and /dev/null 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 Binary files a/daq_system/devices/__pycache__/serial_device.cpython-312.pyc and /dev/null 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 Binary files a/daq_system/devices/__pycache__/temperature.cpython-312.pyc and /dev/null 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 -- cgit v1.2.3