- 我们项目要求以最高采集率运行，10k/s，找到的列子都不太好用。
- 直接上效果图：单通道下任意定量采集,测量一枚3V纽扣式锂电池电压
-
安装：
LabView:稍后贴出网盘链接 -
驱动 NI-DAQmx 博主自己网上找的18点几版本，如需最新版，可去官网（对着官网呵呵）
网盘链接--提取码：l5dx
下载解压后出来后，再解压使用第一个压缩包就好
import nidaqmx
import pprint
import numpy as np
from matplotlib import pyplot as plt
pp = pprint.PrettyPrinter(indent=4)
with nidaqmx.Task() as task:
task.ai_channels.add_ai_voltage_chan("Dev1/ai0")
print('1 Channel 1 Sample Read: ')
data = task.read()
pp.pprint(data)
data = task.read(number_of_samples_per_channel=1)
pp.pprint(data)
print('1 Channel N Samples Read: ')
data = task.read(number_of_samples_per_channel=10)
x=np.arange(0,len(data))
pp.pprint(data)
plt.plot(x,data)
task.ai_channels.add_ai_voltage_chan("Dev1/ai1")
print('N Channel 1 Sample Read: ')
data = task.read()
pp.pprint(data)
print('N Channel N Samples Read: ')
data = task.read(number_of_samples_per_channel=2)
pp.pprint(data)
nidaqmx 下 stream_readers.py文件下有多个读取类，
比如 AnalogSingleChannelReader 模拟单路通道读取
或者 AnalogMultiChannelReader 模拟多通道读取
这儿博主因为需要需要使用两个物理通道，所以就使用的第二个，当然你也可以使用第二个类去读取单通道，只需要在任务注册添加通道时只添加一个通道，并且缓冲区Buffer设置，给一维数组
我调试使用的一段代码：
import time
import nidaqmx
import numpy as np
from nidaqmx import constants
from nidaqmx import stream_readers
from nidaqmx import stream_writers
NUM_CHANNELS = 2
RATE = 10000/NUM_CHANNELS
with nidaqmx.Task() as task:
for i in range(NUM_CHANNELS):
task.ai_channels.add_ai_voltage_chan("Dev1/ai{}".format(i),name_to_assign_to_channel="AI{}".format(i),max_val=10,min_val=-10)
task.timing.cfg_samp_clk_timing(RATE,sample_mode=constants.AcquisitionType.CONTINUOUS,samps_per_chan=100000) #一直采，直到停止任务
#task.timing.cfg_samp_clk_timing(RATE,sample_mode=constants.AcquisitionType.FINITE,samps_per_chan=300000) #采集指定数量的样本
non_local_var = {'All samples': []}
channel1_data = []
channel2_data = []
#################使用 stream_readers 来提高性能 ##########################
# read_task = stream_readers.AnalogMultiChannelReader(task.in_stream)
# write_task = stream_writers.AnalogMultiChannelWriter(task.out_stream)
def read_callback(task_handle, every_n_samples_event_type,
number_of_samples, callback_data):
#################使用 stream_readers 来提高性能 ##########################
# buffer = np.empty((NUM_CHANNELS, number_of_samples), dtype=np.float64,order="C")
# read_task.read_many_sample(buffer, number_of_samples, timeout=constants.WAIT_INFINITELY)
# data = buffer.T.astype(np.float64)
####################直接使用 read 函数 #############
data = task.read(number_of_samples_per_channel=number_of_samples)
# non_local_var['All samples'].extend(data)
channel1_data.extend(data[0])
channel2_data.extend(data[1])
return 0
task.register_every_n_samples_acquired_into_buffer_event(
1000, read_callback)
task.start()
startTime = time.time()
while True:
if len(channel1_data)>=300000:
break
print("耗费时间:",time.time()-startTime)
task.stop()
print("数据总量:",len(channel1_data))
print(channel1_data[1:100])
print(channel1_data[-50:])
-
read_task = stream_readers.AnalogMultiChannelReader(task.in_stream)
-
write_task = stream_writers.AnalogMultiChannelWriter(task.out_stream)
将任务添加到流读取中，以提高性能 -
buffer = np.empty((NUM_CHANNELS, number_of_samples), dtype=np.float64,order="C")
-
read_task.read_many_sample(buffer, number_of_samples, timeout=constants.WAIT_INFINITELY)
设定缓冲区buffer，选用numpy.empty原因是这个初始化所需矩阵比较快,比np.zero快，因为他是随机值，比如博主需要的就是np.empty((2,300000),dtype=np.float64)
然后通过将buffer 传递给流读取对象,read_task去读取 -
data = task.read(number_of_samples_per_channel=number_of_samples)
不使用stream_readers正常读取 -
task.register_every_n_samples_acquired_into_buffer_event(1000, read_callback)
注册回调函数，就如函数名称，每采集n个样品到buffer触发事件，调用回到函数 -
task.start()
最后记得开启任务 -
task.stop()关闭任务
读取到所需数据后