National Instrument Long Monitoring Python
I don't know if you solved your problem, or stopped trying, either way, this is a basic solution. I am using an ethernet controlled NI 9234 card, and I'm storing my data with csv write:
# This works for a NI9234 card
import nidaqmx
from nidaqmx.constants import AcquisitionType
from nidaqmx import stream_readers
import numpy as np
import csv
import time
from datetime import datetime
sample_rate = 2048 # Usually sample rate goes 2048 Hz, 2560 Hz, 3200 Hz, 5120 Hz, 6400 Hz (at least for a NI9234 card)
samples_to_acq = 2048 # At least for vibration matters, Samples to acq go from 2048, 4096, 8192
wait_time = samples_to_acq/sample_rate # Data Acquisition Time in s, very important for the NI task function, and since we have 2048 on both sides, time is 1 s
cont_mode = AcquisitionType.CONTINUOUS # There is also FINITE for sporadic measurements
iterations = 10
with nidaqmx.Task() as task:
now = datetime.now()
military = now.strftime('%H:%M:%S') # Just recording the time like 20:32:54 instead of 8:32:54 pm
first_header = ['Some title in here']
second_header = [f'T. Captura: {military}']
# Create accelerometer channel and configure sample clock. current_excit_val=0.002 enables IEPE, velocity measured in ips and accel in m/s2,
# both prox and tachometer can be used with a simple voltage channel.
# For more info about channels: https://nidaqmx-python.readthedocs.io/en/latest/ai_channel_collection.html
# Two accelerometer channels
task.ai_channels.add_ai_accel_chan(physical_channel = "cDAQ9181-1E3866DMod1/ai0", sensitivity = 100, current_excit_val = 0.002)
task.ai_channels.add_ai_accel_chan(physical_channel = "cDAQ9181-1E3866DMod1/ai1", sensitivity = 100, current_excit_val = 0.002)
# Two voltage channels
task.ai_channels.add_ai_voltage_chan(physical_channel = "cDAQ9181-1E3866DMod1/ai2")
task.ai_channels.add_ai_voltage_chan(physical_channel = "cDAQ9181-1E3866DMod1/ai3")
total_wait_time = wait_time * iterations # We will only take 10 measurements, 10 s for this example
samples_to_acq_new = samples_to_acq * iterations # Also multiply by 10 to keep the same ratio, it should be
# Sets source of sample clock, its rate, and number of samples to aquire = buffer size
task.timing.cfg_samp_clk_timing(sample_rate, sample_mode = cont_mode, samps_per_chan = samples_to_acq_new)
start = time.time()
print ('Starting task...') # Just to keep a control in your console
# Saving the 4 channels to a csv file. This file is overwritten everytime the program is executed.
# It should appear in the same folder that your program is located.
with open('data_10times.csv', 'w', newline = '') as f:
writer = csv.writer(f)
writer.writerow(first_header)
writer.writerow(second_header)
# adding some blank spaces in btw
writer.writerow('')
writer.writerow('')
x = np.linspace(0, total_wait_time, samples_to_acq_new) # Your x axis (ms), starts from 0, final time is total_wait_time, equally divided by the number of samples you'll capture
data = np.ndarray((4, samples_to_acq_new), dtype = np.float64) #Creates an array, 4 columns each one of 20480 rows
nidaqmx.stream_readers.AnalogMultiChannelReader(task.in_stream).read_many_sample(data, samples_to_acq_new, timeout = 14) # it should't take that long for this example, check out time for other exercises
for value in range(len(x)):
writer.writerow([x[value], data[0][value], data[1][value], data[2][value], data[3][value]])
elapsed_time = (time.time() - start)
print (f'done in {elapsed_time}')
so... 5 months later, but I hope it might help.
I finnaly outcome my problems, here a solution for large array of sensor with hight frequency rate to be continuously monitored
import matplotlib.pyplot as plt
import numpy as np
import nidaqmx
from nidaqmx.stream_readers import AnalogMultiChannelReader
from nidaqmx import constants
import threading
from datetime import datetime
import os
# Parameters
sampling_freq_in = 200000 # in Hz
buffer_in_size = 800000
bufsize_callback = 200000
buffer_in_size_cfg = round(buffer_in_size) * 10 # clock configuration * 10 ?
chans_in = 32 # number of chan
refresh_rate_plot = 100000 # in Hz
crop = 0 # number of seconds to drop at acquisition start before saving
# Initialize data placeholders
buffer_in = np.zeros((chans_in, buffer_in_size))
data = np.zeros(
(chans_in, 1)) # will contain a first column with zeros but that's fine
print(data.size)
# Definitions of basic functions
def ask_user():
global running
input("Press ENTER/RETURN to stop acquisition and coil drivers.")
running = False
def cfg_read_task(acquisition):
acquisition.ai_channels.ai_gain = 100
acquisition.ai_channels.ai_max = 100
acquisition.ai_channels.add_ai_accel_chan("Dev3/ai0:15",
sensitivity=1000.0,
max_val=1,
min_val=-1) # Cards1
acquisition.ai_channels.add_ai_accel_chan("Dev4/ai0:15",
sensitivity=1000.0,
max_val=1,
min_val=-1) # Cards2
acquisition.timing.cfg_samp_clk_timing(rate=sampling_freq_in,
sample_mode=constants.AcquisitionType.CONTINUOUS,
samps_per_chan=buffer_in_size_cfg)
def reading_task_callback(task_idx, event_type, num_samples, callback_data):
global data
global buffer_in
if running:
path = r'D:\Experience\caca\Acc/'
isExist = os.path.exists(path)
if not isExist:
os.makedirs(path)
buffer_in = np.zeros((chans_in, num_samples)) # double definition ???
stream_in.read_many_sample(buffer_in, num_samples,
timeout=constants.WAIT_INFINITELY)
data = np.append(data, buffer_in,
axis=1) # appends buffered data to total variable data
filename = path + 'Acc_' + str(
datetime.now().strftime("%m%d%h%H%M%S%f"))
extension = '.npy'
np.save(filename + extension, data)
# f=np.fft.rfftfreq(data[4][1::].size,d=1/100000)
# P=abs(np.fft.rfft(data[4][1::]))
# plt.plot(f[f>200] ,P[f>200],'k')
# plt.xlim(100,10000)
# plt.plot(data[0][1::],'k--')
data = np.zeros((chans_in, 1))
return 0
# Configure and setup the tasks
task_in = nidaqmx.Task()
cfg_read_task(task_in)
stream_in = AnalogMultiChannelReader(task_in.in_stream)
task_in.register_every_n_samples_acquired_into_buffer_event(bufsize_callback,
reading_task_callback)
# Start threading to prompt user to stop
thread_user = threading.Thread(target=ask_user)
thread_user.start()
# Main loop
running = True
time_start = datetime.now()
task_in.start()
# Plot a visual feedback for the user's mental health
# f, (ax1, ax2, ax3) = plt.subplots(3, 1, sharex='all', sharey='none')
while running: # make this adapt to number of channels automatically
a = 0
# ax1.clear()
# ax2.clear()
# ax3.clear()
# ax1.plot(data[0, -sampling_freq_in * 5:].T,'r') # 5 seconds rolling window
# ax2.plot(data[1, -sampling_freq_in * 5:].T,'k')
# ax3.plot(data[2, -sampling_freq_in * 5:].T,'b')
# Label and axis formatting
# ax3.set_xlabel('time [s]')
# ax1.set_ylabel('m/s**2')
# ax2.set_ylabel('m/s**2')
# ax3.set_ylabel('m/s**2')
# xticks = np.arange(0, data[0, -sampling_freq_in * 5:].size, sampling_freq_in)
# xticklabels = np.arange(0, xticks.size, 1)
# ax3.set_xticks(xticks)
# ax3.set_xticklabels(xticklabels)
# plt.pause(1/refresh_rate_plot) # required for dynamic plot to work (if too low, nulling performance bad)
# Close task to clear connection once done
task_in.close()
duration = datetime.now() - time_start
print(duration, 'Pan t es mort')
experience_file = [duration, sampling_freq_in]
np.save('info.npy', experience_file)
# savefile:
# Some messages at the end
# print("\n")
# print("OPM acquisition ended.\n")
# print("Acquisition duration: {}.".format(duration))