Python Mayaviを他のスレッドを実行中にタイマーを使用して更新する

Question

Traitイベントではなくタイムベースで更新されるMayaviビジュアライゼーションを作成できますか？

私は継続的に更新する必要がある視覚化がありますが、私が更新しているデータは外部ソースから来ています（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

次のリンクで見つかったソリューション：呼び出すために、@のmlab.animatorを使用することによって解決 Animating a mayavi points3d plot

ユーザが操作できるようにアニメーションを解放するために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()