Блог

Я пишу приложение с графическим интерфейсом Python 3.9, используя wxPython. Приложение позволяет пользователю ввести информацию и нажать кнопку «рассчитать». Это приведет к запуску продолжительного потока, который обновляет главное окно приложения обновлениями статуса. Приложение успешно запускает поток вычислений и остается отзывчивым.

Теперь я хотел бы добавить кнопку «прервать», которая остановит/отменит/прервет ранее запущенный поток, когда пользователи решат нажать кнопку «прервать».

Поискав некоторое время, я нашел приведенный ниже код на веб-сайте wxPython, на котором есть общая структура, которую я ищу. Это следует основной логике моего приложения в более простой форме. В нем отсутствует текущее обновление статуса графического интерфейса, поэтому я сделал одно дополнение для имитации обновлений статуса графического интерфейса — см. «lt;lt;lt;lt;lt; wx.PostEvent(self._notify_window, ResultEvent(i))

Однако после добавления этой строки кажется, что обработка событий прерывается, когда пользователь нажимает кнопку «стоп» во время запуска потока обработки. При нажатии кнопки «стоп» через 3-4 секунды поток подсчета в приведенном ниже примере кода продолжает выполняться и игнорирует событие кнопки «стоп».

Когда эта строка удаляется, функция «стоп» в приведенном ниже примере кода работает, и цикл прерывается.

Как я могу сохранить функцию обновления графического интерфейса с обновлениями статуса во время выполнения потока подсчета, а также использовать кнопку «остановить» для завершения активно работающего потока?

Я уверен, что есть способы оптимизировать приведенный ниже код. Пожалуйста, не стесняйтесь делиться своими мыслями/знаниями. Спасибо.

 import time from threading import * import wx  # Button definitions ID_START = wx.NewId() ID_STOP = wx.NewId()  # Define notification event for thread completion EVT_RESULT_ID = wx.NewId()  def EVT_RESULT(win, func):  """Define Result Event."""  win.Connect(-1, -1, EVT_RESULT_ID, func)  class ResultEvent(wx.PyEvent):  """Simple event to carry arbitrary result data."""  def __init__(self, data):  """Init Result Event."""  wx.PyEvent.__init__(self)  self.SetEventType(EVT_RESULT_ID)  self.data = data  # Thread class that executes processing class WorkerThread(Thread):  """Worker Thread Class."""  def __init__(self, notify_window):  """Init Worker Thread Class."""  Thread.__init__(self)  self._notify_window = notify_window  self._want_abort = 0  # This starts the thread running on creation, but you could  # also make the GUI thread responsible for calling this  #self.start()   def run(self):  """Run Worker Thread."""  # This is the code executing in the new thread. Simulation of  # a long process (well, 10s here) as a simple loop - you will  # need to structure your processing so that you periodically  # peek at the abort variable  for i in range(10):  if self._want_abort:  # Use a result of None to acknowledge the abort (of  # course you can use whatever you'd like or even  # a separate event type)  wx.PostEvent(self._notify_window, ResultEvent(None))  return  time.sleep(2)  wx.PostEvent(self._notify_window, ResultEvent(i)) lt;lt;lt;lt;lt;lt;lt; Added to simulate "status update" feedback of my application.    # Here's where the result would be returned (this is an  # example fixed result of the number 10, but it could be  # any Python object)  wx.PostEvent(self._notify_window, ResultEvent(10))   def abort(self):  """abort worker thread."""  # Method for use by main thread to signal an abort  self._want_abort = 1  # GUI Frame class that spins off the worker thread class MainFrame(wx.Frame):  """Class MainFrame."""  def __init__(self, parent, id):  """Create the MainFrame."""  wx.Frame.__init__(self, parent, id, 'Thread Test')   # Dumb sample frame with two buttons  wx.Button(self, ID_START, 'Start', pos=(0,0))  wx.Button(self, ID_STOP, 'Stop', pos=(0,50))  self.status = wx.StaticText(self, -1, '', pos=(0,100))   self.Bind(wx.EVT_BUTTON, self.OnStart, id=ID_START)  self.Bind(wx.EVT_BUTTON, self.OnStop, id=ID_STOP)   # Set up event handler for any worker thread results  EVT_RESULT(self,self.OnResult)   # And indicate we don't have a worker thread yet  self.worker = None   def OnStart(self, event):  """Start Computation."""  # Trigger the worker thread unless it's already busy  if not self.worker:  self.status.SetLabel('Starting computation')  self.worker = WorkerThread(self)  self.worker.start()   def OnStop(self, event):  """Stop Computation."""  # Flag the worker thread to stop if running  if self.worker:  self.status.SetLabel('Trying to abort computation')  self.worker.abort()   def OnResult(self, event):  """Show Result status."""  if event.data is None:  # Thread aborted (using our convention of None return)  self.status.SetLabel('Computation aborted')  else:  # Process results here  self.status.SetLabel('Computation Result: %s' % event.data)  # In either event, the worker is done  self.worker = None  class MainApp(wx.App):  """Class Main App."""  def OnInit(self):  """Init Main App."""  self.frame = MainFrame(None, -1)  self.frame.Show(True)  self.SetTopWindow(self.frame)  return True  if __name__ == '__main__':  app = MainApp(0)  app.MainLoop()  

Вы просто находитесь self.worker = None в неправильной точке.
Он должен быть установлен только None на остановку или когда он завершается естественным образом, кроме того, что код в порядке.
напр.

 import time from threading import * import wx  # Button definitions ID_START = wx.NewId() ID_STOP = wx.NewId()  # Define notification event for thread completion EVT_RESULT_ID = wx.NewId()  def EVT_RESULT(win, func):  """Define Result Event."""  win.Connect(-1, -1, EVT_RESULT_ID, func)  class ResultEvent(wx.PyEvent):  """Simple event to carry arbitrary result data."""  def __init__(self, data):  """Init Result Event."""  wx.PyEvent.__init__(self)  self.SetEventType(EVT_RESULT_ID)  self.data = data  # Thread class that executes processing class WorkerThread(Thread):  """Worker Thread Class."""  def __init__(self, notify_window):  """Init Worker Thread Class."""  Thread.__init__(self)  self._notify_window = notify_window  self._want_abort = 0  # This starts the thread running on creation, but you could  # also make the GUI thread responsible for calling this  #self.start()   def run(self):  """Run Worker Thread."""  # This is the code executing in the new thread. Simulation of  # a long process (well, 10s here) as a simple loop - you will  # need to structure your processing so that you periodically  # peek at the abort variable  for i in range(10):  if self._want_abort:  # Use a result of None to acknowledge the abort (of  # course you can use whatever you'd like or even  # a separate event type)  wx.PostEvent(self._notify_window, ResultEvent(None))  return  time.sleep(1)  wx.PostEvent(self._notify_window, ResultEvent(i)) # lt;lt;lt;lt;lt;lt;lt; Added to simulate "status update" feedback of my application.    # Here's where the result would be returned (this is an  # example fixed result of the number 10, but it could be  # any Python object)  time.sleep(2)  wx.PostEvent(self._notify_window, ResultEvent(10))   def abort(self):  """abort worker thread."""  # Method for use by main thread to signal an abort  self._want_abort = 1  # GUI Frame class that spins off the worker thread class MainFrame(wx.Frame):  """Class MainFrame."""  def __init__(self, parent, id):  """Create the MainFrame."""  wx.Frame.__init__(self, parent, id, 'Thread Test')   # Dumb sample frame with two buttons  wx.Button(self, ID_START, 'Start', pos=(0,0))  wx.Button(self, ID_STOP, 'Stop', pos=(0,50))  self.status = wx.StaticText(self, -1, '', pos=(0,100))   self.Bind(wx.EVT_BUTTON, self.OnStart, id=ID_START)  self.Bind(wx.EVT_BUTTON, self.OnStop, id=ID_STOP)   # Set up event handler for any worker thread results  EVT_RESULT(self,self.OnResult)   # And indicate we don't have a worker thread yet  self.worker = None   def OnStart(self, event):  """Start Computation."""  # Trigger the worker thread unless it's already busy  if self.worker:  return  else:  self.status.SetLabel('Starting computation')  self.worker = WorkerThread(self)  self.worker.start()  self.status.SetLabel('Computation started')   def OnStop(self, event):  """Stop Computation."""  # Flag the worker thread to stop if running  if self.worker:  self.status.SetLabel('Trying to abort computation')  self.worker.abort()  self.worker = None  else:  self.status.SetLabel('Computation not running')   def OnResult(self, event):  """Show Result status."""  if event.data is None:  # Thread aborted (using our convention of None return)  self.status.SetLabel('Computation aborted')  else:  # Process results here  self.status.SetLabel('Computation Result: %s' % event.data)  if event.data == 10:   self.worker = None  self.status.SetLabel('Computation Finished')  class MainApp(wx.App):  """Class Main App."""  def OnInit(self):  """Init Main App."""  self.frame = MainFrame(None, -1)  self.frame.Show(True)  self.SetTopWindow(self.frame)  return True  if __name__ == '__main__':  app = MainApp(0)  app.MainLoop()