Python 在运行其他线程时使用计时器更新 Mayavi

Question

是否可以创建基于定时而不是通过 Trait 事件更新的 Mayavi 可视化？

我有一个需要不断更新的可视化，但我正在更新的数据来自外部源（即不是来自图形界面的用户输入的事件）。

与此同时，我需要运行一个单独的线程 - 所以 Mayavi 可视化无法控制主循环。

这可以吗？如果是的话怎么办？？

任何帮助将不胜感激。

下面是我尝试解决此问题的一些虚拟代码：

import numpy

from mayavi.sources.array_source import ArraySource

from pyface.api import GUI
from mayavi.modules.api import Surface
from mayavi.api import Engine

import threading
import time


# Class runs a given function on a given thread at a given scan time
class TimedThread(threading.Thread):
    def __init__(self, thread, scan_time, funct, *funct_args):
        threading.Thread.__init__(self)

        # Thread for the function to operate in
        self.thread = thread

        # Defines the scan time the function is to be run at
        self.scan_time = scan_time

        # Function to be run
        self.run_function = funct

        # Function arguments
        self.funct_args = funct_args

    def run(self):
        while True:
            # Locks the relevant thread
            self.thread.acquire()

            # Begins timer for elapsed time calculation
            start_time = time.time()

            # Runs the function that was passed to the thread
            self.run_function(*self.funct_args)

            # Wakes up relevant threads to listen for the thread release
            self.thread.notify_all()

            # Releases thread
            self.thread.release()

            # Calculates the elapsed process time & sleeps for the remainder of the scan time
            end_time = time.time()
            elapsed_time = end_time - start_time
            sleep_time = self.scan_time - elapsed_time

            if sleep_time > 0:
                time.sleep(sleep_time)
            else:
                print 'Process time exceeds scan time'


# Function to update the visualisation
def update_visualisation(source):
    print("Updating Visualization...")

    # Pretend the data is being updated externally
    x = numpy.array([0, numpy.random.rand()])
    y = z = x
    data = [x, y, z]
    source.scalar_data = data
    GUI.invoke_later(source.update)


# Function to run the visualisation
def run_main():
    print 'Running Main Controller'


if __name__ == '__main__':
    c = threading.Condition()

    # Create a new Engine for Mayavi and start it
    engine = Engine()
    engine.start()

    # Create a new Scene
    engine.new_scene()

    # Create the data
    x = numpy.linspace(0, 10, 2)
    y = z = x
    data = [x, y, z]

    # Create a new Source, map the data to the source and add it to the Engine
    src = ArraySource()
    src.scalar_data = data
    engine.add_source(src)

    # Create a Module
    surf = Surface()

    # Add the Module to the Engine
    engine.add_module(surf)

    # Create timed thread classes
    visualisation_thread = TimedThread(c, 2.0, update_visualisation, src)
    main_thread = TimedThread(c, 1.0, run_main)

    # Start & join the threads
    main_thread.start()
    visualisation_thread.start()
    main_thread.join()
    visualisation_thread.join()

Answer 1

在以下链接中找到解决方案： Animating a mayavi points3d plot

通过使用@mlab.animator 调用更新函数并使用yield 命令释放动画以允许用户操作来解决。

解决方法如下：

import numpy as np
import threading
import time

from mayavi import mlab
from mayavi.api import Engine

# Class runs a given function on a given thread at a given scan time
class SafeTimedThread(threading.Thread):
    def __init__(self, thread_condition, scan_time, funct, *funct_args):
        threading.Thread.__init__(self)

        # Thread condition for the function to operate with
        self.tc = thread_condition

        # Defines the scan time the function is to be run at
        self.scan_time = scan_time

        # Function to be run
        self.run_function = funct

        # Function arguments
        self.funct_args = funct_args

    def run(self):
        while True:
            # Locks the relevant thread
            self.tc.acquire()

            # Begins timer for elapsed time calculation
            start_time = time.time()

            # Runs the function that was passed to the thread
            self.run_function(*self.funct_args)

            # Wakes up relevant threads to listen for the thread release
            self.tc.notify_all()

            # Releases thread
            self.tc.release()

            # Calculates the elapsed process time & sleep for the remainder of the scan time
            end_time = time.time()
            elapsed_time = end_time - start_time
            sleep_time = self.scan_time - elapsed_time

            if sleep_time > 0:
                time.sleep(sleep_time)
            else:
                print 'Process time exceeds scan time'


# Function to run the main controller
def run_main():
    print 'Running Main Controller'


def init_vis():
    # Creates a new Engine, starts it and creates a new scene
    engine = Engine()
    engine.start()
    engine.new_scene()

    # Initialise Plot
    data = np.random.random((3, 2))
    x = data[0]
    y = data[1]
    z = data[2]

    drawing = mlab.plot3d(x, y, z, np.ones_like(x))

    return drawing


@mlab.animate(delay=500, ui=False)
def update_visualisation(drawing):
    while True:
        print ('Updating Visualisation')
        # Pretend to receive data from external source
        data = np.random.random((3, 2))
        x = data[0]
        y = data[1]
        z = data[2]

        drawing.mlab_source.set(x=x, y=y, z=z)
        yield


if __name__ == '__main__':
    # Create Condition for Safe Threading
    c = threading.Condition()

    # Create display window
    dwg = init_vis()

    # Create safe timed thread for main thread and start
    main_thread = SafeTimedThread(c, 1.0, run_main).start()

    # Update using mlab animator
    vis_thread = update_visualisation(dwg)

    mlab.show()