luminal.managers.tracer

Module Contents

Classes

LoopTask

A wrapper for a coroutine and its arguments which will be started at

LoopTrace

The LoopTrace class serves as a versatile tool for testing loops of
class LoopTask(at_iteration, coroutine, *args)

A wrapper for a coroutine and its arguments which will be started at a specific iteration within an infinite loop, or any loop in general.

Examples

>>> async def do_task(some_arg):
...     await asyncio.sleep(1)
...     print(some_arg)
...
>>> task = LoopTask(at_iteration=5, coroutine=do_task, some_arg='hello')
>>> # Run a loop and start the coroutine when on the 5th iteration.
>>> for i in range(10):
...     if i == task._at_iteration:
...         await task.coroutine(*task.args)
Parameters:

  • at_iteration (int) –

  • coroutine (Awaitable) –

  • args (tuple) –

class LoopTrace(tasks, iteration_limit=0)

The LoopTrace class serves as a versatile tool for testing loops of any type, including infinite loops. It can be effortlessly placed within any codebase and seamlessly integrated with all loops that call its evaluate_tasks() method.

Examples

>>> async def do_task(some_arg):
...     print(some_arg)
...
>>> task = LoopTask(at_iteration=5, coroutine=do_task, some_arg='hello')
>>> trace = LoopTrace(tasks=[task], iteration_limit=10)
>>> # Create an infinite loop and start the coroutine when on the 5th iteration.
>>> while True:
...     try:
...         await trace.evaluate_tasks()
...     except StopIteration:
...         break
Parameters:
property tasks: list[LoopTask]

Returns a list of all available tasks to run.

Returns:

The list of all tasks added to the LoopTrace.

Return type:

List[LoopTask]

property tasks_with_keys: dict[int, LoopTask]

Returns a list of all available tasks to run and their respective iteration as a dict[int, LoopTask].

Returns:

A dictionary of tasks along with their specific iteration number.

Return type:

dict[int, LoopTask]

property iteration_limit: int

Returns the maximum limit of iterations for the current LoopTrace instance.

Returns:

The maximum number of iterations after which the evaluation will stop.

Return type:

int

property current_iteration: int

Returns the current loop iteration of the LoopTrace instance.

Returns:

The current iteration that is being evaluated.

Return type:

int

add_task(task)

Adds a new task to the already existing tasks list.

Parameters:

task (LoopTask) – A task instance that needs to be added to the tasks list.

Returns:

Returns True if the task was added, False otherwise.

Return type:

bool

Raises:

KeyError – If the task is already present in the tasks list.

Examples

>>> def do_task(args):
...    print(args)
>>> task = LoopTask(at_iteration=5, coroutine=do_task, some_arg='hello')
>>> trace = LoopTrace(tasks=[task])
>>> trace.add_task(task) # Raises KeyError as the task is already present in the tasks list.
remove_task(task)

Removes a task from the already existing tasks list.

Parameters:

task (LoopTask) – The task instance that needs to be removed from the tasks list.

Returns:

Returns True if the task was removed, False otherwise.

Return type:

bool

Examples

>>> def do_task(args):
...    print(args)
>>> task = LoopTask(at_iteration=5, coroutine=do_task, some_arg='hello')
>>> trace = LoopTrace(tasks=[task])
>>> trace.remove_task(task) # Returns True as the task was removed from the tasks list.
async evalutate_tasks()

Evaluates and executes the tasks that are supposed to be started at the current iteration.

Raises:

StopIteration – If the iteration limit has been reached.

Return type:

None

Example

>>> def do_task(args):
...    print(args)
>>> task = LoopTask(at_iteration=5, coroutine=do_task, some_arg='hello')
>>> trace = LoopTrace(tasks=[task], iteration_limit=10)
>>> # Create an infinite loop and start the coroutine when on the 5th iteration.
>>> while True:
...     try:
...         await lt.evaluate_tasks()
...     except StopIteration:
...         break