Python正在更新Mayavi在運行其他線程時使用定時器

Question

是否可以創建一個Mayavi可視化，該應用程序是在定時基礎上而不是通過Trait事件更新的？

我有一個可視化，我需要不斷更新，但我更新的數據來自外部來源（I.E.不是來自圖形界面的用戶輸入的事件）。與此同時，我需要運行一個單獨的線程 - 所以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()