An example plugin showing basic usage of task_lib for background tasks.

task_lib_example.py

"""
(*)~---------------------------------------------------------------------------
Pupil - eye tracking platform
Copyright (C) 2012-2019 Pupil Labs

Distributed under the terms of the GNU
Lesser General Public License (LGPL v3.0).
See COPYING and COPYING.LESSER for license details.
---------------------------------------------------------------------------~(*)

This demo shows you how to use tasklib in your plugin.

The following things are NOT shown in this demo:
- Usage of task events like on_finished, on_canceled, ...
  See tasklib.interface for all available events and for documentation about them.
- How to cancel tasks before they are done.
- How to report the progress of a background process to the foreground.

How to use (copied from tasklib.manager):
    1) Make your plugin an Observable.
    2) Create a PluginTaskManager() in your plugin.
    3) Give the manager e.g. to controllers and other parts of your plugin that need
       to start tasks.
    4) Call create_background_task() to add tasks, or add you own task that inherits
       from TaskInterface via add_task().

"""

import logging
import time

import plugin
import tasklib
from observable import Observable

logger = logging.getLogger(__name__)


# Every function, generator or lambda can be run in the background.
# They don't need to have any specific format, parameters etc.
def dummy_generator():
    for i in range(10):
        time.sleep(0.5)
        yield i


# Your plugin needs to inherit from Observable, otherwise the PluginTaskManager can
# not work. Don't worry, if you forget this, you will get a corresponding exception!
# Note that Observable appears BEFORE plugin.Plugin in the list. Some plugins change
# behavior based on their base class, which is the rightmost class in the list.
# Hence, having observable last might break plugins.
class DummyPlugin(Observable, plugin.Plugin):
    def __init__(self, g_pool):
        super().__init__(g_pool)
        # the manager allows you to create tasks and automatically fetches background
        # results for you. It also terminates all tasks if your plugin gets closed.
        self.task_manager = tasklib.manager.PluginTaskManager(plugin=self)

    def on_click(self, pos, button, action):
        task1 = self.task_manager.create_background_task(
            name="my dummy task",
            routine_or_generator_function=self._dummy_task,
            args=(10, 2),
        )
        # If the background task raises an exception it would just die without
        # outputting the exception. Because we would like to know about any
        # exceptions, we add a predefined observer that raises the exception also in
        # the foreground.
        # If you actually expect an exception than you can also define your own handler.
        task1.add_observer("on_exception", tasklib.raise_exception)
        # on_completed is called when your task finished successfully. Your handler
        # needs to accept exactly one argument (which will be the returned value of
        # the task).
        task1.add_observer("on_completed", self._on_dummy_task_done)
        # tasks can be started manually, but there is no need to. Any unstarted tasks
        # will be started when the PluginTaskManager does its next update (which it
        # does every couple of milliseconds)
        # task1.start()

        task2 = self.task_manager.create_background_task(
            name="my dummy task",
            routine_or_generator_function=dummy_generator,
        )
        task2.add_observer("on_exception", tasklib.raise_exception)
        # on_yield is called every time a generator yields. There is also
        # on_completed for generators, but that's called with None as return value
        # because there is none in this case.
        task2.add_observer("on_yield", self._on_dummy_generator_yields)

    def _dummy_task(self, a, b):
        time.sleep(1.0)
        return a + b

    def _on_dummy_task_done(self, result):
        logger.info("Dummy task done with result {}".format(result))

    def _on_dummy_generator_yields(self, yield_value):
        logger.info("Dummy generator yields value {}".format(yield_value))