Довольно часто стоит задача обновлять индикатор чрезвычайно часто. В данной статей мы создадим рамку с кнопкой. Когда кнопка нажимается, появляется диалог, который содержит наш индикатор, а он, в свою очередь, запустит поток. Поток – это предельно простой поток, который ничего толком не делает, кроме того, чтоб отправляет обновление обратно в диалог, при чём делает это раз в 20 секунд.
После чего диалог уничтожается. Теперь давайте посмотрим, как мы можем выполнить данную задачу, используя wxPython 2.8.12.1, который является довольно популярной версией wxPython, даже несмотря на то, что она уже устарела.
|
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 |
# This code works with wxPython 2.8.12 and Python 2 import time import wx from threading import Thread from wx.lib.pubsub import Publisher class TestThread(Thread): """Test Worker Thread Class.""" def __init__(self): """Init Worker Thread Class.""" Thread.__init__(self) self.start() # start the thread def run(self): """Run Worker Thread.""" # This is the code executing in the new thread. for i in range(20): time.sleep(1) wx.CallAfter(Publisher().sendMessage, "update", "") class MyProgressDialog(wx.Dialog): """""" def __init__(self): """Constructor""" wx.Dialog.__init__(self, None, title="Progress") self.count = 0 self.progress = wx.Gauge(self, range=20) sizer = wx.BoxSizer(wx.VERTICAL) sizer.Add(self.progress, 0, wx.EXPAND) self.SetSizer(sizer) # Создаём пабсаб-слушателя Publisher().subscribe(self.updateProgress, "update") def updateProgress(self, msg): """ Update the progress bar """ self.count += 1 if self.count >= 20: self.Destroy() self.progress.SetValue(self.count) class MyFrame(wx.Frame): def __init__(self): wx.Frame.__init__(self, None, title="Progress Bar Tutorial") # Добавляем панель таким образом, чтобы она корректно отображалась на всех платформах panel = wx.Panel(self, wx.ID_ANY) self.btn = btn = wx.Button(panel, label="Start Thread") btn.Bind(wx.EVT_BUTTON, self.onButton) sizer = wx.BoxSizer(wx.VERTICAL) sizer.Add(btn, 0, wx.ALL|wx.CENTER, 5) panel.SetSizer(sizer) def onButton(self, event): """ Запускаем поток """ btn = event.GetEventObject() btn.Disable() TestThread() dlg = MyProgressDialog() dlg.ShowModal() btn.Enable() # Запускает программу if __name__ == "__main__": app = wx.App(False) frame = MyFrame() frame.Show() app.MainLoop() |
Давайте потратим несколько минут на то, чтобы с этим разобраться. Мы начнём с конца. Класс MyFrame запускается в первую очередь. Когда этот скрипт запустится, вы должны будете увидеть что-то вроде этого:
Как вы видите, всё что делает этот код – создаёт простую рамку, в которой находится кнопка. Если вы нажмёте на эту кнопку, следующий диалог будет создан и запустится новый поток.
Давайте взглянем на часть кода, которая создаёт диалог:
|
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 |
class MyProgressDialog(wx.Dialog): """""" def __init__(self): """Constructor""" wx.Dialog.__init__(self, None, title="Progress") self.count = 0 self.progress = wx.Gauge(self, range=20) sizer = wx.BoxSizer(wx.VERTICAL) sizer.Add(self.progress, 0, wx.EXPAND) self.SetSizer(sizer) # Создаёт пабсаб-слушателя Publisher().subscribe(self.updateProgress, "update") def updateProgress(self, msg): """ Обновляет индикатор """ self.count += 1 if self.count >= 20: self.Destroy() self.progress.SetValue(self.count) |
Этот код просто создаёт диалог с помощью виджета wx.Gauge. Gauge – это виджет, который, по сути, обеспечивает работу индикатора. В любом случае, мы создаём слушателя пабсаб в самом конце метода __init__ нашего диалога. Этот слушатель будет принимать сообщения, которые будут создаваться методом updateProgress. Вы заметите, что все сообщения отправляются в класс потока.
В методе updateProgress мы демонстрируем движение счётчика, обновляя wx.Gauge посредством того, что задаём ему значение. Также мы проверяем, не является ли это значение равно или больше 20, ведь именно таков размер шага – 20 секунд. Если это действительно так, то диалог уничтожается.
Теперь мы готовы разобраться с кодом потока
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
class TestThread(Thread): """Проверка класса рабочего потока.""" def __init__(self): """Запуск класса рабочего потока.""" Thread.__init__(self) self.start() # Запустить поток def run(self): """Запуск рабочего потока.""" # Данный код исполняется в новом потоке. for i in range(20): time.sleep(1) wx.CallAfter(Publisher().sendMessage, "update", "") |
Здесь мы создаём поток и немедленно его запускаем. Поток запускает цикл, который длится 20 секунд и использует модуль времени, чтобы засыпать на секунду между каждым повторением. После каждой секунды сна, он направляет сообщение в диалог. Это сообщение приказывает обновить индикатор.
Обновляем код для того, чтобы он работал в wxPython 3.0.2.0 + Phoenix
Код в прошлом примере был написан с использованием старого API пабсаба, а им перестали пользовался в ту же секунду, когда большинство перешло на wxPython 2.9. Так что, если вы запустите тот код в версии 2.9 или новее, то, скорее всего, нарвётесь на неприятности. Таким образом, для полноты картины, вот версия кода, которая использует новый API пабсаба и работает на wxPython 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 82 83 84 85 86 |
# This code works for wxPython 3.0+ and Phoenix # So you can run this code with Python 2 or 3 import time import wx from threading import Thread from wx.lib.pubsub import pub class TestThread(Thread): """Test Worker Thread Class.""" def __init__(self): """Проверка класса работающего потока.""" Thread.__init__(self) self.start() # start the thread def run(self): """Запуск рабочего потока.""" # Данный код исполняется в новом потоке. for i in range(20): time.sleep(1) wx.CallAfter(pub.sendMessage, "update", msg="") class MyProgressDialog(wx.Dialog): """""" def __init__(self): """Constructor""" wx.Dialog.__init__(self, None, title="Progress") self.count = 0 self.progress = wx.Gauge(self, range=20) sizer = wx.BoxSizer(wx.VERTICAL) sizer.Add(self.progress, 0, wx.EXPAND) self.SetSizer(sizer) # Создаётся ресивер пабсаба pub.subscribe(self.updateProgress, "update") def updateProgress(self, msg): """""" self.count += 1 if self.count >= 20: self.Destroy() self.progress.SetValue(self.count) class MyForm(wx.Frame): def __init__(self): wx.Frame.__init__(self, None, wx.ID_ANY, "Tutorial") # Добавляем панель таким образом, чтобы она корректно отображалась на всех платформах panel = wx.Panel(self, wx.ID_ANY) self.btn = btn = wx.Button(panel, label="Start Thread") btn.Bind(wx.EVT_BUTTON, self.onButton) sizer = wx.BoxSizer(wx.VERTICAL) sizer.Add(btn, 0, wx.ALL|wx.CENTER, 5) panel.SetSizer(sizer) def onButton(self, event): """ Запускает поток """ btn = event.GetEventObject() btn.Disable() TestThread() dlg = MyProgressDialog() dlg.ShowModal() btn.Enable() # Запускает программу if __name__ == "__main__": app = wx.App(False) frame = MyForm().Show() app.MainLoop() |
Обратите внимание на то, что теперь вы импортируете модуль pub вместо модуля Publisher. Также обратите внимание на то, что вы используете ключевые слова в качестве аргументов. Чтобы узнать об этом больше, ознакомьтесь с документацией пабсаба.
Итоги
Теперь вы должны знать, как создать собственный диалог индикатора и обновлять его прямо из потока. Вы можете использовать одну из вариаций этого кода для создания загрузчика файлов.
Если вы это сделаете, вам нужно будет проверят размер файла, и, если он превышает установленный вами шаг, загружать его по фрагментам, используя всё тот же wx.Gauge. Таким образом, индикатор будет обновляться после загрузки каждого фрагмента. Надеюсь, что у вас также появились идеи использования данного виджета в ваших проектах.
Являюсь администратором нескольких порталов по обучению языков программирования Python, Golang и Kotlin. В составе небольшой команды единомышленников, мы занимаемся популяризацией языков программирования на русскоязычную аудиторию. Большая часть статей была адаптирована нами на русский язык и распространяется бесплатно.
E-mail: vasile.buldumac@ati.utm.md
Образование
Universitatea Tehnică a Moldovei (utm.md)
- 2014 — 2018 Технический Университет Молдовы, ИТ-Инженер. Тема дипломной работы «Автоматизация покупки и продажи криптовалюты используя технический анализ»
- 2018 — 2020 Технический Университет Молдовы, Магистр, Магистерская диссертация «Идентификация человека в киберпространстве по фотографии лица»