1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
|
"""
api_layers/nidaqmx_layer.py
NI-DAQmx API abstraction layer.
• When nidaqmx package + NI runtime are present → uses real hardware
• Otherwise → falls back to simulation
Swap this layer by changing the `backend` parameter on AnalogInputDevice
or by subclassing NidaqmxLayer and overriding _hw_read().
Usage example:
from api_layers.nidaqmx_layer import NidaqmxLayer
layer = NidaqmxLayer(device_name="Dev1", channels=["ai0","ai1"], simulate=False)
layer.start()
values = layer.read() # {"ai0": 1.23, "ai1": -0.45}
layer.stop()
"""
import math
import random
import time
from typing import Dict, List, Optional
# ── Try to import real nidaqmx ──────────────────────────────────────────────
try:
import nidaqmx # type: ignore
from nidaqmx.constants import TerminalConfiguration # type: ignore
_NI_AVAILABLE = True
except ImportError:
_NI_AVAILABLE = False
class NidaqmxLayer:
"""
Thin wrapper around nidaqmx.Task for analog input.
Parameters
----------
device_name : str
NI device identifier, e.g. "Dev1"
channels : list of str
Physical channel names relative to device, e.g. ["ai0", "ai1", "ai2"]
sample_rate : float
Samples per second (hardware mode only; ignored in sim)
min_val / max_val : float
Expected voltage range for hardware task configuration
simulate : bool
Force simulation even if nidaqmx is available
"""
def __init__(
self,
device_name: str = "Dev1",
channels: List[str] = None,
sample_rate: float = 1000.0,
min_val: float = -10.0,
max_val: float = 10.0,
simulate: bool = True,
):
self.device_name = device_name
self.channels = channels or ["ai0", "ai1", "ai2", "ai3"]
self.sample_rate = sample_rate
self.min_val = min_val
self.max_val = max_val
self.simulate = simulate or not _NI_AVAILABLE
self._task = None
self._started = False
self._t0 = 0.0
# Sim waveform params per channel
self._sim_params = [
{
"freq": 0.3 + i * 0.17,
"amp": (max_val - min_val) * 0.4,
"offset": (max_val + min_val) / 2,
"noise": 0.02,
"phase": i * 0.8,
}
for i in range(len(self.channels))
]
# ── Lifecycle ───────────────────────────────────────────────────────
def start(self) -> bool:
"""Configure and start acquisition. Returns True on success."""
self._t0 = time.time()
if self.simulate:
self._started = True
return True
try:
self._task = nidaqmx.Task()
for ch in self.channels:
physical = f"{self.device_name}/{ch}"
self._task.ai_channels.add_ai_voltage_chan(
physical,
min_val=self.min_val,
max_val=self.max_val,
terminal_config=TerminalConfiguration.RSE,
)
self._task.timing.cfg_samp_clk_timing(
rate=self.sample_rate,
sample_mode=nidaqmx.constants.AcquisitionType.CONTINUOUS,
samps_per_chan=int(self.sample_rate),
)
self._task.start()
self._started = True
return True
except Exception as e:
print(f"[NidaqmxLayer] start() failed: {e}")
self._started = False
return False
def stop(self) -> None:
self._started = False
if self._task is not None:
try:
self._task.stop()
self._task.close()
except Exception:
pass
self._task = None
# ── Read ────────────────────────────────────────────────────────────
def read(self) -> Dict[str, float]:
"""Return latest sample per channel as {channel_name: voltage}."""
if not self._started:
return {}
if self.simulate:
return self._sim_read()
return self._hw_read()
def _hw_read(self) -> Dict[str, float]:
"""Read one sample per channel from hardware."""
try:
samples = self._task.read(number_of_samples_per_channel=1)
# nidaqmx returns list-of-lists when multiple channels
if len(self.channels) == 1:
samples = [samples]
return {ch: float(samples[i][0]) for i, ch in enumerate(self.channels)}
except Exception as e:
print(f"[NidaqmxLayer] read() failed: {e}")
return {}
def _sim_read(self) -> Dict[str, float]:
t = time.time() - self._t0
result = {}
for i, ch in enumerate(self.channels):
p = self._sim_params[i]
val = p["amp"] * math.sin(2 * math.pi * p["freq"] * t + p["phase"]) + p["offset"]
val += random.gauss(0, p["noise"] * p["amp"])
val = max(self.min_val, min(self.max_val, val))
result[ch] = round(val, 5)
return result
# ── Introspection ───────────────────────────────────────────────────
@staticmethod
def list_devices() -> List[str]:
"""Return list of detected NI device names, or [] if unavailable."""
if not _NI_AVAILABLE:
return []
try:
system = nidaqmx.system.System.local()
return [d.name for d in system.devices]
except Exception:
return []
@property
def is_simulated(self) -> bool:
return self.simulate
@property
def ni_available(self) -> bool:
return _NI_AVAILABLE
def __repr__(self):
mode = "SIM" if self.simulate else "HW"
return f"<NidaqmxLayer {self.device_name} ch={self.channels} [{mode}]>"
|
