1

我有一个 GUI 脚本，其中包含我所有的 wxPython 代码，以及一个单独的 testSequences 模块，它有一堆我根据来自 GUI 的输入运行的任务。这些任务需要很长时间才能完成（从 20 秒到 3 分钟），所以我想线程化它们，否则 GUI 在它们运行时会锁定。我还需要它们一个接一个地运行，因为它们都使用相同的硬件。（我在线程背后的基本原理只是为了防止 GUI 锁定。）我想要一个“运行”消息（在它之后有不同数量的句点，即“运行”、“运行”、“运行.. "等），因此用户知道正在发生进度，即使它不可见。我希望这个脚本在单独的线程中运行测试序列，但顺序是，这样第二个线程就不会了 t 被创建并运行，直到第一个完成。由于这与线程的目的相反，我真的找不到任何关于如何做到这一点的信息......任何帮助将不胜感激。

提前致谢！

gui.py

import testSequences
from threading import Thread

#wxPython code for setting everything up here...
for j in range(5):
    testThread = Thread(target=testSequences.test1)
    testThread.start()
    while testThread.isAlive():
        #wait until the previous thread is complete
        time.sleep(0.5)
        i = (i+1) % 4
        self.status.SetStatusText("Running"+'.'*i)

测试序列.py

import time
def test1():
    for i in range(10):
        print i
        time.sleep(1)

（显然这不是实际的测试代码，但想法是一样的。）

4

2 回答 2

2

想出了一个办法来做到这一点。我没有在我的 gui.py 中创建线程，而是创建了一个继承自 Thread 的类，并运行该类中的所有测试，然后在一个测试完成时发布 wxPython 事件（以便我可以更新状态栏）以及所有测试完成（所以我可以通知用户所有测试都已完成。

myEVT_TESTDONE = wx.NewEventType()
EVT_TESTDONE = wx.PyEventBinder(myEVT_TESTDONE , 1)
myEVT_ALLDONE = wx.NewEventType()
EVT_ALLDONE = wx.PyEventBinder(myEVT_ALLDONE, 1)

class TestDone(wx.PyCommandEvent):
    def __init__(self, etype, eid, val=None):
        wx.PyCommandEvent.__init__(self, etype, eid)
        self._val = val
    def GetValue(self):
        return self._val

class AllDone(wx.PyCommandEvent):
    def __init__(self, etype, eid):
        wx.PyCommandEvent.__init__(self, etype, eid)

class TestSequence(Thread):
    def __init__(self, parent, queue):
        Thread.__init__(self)
        self._queue = queue
        self._parent = parent
        self.start()
    def run(self):
        testCount = 0
        for test in self._queue:
            #Time-intensive task goes here
            for i in range(10):
                print i
                sleep(1)
            evt = TestDone(myEVT_TESTDONE, -1, i)
            wx.PostEvent(self._parent, evt)
        evt = AllDone(myEVT_ALLDONE, -1)
        wx.PostEvent(self._parent, evt)

class MainSequence(wx.Frame):
    def __init__(self, parent, id, title):
        self.Bind(EVT_TESTDONE, self.testDoneEvt)
        self.Bind(EVT_ALLDONE, self.allDoneEvt)
        #...the rest of the wxPython code
    def testDoneEvt(self, event):
        #Set what to be done after every test, e.g. update progress bar
        step = event.GetValue()
    def allDoneEvt(self, event):
        #Set what to be done after all tests, e.g. display "Tests complete"

program = wx.App()
window = MainSequence(None, -1, 'App title')
program.MainLoop()
于 2012-10-15T18:30:27.013 回答
2

您不能在带有 while 循环的 GUI 线程中等待，因为您阻塞了事件队列的处理。一种解决方案是使用计时器轮询线程的状态：

import wx 
import time
from threading import Thread

def test1():
    for i in range(10):
        print i
        time.sleep(1)

class TestFrame(wx.Frame): 
    def __init__(self): 
        wx.Frame.__init__(self, None, -1, "Test") 
        panel = wx.Panel(self, -1) 
        sizer = wx.BoxSizer(wx.VERTICAL) 
        panel.SetSizer(sizer) 

        self.button = wx.Button(panel, 0, "Start")
        sizer.Add(self.button, 0, wx.ALIGN_LEFT) 
        self.button.Bind(wx.EVT_BUTTON, self.OnButton)

        self.text = wx.StaticText(panel, 0, "No test is running")
        sizer.Add(self.text, 0, wx.ALIGN_LEFT) 

        self.timer = wx.Timer(self)

    def OnButton(self, event):
        self.testThread = Thread(target=test1)
        self.testThread.start()
        self.text.SetLabel("Running")
        self.button.Disable()
        self.Bind(wx.EVT_TIMER, self.PollThread)
        self.timer.Start(20, oneShot=True)
        event.Skip()

    def PollThread(self, event):
        if self.testThread.isAlive():
            self.Bind(wx.EVT_TIMER, self.PollThread)
            self.timer.Start(200, oneShot=True)
            self.text.SetLabel(self.text.GetLabel() + ".")
        else:
            self.button.Enable()
            self.text.SetLabel("Test completed")


app = wx.PySimpleApp() 
TestFrame().Show() 
app.MainLoop()
于 2012-10-13T18:38:50.710 回答