NATIONAL INSTRUMENTS 美国 NI-USB6000 数据采集卡，Python编程

一、简单介绍

- 最近公司一个项目用到这款数据采集卡，这两天把我头都搞炸了，特做一个记录，给需要使用Python采集后面有需要的人少走点坑

• 我们项目要求以最高采集率运行，10k/s，找到的列子都不太好用。
• 直接上效果图：单通道下任意定量采集,测量一枚3V纽扣式锂电池电压
  

1.1 NI-USB6000

NI USB-6000提供USB接口、八个单端模拟（AI）输入通道、4个数字输入输出（DIO）通道和一个32位计数器。

• 安装：
  LabView:稍后贴出网盘链接

• 驱动 NI-DAQmx 博主自己网上找的18点几版本，如需最新版，可去官网（对着官网呵呵）
  网盘链接--提取码：l5dx
  下载解压后出来后，再解压使用第一个压缩包就好

二、如何通过Python与采集卡通信获取数据

- nidaqmx 模块，这是官方提供的一个模块，可与采集卡驱动通信，非常好用 - 安装：pip install nidaqmx 下载慢的可以 -i 使用国内镜像源，清华源、阿里源、豆瓣，都非常好用，速度杠杆的

NI DAQmx Python API参考，及使用教程

2.1 简单使用

- 按照好驱动后，重启，NI设备监视器会自动运行，如图点看查看设备 “Dev1”就是设备名称了，而 “ai0” "ai1" "ai2" 就是设备不同的通道，看自己需要而定了 

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)

官方提示

2.2 通过stream_readers来提高性能

- 简单理解，因为博主这个设备是个旧设备，说明、光盘啥的全丢了，然后NI官方只有会员才能享用他们的技术咨询服务，（坑爹的NI 买了硬件不行，还要买软件，会员） - 言归正传：据我这两天网络查找资料和查看源码，应当是我们创建虚拟通道对应物理通道与驱动通信，给与驱动Buffer，驱动会将数据读取回来写入Buffer，如果读回来的数据大于Buffer，会自动将旧数据覆盖

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:])

2.2.2 几个函数说明

- task.ai_channels.add_ai_voltage_chan("Dev1/ai0",name_to_assign_to_channel="AI0",max_val=10,min_val=-10) 选择ai_channels（输入）添加通道0 也就是"Dev1/ai0"，设备名称不一样的自己在上面所示的NI监听设备中去查看哦。 - task.timing.cfg_samp_clk_timing(RATE,sample_mode=constants.AcquisitionType.CONTINUOUS,samps_per_chan=100000) #一直采，直到停止任务 设定采集率，采集模式，每通道采集数量等...，（注意：采集率最大10k,计算方式应该为最大采集率除以你所需的通道数，10k/channels）比如博主使用2通道，最大就只能设置5000/s的采集率了。

• 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()关闭任务
  读取到所需数据后

一些事列的Git地址
stream_readers
关于NI设备的一些知识