diff options
Diffstat (limited to 'daq_system/devices/analog_input.py')
| -rw-r--r-- | daq_system/devices/analog_input.py | 184 |
1 files changed, 184 insertions, 0 deletions
diff --git a/daq_system/devices/analog_input.py b/daq_system/devices/analog_input.py new file mode 100644 index 0000000..8aaf9db --- /dev/null +++ b/daq_system/devices/analog_input.py @@ -0,0 +1,184 @@ +""" +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() |