Работа с длительными операциями

Задачи

Скрипты на 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 потоками. Типичные примеры таких задач это загрузка ресурсов с диска или скачивание файлов по сети.