python - 无法从另一个线程停止计时器(带有nidaqmx-python和回调的简短示例)

Question

我在该论坛上看到了有关该主题的其他问题，但没有一个问题帮助我了解如何处理此问题。在我看来，它们中的大多数似乎也都是相当复杂且冗长的代码。我相信我正在做一些相当简单的事情/想要做一些相当简单的事情。我希望有人能帮帮忙!下面是广泛的解释，然后是我当前的代码。

注意:请不要删除此问题。我考虑了以下许多问题，并仔细地浏览了相关主题，但无济于事。我还认为发布此消息是有道理的，因为它部分与一个更通用的问题有关:即如何在后台运行回调的同时实时绘制(请参见最后的摘要)，可以总结为:我的总体目标。

设置和目标: National Instruments采集模块(这很重要)NI cDAQ9178，通过nidaqmx-python(NI维护的带有文档here的软件包)进行接口(interface)。一些模拟信号输入到那里，目标是在实时绘制信号的同时以一定的采样率(大约1000 Hz)连续采集(直到我决定停止采集)。绘图几乎不需要经常刷新(10Hz刷新率甚至可以)。我在conda虚拟环境中使用带有Python 3.7的Windows 10，并且在PyCharm中完成了编辑。理想情况下，事物应该在PyCharm和任何终端中都可以工作。

情况: nidaqmx-python提供了高级功能，允许人们注册回调(一个定义为一个愿望)，每当一定数量的样本(在我的情况下为100，但不严格)填充PC缓冲区时，就会调用该回调。这个想法是，下面定义的回调会在那时读取缓冲区，然后执行某些操作(在我的情况下，为简洁起见，我进行了一些低通滤波，其中一些存储在全局变量data中，并且可能绘制-见下文)。

问题:我一直在鬼混要在回调中包含实时绘制数据的任何内容，但是对于matplotlib来说是一场噩梦，因为回调使用的不是主线程，而matplotlib不喜欢被调用从主线程之外的任何地方。我已经搜索了其他针对实时绘图优化的库(并且，我在想​​，希望是线程安全的)，但这并不是那么容易:我无法获得vispy的工作，也无法获得pyqtgraph的安装，甚至只是为了给你一些例子。然后，我看到人们网上的一些帖子实际上是用matplotlib管理相当不错的实时动画，尽管它的开发是出于发布而不是这些应用程序的考虑。所以我想让我们尝试一下。

我的看法:由于无法让matplotlib从回调内部完成工作，因此我执行了以下操作(这是您在下面看到的代码):在回调之后以及使用task.start()启动任务之后(特定于nidaqmx-python) ，我只是创建了一个while循环，该循环绘制了全局变量buffer。我认为这是一个不错的技巧:瞧，buffer由回调每0.1秒左右更新一次(无关紧要)(无关紧要)，从侧面看，while循环一遍又一遍地绘制buffer变量，将其擦除每次在绘制之前，都可以有效地产生类似实时的图形。

注意:我完全知道绘图部分不尽如人意(我可能应该使用matplotlib的斧头API和subplots，更不用说动画了)，但是我暂时不在乎。我将在稍后处理，并对其进行改进以提高效率。

我想要的东西:这实际上可以实现我想要的……除了，为了阻止它，我在try:循环周围引入了except:和while语句，如下面的代码所示。自然地，按CTRL+C确实会破坏循环……但同时也会破坏整个正在运行的脚本，并给我留下以下错误:PyCharm中的forrtl: error (200): program aborting due to control-C event，以及从终端运行时的以下精度:

Image              PC                Routine            Line        Source
libifcoremd.dll    00007FFECF413B58  Unknown               Unknown  Unknown
KERNELBASE.dll     00007FFF219F60A3  Unknown               Unknown  Unknown
KERNEL32.DLL       00007FFF23847BD4  Unknown               Unknown  Unknown
ntdll.dll          00007FFF240CCED1  Unknown               Unknown  Unknown
QObject::~QObject: Timers cannot be stopped from another thread

不便之处在于，我别无选择，只能关闭python shell(再次考虑PyCharm)，而且我无法访问我的宝贵变量 data，该变量包含...好吧，我的数据。

猜猜:显然，回调不喜欢在此功能中停止。 nidaqmx_python任务应使用 task.stop()停止。我尝试将 task.stop()放在KeyboardInterrupt except:的紧后，但这并没有帮助，因为 CTRL+C将脚本停在最上层/而不是打破while循环。我认为需要一些更老练的方法来停止我的任务。我已经考虑了好几天，但是却找不到解决这两种问题的方法:可以停止的任务，以及实时绘图。请注意，如果不进行绘图，很容易在 ENTER按键时停止任务:一个简单地写在最后

input('Press ENTER to stop task')
task.stop()

但是，当然，简单地做上述事情并不能使我包括实时绘图部分。

摘要:我无法从连续读取数据的回调中调用matplotlib，因此我在一个单独的块中编写了 while循环以进行实时绘图，但是如果没有上述错误，我将无法停止该 while循环(我认为它提示回调是从另一个线程停止的)。

我希望我很清楚，如果没有，请一定要问!

代码:我已经对其进行了清理，以使其尽可能接近显示问题的MWE，尽管我当然知道你们中的大多数人都没有NI daq可以玩并连接以便能够运行它。无论如何...这里是:

import matplotlib.pyplot as plt
import numpy as np

import nidaqmx
from nidaqmx import stream_readers
from nidaqmx import constants

sfreq = 1000
bufsize = 100

with nidaqmx.Task() as task:

    # Here we set up the task ... nevermind
    task.ai_channels.add_ai_voltage_chan("cDAQ2Mod1/ai1")
    task.timing.cfg_samp_clk_timing(rate=sfreq, sample_mode=constants.AcquisitionType.CONTINUOUS,
                                    samps_per_chan=bufsize)
    # Here we define a stream to be read continuously
    stream = stream_readers.AnalogMultiChannelReader(task.in_stream)

    data = np.zeros((1, 0))  # initializing an empty numpy array for my total data
    buffer = np.zeros((1, bufsize))  # defined so that global buffer can be written to by the callback

    # This is my callback to read data continuously
    def reading_task_callback(task_idx, event_type, num_samples, callback_data):  # bufsize is passed to num_samples when this is called
        global data
        global buffer

        buffer = np.zeros((1, num_samples))

        # This is the reading part
        stream.read_many_sample(buffer, num_samples, timeout=constants.WAIT_INFINITELY)
        data = np.append(data, buffer, axis=1)  # appends buffered data to variable data

        return 0  # Absolutely needed for this callback to be well defined (see nidaqmx doc).

    # Here is the heavy lifting I believe: the above callback is registered
    task.register_every_n_samples_acquired_into_buffer_event(bufsize, reading_task_callback)
    task.start()  # The task is started (callback called periodically)

    print('Acquiring sensor data. Press CTRL+C to stop the run.\n')  # This should work ...

    fig = plt.figure()
    try:
        while True:
            # Poor's man plot updating
            plt.clf()
            plt.plot(buffer.T)
            plt.show()
            plt.pause(0.01)  # 100 Hz refresh rate
    except KeyboardInterrupt:  # stop loop with CTRL+C ... or so I thought :-(
        plt.close(fig)
        pass

    task.stop()  # I believe I never get to this part after pressing CTRL+C ...

    # Some prints at the end ... nevermind
    print('Total number of acquired samples: ', len(data.T),'\n')
    print('Sampling frequency: ', sfreq, 'Hz\n')
    print('Buffer size: ', bufsize, '\n')
    print('Acquisition duration: ', len(data.T)/sfreq, 's\n')

任何输入将不胜感激。预先感谢大家!

编辑:在下面的可接受的答案之后，我重写了上面的代码，并提出了以下内容，该代码现在可以按预期工作(对不起，这次我没有清理它，有些行与当前问题无关):

# Stream read from a task that is set up to read continuously
import matplotlib.pyplot as plt
import numpy as np

import nidaqmx
from nidaqmx import stream_readers
from nidaqmx import constants

from scipy import signal

import threading

running = True

sfreq = 1000
bufsize = 100
bufsizeb = 100

global task

def askUser():  # it might be better to put this outside of task
    global running
    input("Press return to stop.")
    running = False

def main():
    global running

    global data
    global buffer
    global data_filt
    global buffer_filt

    global b
    global z

    print('Acquiring sensor data...')

    with nidaqmx.Task() as task:  # maybe we can use target as above

        thread = threading.Thread(target=askUser)
        thread.start()

        task.ai_channels.add_ai_voltage_chan("cDAQ2Mod1/ai1")
        task.timing.cfg_samp_clk_timing(rate=sfreq, sample_mode=constants.AcquisitionType.CONTINUOUS,
                                        samps_per_chan=bufsize)
        # unclear samps_per_chan is needed here above or why it would be different than bufsize
        stream = stream_readers.AnalogMultiChannelReader(task.in_stream)

        data = np.zeros((1, 0))  # probably not the most elegant way of initializing an empty numpy array
        buffer = np.zeros((1, bufsizeb))  # defined so that global buffer can be written in the callback
        data_filt = np.zeros((1, 0))  # probably not the most elegant way of initializing an empty numpy array
        buffer_filt = np.zeros((1, bufsizeb))  # defined so that global buffer can be written in the callback

        b = signal.firwin(150, 0.004)
        z = signal.lfilter_zi(b, 1)

        def reading_task_callback(task_idx, event_type, num_samples, callback_data):  # bufsizeb is passed to num_samples
            global data
            global buffer
            global data_filt
            global buffer_filt
            global z
            global b

            if running:
                # It may be wiser to read slightly more than num_samples here, to make sure one does not miss any sample,
                # see: https://documentation.help/NI-DAQmx-Key-Concepts/contCAcqGen.html
                buffer = np.zeros((1, num_samples))
                stream.read_many_sample(buffer, num_samples, timeout=constants.WAIT_INFINITELY)
                data = np.append(data, buffer, axis=1)  # appends buffered data to variable data

                # IIR Filtering, low-pass
                buffer_filt = np.zeros((1, num_samples))
                for i, x in enumerate(np.squeeze(buffer)):  # squeeze required for x to be just a scalar (which lfilter likes)
                    buffer_filt[0,i], z = signal.lfilter(b, 1, [x], zi=z)

                data_filt = np.append(data_filt, buffer_filt, axis=1)  # appends buffered filtered data to variable data_filt

            return 0  # Absolutely needed for this callback to be well defined (see nidaqmx doc).

        task.register_every_n_samples_acquired_into_buffer_event(bufsizeb, reading_task_callback)  # bufsizeb instead

        task.start()
        while running:  # this is perfect: it "stops" the console just like sleep in a way that the task does not stop
            plt.clf()
            plt.plot(buffer.T)
            plt.draw()
            plt.pause(0.01)  # 100 Hz refresh rate
        # plt.close(fig)  # maybe no need to close it for now

        # task.join()  # this is for threads I guess ... (seems useless to my case?)

        # Some prints at the end ...
    print('Total number of acquired samples:', len(data.T))
    print('Sampling frequency:', sfreq, 'Hz')
    print('Buffer size:', bufsize)
    print('Acquisition duration:', len(data.T)/sfreq, 's')

if __name__ == '__main__':
    main()

请注意，我毕竟不需要 task.stop()，因为连续获取任务与此程序包一起工作的方式是，在 task.start()之后读取不是 sleep或类似内容的任何代码行都会使任务停止(至少这是我的理解) 。

Answer 1

我做的第一件事是摆脱键盘中断循环。我将其替换为全局变量running，另一个线程将其从返回时设置为False。

def askUser():
  global running
  input("Press return to stop.")
  running = False

然后，在 while loop之前，创建一个将执行此功能的新线程。

askUserThread = threading.Thread(target=askUser)
askUserThread.start()

对于while循环，摆脱 try catch语句:

while running:
  plt.clf()
  plt.plot(buffer.T)
  plt.draw()          # Note: this got changed because .show wasn't working.
  plt.pause(0.01)

这对我仍然不起作用，因为我必须关闭绘图窗 Eloquent 能显示一个新窗口。因此，从 this answer，我将其从 .show更改为 .draw。

我的最终代码有些不同(因为我对随机数据进行了采样)，但这确实如此。

# sampling.py
# by Preston Hager

import matplotlib.pyplot as plt
import numpy as np

import threading

sfreq = 1000
bufsize = 100

running = True

data = np.zeros((1, 0))  # initializing an empty numpy array for my total data
buffer = np.zeros((1, bufsize))  # defined so that global buffer can be written to by the callback

def askUser():
    global running

    input("Press return to stop.")
    running = False

def readingTask():
    global data
    global buffer

    while running:
        buffer = np.random.rand(1, bufsize)
        # This is the reading part
        data = np.append(data, buffer, axis=1)  # appends buffered data to variable data

def main():
    global running

    print('Acquiring sensor data.')

    thread = threading.Thread(target=askUser)
    thread.start()
    task = threading.Thread(target=readingTask)
    task.start()

    fig = plt.figure()
    while running:
        # Poor's man plot updating
        plt.clf()
        plt.plot(buffer.T)
        plt.draw()
        plt.pause(0.01)  # 100 Hz refresh rate
    plt.close(fig)

    task.join()

    # Some prints at the end ... nevermind
    print('Total number of acquired samples:', len(data.T))
    print('Sampling frequency:', sfreq, 'Hz')
    print('Buffer size:', bufsize)
    print('Acquisition duration:', len(data.T)/sfreq, 's')

if __name__ == '__main__':
    main()