qmi.core.task

Background tasks for QMI.

This module provides support for tasks which run asynchronously in the background.

Defining tasks

To define a new background task, create a subclass of QMI_Task. The new class must define a run() method, containing the main code of the task. It may also be helpful to redefine the __init__() method in order to accept arguments and initialize the task.

A task may define two types of special variables:

settings: data that affects the behaviour of the task. Settings can be modified by other tasks or scripts while the task is still running.
status: data that represents the current activity of the task. The task can change its own status while it runs, but other tasks or scripts can not change this task’s status.

Declare two custom named tuple types to contain the settings resp. the status of your task. Instances of these named tuples will be used to store the current settings in self.settings and the current status in self.status.

Example of a custom task definition:

MySettings = namedtuple("MySettings", "param other_param")
MyStatus = namedtuple("MyStatus", "field other_field")

class MyTask(QMI_Task):
    def __init__(self, task_runner, name, my_arg, my_other_arg):
        super().__init__(task_runner, name)
        self.settings = MySettings(...)
        self.status = MyStatus(...)
        ...
    def run(self):
        while not self.stop_requested():
            self.update_settings()
            ...

Running tasks

Tasks are created via the QMI context. The context then returns a proxy for the new task instance. A newly created task must be explicitly started by calling the start() method through the proxy.

A task stops when its run() method returns. Alternatively, the stop() method may be called through the proxy to request that the task stops itself as soon as possible. Even in this case, the task will still continue to run until its run() method honours the stop request by returning.

To clean up a stopped task, call the join() method through the proxy. The join() method will wait until the task has stopped, if necessary, then clean up the background thread.

Example of starting and stopping a task:

proxy = qmi.make_task("my_task", MyTask, "arg1", "arg2")
proxy.start()
...
new_settings = MySettings(...)
proxy.set_settings(new_settings)
...
status = proxy.get_status()
print("Status:", status)
...
proxy.stop()
proxy.join()

Custom Task Runner

It’s possible to define a custom task runner and have qmi use _it_ instead of a default QMI_TaskRunner instance. This will allow custom RPC methods and state to be added to the runner, which can be useful in some cases.

Example of a custom task runner:

class MyTaskRunner(QMI_TaskRunner):
    @rpc_method
    def set_param(self, param):
        settings = self.get_pending_settings()
        settings._replace(param=param)
        self.set_settings(settings)

proxy = qmi.make_task("my_task", MyTask, "arg1", "arg2", task_runner=MyTaskRunner)
proxy.start()
...
proxy.set_param(...)
...
proxy.stop()
proxy.join()

Reference

Classes

`QMI_LoopTask`(task_runner, name[, ...])	QMI_LoopTask is a subclass of QMI_Task, implementing specific background loop task.
`QMI_LoopTaskMissedLoopPolicy`(value)	Possible policies of QMI_LoopTask instance missing a loop period.
`QMI_Task`(task_runner, name)	Base class for all background tasks.
`QMI_TaskRunner`(context, name, task_class, ...)	Manager for a single task.

class qmi.core.task.QMI_LoopTaskMissedLoopPolicy(value): Possible policies of QMI_LoopTask instance missing a loop period.

class qmi.core.task.QMI_Task(task_runner: QMI_TaskRunner, name: str)

Base class for all background tasks.

Each instance of QMI_Task runs in a separate thread where it performs background activities without blocking the rest of the application.

Subclasses of QMI_Task implement specific background tasks. Each subclass must implement a special method run(), containing the code which runs in the background.

Tasks may publish QMI signals.

sig_settings_updated: Signal used to publish the latest settings when updated. Can be used by clients who need to know when the settings might have changed by another client. Users need to re-define this signal with appropriate type, if necessary.

stop_requested() → bool

Return True if the task should stop.

A long running background task should regularly call this function to check whether a stop request has been sent to the task.

If this function returns True, the task should stop as soon as possible (by returning from its run() function).

sleep(duration: float) → None

Sleep for the specified duration (in seconds).

This should be used by the task code instead of time.sleep(). If the task receives a stop request while sleeping, this method ends immediately and raises QMI_TaskStopException.

Parameters:: duration – Sleep duration in seconds.
Raises:: QMI_TaskStopException – If the task received a stop request.

update_settings() → bool

Update to the latest settings that were sent to the task.

If another routine called set_settings() to send new settings to this task, this method copies the new settings to self.settings and returns True.

Otherwise, the settings remain the same and this method returns False.

Returns:: True if there are new settings, False if the settings are unchanged.

run() → None

Main function of the task.

This function runs in the background as long as the task is active. Subclasses must provide their own implementation.

class qmi.core.task.QMI_TaskRunner(context: qmi.core.context.QMI_Context, name: str, task_class: Type[QMI_Task], task_args: tuple, task_kwargs: dict)

Manager for a single task.

A QMI_TaskRunner manages a single instance of QMI_Task.

A dedicated QMI_TaskRunner is created by the context for each instance of QMI_Task. The QMI_TaskRunner creates a background thread to run the actual QMI_Task instance.

The QMI_TaskRunner supports a fixed set of RPC methods. To interact with the task, other tasks or routines call these methods.

A QMI_TaskRunner instance is created by the context when the application calls context.make_task(). Application code should not explicitly call the QMI_TaskRunner() constructor.

classmethod get_category() → str | None

Return the optional name of the category this object belongs to.

A category name is a free-form string that has no special significance. Its purpose is to distinguish between groups of RPC objects that fulfill similar roles.

release_rpc_object() → None: Ensure the task is joined before it is removed from the context.

start() → None

Start the task.

This function triggers a call to the run() method of the task. Note that the task runs in a separate thread. This function returns as soon as the task is started, while the task continues to run in the background.

This function may be called only once per task.

stop() → None

Stop the task.

This function tells the task to stop. From this point on, if the task calls stop_requested() it will return True and if the task sleeps it will be stopped via QMI_TaskStopException.

This function returns immediately without waiting until the task is stopped. The task may continue to run in the background for some time, depending on the code inside the task.

join() → None

Wait until the task is fully stopped.

This function blocks until the task is fully stopped. This may take some time, depending on the code inside the task. If the task does not stop spontaneously and stop() has not been called, this function may never return.

If the task code raised an exception, the exception will be re-raised from this method.

Each task must be joined before it can be removed from the context. Joining the task will also stop the associated thread and release its resources.

is_running() → bool

Return True if the task is currently running.

This method returns True if the task has been started, has not yet stopped and has not raised an exception.

set_settings(new_settings: Any) → None

Send new settings to the task.

Other parts of the software may call this method to change the settings used by the task.

Note that the task may not immediately use the new settings. The task continues to use its current settings until it explicitly updates the settings by calling update_settings().

get_settings() → Any

Return the current value of the task’s self.settings variable.

The task can receive settings from other parts of the software. This method returns a snapshot of the settings that are currently in effect.

Note that calling get_settings() immediately after set_settings() may still return the old settings until the task explicitly updates its settings by calling update_settings().

get_pending_settings() → Any

Return pending settings stored in the task runner’s fifo.

Only once the internal task calls update_settings(), is the pending settings from the fifo assigned to the task’s actual settings.

get_status() → Any

Return the current value of the task’s self.status variable.

The task code can update its self.status variable at any time to expose its current status to other parts of the software. This method returns a current snapshot of the status variable.

get_task_class_name() → str: Return the name of the QMI_Task class managed by this task runner.

force_unlock() → None

Forcefully unlock the remote object.

This unlocks the object, regardless of who owns the lock. This allows you to unlock an object if the locking proxy has been destroyed without unlocking.

Use this with care.

Do not override this stub method in subclasses. It has already been implemented in QMI_RpcProxy.

get_name() → str

Return the name of this object.

Returns:: name attribute.

get_signals() → list[SignalDescription]

Return a list of signals that can be published by this object.

Returns:: List consisting of qmi_signals attributes.

is_locked() → bool

Query if the remote object is locked.

Do not override this stub method in subclasses. It has already been implemented in QMI_RpcProxy.

lock(timeout: float = 0.0, lock_token: str | None = None) → bool

Lock the remote object. If timeout is given, try every 0.1s within the given timeout value. The remote object can be locked with an optional custom lock token by giving a string into lock_token keyword argument.

If successful, this proxy is the only proxy that can invoke RPC methods on the remote object; other proxies will receive an “object is locked” response. The return value indicates if the lock was granted; a denied lock means the object was already locked by another proxy.

Do not override this stub method in subclasses. It has already been implemented in QMI_RpcProxy.

unlock(lock_token: str | None = None) → bool

Unlock the remote object.

Without optional parameters, this is only allowed by the proxy that initially locked the object. By giving the lock token as an input parameter, the specific object locked by this token can be unlocked. The return value indicates if the unlocking was successful.

Do not override this stub method in subclasses. It has already been implemented in QMI_RpcProxy.

class qmi.core.task.QMI_LoopTask(task_runner, name, loop_period: float = 1.0, policy: QMI_LoopTaskMissedLoopPolicy = QMI_LoopTaskMissedLoopPolicy.IMMEDIATE)

QMI_LoopTask is a subclass of QMI_Task, implementing specific background loop task. It has its own special run() method, containing a sequence of actions that apply to most QMI_Task implementations that will be used.

sig_status_updated: Signal used to publish the latest status when updated. Can be used by clients who need to know when the status might have changed by another client. Users need to re-define this signal with appropriate type, if necessary.

run() → None

Main function inside the task thread. This overrides the base class implementation.

The default implementation creates a skeleton of actions that a task loop will take. The clients using this QMI_LoopTask will have to override the implementations of loop_prepare(), loop_iteration(), loop_finalize(), process_new_settings(), update_status(), and publish_signals().

The behaviour of handling missed loop iterations depend on the QMI_LoopTaskMissedLoopPolicy attribute set.

loop_prepare() → None

Method to prepare the task loop.

Subclasses may override this method to prepare the task loop. The default implementation does nothing.

process_new_settings() → None

Method for processing new settings.

Subclasses may override this method to new task settings. The default implementation does nothing.

loop_iteration() → None

Define work to be done in the task loop.

Subclasses may override this method to define work done in a loop iteration. The default implementation does nothing.

update_status() → bool

Update to the latest status that were set to the task.

If another routine called set_status() to set a new status to this task, this method should copy the new status to self.status and return True. Otherwise the status remains the same and this method returns False.

Subclasses may override this method to set specific status. The default implementation does nothing (returns False).

Returns:: bool - True if there is a new status, False if the status is unchanged.

publish_signals() → None

Method for publishing other signals described by the client for the loop task.

Subclasses may override this method to publish specific signals. The default implementation does nothing.

loop_finalize() → None

The loop should return the task to a specific state and settings at the end.

Subclasses may override this method to specify finalizing actions. The default implementation does nothing.

sleep(duration: float) → None

Sleep for the specified duration (in seconds).

This should be used by the task code instead of time.sleep(). If the task receives a stop request while sleeping, this method ends immediately and raises QMI_TaskStopException.

Parameters:: duration – Sleep duration in seconds.
Raises:: QMI_TaskStopException – If the task received a stop request.

stop_requested() → bool

Return True if the task should stop.

A long running background task should regularly call this function to check whether a stop request has been sent to the task.

If this function returns True, the task should stop as soon as possible (by returning from its run() function).

update_settings() → bool

Update to the latest settings that were sent to the task.

If another routine called set_settings() to send new settings to this task, this method copies the new settings to self.settings and returns True.

Otherwise, the settings remain the same and this method returns False.

Returns:: True if there are new settings, False if the settings are unchanged.