Работа с длительными операциями
Задачи
Скрипты на python выполняются в основном потоке приложения или по другому в потоке интeрфейса. Если операция будет выполняться слишком долго, то интерфейс «зависнет» и будет невозможно со стороны пользователя понять это ошибка или программа все-таки продолжает выполнять свою работу. Чтобы этого избежать длительные вычисления следует выносить в фоновые потоки. В потоке интерфейса во время выполнения фоновой задачи есть возможность показывать прогресс операции.
Чтобы превратить пользовательскую функцию в задачу, нужно использовать класс axipy.Task
.
import time
import axipy
def user_heavy_function(t: axipy.Task, arg1: int, arg2: str):
print(f"Переданные аргументы: {arg1}, {arg2}.")
time.sleep(2)
return 1
task = axipy.Task(user_heavy_function)
result = task.run_and_get("Hello", "world")
print(result)
"""
>>> Переданные аргументы: Hello, world.
>>> 1
"""
При использовании метода axipy.Task.start()
, для получения результата, нужно подписаться на сигнал axipy.Task.finished
.
import time
import axipy
def user_heavy_function(t: axipy.Task, arg1: int, arg2: str):
print(f"Переданные аргументы: {arg1}, {arg2}.")
time.sleep(2)
return 1
task = axipy.Task(user_heavy_function)
def finished(t: axipy.Task):
print(t.status)
print(t.result)
task.finished.connect(finished)
task.start("Hello", "world")
"""
>>> Переданные аргументы: Hello, world.
>>> Status.SUCCESS
>>> 1
"""
Чтобы сделать поддержку типизированных аргументов и возвращаемого значения для задачи, рекомендуется
наследование от базового класса axipy.Task
.
import axipy
class SummTask(axipy.Task):
def __init__(self, *args, **kwargs) -> None:
super().__init__(self.run, *args, **kwargs)
def run(self, arg1: float, arg2: float) -> float:
self.value += 1
return arg1 + arg2
def run_and_get(self, arg1: float, arg2: float) -> float:
return super().run_and_get(arg1, arg2)
result = SummTask().run_and_get(1.0, 2.3)
print(result)
"""
>>> 3.3
"""
Представление прогресса операции
Класс axipy.DialogTask
дополняет класс axipy.Task
наличием диалога для отображения прогресса операции.
Для обмена информацией между выполняемой задачей и элементом, отображающим прогресс,
используется ссылка на экземпляр задачи (axipy.Task
), передаваемая первым аргументом в пользовательскую функцию.
Типичный вариант использования выглядит следующим образом:
import axipy
import time
def user_heavy_function(dt: axipy.DialogTask, loop_range: int):
dt.range = dt.Range(0, loop_range)
for i in range(loop_range):
time.sleep(0.5)
dt.value += 1
return 1
dialog_task = axipy.DialogTask(user_heavy_function)
result = dialog_task.run_and_get(10)
print(result)
"""
>>> 1
"""
Вместо axipy.Task.is_canceled
можно
использовать axipy.Task.raise_if_canceled
если нужно выйти из нескольких вложенных вызовов функций или циклов.
Создание пользовательского виджета для отображения прогресса
import axipy
import time
import PySide2.QtWidgets
import PySide2.QtCore
task_max_range = 20
def heavy_function(t: axipy.Task):
for i in range(task_max_range):
time.sleep(0.5)
t.value += 1
t.raise_if_canceled()
return 1
class CustomDialog(PySide2.QtWidgets.QDialog):
def __init__(self, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)
vbox = PySide2.QtWidgets.QVBoxLayout()
self.pbar = PySide2.QtWidgets.QProgressBar()
self.pbar.setRange(0, task_max_range)
vbox.addWidget(self.pbar)
bbox = PySide2.QtWidgets.QDialogButtonBox()
bbox.setStandardButtons(
PySide2.QtWidgets.QDialogButtonBox.Cancel
)
bbox.rejected.connect(self.bbox_reject)
vbox.addWidget(bbox)
self.setLayout(vbox)
@PySide2.QtCore.Slot()
def bbox_reject(self) -> None:
def ask() -> PySide2.QtWidgets.QMessageBox.StandardButton:
return PySide2.QtWidgets.QMessageBox.question(
axipy.view_manager.global_parent,
"Отмена",
"Отменить задачу?",
)
answer = ask()
if answer == PySide2.QtWidgets.QMessageBox.Yes:
self.reject()
@PySide2.QtCore.Slot()
def add_progress(self) -> None:
self.pbar.setValue(self.pbar.value() + 1)
dialog = CustomDialog(axipy.view_manager.global_parent)
task = axipy.Task(heavy_function)
task.value_changed.connect(dialog.add_progress)
task.finished.connect(dialog.close)
dialog.rejected.connect(task.cancel)
task.start()
if task.status not in (
task.Status.SUCCESS,
task.Status.CANCELED,
task.Status.ERROR,
):
dialog.exec_()
Выполнение задач и многопоточность
Важно понимать, что задачи будут выполняться не в потоке интерфейса, поэтому внутри
этих задач нельзя отображать никакие графические элементы (PySide2.QtWidgets.QWidget
).
Так же нужно внимательно следить за тем какие ресурсы могут использоваться несколькими
потоками и при необходимости использовать различные механизмы синхронизации ( мьютексы,
локи и т.д.). Общее правило при работе с несколькими потоками следующее:
старайтесь чтобы каждая задача содержала в себе все необходимые данные для
выполнения. Синхронизацию, если она необходима, следует использовать только в
момент получения результата. Это упростит код и сведет к минимум количество ошибок.
Переданные на выполнения задачи ставятся в общую очередь, которая распределяется между физическими ядрами процессора в порядке добавления. Поэтому нет гарантии, что переданная задача выполнится мгновенно, а так же то что группа задач будет выполняться именно в порядке добавления. Если задача предполагает долгое ожидание без интенсивных нагрузок на процессор, то лучше воспользоваться стандартными python потоками. Типичные примеры таких задач это загрузка ресурсов с диска или скачивание файлов по сети.