Если вы довольно давно используете графический интерфейс wxPython, то вы знаете, что иногда нам нужно запускать долгосрочные процессы, при чём делать это постоянно. Конечно, если вы делаете это как вам хотелось бы, посредством командной строки, вы будете удивлены.
В большинстве случаев, всё заканчивается блокировкой цикла событий графического интерфейса, и пользователь видит, что ваша программа немного фризит. Что нужно сделать, чтобы избавиться от этих неприятных моментов? Конечно же, запускать задачу в другом потоке или процессе. В данной статье я рассмотрю, как сделать это с помощью wxPython и его потокового модуля.
Методы потоковой безопасности wxPython
В мире wxPython существует три связанных метода потоковой безопасности. Если вы не используете один из этих трёх методов во время обновления вашего пользовательского интерфейса, вы можете напороться на странные проблемы. В то же время, иногда ваш графический интерфейс работает вполне нормально.
В других случаях, Python крашнется без видимой на то причины. Этим и объясняется потребность в методах потоковой безопасности: wx.PostEvent, wx.CallAfter and wx.CallLater. По словам Робина Дана (создателя wxPython) wx.CallAfter использует wx.PostEvent, чтобы отправить событие в объект приложения. В этом приложении будет хэндлер, привязанный к этому событию, поэтому оно будет считано, а затем произойдёт то, что кодер прописал в скрипте. Я понимаю это так: wx.CallLater вызывает wx.CallAfter с определёнными ограничениями по времени, так что вы можете указать сколько нужно ждать до отправки события.
Робин Дан также указал на то, что Python Global Interpreter Lock будет делать так, чтобы только один поток исполнял один и тот же фрагмент кода за раз, что может повлияет на количество ядер, используемых вашей программой. С другой стороны, он также сказал «wxPython запускает GIL, когда обращается к API wx, так что другие потоки можно запускать в то же время». Другими словами, ваш mileage (пробег или расстояние?) может отличаться, если вы используете потоки или машины с несколькими ядрами. Я счёл данное обсуждение интересным, но, в то же время, удручающим.
Как бы то ни было, в отношении этих трёх методов это значит, что wx.CallLater является наиболее абстрактным методом потоковой безопасности, за ним идёт wx.CallAfter, а wx.PostEvent оказался наиболее низкоуровневым. В следующих примерах вы увидите, как нужно использовать wx.CallAfter и wx.PostEvent, чтобы обновлять вашу программу, написанную с помощью wxPython.
wxPytnon, потоки, wx.CallAfter и PubSub
Если вы подписаны на рассылку от wxPython, то увидите, что эксперты советуют остальным использовать wx.CallAfter одновременно с PubSub, таким образом создавая возможность общения с другими их приложениями wxPython из другого потока. Возможно даже я говорил людям так делать. Так что, в следующем примере, это именно то, что мы собираемся сделать. Учтите, что данный код использует старую версию пабсаба и будет работать только в wxPython 2.8.12 или старее.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 |
# wxPython 2.8.12 import time import wx from threading import Thread from wx.lib.pubsub import Publisher class TestThread(Thread): """Проверка класса рабочего потока.""" def __init__(self): """Запуск класса рабочего потока.""" Thread.__init__(self) self.start() # Начало потока def run(self): """Запуск рабочего потока.""" # Данный код исполняется в новом потоке. for i in range(6): time.sleep(10) wx.CallAfter(self.postTime, i) time.sleep(5) wx.CallAfter(Publisher().sendMessage, "update", "Thread finished!") def postTime(self, amt): """ Отправляет время в графический интерфейс """ amtOfTime = (amt + 1) * 10 Publisher().sendMessage("update", amtOfTime) class MyForm(wx.Frame): def __init__(self): wx.Frame.__init__(self, None, wx.ID_ANY, "Tutorial") #Добавляет панель таким образом, чтобы она выглядела корректно на всех платформах panel = wx.Panel(self, wx.ID_ANY) self.displayLbl = wx.StaticText(panel, label="Amount of time since thread started goes here") self.btn = btn = wx.Button(panel, label="Start Thread") btn.Bind(wx.EVT_BUTTON, self.onButton) sizer = wx.BoxSizer(wx.VERTICAL) sizer.Add(self.displayLbl, 0, wx.ALL|wx.CENTER, 5) sizer.Add(btn, 0, wx.ALL|wx.CENTER, 5) panel.SetSizer(sizer) # Создаёт ресивер пабсаба Publisher().subscribe(self.updateDisplay, "update") def onButton(self, event): """ Запускает поток """ TestThread() self.displayLbl.SetLabel("Thread started!") btn = event.GetEventObject() btn.Disable() def updateDisplay(self, msg): """ Получает данные из потока и обновляет дисплей """ t = msg.data if isinstance(t, int): self.displayLbl.SetLabel("Time since thread started: %s seconds" % t) else: self.displayLbl.SetLabel("%s" % t) self.btn.Enable() # Запускает программу if __name__ == "__main__": app = wx.App(False) frame = MyForm().Show() app.MainLoop() |
Мы будем использовать модуль времени Python для того, чтобы имитировать наш длительный процесс. Как бы то ни было, не стесняйтесь вставлять на его место что-то позатейливее. В примере из реальной жизни я использовал, я использовал поток, который открывает Adobe Reader и отправляет PDF-файл на принтер.
В этом, на первый взгляд, нет ничего особенного, но, когда я не использовал поток, кнопка отправки в печать зависала на тот период, пока файл не попадал в принтер. Даже секунда или две могут иметь значение для конечного пользователя!
В любом случае, давайте выясним как это работает. В нашем классе потока (воспроизведено ниже), мы перенастраиваем метод «run», так что он делает всё, что мы захотим. Данный поток запускает тогда, когда мы подтверждаем его, так как в его методе __init__ есть «self.start()». В методе «run» мы запускаем цикл с шагом в 6, сном в 10 секунд между повторениями, а затем обновляем наш интерфейс, используя wx.CallAfter and PubSub. Когда цикл завершается, мы отправляем последнее сообщение в наше приложение, чтобы сообщить пользователю о том, что произошло.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 |
class TestThread(Thread): """Проверка класса рабочего потока.""" def __init__(self): """Запуск класса рабочего потока.""" Thread.__init__(self) self.start() # Начало потока def run(self): """Запуск рабочего потока.""" # Данный код запускается в новом потоке. for i in range(6): time.sleep(10) wx.CallAfter(self.postTime, i) time.sleep(5) wx.CallAfter(Publisher().sendMessage, "update", "Thread finished!") def postTime(self, amt): """ Отправка времени в графический интерефейс """ amtOfTime = (amt + 1) * 10 Publisher().sendMessage("update", amtOfTime) |
Вы заметите, что в нашем коде wxPython мы запускаем поток, используя хэндлер событий кнопки. Мы также отключаем кнопку, чтобы пользователь не мог запустить лишних потоков. Было бы очень неловко, если бы несколько из них запустились одновременно, и пользовательский интерфейс случайно говорил о том, что событие произошло, хотя на самом деле этого не было.
Это также отличное упражнение для читателей. Вы можете отображать PID потока, чтобы знать где какой из них…и вы можете захотеть сделать так, чтобы вся информация попадала в текстовый контроль, который вы можете листать и отслеживать активность различных потоков.
Последним интересным фрагментом здесь, пожалуй, является ресивер пабсаба и его же хэндлер событий:
|
1 2 3 4 5 6 7 8 9 10 |
def updateDisplay(self, msg): """ Receives data from thread and updates the display """ t = msg.data if isinstance(t, int): self.displayLbl.SetLabel("Time since thread started: %s seconds" % t) else: self.displayLbl.SetLabel("%s" % t) self.btn.Enable() |
Видите, как мы извлекаем сообщение из нашего потока, а затем используем его для обновления дисплея? Мы также можем тип получаемых данных, сообщающий нам, что показывать пользователю. Круто, не так ли?
Делаем так, чтобы потоки и пабсаб работали в Phoenix
Как вы могли узнать, читая мои предыдущие статьи, модуль пабсаба был изменён в wxPython 2.9, так что код из прошлого фрагмента статьи не будет работать в текущей версии wxPython. Так что давайте немного изменим код так, чтобы он работал на wxPython 3.0 Classic и Phoenix.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 |
# wxPython 3.0 and Phoenix import time import wx from threading import Thread from wx.lib.pubsub import pub class TestThread(Thread): """Проверка класса рабочего потока.""" def __init__(self): """Запуск класса рабочего потока.""" Thread.__init__(self) self.start() # Начало потока def run(self): """Запуск потока.""" # Данный код будет исполнятся в новом потоке. for i in range(6): time.sleep(10) wx.CallAfter(self.postTime, i) time.sleep(5) wx.CallAfter(pub.sendMessage, "update", msg="Thread finished!") def postTime(self, amt): """ Отправляем время в графический интерфейс """ amtOfTime = (amt + 1) * 10 pub.sendMessage("update", msg=amtOfTime) class MyForm(wx.Frame): def __init__(self): wx.Frame.__init__(self, None, wx.ID_ANY, "Tutorial") # Добавляет панель так, чтобы она корректно отображалась на всех платформах. panel = wx.Panel(self, wx.ID_ANY) self.displayLbl = wx.StaticText(panel, label="Amount of time since thread started goes here") self.btn = btn = wx.Button(panel, label="Start Thread") btn.Bind(wx.EVT_BUTTON, self.onButton) sizer = wx.BoxSizer(wx.VERTICAL) sizer.Add(self.displayLbl, 0, wx.ALL|wx.CENTER, 5) sizer.Add(btn, 0, wx.ALL|wx.CENTER, 5) panel.SetSizer(sizer) # Создаёт ресивер пабсаба pub.subscribe(self.updateDisplay, "update") def onButton(self, event): """ Запускает поток """ TestThread() self.displayLbl.SetLabel("Thread started!") btn = event.GetEventObject() btn.Disable() def updateDisplay(self, msg): """ Получает данные из потока и обновляет дисплей """ t = msg if isinstance(t, int): self.displayLbl.SetLabel("Time since thread started: %s seconds" % t) else: self.displayLbl.SetLabel("%s" % t) self.btn.Enable() # Запускает программу if __name__ == "__main__": app = wx.App(False) frame = MyForm().Show() app.MainLoop() |
Вы заметили, что мы только закончили импортировать pub и замещаем все отсылки к Publisher() на отсылки к pub. На также нужно значительно изменить вызов sendMessage, так как нам нужно использовать в качестве аргументов ключевые слова, которые соответствуют функции, которую вызывает подписчик. Это всё небольшие изменения, но они необходимы для того, чтобы данный код работал в новых версиях wxPython.
wx.PostEvent и потоки
Следующий код основан на примере, взятом из wxPython вики. Он немного сложнее чем код с wx.CallAfter, который мы только что просмотрели, но я уверен, что вы справитесь с ним.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 |
import time import wx from threading import Thread # Определение события-уведомления, которое ознаменует завершение потока EVT_RESULT_ID = wx.NewId() def EVT_RESULT(win, func): """Определение события-результата .""" win.Connect(-1, -1, EVT_RESULT_ID, func) class ResultEvent(wx.PyEvent): """Простое событие, содержащие случайные данные результата.""" def __init__(self, data): """Запустить событие-результат.""" wx.PyEvent.__init__(self) self.SetEventType(EVT_RESULT_ID) self.data = data class TestThread(Thread): """Проверка класса рабочего потока.""" def __init__(self, wxObject): """Запуск класса рабочего потока.""" Thread.__init__(self) self.wxObject = wxObject self.start() # Начало потока def run(self): """Запуск рабочего потока.""" # Данный код исполняется в новом потоке. for i in range(6): time.sleep(10) amtOfTime = (i + 1) * 10 wx.PostEvent(self.wxObject, ResultEvent(amtOfTime)) time.sleep(5) wx.PostEvent(self.wxObject, ResultEvent("Thread finished!")) class MyForm(wx.Frame): def __init__(self): wx.Frame.__init__(self, None, wx.ID_ANY, "Tutorial") # Добавляем панель таким образом, чтобы она корректно отображалась на всех платформах. panel = wx.Panel(self, wx.ID_ANY) self.displayLbl = wx.StaticText(panel, label="Amount of time since thread started goes here") self.btn = btn = wx.Button(panel, label="Start Thread") btn.Bind(wx.EVT_BUTTON, self.onButton) sizer = wx.BoxSizer(wx.VERTICAL) sizer.Add(self.displayLbl, 0, wx.ALL|wx.CENTER, 5) sizer.Add(btn, 0, wx.ALL|wx.CENTER, 5) panel.SetSizer(sizer) # Настройка хэндлера событий для любых результатов рабочего потока EVT_RESULT(self, self.updateDisplay) def onButton(self, event): """ Запускает поток """ TestThread(self) self.displayLbl.SetLabel("Thread started!") btn = event.GetEventObject() btn.Disable() def updateDisplay(self, msg): """ Получает данные из потока и обновляет дисплей """ t = msg.data if isinstance(t, int): self.displayLbl.SetLabel("Time since thread started: %s seconds" % t) else: self.displayLbl.SetLabel("%s" % t) self.btn.Enable() # Запускает программу if __name__ == "__main__": app = wx.App(False) frame = MyForm().Show() app.MainLoop() |
Давайте разберёмся с этим. Для меня наиболее удручающими являются первые три фрагмента:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
# 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 |
Ключевым объектом здесь является EVT_RESULT_ID. Он делает ссылку потока на wx.PyEvent и на эту странную функцию «EVT_RESULT». В коде wxPython, мы привязываем хэндлер события к функции EVT_RESULT. Это позволяет нам использовать wx.PostEvent в потоке, для того, чтобы отправить событие в наш обычный класс события, который называется ResultEvent. Что он делает?
Он отправляет данные дальше в программу wxPython, посредством запуска этого обычного EVT_RESULT, к которому мы и делали привязку. Я надеюсь, что это звучит так, будто в нём есть смысл.
Как только вы стали уверенны в том, что поняли написанное выше, продолжайте читать. Вы готовы? Отлично! Вы заметите что ваш класс TestThread точно такой же, за исключением того, что мы используем для отправки сообщений wx.PostEvent вместо пабсаба. API нашего обновления дисплея графического интерфейса остаётся неизменённым. Мы по-прежнему всего лишь используем свойства данных сообщения, чтобы извлечь необходимую нам информацию. Вот и всё, ребята!
Итоги
Я надеюсь, что теперь вы знаете как использовать простые потоковые техники в ваших программах, написанных с помощью wxPython. Также есть ещё несколько других потоковых методов, шанса рассказать о которых в данной статье у меня не было. Речь идёт о wx.Yield и Queues. К счастью, на wxPython вики данная тема рассмотрена довольно неплохо, так что убедитесь, что вы ознакомились с соответствующими статьями. Они помогут вам освоить эти методы.
Являюсь администратором нескольких порталов по обучению языков программирования Python, Golang и Kotlin. В составе небольшой команды единомышленников, мы занимаемся популяризацией языков программирования на русскоязычную аудиторию. Большая часть статей была адаптирована нами на русский язык и распространяется бесплатно.
E-mail: vasile.buldumac@ati.utm.md
Образование
Universitatea Tehnică a Moldovei (utm.md)
- 2014 — 2018 Технический Университет Молдовы, ИТ-Инженер. Тема дипломной работы «Автоматизация покупки и продажи криптовалюты используя технический анализ»
- 2018 — 2020 Технический Университет Молдовы, Магистр, Магистерская диссертация «Идентификация человека в киберпространстве по фотографии лица»