国家仪器长监测Python

Question

我正在通过 NI 开发的 nidaqmx 模块构建一些脚本例程。 我使用 2 个 NI PXI 44-98（32 通道）采集卡。

我们想开发一个长时间（10 小时）的监测实验，采样率为 200k Hz。

目前，我正在努力解决 python 的内在限制和 Nidaqmx 模块的逻辑。

到目前为止，我已经用有限数量的传感器编写了一个连续采集例程。

import nidaqmx
from nidaqmx.constants import AcquisitionType
import matplotlib.pyplot as plt
import numpy as np

sample_time = 600  # units = seconds
s_freq = 200000
num_samples = sample_time*s_freq
dt = 1/s_freq
print('go acquisition !')
with nidaqmx.Task() as task:
    task.ai_channels.add_ai_accel_chan("PXI1Slot2_3/ai1:3",
                                       sensitivity=10000.0,
                                       max_val=1,
                                       min_val=-1)
    task.timing.cfg_samp_clk_timing(s_freq,
                                   sample_mode = AcquisitionType.CONTINUOUS)
    data = task.read(number_of_samples_per_channel=num_samples, timeout = nidaqmx.constants.WAIT_INFINITELY)
print('I do it right !')

但是使用这个非常简单的例程，我无法记录监控 > 10 分钟。 python的内存是不够的。这对我来说是完全合乎逻辑的。

我在 NI 网站上查看了缓冲区逻辑，但我并不清楚如何在这里实现它......

我无法理解如何在这个小例程中适应任务记录的每 X MB 数据写入磁盘，同时仍然监视和清空“数据”目录以避免溢出，我没有在stackoverflow上查看我的一些正确答案。

如果你已经遇到过这个问题并且有解决方案，我很感兴趣，

感谢阅读

Answer 1

我不知道你是否解决了你的问题，或者停止尝试，无论哪种方式，这是一个基本的解决方案。我正在使用以太网控制的 NI 9234 卡，并且我正在使用 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}')

所以... 5 个月后，但我希望它可能会有所帮助。

Answer 2

我终于解决了我的问题，这里有一个解决方案，可以连续监控大量高频率传感器阵列

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