examples/NI DAQmx/README.md
A Python bridge that streams real-time data from NI DAQ devices to Serial Studio (or any UDP listener) in CSV format. Works with any NI DAQ hardware that supports analog input.
pip install numpy nidaqmx
Configure the connection:
UDP.127.0.0.1.9000.Connect and run DAQBridge:
python daqbridge.py
Edit the top of daqbridge.py.
DEVICE_NAME = "Dev1" # Check NI MAX for your device name
SAMPLE_RATE = 3000.0 # Check your device specs for max rate
SAMPLES_PER_READ = 1 # 1 = lowest latency
Use NI MAX (Measurement & Automation Explorer) to find your device name and confirm supported sample rates.
The default configuration declares six channels (four voltage, two current). One of each:
CHANNEL_CONFIG = [
# Voltage measurement
{"channel": "ai0", "type": "voltage", "voltage_range": (-10.0, 10.0), "terminal": "RSE", "name": "Voltage_0"},
# Current measurement (I = V / shunt_resistor)
{"channel": "ai2", "type": "current", "voltage_range": (-10.0, 10.0), "terminal": "RSE", "name": "Current_0", "shunt_resistor": 500},
]
| Parameter | Values | Description |
|---|---|---|
channel | ai0, ai1, and so on | Physical channel (device-dependent) |
type | voltage, current | Measurement type |
voltage_range | (-10,10), (-5,5), and so on | Input range (device-dependent) |
terminal | RSE, DIFF | RSE = single-ended, DIFF = differential |
shunt_resistor | ohms | For current type only |
UDP_HOST = "127.0.0.1"
UDP_PORT = 9000
Sample rate limits vary by device. Check your device specs in NI MAX.
Example: USB-6001 (20 kS/s aggregate):
| Channels | Max rate/channel |
|---|---|
| 1 | 20,000 Hz |
| 4 | 5,000 Hz |
| 8 | 2,500 Hz |
Example: USB-6210 (250 kS/s aggregate):
| Channels | Max rate/channel |
|---|---|
| 1 | 250,000 Hz |
| 8 | 31,250 Hz |
| 16 | 15,625 Hz |
Edit process_readings() in daqbridge.py to add your own calculations:
def process_readings(raw_data: np.ndarray, channel_config: list) -> np.ndarray:
output = raw_data.copy()
# Default: convert current channels (I = V / R)
for ch_idx, ch_cfg in enumerate(channel_config):
if ch_cfg.get("type") == "current":
shunt = ch_cfg.get("shunt_resistor", 1.0)
output[ch_idx, :] = raw_data[ch_idx, :] / shunt
# ADD YOUR CUSTOM PROCESSING HERE
# Example:
# output[0, :] = raw_data[0, :] * 2.0 # Scale channel 0
# output[1, :] = np.clip(raw_data[1, :], -5, 5) # Clamp channel 1
return output
Load
│
├───[Rshunt]───┬─── AI channel
│ │
GND GND
Current = V_measured / Rshunt
| Error | Fix |
|---|---|
| Device identifier is invalid | Check the device name in NI MAX |
| Sample rate exceeds maximum | Lower SAMPLE_RATE or the number of channels |
| No data in Serial Studio | Check UDP port 9000, and check firewall rules |
| Resource is reserved | Close NI MAX or other DAQ applications |
| Specified route cannot be satisfied | Check terminal config (RSE/DIFF) supported by your device |
Any NI DAQ device compatible with NI-DAQmx, including: