neurotorchmz.core.task_system

neurotorchmz.core.task_system#

Classes

`Task`(function, name[, run_async, ...])	A class, which allows to efficiently handle work in the background by running them in a thread.
`TaskState`(*values)

Exceptions

TaskError

class neurotorchmz.core.task_system.Task(function: Callable[[...], Any] | _TaskFunction, name: str, run_async: bool = True, keep_alive: bool = False, background: bool = False)#

Bases: object

A class, which allows to efficiently handle work in the background by running them in a thread. Please note: Due to the Python Interpreter Lock, the function will still be running effectively on one core

_id_count = count(0)#

_task_wrapper(**kwargs)#

Internal wrapper function for the task function to catch errors, measure time and fire the callback

_tasks: list[Task] = []#

add_callback(callback: Callable[[], Any]) → Self#: Add a callback for this task. Can be called even after the task finished

property finished: bool#: Returns true if the task finished (successfully or with an error)

static gc_task_list() → None#: The task class stores all created tasks. Calling this function will garbage collect this list and remove inactive tasks

static get_active_tasks() → list[Task]#: Get a list of all currently running tasks

static get_recently_ended_tasks() → list[Task]#: Get a list of task which ended either successfully or with an error not more then 5 seconds ago

static get_recently_failed_tasks() → list[Task]#

static get_tasks() → list[Task]#: Get a list of all tasks ever created

property inactive: bool#: Returns true if the task finished more then 5 seconds ago

is_determinate() → bool#: Returns if the task is determinate or indeterminate

join() → bool#: Join the task. Returns False if in sync mode or the task has not been started yet, otherwise True

property progress: float | None#: The progress of the task as a float between 0 and 1. Set to None if the task hasn’t been started or the task is indeterminate

reinitalize() → Self#: Reset all properties of this task

reset(function: Callable[[...], None], name: str, run_async: bool = True, keep_alive: bool = False, background: bool = False)#: Reinitalize the task object as it would have been newly created but keeps the resources like the thread for example

reset_progress() → Self#: Reset the current progress to zero

property running: bool#: Returns true if the task is currently running

property runtime: float | None#: Returns the runtime of the task in seconds or None if not finished yet

set_error_callback(error_callback: Callable[[Exception], Any]) → Self#: Provide a function which is called when an error happens in the task. The exception will be passed to the supplied function. Note that when setting an error callback, the exception not raised anymore by the task object. The function works even after a task finished.

set_finished() → Self#: Set the percentage to 100% or the step to step count

set_indeterminate() → Self#: Marks the task as indeterminate; task.progress will now always return None

set_message(description: str | None = None) → Self#: Set the given text as short info message about the current state

set_percentage_mode() → Self#: Report progress in percent (default)

set_progress(val: float, description: str | None = None) → Self#

Set the progress in percent. Use the description parameter to supply a short message for the current state. Can only be called when in percent mode

Raises:

ValueError – the value is not a float between 0 and 1
RuntimeError – trying to set a progress value on an step task or in indeterminate mode

set_step_mode(step_count: int) → Self#

Calling this function will report a progress in form of steps instead of percentage. Can also be used to update the step count

Parameters:

step_count (int)

Raises:

ValueError – step_count is not an positve integer
ValueError – step_count is higher then the current step

set_step_progress(step: int, description: str | None = None) → Self#

When in step mode, set the current progress. Steps are counted from zero. Use the description parameter to supply a short message for the current step

Raises:

RuntimeError – trying to set a step progress when not in step_mode
ValueError – step is not a non negative integer
ValueError – step is greater than step_count

start(**kwargs) → Self#: Run the task. Any arguments given are passed to the function. If the task is already running, do nothing. Otherwise start a new one or wake it up from standby

property state: TaskState#: Returns the current state of the Task

property time_since_end: float | None#: The seconds since the task ended or None if not ended yet

property time_since_start: float | None#: The seconds since the task has been started or None if not started yet

exception neurotorchmz.core.task_system.TaskError#: Bases: Exception

class neurotorchmz.core.task_system.TaskState(*values)#

Bases: Enum

CREATED = 0#: The task has been created and is awaiting to be started

ERROR = 30#: The task failed

FINISHED = 25#: The task has finished. If the keep_alive parameter is set, this state will never be reached

RUNNING = 10#: The task is currently running

STANDBY = 20#: The task has ended but is staying in standby such that no new task needs to be created when calling start() again. Useful in async mode as it keeps the thread alive

class neurotorchmz.core.task_system._TaskFunction(*args, **kwargs)#

Bases: Protocol

The protocoll for a valid function inside a task

_abc_impl = <_abc._abc_data object>#

_is_protocol = True#

neurotorchmz.core.task_system

Contents

neurotorchmz.core.task_system#