cotask module

! @file cotask.py This file contains classes to run cooperatively scheduled tasks in a multitasking system.

Tasks are created as generators, functions which have infinite loops and call @c yield at least once in the loop. References to all the tasks to be run in the system are kept in a list maintained by class @c CoTaskList; the system scheduler then runs the tasks’ @c run() methods according to a chosen scheduling algorithm such as round-robin or highest-priority-first.

@author JR Ridgely @date 2017-Jan-01 JRR Approximate date of creation of file @date 2021-Dec-18 JRR Docstrings modified to work without DoxyPyPy @copyright This program is copyright (c) 2017-2022 by JR Ridgely and released

under the GNU Public License, version 3.0.

It is intended for educational use only, but its use is not limited thereto. THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS “AS IS” AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

class cotask.Task(run_fun, name='NoName', priority=0, period=None, profile=False, trace=False, shares=())

Bases: object

! 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

@code

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 ()

@endcode

get_trace(): ! 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. @return A possibly quite large string showing state transitions

go(): ! 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: ! 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(): ! This method resets the variables used for execution time profiling. This method is also used by @c __init__() to create the variables.

schedule() → bool

! 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 @c False immediately; if this task is ready to run, it runs the task’s generator up to the next @c yield() and then returns @c True.

@return @c True if the task ran or @c False if it did not

set_period(new_period): ! This method sets the period between runs of the task to the given number of milliseconds, or @c None if the task is triggered by calls to @c go() rather than time. @param new_period The new period in milliseconds between task runs

class cotask.TaskList

Bases: object

! A list of tasks used internally by the task scheduler. This class holds the list of tasks which will be run by the task scheduler. The task list is usually not directly used by the programmer except when tasks are added to it and the scheduler is called. An example showing the use of the task list is given in the last few lines of the documentation for class @c Task.

The task list is sorted by priority so that the scheduler can efficiently look through the list to find the highest priority task which is ready to run at any given time. Tasks can also be scheduled in a simpler “round-robin” fashion.

append(task): ! Append a task to the task list. The list will be sorted by task priorities so that the scheduler can quickly find the highest priority task which is ready to run at any given time. @param task The task to be appended to the list

pri_sched()

! Run tasks according to their priorities.

This scheduler runs tasks in a priority based fashion. Each time it is called, it finds the highest priority task which is ready to run and calls that task’s @c run() method.

rr_sched()

! Run tasks in order, ignoring the tasks’ priorities.

This scheduling method runs tasks in a round-robin fashion. Each time it is called, it goes through the list of tasks and gives each of them a chance to run. Although this scheduler first runs higher priority tasks first, that has no significant effect in the long run, as all the tasks are given a chance to run each time through the list, and it takes about the same amount of time before each is given a chance to run again.