Implements multitasking with scheduling and some performance logging. More...

Public Member Functions
def	__init__ (self, run_fun, name="NoName", priority=0, period=None, profile=False, trace=False, shares=())
	Initialize a task object so it may be run by the scheduler. More...

bool	schedule (self)
	This method is called by the scheduler; it attempts to run this task. More...

bool	ready (self)
	This method checks if the task is ready to run. More...

def	set_period (self, new_period)
	This method sets the period between runs of the task to the given number of milliseconds, or `None` if the task is triggered by calls to `go()` rather than time. More...

def	reset_profile (self)
	This method resets the variables used for execution time profiling. More...

def	get_trace (self)
	This method returns a string containing the task's transition trace. More...

def	go (self)
	Method to set a flag so that this task indicates that it's ready to run. More...

def	__repr__ (self)
	This method converts the task to a string for diagnostic use. More...

Public Attributes
	name
	The name of the task, hopefully a short and descriptive string.

	priority
	The task's priority, an integer with higher numbers meaning higher priority.

	period
	The period, in milliseconds, between runs of the task's `run()` method. More...

	go_flag
	Flag which is set true when the task is ready to be run by the scheduler.

Detailed Description

Implements multitasking with scheduling and some performance logging.

This class implements behavior common to tasks in a cooperative multitasking system which runs in MicroPython. The ability to be scheduled on the basis of time or an external software trigger or interrupt is implemented, state transitions can be recorded, and run times can be profiled. The user's task code must be implemented in a generator which yields the state (and the CPU) after it has run for a short and bounded period of time.

Example:

def task1_fun ():
     '''! This function switches states repeatedly for no reason '''
     state = 0
     while True:
         if state == 0:
             state = 1
         elif state == 1:
             state = 0
         yield (state)
 
 # In main routine, create this task and set it to run twice per second
 task1 = cotask.Task (task1_fun, name = 'Task 1', priority = 1, 
                      period = 500, profile = True, trace = True)
 
 # Add the task to the list (so it will be run) and run scheduler
 cotask.task_list.append (task1)
 while True: 
     cotask.task_list.pri_sched ()

Constructor & Destructor Documentation

◆ init()

def cotask.Task.__init__	(	self,
		run_fun,
		name = `"NoName"`,
		priority = `0`,
		period = `None`,
		profile = `False`,
		trace = `False`,
		shares = `()`
	)

Initialize a task object so it may be run by the scheduler.

This method initializes a task object, saving copies of constructor parameters and preparing an empty dictionary for states.

Parameters

run_fun	The function which implements the task's code. It must be a generator which yields the current state.
name	The name of the task, by default `NoName`. This should be overridden with a more descriptive name by the programmer.
priority	The priority of the task, a positive integer with higher numbers meaning higher priority (default 0)
period	The time in milliseconds between runs of the task if it's run by a timer or `None` if the task is not run by a timer. The time can be given in a `float` or `int`; it will be converted to microseconds for internal use by the scheduler.
profile	Set to `True` to enable run-time profiling
trace	Set to `True` to generate a list of transitions between states. Note: This slows things down and allocates memory.
shares	A list or tuple of shares and queues used by this task. If no list is given, no shares are passed to the task

Member Function Documentation

◆ repr()

def cotask.Task.__repr__ ( self )

This method converts the task to a string for diagnostic use.

It shows information about the task, including execution time profiling results if profiling has been done.

Returns: The string which represents the task

◆ get_trace()

def cotask.Task.get_trace ( self )

This method returns a string containing the task's transition trace.

The trace is a set of tuples, each of which contains a time and the states from and to which the system transitioned.

Returns: A possibly quite large string showing state transitions

◆ go()

def cotask.Task.go ( self )

Method to set a flag so that this task indicates that it's ready to run.

This method may be called from an interrupt service routine or from another task which has data that this task needs to process soon.

◆ ready()

bool cotask.Task.ready ( self )

This method checks if the task is ready to run.

If the task runs on a timer, this method checks what time it is; if not, this method checks the flag which indicates that the task is ready to go. This method may be overridden in descendent classes to implement some other behavior.

◆ reset_profile()

def cotask.Task.reset_profile ( self )

This method resets the variables used for execution time profiling.

This method is also used by init() to create the variables.

◆ schedule()

bool cotask.Task.schedule ( self )

This method is called by the scheduler; it attempts to run this task.

If the task is not yet ready to run, this method returns False immediately; if this task is ready to run, it runs the task's generator up to the next yield() and then returns True.

Returns: True if the task ran or False if it did not

◆ set_period()

def cotask.Task.set_period	(	self,
		new_period
	)

This method sets the period between runs of the task to the given number of milliseconds, or None if the task is triggered by calls to go() rather than time.

Parameters

new_period The new period in milliseconds between task runs

Member Data Documentation

◆ period

cotask.Task.period

The period, in milliseconds, between runs of the task's run() method.

If the period is None, the run() method won't be run on a time basis but will instead be run by the scheduler as soon as feasible after code such as an interrupt handler calls the go() method.

The documentation for this class was generated from the following file:

src/cotask.py

Public Member Functions

Public Attributes