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



是否可以创建按时间更新的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()

在以下链接中找到解决方案: 对马亚维点3d 绘图进行动画处理

通过使用 @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()

相关内容

  • 没有找到相关文章

最新更新