diff options
Diffstat (limited to 'devices/nidaqmx_device.py')
| -rw-r--r-- | devices/nidaqmx_device.py | 44 |
1 files changed, 22 insertions, 22 deletions
diff --git a/devices/nidaqmx_device.py b/devices/nidaqmx_device.py index 9ebec0f..652b7e2 100644 --- a/devices/nidaqmx_device.py +++ b/devices/nidaqmx_device.py @@ -13,7 +13,7 @@ from PyQt6.QtWidgets import ( ) from PyQt6.QtCore import Qt, QThread, pyqtSignal -from devices.base_device import BaseDevice, ChannelConfig, DeviceInfo, DeviceStatus +from devices.base_device import BaseDevice, SignalConfig, DeviceInfo, DeviceStatus from api_layers.nidaqmx_layer import NidaqmxLayer _ANALOG_COLORS = [ @@ -53,20 +53,20 @@ class NidaqmxDevice(BaseDevice): channels = [] for i in range(num_analog): - channels.append(ChannelConfig( - channel_id=f"ai{i}", name=f"AI{i}", unit="V", + channels.append(SignalConfig( + signal_id=f"ai{i}", name=f"AI{i}", unit="V", min_value=min_v, max_value=max_v, color=_ANALOG_COLORS[i % len(_ANALOG_COLORS)], )) for i in range(num_di): - channels.append(ChannelConfig( - channel_id=f"di{i}", name=f"DI {i}", unit="", + channels.append(SignalConfig( + signal_id=f"di{i}", name=f"DI {i}", unit="", min_value=0.0, max_value=1.0, color=_DI_COLORS[i % len(_DI_COLORS)], )) for i in range(num_do): - channels.append(ChannelConfig( - channel_id=f"do{i}", name=f"DO {i}", unit="", + channels.append(SignalConfig( + signal_id=f"do{i}", name=f"DO {i}", unit="", min_value=0.0, max_value=1.0, color=_DO_COLORS[i % len(_DO_COLORS)], writable=True, @@ -79,7 +79,7 @@ class NidaqmxDevice(BaseDevice): description=f"NI-DAQmx — {num_analog} AI, {num_di} DI, {num_do} DO", manufacturer="National Instruments", icon=self.ICON, - channels=channels, + signals=channels, ) super().__init__(info) @@ -99,8 +99,8 @@ class NidaqmxDevice(BaseDevice): ok = True try: if self._num_analog > 0: - ai_ids = [ch.channel_id for ch in self.info.channels - if ch.channel_id.startswith("ai")] + ai_ids = [ch.signal_id for ch in self.info.signals + if ch.signal_id.startswith("ai")] self._ai_layer = NidaqmxLayer( device_name=self._ni_device, channels=ai_ids, @@ -179,14 +179,14 @@ class NidaqmxDevice(BaseDevice): try: if self._ni_in_task: vals = self._ni_in_task.read() - di_chs = [c for c in self.info.channels if c.channel_id.startswith("di")] + di_chs = [c for c in self.info.signals if c.signal_id.startswith("di")] for i, ch in enumerate(di_chs): - result[ch.channel_id] = float(vals[i] if isinstance(vals, list) else vals) + result[ch.signal_id] = float(vals[i] if isinstance(vals, list) else vals) except Exception: pass - for ch in (c for c in self.info.channels if c.channel_id.startswith("do")): - result[ch.channel_id] = float(self._output_state.get(ch.channel_id, 0)) + for ch in (c for c in self.info.signals if c.signal_id.startswith("do")): + result[ch.signal_id] = float(self._output_state.get(ch.signal_id, 0)) return result @@ -194,12 +194,12 @@ class NidaqmxDevice(BaseDevice): import math, time t = time.time() result = {} - for i, ch in enumerate([c for c in self.info.channels if c.channel_id.startswith("ai")]): - result[ch.channel_id] = math.sin(t + i) * (self._max_v * 0.5) - for ch in (c for c in self.info.channels if c.channel_id.startswith("di")): - result[ch.channel_id] = 0.0 - for ch in (c for c in self.info.channels if c.channel_id.startswith("do")): - result[ch.channel_id] = float(self._output_state.get(ch.channel_id, 0)) + for i, ch in enumerate([c for c in self.info.signals if c.signal_id.startswith("ai")]): + result[ch.signal_id] = math.sin(t + i) * (self._max_v * 0.5) + for ch in (c for c in self.info.signals if c.signal_id.startswith("di")): + result[ch.signal_id] = 0.0 + for ch in (c for c in self.info.signals if c.signal_id.startswith("do")): + result[ch.signal_id] = float(self._output_state.get(ch.signal_id, 0)) return result def write_channel(self, channel_id: str, value: Any) -> bool: @@ -208,8 +208,8 @@ class NidaqmxDevice(BaseDevice): self._output_state[cid] = int(bool(value)) if not self.simulate and self._ni_out_task: try: - do_chs = [c for c in self.info.channels if c.channel_id.startswith("do")] - states = [self._output_state.get(c.channel_id, 0) for c in do_chs] + do_chs = [c for c in self.info.signals if c.signal_id.startswith("do")] + states = [self._output_state.get(c.signal_id, 0) for c in do_chs] self._ni_out_task.write(states) except Exception as e: print(f"[NidaqmxDevice] write failed: {e}") |