Lesson 8 - Organizing Tasks

Introduction

After the sightseeing tour of our last lesson we come now to the real thing, tools that implement multithreading and help to organize multiple tasks. If you didn't read lesson 7, you need profound knowledge of multithreading. If you are not this expert, please visit lesson 7 and then come back.

The basic idea we build our tools upon are linked lists. We name them tasks. A task is an object, that can be started and then executes a sequence of callables. The final result will be somewhat complex (only inside, its outside API will be very simple). Therefore we start with a basic version and add the functionality step by step.

Combining linked lists with multithreading

Let's look at a first version:

#!/usr/bin/env python3

import threading, typing

STATE_INIT = 'INIT'
STATE_STARTED = 'STARTED'
STATE_FINISHED = 'FINISHED'

def concat(*tasks) -> 'Task':
    chain = None
    for task in tasks:
        assert isinstance(task, Task), 'tasks must be instances of class Task'
        if not chain:
            chain = task
        else:
            chain.append(task)
    return chain

class Task:
    def __init__(
            self,
            action: typing.Callable,
            args: tuple=(),
            kwargs: dict={},
    ):
        assert isinstance(action, typing.Callable), \
            "action needs to be a callable"
        assert isinstance(args, tuple), 'args needs to be a tuple'
        assert isinstance(kwargs, dict), 'kwargs needs to be a dictionary'
        self._action = action
        self._args = args
        self._kwargs = kwargs
        self._next = None
        self._root = self
        # the following are root only attributes
        self._state = STATE_INIT
        self._thread = None
        self._lock = threading.Lock()
        self._last = self

    def append(self, task: 'Task') -> 'Task':
        self._lock.acquire()
        assert isinstance(task, Task), "task needs to be a Task instance"
        assert self._root is self, "only root tasks can be appended"
        assert task._root is task, "both tasks need to be root tasks"
        assert self._state in [STATE_INIT, STATE_FINISHED], "can't append to tasks in state " + self._state
        assert task._state in [STATE_INIT, STATE_FINISHED], "can't append tasks in state " + task._state
        self._last._next = task
        self._last = task._last
        task.root = self
        self._lock.release()
        return self

    def start(self) -> 'Task':
        self._lock.acquire()
        assert self._root is self, "only root tasks can be started"
        assert self._state in [STATE_INIT, STATE_FINISHED], "can't start from state " + self._state
        self._state = STATE_STARTED
        self._thread = threading.Thread(target=self._execute)
        self._thread.start()
        return self

    def join(self) -> None:
        self._thread.join()
        
    def _execute(self) -> None:
        self._root._lock.release()
        self._action(*self._args, **self._kwargs)
        self._root._lock.acquire()
        if self._next:
            self._next._execute()
        else:
            self._root._state = STATE_FINISHED
            self._root._lock.release()
            return

    @property
    def state(self) -> str:
        self._lock.acquire()
        value = self.state_no_lock
        self._lock.release()
        return value
    
    @property
    def state_no_lock(self) -> str:
        assert self._root is self, "only root tasks can be asked about their state"
        return self._state
    
    @property
    def lock(self) -> threading.Lock:
        assert self._root is self, "only root tasks can be asked about their lock"
        return self._lock

    @property
    def root(self):
        return self._root
    @root.setter
    def root(self, task):
        self._root = task
        if self._next:
            self._next.root = task
      

Remarks:

• A task object is a linked list (chain of tasks), that starts with a root task, which is followed by an unlimited number of links. All of them are Task objects.
• All Task objects are created as root tasks. Method append does what its name says, it appends a root task (Task object) to another root task. The result is a chain of tasks, where the appended task is no more a root task.
• Function concat is syntactic sugar, which allows to build a chain of tasks from a tuple of Task objects.
• All comunication from and to the outside world is done via the root task, which represents the whole chain of tasks. The following tasks in the linked list become inaccessible for the outside world.
• Attributes of all tasks:
  • _action: callable object (f.i. a function or a method).
  • _args: argument list of _action.
  • _kwargs: keyword arguments of _action.
  • _next: pointer to the next Task object in the linked list.
  • _root: pointer to the root task (also a Task object).
• Additional attributes of root tasks:
  • _state: actual state of the task (or chain of tasks).
  • _thread: The thread, that does the execution (Thread instance from module threading).
  • _lock: The Lock object (from module threading).
  • _last: pointer to the last task in the chain (method append needs it).
• A root task can be identified by self._root is self.
• Method start creates a new thread, that does the execution. start itself returns immediately. If you need the started thread for your timing, you can call method join (f.i. task.start().join()).
• Method _execute is the heart of the execution. Every task executes its action and then calls method _execute of the next link in the chain.
• Nearly all methods and properties start with aquiring the Lock object. This guaranties an exclusive access and finds an object in a consitent state. At the end of the method (or property logic), the lock is released.
• The lock is released, when _action starts execution and again acquired, after _action is finished. This says another thread gets access to a Task object, when:
  • its state is STATE_INIT (before started),
  • its state is STATE_FINISHED,
  • one of its actions is actually executed.
• All the properties, that read changeable data need a locking mechanism. If the outside world has to read more than one property, it can explicitly lock and then call the no-lock-version of the properties:

  
  ...
  my_task.lock.acquire()
  if my_task.state_no_lock is task.STATE_FINISHED:
      print("finished")
  elif my_task.state_no_lock is task.STATE_STARTED:
      print("started")
  else:
      print("initial")
  my_task.lock.release()
  ...   
     

  If no explicit locking would be done and the state changed just between the two accesses (STARTED -> FINISHED), it printed an incorrect state (INIT). This case is for demonstration only, a better solution would be:

  
  ...
  state = my_task.state
  if state is task.STATE_FINISHED:
      print("finished")
  elif state is task.STATE_STARTED:
      print("started")
  else:
      print("initial")
  ...   
     

• the setter of property root is recursive. If set in a former root task, it sets the new root for the whole chain of tasks.

Task objects versus Thread objects

Actually Task objects are very similar to Thread objects. Both have methods start and join. The main difference is that Task objects allow to build chains and handle the chains like single Task objects.

Parallel executed tasks

As a first test, we run two tasks parallel:

#!/usr/bin/env python3

import ev3, ev3_sound, ev3_vehicle, task

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
vehicle = ev3_vehicle.TwoWheelVehicle(0.02128, 0.1175, ev3_obj=jukebox)

speed = 10
t1 = task.Task(
    vehicle.drive_turn,
    args=(speed, 0, 720)
)
t2 = task.Task(
    jukebox.play_song,
    args=(ev3_sound.HAPPY_BIRTHDAY,)
)
t1.start()
t2.start()
t1.join()
t2.join()
vehicle.stop()
      

The vehicle turns two circles on place and plays Happy Birthday. The longer of both tasks does the timing.

If we use Thread objects instead of Task objects, this reads as:

#!/usr/bin/env python3

import ev3, ev3_sound, ev3_vehicle, threading

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
vehicle = ev3_vehicle.TwoWheelVehicle(0.02128, 0.1175, ev3_obj=jukebox)

speed = 10
t1 = threading.Thread(
    target=vehicle.drive_turn,
    args=(speed, 0, 720)
)
t2 = threading.Thread(
    target=jukebox.play_song,
    args=(ev3_sound.HAPPY_BIRTHDAY,)
)
t1.start()
t2.start()
t1.join()
t2.join()
vehicle.stop()
    

There is no real difference, the handling of both alternatives is very similar.

Chains of tasks

We run a chain of tasks:

#!/usr/bin/env python3

import ev3, ev3_sound, ev3_vehicle, task

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
vehicle = ev3_vehicle.TwoWheelVehicle(0.02128, 0.1175, ev3_obj=jukebox)

speed = 10
task.concat(
    task.Task(
        vehicle.drive_turn,
        args=(speed, 0, 360)
    ),
    task.Task(
        vehicle.drive_turn,
        args=(speed, 0, 360),
        kwargs={"right_turn": True}
    ),
    task.Task(
        vehicle.stop
    )
).start()
jukebox.play_song(ev3_sound.HAPPY_BIRTHDAY)
      

The movement of the vehicle is a chain of three links, driving and music run parallel.

We use a Thread object instead of Task objects:

#!/usr/bin/env python3

import ev3, ev3_sound, ev3_vehicle, threading

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
vehicle = ev3_vehicle.TwoWheelVehicle(0.02128, 0.1175, ev3_obj=jukebox)

speed = 10

def do_it():
    vehicle.drive_turn(speed, 0, 360)
    vehicle.drive_turn(speed, 0, 360, right_turn=True)
    vehicle.stop()
    
threading.Thread(
    target=do_it
).start()
jukebox.play_song(ev3_sound.HAPPY_BIRTHDAY)
      

O.k. it works, but in its actual state it is no real progress to use Task objects instead of Thread objects. It needs additional functionality to make it a valuable tool.

Repeated actions

Let's look at some scenarios and prove if our class Task is able to execute them. Here they are:

• Every 0.2 sec. your EV3 device reads the actual free distance from its infrared sensor. This is repeated 100 times. The times and data are written to a file.
• Your application drives a vehicle: Every 0.5 sec. it tests if there is no barrier in front of the vehicle, if the distance is smaller than 30 cm, it stops the motors.
• Your application watches a door: Every 5 sec. it tests, if the touch sensor is touched, if not, it plays some sound signal.
• Your application is a vehicle, that follows a light. Every 20 sec. its light sensor scans all directions for the brightest source of light. Then the orientation of the vehicle is changed to this direction.

All these scenarios don't fit our tool. We need some modifications to deal with repeated actions. Hopefully you remember class TwoWheelVehicle of lesson 6. We had a very similar situation, where functions _test_pos and _test_o returned numbers:

• value == -1: signals the caller to finish the loop.
• value > 0: value is the time to wait until the next call (in sec.).
• value == 0: call again without any waiting.

We need some new constants and we import modules time and numbers:

#!/usr/bin/env python3

import threading, typing, time, numbers

...

ACTIVITY_NONE = 'NONE'
ACTIVITY_BUSY = 'BUSY'
ACTIVITY_SLEEP = 'SLEEP'
  

We change the constructor of class Task:

    def __init__(
            self,
            action: typing.Callable,
            args: tuple=(),
            kwargs: dict={},
    ):
        ...
        self._num = 0
        self._cnt = 0
        # the following are root only attributes
        ...
        self._cond = threading.Condition(self._lock)
        self._activity = ACTIVITY_NONE
        self._time_action = None
      

The meaning of the new attributes:

• _num: maximum calls number of _action (value 0 stands for unlimited).
• _cnt: counter of _action calls.
• _cond: used for interruptable sleeping.
• _activity: actual activity type (ACTIVITY_NONE, ACTIVITY_BUSY or ACTIVITY_SLEEP).
• _time_action: time of the actual call of _action (when _activity is ACTIVITY_BUSY) or the next call of _action (when _activity is ACTIVITY_SLEEP).

We add two lines of code to method start:

    def start(self) -> None:
        ...
        self._cnt = 0
        self._time_action = time.time()
        self._thread = threading.Thread(target=self._execute)
        self._thread.start()
      

The new version of method _execute:

    def _execute(self) -> None:
        while True:
            gap = self._wrapper()
            self._cnt += 1
            if gap == -1 or self._num > 0 and self._cnt >= self._num:
                self._root._time_action = time.time()
                break
            if gap == 0:
                self._root._time_action = time.time()
                continue
            self._root._time_action += gap
            real_gap = self._root._time_action - time.time()
            if real_gap > 0:
                self._root._activity = ACTIVITY_SLEEP
                self._root._cond.wait(real_gap)
                self._root._activity = ACTIVITY_NONE
        self._root._time_action = time.time()
        gap = 0
        if gap > 0:
            self._root._activity = ACTIVITY_SLEEP
            self._root._cond.wait(gap)
            self._root._activity = ACTIVITY_NONE
        if self._next:
            self._next._cnt = 0
            self._next._execute()
        else:
            self._final()
      

with new methods, that wrapp _action:

    def _wrapper(self) -> int:
        self._wrapper1()
        self._action(*self._args, **self._kwargs)
        self._wrapper2()
        return -1
    
    def _wrapper1(self):
        self._root._activity = ACTIVITY_BUSY
        self._root._lock.release()

    def _wrapper2(self):
        self._root._lock.acquire()
        self._root._activity = ACTIVITY_NONE
      

returning value -1 guaranties that the loop is executed once and no sleeping will happen. This sounds funny, we code some new logic which is not used. The explanation is: Task objects will never use it, but their subclasses! It allows to subclass Task with unchanged method _execute but new method _wrapper.

We add a new method, that prepares a consistent combination of attributes in case of returning:

    def _final(self) -> None:
        self._root._state = STATE_FINISHED:
        self._root._time_action = None
        self._root._lock.release()
      

We add some properties:

    @property
    def time_action(self) -> float:
        self._root._lock.acquire()
        assert self._root is self, 'only root tasks can be asked about their time_action'
        value = self.time_action_no_lock
        self._root._lock.release()
        return value

    @property
    def time_action_no_lock(self) -> float:
        return self._time_action

    @property
    def activity(self) -> str:
        self._root._lock.acquire()
        assert self._root is self, 'only root tasks can be asked about their activity'
        value = self.activity_no_lock
        self._root._lock.release()
        return value

    @property
    def activity_no_lock(self) -> str:
        return self._activity
      

Class Repeated

We subclass Task:

class Repeated(Task):
    def __init__(
            self,
            action: typing.Callable,
            args: tuple=(),
            kwargs: dict={},
            num: int=0
    ):
        super().__init__(action, args, kwargs)
        assert isinstance(num, int), 'num must be an integer'
        assert num >= 0, 'num must be positive'
        self._num = num

    def _wrapper(self):
        self._wrapper1()
        value = self._action(*self._args, **self._kwargs)
        assert isinstance(value, Task) or \
            isinstance(value, numbers.Number) or \
            isinstance(value, bool) or \
            value is None, \
            'action needs to return a task, a number, a boolean or None'
        assert not isinstance(value, numbers.Number) or \
            value == -1 or \
            value >= 0, \
            'if action return a number, it must be positive or -1'
        if value is True:
            rc = -1
        elif  isinstance(value, Task) or value is False or value is None:
            rc = 0
        else:
            rc = value
        self._wrapper2()
        return rc
      

It's the callable _action that controls the execution by its return value. There are three types of return values (None, bool, int):

• None: There is no waiting and the number of calls of _action must be limited by num.
• bool: There is no waiting, the calling ends, when value True is returned.
• int:
  • -1: no further call (of _action).
  • 0: no waiting, call again as soon as possible.
  • > 0: wait, then call again.

num is an alternative to control the execution. If set, it is an upper limit for the number of calls of _action. It must be set when the Repeated object is constructed.

Tests

We play the trias three times:

#!/usr/bin/env python3

import ev3, ev3_sound, task

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
task.Repeated(
    jukebox.play_song,
    args=(ev3_sound.TRIAS,),
    num=3
).start()
      

Task factories

I feel confident in the task concept and try to organize as much actions as possible in tasks. The best way to this target is implementing tasks into the subclasses of class EV3.

Method song of class Jukebox

We start with class Jukebox and add a method song:

    def song(self, song: dict) -> task.Task:
        return task.concat(
            task.Task(
                self._init_tone
            ),
            task.Repeated(
                self._next_tone,
                args=(song,)
            ),
            task.Task(
                self.stop
            )                
        )
      

This method returns a Task object that plays the song, when started. It is a chain of tasks with three links, but this is an internal information. The ouside world knows: it's a task, which can be concatenated or started. We test it with this program:

#!/usr/bin/env python3

import ev3, ev3_sound, task

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
jukebox.song(ev3_sound.HAPPY_BIRTHDAY).start()
      

We miss the lightshow, which will be added soon. Everything seems to work. If you look at the above described scenarios, we already have the tools to realize them. Periodic, our next subclass of Task will be syntactic sugar. It prevents from writing wrappers around callables.

Periodic actions

All of our scenarios knew the time distance between two calls of _action from the beginning. This allows to give them as sttributes into the constructor of the objects. It does not help for transparency, when the time distance is returned by method _action. Our new subclass Periodic will better fit this situation:

class Periodic(Task):
    def __init__(
            self,
            intervall: float,
            action: typing.Callable,
            args: tuple=(),
            kwargs: dict={},
            num: int=0
    ):
        super().__init__(action, args, kwargs)
        assert isinstance(intervall, numbers.Number), 'intervall must be a number'
        assert intervall >= 0, 'intervall must be positive'
        assert isinstance(num, int), 'num must be an integer'
        assert num >= 0, 'num must be positive'
        self._intervall = intervall
        self._num = num

    def _wrapper(self):
        self._wrapper1()
        value = self._action(*self._args, **self._kwargs)
        assert isinstance(value, Task) or isinstance(value, bool) or value is None, \
            'action needs to return a task, a boolean or None'
        if value is True:
            rc = -1
        else:
            rc = self._intervall
        self._wrapper2()
        return rc
      

The waiting time is set, when the constructor is called (parameter intervall). I think, a good and descriptive variant. Look at method _wrapper, the callable _action may return two types of values (None, bool):

• None: You have to limit the number of calls of _action by setting num.
• bool: The calling ends, when value True is returned.

Test

We play the trias three times, once per minute:

#!/usr/bin/env python3

import ev3, ev3_sound, task

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
task.Periodic(
    60,
    jukebox.play_song,
    args=(ev3_sound.TRIAS,),
    num=3
).start()
      

Exact timing

For the moment, exact timing works for repetitions but not for chains of tasks. We want:

• the possibility to switch to netto-time (that says the waiting for the next call of _action does not depend from the duration of its last execution),
• a chance to set a fixed duration for a task.

This says we have to add some functionality. The new constructor:

    def __init__(
            self,
            action: typing.Callable,
            args: tuple=(),
            kwargs: dict={},
            duration: float=None
    ):
        ...
        assert duration is None or isinstance(duration, numbers.Number), \
            'duration needs to be a number'
        assert duration is None or duration >= 0, \
            'duration needs to be positive'
        ...
        self._duration = duration
        self._time_end = None
        self._netto_time = False
        # the following are root only attributes
        ...
      

The meanings of the new attributes:

• _duration: duration for this task. If the task finishes later, the following tasks have to compensate.
• _time_end: holds the time, when this task has to be finished (is set when executed).
• _netto_time: flag, that waiting is netto (execution of action counts extra)

We change method start:

    def start(self) -> None:
        self._lock.acquire()
        assert self._root is self, "only root tasks can be started"
        assert self._state in [STATE_INIT, STATE_FINISHED], \
            "can't start from state " + self._state
        self._state = STATE_STARTED
        self._cnt = 0
        self._time_action = time.time()
        if self._duration != None:
            self._time_end = self._time_action + self._duration
        self._thread = threading.Thread(target=self._execute)
        self._thread.start()
      

The updated method _execute:

    def _execute(self) -> None:
        while True:
            gap = self._wrapper()
            self._cnt += 1
            if gap == -1 or self._num > 0 and self._cnt >= self._num:
                self._root._time_action = time.time()
                break
            if gap == 0:
                self._root._time_action = time.time()
                continue
            if self._netto_time:
                self._root._time_action = time.time() + gap
                real_gap = gap
            else:
                self._root._time_action += gap
                real_gap = self._root._time_action - time.time()
            if real_gap > 0:
                self._root._activity = ACTIVITY_SLEEP
                self._root._cond.wait(real_gap)
                self._root._activity = ACTIVITY_NONE
        if self._time_end:
            self._root._time_action = self._time_end
            self._time_end = None
            gap = self._root._time_action - time.time()
            if gap > 0:
                self._root._activity = ACTIVITY_SLEEP
                self._root._cond.wait(gap)
                self._root._activity = ACTIVITY_NONE
        else:
            self._root._time_action = time.time()
        if self._next:
            self._next._cnt = 0
            if self._next._duration != None:
                self._next._time_end = self._root._time_action + self._next._duration
            self._next._execute()
        else:
            self._final()
      

There is an additional call of method sleep at the end of the task. This call guaranties, that the task lasts at least the duration, which was set by the constructor.

_netto_time is relevant for the classes Repeated and Periodic only. We need to change their constructors. The new constructor of class Repeated:

    def __init__(
            self,
            action: typing.Callable,
            args: tuple=(),
            kwargs: dict={},
            num: int=0,
            duration: float=0,
            netto_time: bool=False
    ):
        super().__init__(action, args, kwargs, duration=duration)
        assert isinstance(num, int), 'num must be an integer'
        assert num >= 0, 'num must be positive'
        assert isinstance(netto_time, bool), 'netto_time must be a bool value'
        self._num = num
        self._netto_time = netto_time
      

The new constructor of class Periodic:

    def __init__(
            self,
            intervall: numbers.Number,
            action: typing.Callable,
            args: tuple=(),
            kwargs: dict={},
            num: int=0,
            duration: float=0,
            netto_time: bool=False
    ):
        super().__init__(action, args, kwargs, duration=duration)
        assert isinstance(intervall, numbers.Number), 'intervall must be a number'
        assert intervall >= 0, 'intervall must be positive'
        assert isinstance(num, int), 'num must be an integer'
        assert num >= 0, 'num must be positive'
        assert isinstance(netto_time, bool), 'netto_time must be a bool value'
        self._intervall = intervall
        self._num = num
        self._netto_time = netto_time
      

Subclass Sleep

This allows to create another subclass:

class Sleep(Task):
    def __init__(self, seconds: float):
        super().__init__(self._do_nothing, duration=seconds)

    def _do_nothing(self): return -1
      

Class Sleep will allow to be stopped, but this is the future. For the moment it is just an alternative for:

Task(time.sleep, args=(seconds,))
      

which is able to compensate the time deviations of previous tasks.

Tests

We test the compensation:

#!/usr/bin/env python3

import ev3, ev3_sound, task, time

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
jukebox.verbosity = 1

task.concat(
    task.Task(
        time.sleep,
        args=(1,),
        duration=0.5
    ),
    task.Periodic(
        1,
        jukebox.play_tone,
        args=("a'", 0.1),
        num=4
    )
).start()
      

Sleeping for one second with a duration of 0.5 sec.! The chain of tasks will be half a sec. late, when it has finished its first task (sleep). The second task (play_tone) has to compensate this delay. Its output:

10:27:54.356145 Sent 0x|0D:00|2A:00|80|00:00|94:01:01:82:B8:01:81:64|
10:27:54.855296 Sent 0x|0D:00|2B:00|80|00:00|94:01:01:82:B8:01:81:64|
10:27:55.854861 Sent 0x|0D:00|2C:00|80|00:00|94:01:01:82:B8:01:81:64|
10:27:56.854826 Sent 0x|0D:00|2D:00|80|00:00|94:01:01:82:B8:01:81:64|
      

Between the first call of play_tone and the second we see a time distance of half a sec. This compensates the delay. From then on the time distance is one sec. We recognize, that the usage of duration allows exact timing and also can prevent exact timing.

We test class Sleep with a program, that uses tasks for a little lightshow:

#!/usr/bin/env python3

import task, ev3, ev3_sound

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
jukebox.verbosity = 1

t_green = task.Periodic(2, jukebox.change_color, args=(ev3.LED_GREEN,), num=4)
t_red = task.Periodic(4, jukebox.change_color, args=(ev3.LED_RED,), num=2)
t_orange = task.Periodic(4, jukebox.change_color, args=(ev3.LED_ORANGE,), num=2)

t = task.concat(
    task.Task(t_red.start),
    task.Sleep(1),
    task.Task(t_green.start),
    task.Sleep(1),
    task.Task(t_orange.start),
    task.Task(t_green.join)
)
t.start().join()
print("done")
      

Here we have four tasks, t_green, t_red, t_orange and t. t is the master, that schedules the starting of the three other tasks. When t_orange is started, t joins the task of t_green. We want its thread to be joinable (for timing purpose, the join has to end when the lightshow ends). Its output:

11:26:32.103894 Sent 0x|08:00|2A:00|80|00:00|82:1B:02|
11:26:33.105693 Sent 0x|08:00|2B:00|80|00:00|82:1B:01|
11:26:34.107247 Sent 0x|08:00|2C:00|80|00:00|82:1B:03|
11:26:35.106055 Sent 0x|08:00|2D:00|80|00:00|82:1B:01|
11:26:36.104310 Sent 0x|08:00|2E:00|80|00:00|82:1B:02|
11:26:37.106052 Sent 0x|08:00|2F:00|80|00:00|82:1B:01|
11:26:38.107876 Sent 0x|08:00|30:00|80|00:00|82:1B:03|
11:26:39.106062 Sent 0x|08:00|31:00|80|00:00|82:1B:01|
done
      

It works and we learned, that tasks can start other tasks. We plan to make our tasks stoppable, which says that tasks inside of tasks (contained tasks) need to be stopped too. The conclusion is, we have to change class Task once more.

Contained tasks

We add a class attribute _contained_register to class Task:

    class Task:
        _contained_register = {}
      

Remarks:

• This will hold all contained tasks with their parent tasks. It's a class attribute that exists once and can be accessed from all instances of class Task. It allows to ask if task1 is cild of (contained in) task2:

  
      if task1 in task.Task._contained_register and task.Task._contained_register[task1] is task2:
          print(task1, "is child of", task2)
     

• The access is from the child to the parent. We delete an entry, when a task is finished:

  
      self._root._state = STATE_FINISHED
      if self._root in self._contained_register:
          self._contained_register.pop(self._root)
     

We add some code to the constructor:

class Task:
    def __init__(
            self,
            action: typing.Callable,
            args: tuple=(),
            kwargs: dict={},
            join: bool=False,
            duration: float=None
    ):
        ...
        assert isinstance (join, bool), 'join needs to be a bool value'
        assert not join or hasattr(action, '__self__'), 'only bounded methods can be joined'
        assert not join or isinstance(action.__self__, Task), 'only instances of Task can be joined'
        assert not join or action.__name__ == "start", 'only method start can be joined'
        ...
        self._join = join
        ...
        # the following are root only attributes
        ...
        self._contained = []
      

The meanings of the new attributes:

• _join: A flag, that signals if the contained task will be joined when starting it. This says the starting task will wait until the started task is finished.
• _contained: Holds the tasks, which where started as contained (cild) tasks and may still run. In case of stopping the task, these have to be stopped too. This is the opposite direction than _contained_register: from the parent to the child. Its a list because a task may have multiple contained tasks.

Methods _wrapper1 and _wrapper2 also need to be changed:

    def _wrapper1(self) -> None:
        if hasattr(self._action, '__self__') and \
           isinstance(self._action.__self__, Task) and \
           self._action.__name__ in ["start", "join"]:
            task = self._action.__self__
            name = self._action.__name__
            if name == "start":
                if not task in self._root._contained:
                    self._root._contained.append(task)
                self._contained_register.update({task: self._root})
        if not hasattr(self._action, '__self__') or \
           not isinstance(self._action.__self__, Task) or \
           not self._action.__name__ == "start" or \
           self._action.__name__ == "start" and self._join:
            self._root._activity = ACTIVITY_BUSY
            self._root._lock.release()

    def _wrapper2(self) -> None:
        if self._join:
            self._action.__self__._thread.join()
        if not hasattr(self._action, '__self__') or \
           not isinstance(self._action.__self__, Task) or \
           not self._action.__name__ == "start" or \
           self._action.__name__ == "start" and self._join:
            self._root._lock.acquire()
            self._root._activity = ACTIVITY_NONE
        if hasattr(self._action, '__self__') and \
           isinstance(self._action.__self__, Task) and \
           self._action.__name__ in ["start", "join"]:
            task = self._action.__self__
            state = task.state
            if state == STATE_FINISHED and \
               task in self._root._contained:
                self._root._contained.remove(task)
      

This adds all started tasks to the root tasks list _contained and guaranties that the root task knows all contained tasks which run parallel. A few remarks:

• Attribute __self__ of a method (bounded callable) holds the object, the method belongs to.
• Attribute __name__ of a callable holds its name.
• These two attributes help to identify contained tasks (_action calls method start of a Task instance):

  
          if hasattr(self._action, '__self__') and \
             isinstance(self._action.__self__, Task) and \
             self._action.__name__ == "start":
     

• Starting a task is not time consuming, therefore we do not release the lock. Only if we join the started task.

Method _final needs to be changed:

    def _final(self) -> None:
        self._root._contained = self._join_contained()
        self._root._state = STATE_FINISHED
        if self._root in self._contained_register:
            self._contained_register.pop(self._root)
        self._root._time_action = None
        self._root._lock.release()
      

with a new method:

    def _join_contained(self) -> list:
        contained = self._root._contained
        self._root._activity = ACTIVITY_JOIN
        self._root._lock.release()
        not_finished = []
        for task in contained:
            if not task in self._contained_register or \
               not self._contained_register[task] is self._root:
                continue
            task.join()
            if task.state != STATE_FINISHED:
                not_finished.append(task)
        self._root._lock.acquire()
        self._root._activity = ACTIVITY_NONE
        return not_finished
      

The call of method _join_contained guaranties, that a task never is finished, before all its contained tasks are finished. We add a new activity:

ACTIVITY_NONE = 'NONE'
ACTIVITY_BUSY = 'BUSY'
ACTIVITY_SLEEP = 'SLEEP'
ACTIVITY_JOIN = 'JOIN'
      

We change property time_action_no_lock:

    @property
    def time_action_no_lock(self) -> float:
        min = None
        if not hasattr(self._action, '__self__') or \
           not isinstance(self._action.__self__, Task) or \
           not self._action.__name__ in ["start", "join"]:
            min = self._time_action
        for task in self._contained:
            act = task.time_action
            if min is None:
                min = act
            elif act != None and act < min:
                min = act
        return min
      

The earliest time of any action is returned, may it be of the tasks own action or one of its direct or indirect contained tasks (recursion). Starting or joining a task is no action, we ignore them. What counts is the next action of the contained task, which is considered by recursion.

I think, it is not a good idea, when instances of Repeated or Periodic start unjoined tasks. Therefore we don't add a parameter join to their constructors. This says, they can contain tasks but these are always joined tasks. The constructor of Repeated:

    def __init__(
            self,
            action: typing.Callable,
            args: tuple=(),
            kwargs: dict={},
            num: int=0,
            duration: float=0,
            netto_time: bool=False
    ):
        if hasattr(action, '__self__') and \
           isinstance(action.__self__, Task) and \
           action.__name__ == "start":
            super().__init__(action, args, kwargs, join=True, duration=duration)
        else:
            super().__init__(action, args, kwargs, duration=duration)
        ...
      

The constructor of Periodic:

    def __init__(
            self,
            intervall: float,
            action: typing.Callable,
            args: tuple=(),
            kwargs: dict={},
            num: int=0,
            duration: float=0,
            netto_time: bool=False
    ):
        if hasattr(action, '__self__') and \
           isinstance(action.__self__, Task) and \
           action.__name__ == "start":
            super().__init__(action, args, kwargs, join=True, duration=duration)
        else:
            super().__init__(action, args, kwargs, duration=duration)
        ...
      

Testing contained classes

We change our lightshow-program:

#!/usr/bin/env python3

import task, ev3, ev3_sound

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
jukebox.verbosity = 1

t_green = task.Periodic(2, jukebox.change_color, args=(ev3.LED_GREEN,), num=4)
t_red = task.Periodic(4, jukebox.change_color, args=(ev3.LED_RED,), num=2)
t_orange = task.Periodic(4, jukebox.change_color, args=(ev3.LED_ORANGE,), num=2)

task.concat(
    task.Task(t_red.start),
    task.Sleep(1),
    task.Task(t_green.start),
    task.Sleep(1),
    task.Task(t_orange.start)
).start().join()
print("done")
      

There is no need for joining t_green any more, this is done inside. This programs output:

12:39:08.368138 Sent 0x|08:00|2A:00|80|00:00|82:1B:02|
12:39:09.370120 Sent 0x|08:00|2B:00|80|00:00|82:1B:01|
12:39:10.373537 Sent 0x|08:00|2C:00|80|00:00|82:1B:03|
12:39:11.371645 Sent 0x|08:00|2D:00|80|00:00|82:1B:01|
12:39:12.368623 Sent 0x|08:00|2E:00|80|00:00|82:1B:02|
12:39:13.370486 Sent 0x|08:00|2F:00|80|00:00|82:1B:01|
12:39:14.374323 Sent 0x|08:00|30:00|80|00:00|82:1B:03|
12:39:15.370430 Sent 0x|08:00|31:00|80|00:00|82:1B:01|
done
      

Modifying the task factory

We add another method to class Jukebox, which also is a task factory.

Method sound

Method sound plays a sound file. It's API:

     |  sound(self, path:str, duration:float=None, repeat:bool=False) -> task.Task
     |      returns a Task object, that plays a sound file
     |      
     |      Attributes:
     |      path: name of the sound file (without extension ".rsf")
     |      
     |      Keyword Attributes:
     |      duration: duration of the sound file (in sec.)
     |      repeat: flag, if repeatedly playing
      

The code:

    def sound(self, path: str, duration: float=None, repeat: bool=False) -> task.Task:
        if repeat:
            ops = b''.join([
                ev3.opSound,
                ev3.REPEAT,
                ev3.LCX(self._volume), # VOLUME
                ev3.LCS(path)          # NAME
            ])
        else:
            ops = b''.join([
                ev3.opSound,
                ev3.PLAY,
                ev3.LCX(self._volume), # VOLUME
                ev3.LCS(path)          # NAME
            ])
        if not duration:
            return task.Task(
                self.send_direct_cmd,
                args=(ops,)
            )
        elif not repeat:
            return task.Task(
                self.send_direct_cmd,
                args=(ops,),
                duration=duration
            )
        else:
            return task.concat(
                task.Task(
                    self.send_direct_cmd,
                    args=(ops,),
                    duration=duration
                ),
                task.Task(self.stop)
            )
      

We test it with this program:

#!/usr/bin/env python3

import ev3, ev3_sound

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
jukebox.verbosity = 1

t = jukebox.sound('./ui/DownloadSucces', duration=5, repeat=True).start()
      

It's output:

12:43:28.528101 Sent 0x|1D:00|2A:00|80|00:00|94:03:01:84:2E:2F:75:69:2F:44:6F:77:6E:6C:6F:61:64:53:75:63:63:65:73:00|
12:43:33.528569 Sent 0x|07:00|2B:00|80|00:00|94:00|
      

A second test where we wrap a Repeated around the Task:

#!/usr/bin/env python3

import ev3, ev3_sound, task

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
jukebox.verbosity = 1

t = task.Repeated(
    jukebox.sound('./ui/DownloadSucces', duration=2).start,
    num=3
).start()
      

The output:

12:45:46.594595 Sent 0x|1D:00|2A:00|80|00:00|94:02:01:84:2E:2F:75:69:2F:44:6F:77:6E:6C:6F:61:64:53:75:63:63:65:73:00|
12:45:48.612966 Sent 0x|1D:00|2B:00|80|00:00|94:02:01:84:2E:2F:75:69:2F:44:6F:77:6E:6C:6F:61:64:53:75:63:63:65:73:00|
12:45:50.614405 Sent 0x|1D:00|2C:00|80|00:00|94:02:01:84:2E:2F:75:69:2F:44:6F:77:6E:6C:6F:61:64:53:75:63:63:65:73:00|
      

Method song

Now we add the lightshow to method song. First we add an attribute _pos_led to the constructor of class Jukebox:

class Jukebox(ev3.EV3):
    def __init__(self, protocol: str=None, host: str=None, ev3_obj=None):
        super().__init__(protocol=protocol, host=host, ev3_obj=ev3_obj)
        self._volume = 1
        self._temperament = 440
        self._pos_tone = None
        self._pos_led = None
        self._plays = False
      

Then we add the methods _init_color and _next_color to class Jukebox:

    def _init_color(self) -> None:
        self._pos_led = 0

    def _next_color(self, song) -> bool:
        if not self._plays:
            return True
        self.change_color(song["led_sequence"][self._pos_led])
        self._pos_led += 1
        self._pos_led %= len(song["led_sequence"])
      

We modify method stop:

    def stop(self) -> None:
        self.send_direct_cmd(ev3.opSound + ev3.BREAK)
        if self._plays:
            self._plays = False
            self.change_color(ev3.LED_GREEN)
      

We modify method song:

    def song(self, song: dict) -> task.Task:
        tones = task.concat(
            task.Task(self._init_tone),
            task.Repeated(
                self._next_tone,
                args=(song,)
            ),
            task.Task(self.stop)
        )
        colors = task.Periodic(
            60 * song["beats_per_bar"] / song["tempo"],
            self._next_color,
            args=(song,)
        )
        if "upbeat" in song:
            colors = task.concat(
                task.Sleep(60 * song["upbeat"] / song["tempo"]),
                colors
            )
        colors = task.concat(
            task.Task(self._init_color),
            colors
        )
        return task.concat(
            task.Task(tones.start),
            task.Task(colors.start),
            task.Task(tones.join)
        )
      

There is a little trick in this code. Up to now, we can't stop a Task object. Here we need to stop task colors when task tones is finished. This is done by method stop, which sets attribute _plays = False. Attribute _plays signals method _next_color to return value True. This isn't good style and we will simplify the code, when our tasks can be stopped. For the moment we are happy with the solution and we test it:

#!/usr/bin/env python3

import task, ev3, ev3_sound_tmp as ev3_sound

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
jukebox.verbosity = 1
task.concat(
    jukebox.song(ev3_sound.HAPPY_BIRTHDAY),
    task.Sleep(2),
    jukebox.song(ev3_sound.TRIAS)
).start()
      

Last Test

As the last test of this lesson, we pack a Periodic into a Periodic into a Periodic:

#!/usr/bin/env python3

import task, ev3, ev3_sound

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
jukebox.verbosity = 1

t_inner = task.Periodic(0.25, jukebox.play_tone, args=("c", 0.1), duration=1, num=4)
t_middle = task.Periodic(2, t_inner.start, num=2)
t_outer = task.Periodic(2, t_middle.start, num=2, netto_time=True)
t_outer.start()
      

its output:

12:54:55.427018 Sent 0x|0D:00|2A:00|80|00:00|94:01:01:82:83:00:81:64|
12:54:55.678632 Sent 0x|0D:00|2B:00|80|00:00|94:01:01:82:83:00:81:64|
12:54:55.928161 Sent 0x|0D:00|2C:00|80|00:00|94:01:01:82:83:00:81:64|
12:54:56.177827 Sent 0x|0D:00|2D:00|80|00:00|94:01:01:82:83:00:81:64|
12:54:57.427841 Sent 0x|0D:00|2E:00|80|00:00|94:01:01:82:83:00:81:64|
12:54:57.678454 Sent 0x|0D:00|2F:00|80|00:00|94:01:01:82:83:00:81:64|
12:54:57.928489 Sent 0x|0D:00|30:00|80|00:00|94:01:01:82:83:00:81:64|
12:54:58.178475 Sent 0x|0D:00|31:00|80|00:00|94:01:01:82:83:00:81:64|
12:55:00.430093 Sent 0x|0D:00|32:00|80|00:00|94:01:01:82:83:00:81:64|
12:55:00.680987 Sent 0x|0D:00|33:00|80|00:00|94:01:01:82:83:00:81:64|
12:55:00.930972 Sent 0x|0D:00|34:00|80|00:00|94:01:01:82:83:00:81:64|
12:55:01.181053 Sent 0x|0D:00|35:00|80|00:00|94:01:01:82:83:00:81:64|
12:55:02.430725 Sent 0x|0D:00|36:00|80|00:00|94:01:01:82:83:00:81:64|
12:55:02.681313 Sent 0x|0D:00|37:00|80|00:00|94:01:01:82:83:00:81:64|
12:55:02.931478 Sent 0x|0D:00|38:00|80|00:00|94:01:01:82:83:00:81:64|
12:55:03.181468 Sent 0x|0D:00|39:00|80|00:00|94:01:01:82:83:00:81:64|
      

Great, it works! Maybe you get the feeling, that using tasks will result in perls motto There's more than one way to do it. Yes I think so and I'm a bit concerned about it. But for the moment I see this freedom as a chance.

At the beginning of this lesson, we compared class Task and class Thread. We did not see any real advantage of class Task. Then we developed it further and it became a usefull tool to organize tasks.

Conclusion

We coded a family of task classes with the following relatives:

• Task: encapsulates a callable, which will be executed once.

  
      class Task(builtins.object)
       |  Uses multithreading for tasks or chains of tasks.
       |  In standard case it's an action, which is executed by a single callable.
       |  Subsequent tasks or chains of tasks can be added with method append().
       |  
       |  Methods defined here:
       |  
       |  __init__(self, action, args:tuple=(), kwargs:dict={}, duration:float=None, join:bool=False)
       |      action: callable object (f.i. a function)
       |      args: argument list of action
       |      kwargs: keyword arguments of action
       |      duration: duration of task (if action returns earlier, task will wait)
       |      join: flag if contained task will be joined
     

• Repeated: encapsulates a callable that will run multiple times.

  
      class Repeated(Task)
       |  Organizes repeated actions with multithreading (control comes back immediately).
       |  think of task as:
       |      while True:
       |          gap = action(*args, **kwargs)
       |          if gap is False or gap is None:
       |              pass
       |          elif gap is True or gap == -1:
       |              break
       |          else:
       |              time.sleep(gap)
       |  
       |  Methods defined here:
       |  
       |  __init__(self, action, args:tuple=(), kwargs:dict={}, num:int=0, duration:float=0, netto_time:bool=False)
       |      action: callable object, which is repeatedly called (f.i. a function)
       |                If callable it must return a number, a bool or None:
       |                True, -1: end the loop
       |                False, None: next call directly follows (if not reached limit of num)
       |                positive number: time gap between the actual and the next call
       |      args: argument list of action
       |      kwargs: keyword arguments of action
       |      num: number of calls (0 stands for unlimited)
       |      duration: duration of task (if execution ends earlier, task will wait)
       |      netto_time: flag, that waiting is netto (execution of action counts extra)
     

• Periodic: encapsulates a callable to run it multiple times with a fixed time-intervall.

  
      class Periodic(Task)
       |  Uses multithreading for periodic actions (control comes back immediately).
       |  think of task as:
       |      while not action(*args, **kwargs):
       |          time.sleep(intervall)
       |  
       |  Methods defined here:
       |  
       |  __init__(self, intervall:float, action, args:tuple=(), kwargs:dict={}, num:int=0, duration:float=0, netto_time:bool=False)
       |      intervall: intervall between two calls of action (in seconds)
       |      action: callable object, which is repeatedly called (f.i. a function)
       |          It returns a bool or None:
       |              True: end the loop
       |              False, None: next call will follow (if not reached limit of num)
       |      args: argument list of action
       |      kwargs: keyword arguments of action
       |      num: number of calls (0 stands for unlimited)
       |      duration: duration of task (if execution ends earlier, task will wait)
       |      netto_time: flag, that waiting is netto (execution of action counts extra)
     

• Sleep: sleeps

  
      class Sleep(Task)
       |  Sleeps
       |  
       |  Methods defined here:
       |  
       |  __init__(self, seconds:float)
       |      seconds: duration of sleeping
     

All of them are instances of class Task and can be combined (in any order) to build chains of tasks. Behind the scenes, they use multithreading, which allows starting parallel tasks (even inside a task one can start parallel tasks). All tasks can be parametrized for exact timing. This helps for a high level of time control. Locking also is done in the background. Only for seldom cases one needs to think about the locking mechanism.

task objects have the following methods:

     |  append(self, task) -> 'Task'
     |      appends a task or a chain of tasks (both must be root tasks)
     |  
     |  join(self) -> None
     |      joins the thread of the task 
     |      think as: my_task.thread.join(), but evaluated when called
     |  
     |  start(self) -> 'Task'
     |      starts execution of task (finished tasks may be started again)
      

Tasks have an easy to handle API and hide the details of multithreading and locking but they are flexible to use. They are independent from the EV3 device and can be used for multiple kinds of software projects, where multithreading is needed.

Tasks are both, architecture and glue. They allow to code callable atoms and then put them together to tasks with complex functionality. The construction of the task from its atoms often needs the thinking of an architect, but all resulting Task objects have the very same simple API. The users of Task objects need no knowledge of its inner structure.

For the moment, tasks can't be stopped and continued and their error handling must be improved. These will be topics of our next lesson. We will extend the methods to:

     |  append(self, task) -> 'Task'
     |      appends a task or a chain of tasks (both must be root tasks)
     |  
     |  cont(self, gap:float=0) -> 'Task'
     |      continues a stopped task (must be a root task)
     |      gap: sets the waiting time before the next action occurs (in seconds)
     |  
     |  join(self) -> None
     |      joins the thread of the task 
     |      think as: my_task.thread.join(), but evaluated when called
     |  
     |  start(self, gap:float=0) -> 'Task'
     |      starts execution of task (finished or stopped tasks may be started again)
     |      gap: sets the waiting time, before start occurs (in seconds)
     |  
     |  stop(self) -> None
     |      stops execution as fast as possible
     |          allows to continue with method cont or restart with method start
     |          already finished tasks silently do nothing
      

Now time has come to play around with the new tools. Modify your programs The depressed giraffe and The dancing robot and use tasks. Be creative and realize some of your own ideas and develop a feeling for the task concept. When you finished your playing, come back to lesson 9.

Lesson 7 - Multithreading, doing things parallel

Introduction

Last lesson, we have heard of multitasking and multithreading, but we have not seen anything of it. This will be changed now. We create a second subclass of EV3 and name it Jukebox. This class plays tones and music, later we will add some light effects. In this lesson we use it as a playground for multithreading.

Then we will look at different aspects of multithreading. We will write little programs and we will get familiar with it.

Class Jukebox

We code a second subclass of EV3 with a design, that realizes the qualities, we formulated at the end of the last lesson:

#!/usr/bin/env python3

import ev3, time

TRIAS = {
    "tempo": 80,
    "tones": [
        ["c'", 1],
        ["e'", 1],
        ["g'", 1],
        ["c''",3],
    ]
}

HAPPY_BIRTHDAY = {
    "tempo": 100,
    "tones": [
        ["d'", 0.75],
        ["d'", 0.25],
        ["e'", 1],
        ["d'", 1],
        ["g'", 1],
        ["f#'", 2],
        ["d'", 0.75],
        ["d'", 0.25],
        ["e'", 1],
        ["d'", 1],
        ["a'", 1],
        ["g'", 2],
        ["d'", 0.75],
        ["d'", 0.25],
        ["d''", 1],
        ["b'", 1],
        ["g'", 1],
        ["f#'", 1],
        ["e'", 1],
        ["c''", 0.75],
        ["c''", 0.25],
        ["b'", 1],
        ["g'", 1],
        ["a'", 1],
        ["g'", 2]
    ]
}

class Jukebox(ev3.EV3):
    def __init__(self, protocol: str=None, host: str=None, ev3_obj=None):
        super().__init__(protocol=protocol, host=host, ev3_obj=ev3_obj)
        self._volume = 1
        self._temperament = 440
        self._pos_tone = None
        self._plays = False

    @property
    def volume(self):
        return self._volume
    @volume.setter
    def volume(self, value:int):
        self._volume = value

    @property
    def temperament(self):
        return self._temperament
    @temperament.setter
    def temperament(self, value:float):
        self._temperament = value

    def play_tone(self, tone: str, duration: float=0) -> None:
        volume = self._volume
        if tone == "p":
            self.stop()
            return
        elif tone.startswith("c"):
            freq = self._temperament * 2**(-9/12)
        elif tone.startswith("d"):
            freq = self._temperament * 2**(-7/12)
        elif tone.startswith("e"):
            freq = self._temperament * 2**(-5/12)
        elif tone.startswith("f"):
            freq = self._temperament * 2**(-4/12)
        elif tone.startswith("g"):
            freq = self._temperament * 2**(-2/12)
        elif tone.startswith("a"):
            freq = self._temperament
        elif tone.startswith("b"):
            freq = self._temperament * 2**(2/12)
        else:
            raise AttributeError('unknown Tone: ' + tone)

        if len(tone) > 1:
            if tone[1] == "#":
                freq *= 2**(1/12)
            elif tone[1] == "b":
                freq /= 2**(1/12)

        if tone.endswith("'''"):
            freq *= 4
        elif tone.endswith("''"):
            freq *= 2
        elif tone.endswith("'"):
            pass
        else:
            freq /= 2
        ops = b''.join([
            ev3.opSound,
            ev3.TONE,
            ev3.LCX(volume),
            ev3.LCX(round(freq)),
            ev3.LCX(round(1000*duration))
        ])
        self.send_direct_cmd(ops)

    def stop(self) -> None:
        self.send_direct_cmd(ev3.opSound + ev3.BREAK)
        self._plays = False

    def _init_tone(self) -> None:
        self._pos_tone = 0
        self._plays = True

    def _next_tone(self, song) -> float:
        if self._pos_tone == len(song["tones"]):
            return -1
        tone, beats = song["tones"][self._pos_tone]
        self.play_tone(tone)
        self._pos_tone += 1
        return 60 * beats / song["tempo"]

    def play_song(self, song:dict) -> None:
        self._init_tone()
        while self._plays:
            duration = self._next_tone(song)
            if duration == -1:
                break
            time.sleep(duration)
        if self._plays:
            self._plays = False
            self.stop()
      

Remarks:

• You already know all the operations from lesson 2.
• The frequencies of the tones are calculated in the 12 tone equal temperament and then rounded to integers.
• The object attribute _plays is a flag, that signals, that actually a song is played. This class reacts correctly, if method stop is called, while a song is played. It stops playing.
• Method play_song is time consuming but does not block the EV3 device. From this point of view, it behaves like methods drive_straight, drive_turn, rotate_to and drive_to of class TwoWheelVehicle.
• The songs are defined as JSON objects, which is like a poor mans midi notation.
• As all our classes, Jukebox is a layer of abstraction. This one encapsulates the playing of music.

The documentation of module ev3_sound:

Help on module ev3_sound:

NAME
    ev3_sound

CLASSES
    ev3.EV3(builtins.object)
        Jukebox
    
    class Jukebox(ev3.EV3)
     |  plays tones and songs
     |  
     |  Method resolution order:
     |      Jukebox
     |      ev3.EV3
     |      builtins.object
     |  
     |  Methods defined here:
     |  
     |  __init__(self, protocol:str=None, host:str=None, ev3_obj=None)
     |      Establish a connection to a LEGO EV3 device
     |      
     |      Keyword Arguments (either protocol and host or ev3_obj):
     |      protocol: None, 'Bluetooth', 'Usb' or 'Wifi'
     |      host: None or mac-address of the LEGO EV3 (f.i. '00:16:53:42:2B:99')
     |      ev3_obj: None or an existing EV3 object (its connections will be used)
     |  
     |  play_song(self, song:dict) -> None
     |      plays a song
     |      
     |      example:
     |      jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
     |      jukebox.play_song(ev3_sound.HAPPY_BIRTHDAY)
     |  
     |  play_tone(self, tone:str, duration:float=0) -> None
     |      plays a tone
     |      
     |      Attributes:
     |      tone: name of tone f.i. "c'", "cb''", "c#"
     |      
     |      Keyword Attributes:
     |      duration: length (sec.) of the tone (value 0 means forever)
     |  
     |  stop(self) -> None
     |      stops the sound
     |  
     |  ----------------------------------------------------------------------
     |  Data descriptors defined here:
     |  
     |  temperament
     |      temperament of the tones (delfault: 440 Hz)
     |  
     |  volume
     |      volume of sound [0 - 100] (default: 1)
     |  
     |  ----------------------------------------------------------------------
     |  Methods inherited from ev3.EV3:
     |  
     |  __del__(self)
     |      closes the connection to the LEGO EV3
     |  
     |  send_direct_cmd(self, ops:bytes, local_mem:int=0, global_mem:int=0) -> bytes
     |      Send a direct command to the LEGO EV3
     |      
     |      Arguments:
     |      ops: holds netto data only (operations), the following fields are added:
     |        length: 2 bytes, little endian
     |        counter: 2 bytes, little endian
     |        type: 1 byte, DIRECT_COMMAND_REPLY or DIRECT_COMMAND_NO_REPLY
     |        header: 2 bytes, holds sizes of local and global memory
     |      
     |      Keyword Arguments:
     |      local_mem: size of the local memory
     |      global_mem: size of the global memory
     |      
     |      Returns: 
     |        sync_mode is STD: reply (if global_mem > 0) or message counter
     |        sync_mode is ASYNC: message counter
     |        sync_mode is SYNC: reply of the LEGO EV3
     |  
     |  wait_for_reply(self, counter:bytes) -> bytes
     |      Ask the LEGO EV3 for a reply and wait until it is received
     |      
     |      Arguments:
     |      counter: is the message counter of the corresponding send_direct_cmd
     |      
     |      Returns:
     |      reply to the direct command
     |  
     |  ----------------------------------------------------------------------
     |  Data descriptors inherited from ev3.EV3:
     |  
     |  __dict__
     |      dictionary for instance variables (if defined)
     |  
     |  __weakref__
     |      list of weak references to the object (if defined)
     |  
     |  sync_mode
     |      sync mode (standard, asynchronous, synchronous)
     |      
     |      STD:   Use DIRECT_COMMAND_REPLY if global_mem > 0,
     |             wait for reply if there is one.
     |      ASYNC: Use DIRECT_COMMAND_REPLY if global_mem > 0,
     |             never wait for reply (it's the task of the calling program).
     |      SYNC:  Always use DIRECT_COMMAND_REPLY and wait for reply.
     |      
     |      The general idea is:
     |      ASYNC: Interruption or EV3 device queues direct commands,
     |             control directly comes back.
     |      SYNC:  EV3 device is blocked until direct command is finished,
     |             control comes back, when direct command is finished.               
     |      STD:   NO_REPLY like ASYNC with interruption or EV3 queuing,
     |             REPLY like SYNC, synchronicity of program and EV3 device.
     |  
     |  verbosity
     |      level of verbosity (prints on stdout).

DATA
    HAPPY_BIRTHDAY = {'tempo': 100, 'tones': [["d'", 0.75], ["d'", 0.25], ...
    TRIAS = {'tempo': 80, 'tones': [["c'", 1], ["e'", 1], ["g'", 1], ["c''"...
      

We write this little program to test it:

#!/usr/bin/env python3

import ev3, ev3_sound

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
jukebox.verbosity = 1
jukebox.play_song(ev3_sound.HAPPY_BIRTHDAY)
    

Its output:

20:17:14.956954 Sent 0x|0C:00|2A:00|80|00:00|94:01:01:82:26:01:00|
20:17:15.408330 Sent 0x|0C:00|2B:00|80|00:00|94:01:01:82:26:01:00|
20:17:15.559332 Sent 0x|0C:00|2C:00|80|00:00|94:01:01:82:4A:01:00|
20:17:16.160783 Sent 0x|0C:00|2D:00|80|00:00|94:01:01:82:26:01:00|
20:17:16.762240 Sent 0x|0C:00|2E:00|80|00:00|94:01:01:82:88:01:00|
20:17:17.363682 Sent 0x|0C:00|2F:00|80|00:00|94:01:01:82:72:01:00|
20:17:18.565853 Sent 0x|0C:00|30:00|80|00:00|94:01:01:82:26:01:00|
20:17:19.017158 Sent 0x|0C:00|31:00|80|00:00|94:01:01:82:26:01:00|
20:17:19.168137 Sent 0x|0C:00|32:00|80|00:00|94:01:01:82:4A:01:00|
20:17:19.769569 Sent 0x|0C:00|33:00|80|00:00|94:01:01:82:26:01:00|
20:17:20.371090 Sent 0x|0C:00|34:00|80|00:00|94:01:01:82:B8:01:00|
20:17:20.972644 Sent 0x|0C:00|35:00|80|00:00|94:01:01:82:88:01:00|
20:17:22.174288 Sent 0x|0C:00|36:00|80|00:00|94:01:01:82:26:01:00|
20:17:22.625924 Sent 0x|0C:00|37:00|80|00:00|94:01:01:82:26:01:00|
20:17:22.777000 Sent 0x|0C:00|38:00|80|00:00|94:01:01:82:4B:02:00|
20:17:23.378498 Sent 0x|0C:00|39:00|80|00:00|94:01:01:82:EE:01:00|
20:17:23.980124 Sent 0x|0C:00|3A:00|80|00:00|94:01:01:82:88:01:00|
20:17:24.581646 Sent 0x|0C:00|3B:00|80|00:00|94:01:01:82:72:01:00|
20:17:25.183178 Sent 0x|0C:00|3C:00|80|00:00|94:01:01:82:4A:01:00|
20:17:25.784707 Sent 0x|0C:00|3D:00|80|00:00|94:01:01:82:0B:02:00|
20:17:26.236062 Sent 0x|0C:00|3E:00|80|00:00|94:01:01:82:0B:02:00|
20:17:26.387118 Sent 0x|0C:00|3F:00|80|00:00|94:01:01:82:EE:01:00|
20:17:26.988627 Sent 0x|0C:00|40:00|80|00:00|94:01:01:82:88:01:00|
20:17:27.590142 Sent 0x|0C:00|41:00|80|00:00|94:01:01:82:B8:01:00|
20:17:28.191678 Sent 0x|0C:00|42:00|80|00:00|94:01:01:82:88:01:00|
20:17:29.393731 Sent 0x|07:00|43:00|80|00:00|94:00|

Combining multiple tasks without multithreading

Whe combine driving and playing a song:

#!/usr/bin/env python3

import ev3, ev3_sound, ev3_vehicle

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
vehicle = ev3_vehicle.TwoWheelVehicle(
    0.02128,                 # radius_wheel
    0.1175,                  # tread
    ev3_obj=jukebox
)
vehicle.drive_turn(25, 0.2)
jukebox.play_song(ev3_sound.HAPPY_BIRTHDAY)
vehicle.stop()

This program does two independent things, it plays a song, which uses EV3's sound resource and it drives the vehicle, which uses two motors. Two independent actions and no need of multithreading, why that? This is a combination of actions with different character:

• When calling drive_turn without setting an angle, this is an unlimited action and control comes back directly.
• play_song does the timing. Control comes back when the song is finished.
• The call of method stop ends the unlimited movement by interruption. Control directly comes back.

This says, we can split our action in three parts:

• Immediate return:

  
  vehicle.drive_turn(25, 0.2)
   

• Time consuming:

  
  jukebox.play_song(ev3_sound.HAPPY_BIRTHDAY)
   

• Immediate return:

  
  vehicle.stop()
   

The timing is done by the time consuming parts of an action! Please take a look at the program The depressed giraffe of lesson 6 and identify the time consuming actions.

Let's come to a first conclusion. Executing multiple tasks does not necessarily need multithreading. Often it's possible to combine the actions in a sequence so that the correct timing is given and all tasks are done as desired. But this needs a clear understanding of the time consumption and the dependencies. The result is a sequence of actions, some of them are time consuming, others return immediately and are grouped around the time consuming actions, which do the timing.

This does not allow to run time consuming actions parallel! The program always waits until control is back.

Multithreading

Now we use multithreading to execute two independent actions. First we add some information to song HAPY_BIRTHDAY:

HAPPY_BIRTHDAY = {
    "tempo": 100,
    "beats_per_bar": 3,
    "upbeat": 1,
    "led_sequence": [ev3.LED_ORANGE, ev3.LED_GREEN, ev3.LED_RED, ev3.LED_GREEN],
    "tones": [
        ...

Then we run this program:

#!/usr/bin/env python3

import ev3, ev3_sound, threading, time

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
jukebox.verbosity = 1

def change_color(led_pattern: bytes) -> None:
    ops = b''.join([
        ev3.opUI_Write,
        ev3.LED,
        led_pattern
    ])
    jukebox.send_direct_cmd(ops)

def colors(song: dict) -> None:
    if "upbeat" in song:
        time.sleep(60 * song["upbeat"] / song["tempo"])
    pos_led = 0
    while plays:
        change_color(song["led_sequence"][pos_led])
        pos_led += 1
        pos_led %= len(song["led_sequence"])
        time.sleep(60 * song["beats_per_bar"] / song["tempo"])
    change_color(ev3.LED_GREEN)

plays = True  
threading.Thread(
    target=colors,
    args=(ev3_sound.HAPPY_BIRTHDAY,)
).start()
jukebox.play_song(ev3_sound.HAPPY_BIRTHDAY)
plays = False

      

Remarks:

• Both, function colors and method play_song are time consuming. Both run parallel, but play_song does the timing, it stops colors when the song is finished.
• Class Thread allows to run any callable in its own thread. This says, control immediately comes back.
• We call colors with an argument. If you want its return value, this needs some more logic, method start does not return it. The execution is asynchronous and handling return values is one of the drawbacks. We are lucky, there is no return value.
• Please consult the documentation of module threading for details.

This programs output:

20:15:20.846839 Sent 0x|0C:00|2A:00|80|00:00|94:01:01:82:26:01:00|
20:15:21.298260 Sent 0x|0C:00|2B:00|80|00:00|94:01:01:82:26:01:00|
20:15:21.447336 Sent 0x|08:00|2C:00|80|00:00|82:1B:03|
20:15:21.449398 Sent 0x|0C:00|2D:00|80|00:00|94:01:01:82:4A:01:00|
20:15:22.050950 Sent 0x|0C:00|2E:00|80|00:00|94:01:01:82:26:01:00|
20:15:22.652346 Sent 0x|0C:00|2F:00|80|00:00|94:01:01:82:88:01:00|
20:15:23.249905 Sent 0x|08:00|30:00|80|00:00|82:1B:01|
20:15:23.253805 Sent 0x|0C:00|31:00|80|00:00|94:01:01:82:72:01:00|
20:15:24.455873 Sent 0x|0C:00|32:00|80|00:00|94:01:01:82:26:01:00|
20:15:24.907189 Sent 0x|0C:00|33:00|80|00:00|94:01:01:82:26:01:00|
20:15:25.052517 Sent 0x|08:00|34:00|80|00:00|82:1B:02|
20:15:25.058203 Sent 0x|0C:00|35:00|80|00:00|94:01:01:82:4A:01:00|
20:15:25.659172 Sent 0x|0C:00|36:00|80|00:00|94:01:01:82:26:01:00|
20:15:26.260541 Sent 0x|0C:00|37:00|80|00:00|94:01:01:82:B8:01:00|
20:15:26.855334 Sent 0x|08:00|38:00|80|00:00|82:1B:01|
20:15:26.862090 Sent 0x|0C:00|39:00|80|00:00|94:01:01:82:88:01:00|
20:15:28.064249 Sent 0x|0C:00|3A:00|80|00:00|94:01:01:82:26:01:00|
20:15:28.515525 Sent 0x|0C:00|3B:00|80|00:00|94:01:01:82:26:01:00|
20:15:28.657901 Sent 0x|08:00|3C:00|80|00:00|82:1B:03|
20:15:28.666590 Sent 0x|0C:00|3D:00|80|00:00|94:01:01:82:4B:02:00|
20:15:29.268142 Sent 0x|0C:00|3E:00|80|00:00|94:01:01:82:EE:01:00|
20:15:29.869902 Sent 0x|0C:00|3F:00|80|00:00|94:01:01:82:88:01:00|
20:15:30.460544 Sent 0x|08:00|40:00|80|00:00|82:1B:01|
20:15:30.471457 Sent 0x|0C:00|41:00|80|00:00|94:01:01:82:72:01:00|
20:15:31.072930 Sent 0x|0C:00|42:00|80|00:00|94:01:01:82:4A:01:00|
20:15:31.674404 Sent 0x|0C:00|43:00|80|00:00|94:01:01:82:0B:02:00|
20:15:32.125746 Sent 0x|0C:00|44:00|80|00:00|94:01:01:82:0B:02:00|
20:15:32.263208 Sent 0x|08:00|45:00|80|00:00|82:1B:02|
20:15:32.276589 Sent 0x|0C:00|46:00|80|00:00|94:01:01:82:EE:01:00|
20:15:32.878185 Sent 0x|0C:00|47:00|80|00:00|94:01:01:82:88:01:00|
20:15:33.479785 Sent 0x|0C:00|48:00|80|00:00|94:01:01:82:B8:01:00|
20:15:34.065854 Sent 0x|08:00|49:00|80|00:00|82:1B:01|
20:15:34.081116 Sent 0x|0C:00|4A:00|80|00:00|94:01:01:82:88:01:00|
20:15:35.282992 Sent 0x|07:00|4B:00|80|00:00|94:00|
20:15:35.868486 Sent 0x|08:00|4C:00|80|00:00|82:1B:01|
      

Both of them run independently but are thought to work synchronized. We take a closer look to the synchronization and realize, that the synchronization becomes worse. If we played a longer song, we could see and hear the growing time shift. Every command needs some time to execute. And these small durations add up. There are more tones than color changes, this makes that tones fall behind colors.

Class Jukebox with colors

We add colors to class Jukebox. This needs two more methods, change_color and _colors:

    def change_color(self, led_pattern: bytes) -> None:
        ops = b''.join([
            ev3.opUI_Write,
            ev3.LED,
            led_pattern
        ])
        self.send_direct_cmd(ops)

    def _colors(self, song: dict) -> None:
        if "upbeat" in song:
            time.sleep(60 * song["upbeat"] / song["tempo"])
        pos_led = 0
        while self._plays:
            self.change_color(song["led_sequence"][pos_led])
            pos_led += 1
            pos_led %= len(song["led_sequence"])
            time.sleep(60 * song["beats_per_bar"] / song["tempo"])
        self.change_color(ev3.LED_GREEN)
    

We modify method play_song:

    def play_song(self, song:dict) -> None:
        self._init_tone()
        threading.Thread(
            target=self._colors,
            args=(song,)
        ).start()
        while self._plays:
            duration = self._next_tone(song)
            if duration == -1:
                break
            time.sleep(duration)
        if self._plays:
            self._plays = False
            self.stop()
      

We test it:

#!/usr/bin/env python3

import ev3, ev3_sound

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
jukebox.play_song(ev3_sound.HAPPY_BIRTHDAY)
      

We test the stopping:

#!/usr/bin/env python3

import ev3, ev3_sound, time, threading

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
threading.Thread(
    target=jukebox.play_song,
    args=(ev3_sound.HAPPY_BIRTHDAY,)
).start()
time.sleep(5)
jukebox.stop()

This program runs three threads. One changes the colors, one the tones and the base thread stops them.

Exact timing

For a better timing we improve class Jukebox and modify method _colors:

    def _colors(self, song: dict) -> None:
        time_action = time.time()
        if "upbeat" in song:
            time_action += 60 * song["upbeat"] / song["tempo"]
            gap = time_action - time.time()
            if gap > 0:
                time.sleep(gap)
        pos_led = 0
        while self._plays:
            self.change_color(song["led_sequence"][pos_led])
            pos_led += 1
            pos_led %= len(song["led_sequence"])
            time_action += 60 * song["beats_per_bar"] / song["tempo"]
            gap = time_action - time.time()
            if gap > 0:
                time.sleep(gap)
        self.change_color(ev3.LED_GREEN)
      

and method play_song:

    def play_song(self, song:dict) -> None:
        self._init_tone()
        threading.Thread(
            target=self._colors,
            args=(song,)
        ).start()
        time_action = time.time()
        while self._plays:
            duration = self._next_tone(song)
            if duration == -1:
                break
            time_action += duration
            gap = time_action - time.time()
            if gap > 0:
                time.sleep(gap)
        if self._plays:
            self._plays = False
            self.stop()
      

we test it with this program:

#!/usr/bin/env python3

import ev3, ev3_sound

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
jukebox.verbosity = 1
jukebox.play_song(ev3_sound.HAPPY_BIRTHDAY)
      

the output:

09:39:22.873872 Sent 0x|0C:00|2A:00|80|00:00|94:01:01:82:26:01:00|
09:39:23.324479 Sent 0x|0C:00|2B:00|80|00:00|94:01:01:82:26:01:00|
09:39:23.474193 Sent 0x|0C:00|2C:00|80|00:00|94:01:01:82:4A:01:00|
09:39:23.474699 Sent 0x|08:00|2D:00|80|00:00|82:1B:03|
09:39:24.074629 Sent 0x|0C:00|2E:00|80|00:00|94:01:01:82:26:01:00|
09:39:24.674645 Sent 0x|0C:00|2F:00|80|00:00|94:01:01:82:88:01:00|
09:39:25.274656 Sent 0x|0C:00|30:00|80|00:00|94:01:01:82:72:01:00|
09:39:25.275207 Sent 0x|08:00|31:00|80|00:00|82:1B:01|
09:39:26.475232 Sent 0x|0C:00|32:00|80|00:00|94:01:01:82:26:01:00|
09:39:26.924492 Sent 0x|0C:00|33:00|80|00:00|94:01:01:82:26:01:00|
09:39:27.074212 Sent 0x|0C:00|34:00|80|00:00|94:01:01:82:4A:01:00|
09:39:27.074697 Sent 0x|08:00|35:00|80|00:00|82:1B:02|
09:39:27.674645 Sent 0x|0C:00|36:00|80|00:00|94:01:01:82:26:01:00|
09:39:28.274532 Sent 0x|0C:00|37:00|80|00:00|94:01:01:82:B8:01:00|
09:39:28.874644 Sent 0x|0C:00|38:00|80|00:00|94:01:01:82:88:01:00|
09:39:28.875532 Sent 0x|08:00|39:00|80|00:00|82:1B:01|
09:39:30.075237 Sent 0x|0C:00|3A:00|80|00:00|94:01:01:82:26:01:00|
09:39:30.524485 Sent 0x|0C:00|3B:00|80|00:00|94:01:01:82:26:01:00|
09:39:30.674126 Sent 0x|0C:00|3C:00|80|00:00|94:01:01:82:4B:02:00|
09:39:30.674611 Sent 0x|08:00|3D:00|80|00:00|82:1B:03|
09:39:31.274652 Sent 0x|0C:00|3E:00|80|00:00|94:01:01:82:EE:01:00|
09:39:31.874653 Sent 0x|0C:00|3F:00|80|00:00|94:01:01:82:88:01:00|
09:39:32.474641 Sent 0x|0C:00|40:00|80|00:00|94:01:01:82:72:01:00|
09:39:32.475321 Sent 0x|08:00|41:00|80|00:00|82:1B:01|
09:39:33.074650 Sent 0x|0C:00|42:00|80|00:00|94:01:01:82:4A:01:00|
09:39:33.674640 Sent 0x|0C:00|43:00|80|00:00|94:01:01:82:0B:02:00|
09:39:34.124485 Sent 0x|0C:00|44:00|80|00:00|94:01:01:82:0B:02:00|
09:39:34.274186 Sent 0x|0C:00|45:00|80|00:00|94:01:01:82:EE:01:00|
09:39:34.274928 Sent 0x|08:00|46:00|80|00:00|82:1B:02|
09:39:34.874565 Sent 0x|0C:00|47:00|80|00:00|94:01:01:82:88:01:00|
09:39:35.474633 Sent 0x|0C:00|48:00|80|00:00|94:01:01:82:B8:01:00|
09:39:36.074662 Sent 0x|0C:00|49:00|80|00:00|94:01:01:82:88:01:00|
09:39:36.075217 Sent 0x|08:00|4A:00|80|00:00|82:1B:01|
09:39:37.275170 Sent 0x|07:00|4B:00|80|00:00|94:00|
09:39:37.875666 Sent 0x|08:00|4C:00|80|00:00|82:1B:01|
      

This solved the problem! We changed from netto to brutto timing. Now the time distances include the time for execution.

Communication between threads

The base thread and all threads it starts, use the same global data. This allows communication. Every thread can read data of another one. The communication is asynchronous, which is no problem, we are used to asynchronous communication. Think of mails or emails. We already have seen it working. Attribute _plays was used for communication between different threads. If one thread sets this flag, the others read it and react.

Locking

The communication between threads sometimes shows unexpected results. Let's look at a snippet of code, that runs in its own thread. This thread uses a global variable state to tell its actual state to the rest of the world:

STATE_TO_STOP = "TO_STOP"
STATE_STOPPED = "STOPPED"
STATE_FINISHED = "FINISHED"

def finished_or_stopped():
    global state
    if state == STATE_TO_STOP:
        state = STATE_STOPPED
    if state == STATE_STOPPED and not next:
        state = STATE_FINISHED

state = STATE_TO_STOP
next = False
finished_or_stopped()
      

We expect, that there are only two combinations of state and next:

• Before the function is called: state == STATE_TO_STOP and next == False
• After the call of the function: state == STATE_FINISHED and next == False

The experience will show, this is correct in about 99.9999 % of all situations, but not really for all. It may happen, that a foreign thread asks about variable state just after it was changed to value STATE_STOPPED. This seldom case shows a new combination:

• While the function is executed: state == STATE_STOPPED and next == False

The locking mechanism prevents this. We change our function to:

def finished_or_stopped():
    global state
    lock.acquire()
    if state == STATE_TO_STOP:
        state = STATE_STOPPED
    if state == STATE_STOPPED and not next:
        state = STATE_FINISHED
    lock.release()
      

The lock object was created by lock = threading.Lock() and the foreign thread, which asks about the state also must know and use it:

    lock.acquire()
    if state == STATE_STOPPED:
        print("This never happens")
    lock.release()
      

The lock object guaranties, that a second call of method acquire will wait until method release was called (maybe by anyone else). This says either function finished_or_stopped has to wait until the foreign thread has finished its if statement or the foreign thread has to wait until the change of valiable state in function finished_or_stopped is done.

Locking is a very common thechnique. Databases use it to prevent concurrent updates of the same data. Operating systems use locking to manage the usage of hardware resources and so on. But it's never fun to code it. If one forgets a single call of method release, the resource is blocked forever. The good news is, that encapsulation allows to do all this behind the scene. This says, the execution (f.i. usage of the resources) is done through well defined methods, which implement the locking mechanism.

Error handling

Another drawback of asynchronous processing is error handling. If an error is thrown inside a thread, this will not reach the other threads (and not the base thread). Let' look at an example:

#!/usr/bin/env python3

import ev3, ev3_sound, time, threading

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
jukebox.verbosity = 1

def tone(intervall) -> None:
    time_action = time.time()
    for i in range(4):
        jukebox.play_tone("c", 0.1)
        time.sleep(0.5)
        jukebox.play_tone("c'", 0.05)
        time.sleep(0.5)
        jukebox.play_tone("c'", 0.05)
        time.sleep(0.5)
        jukebox.play_tone("c'", 0.05)
        time_action += intervall
        gap = time_action - time.time()
        time.sleep(max(0, gap))
    jukebox.stop()

def led(intervall) -> None:
    time_action = time.time()
    for i in range(2):
        jukebox.change_color(ev3.LED_RED)
        time.sleep(2)
        raise Exception('Something happened')
        jukebox.change_color(ev3.LED_GREEN)
        time_action += intervall
        gap = time_action - time.time()
        time.sleep(max(0, gap))

threading.Thread(target=led, args=(4,)).start()
tone(2)
      

its output:

08:40:18.292905 Sent 0x|08:00|2A:00|80|00:00|82:1B:02|
08:40:18.293778 Sent 0x|0D:00|2B:00|80|00:00|94:01:01:82:83:00:81:64|
08:40:18.795806 Sent 0x|0D:00|2C:00|80|00:00|94:01:01:82:06:01:81:32|
08:40:19.297941 Sent 0x|0D:00|2D:00|80|00:00|94:01:01:82:06:01:81:32|
08:40:19.800143 Sent 0x|0D:00|2E:00|80|00:00|94:01:01:82:06:01:81:32|
08:40:20.294214 Sent 0x|0D:00|2F:00|80|00:00|94:01:01:82:83:00:81:64|
Exception in thread Thread-1:
Traceback (most recent call last):
  File "/usr/lib/python3.4/threading.py", line 920, in _bootstrap_inner
    self.run()
  File "/usr/lib/python3.4/threading.py", line 868, in run
    self._target(*self._args, **self._kwargs)
  File "./test_01.py", line 28, in led
    raise Exception('Something happened')
Exception: Something happend

08:40:20.796906 Sent 0x|0D:00|30:00|80|00:00|94:01:01:82:06:01:81:32|
08:40:21.299067 Sent 0x|0D:00|31:00|80|00:00|94:01:01:82:06:01:81:32|
08:40:21.801280 Sent 0x|0D:00|32:00|80|00:00|94:01:01:82:06:01:81:32|
08:40:22.294128 Sent 0x|0D:00|33:00|80|00:00|94:01:01:82:83:00:81:64|
08:40:22.796186 Sent 0x|0D:00|34:00|80|00:00|94:01:01:82:06:01:81:32|
08:40:23.298252 Sent 0x|0D:00|35:00|80|00:00|94:01:01:82:06:01:81:32|
08:40:23.800432 Sent 0x|0D:00|36:00|80|00:00|94:01:01:82:06:01:81:32|
08:40:24.294131 Sent 0x|0D:00|37:00|80|00:00|94:01:01:82:83:00:81:64|
08:40:24.796197 Sent 0x|0D:00|38:00|80|00:00|94:01:01:82:06:01:81:32|
08:40:25.298273 Sent 0x|0D:00|39:00|80|00:00|94:01:01:82:06:01:81:32|
08:40:25.800424 Sent 0x|0D:00|3A:00|80|00:00|94:01:01:82:06:01:81:32|
08:40:26.294006 Sent 0x|07:00|3B:00|80|00:00|94:00|
      

The exception was handled inside Thread-1. The base thread did not recognize the exception. Maybe this is what we want, maybe not. I prefer a hard stop of everything as the default reaction. Error handling is communication, we add a variable error:

#!/usr/bin/env python3

import ev3, ev3_sound, time, threading, traceback, sys

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
jukebox.verbosity = 1
error = False

def tone(intervall) -> None:
    global error
    time_action = time.time()
    for i in range(4):
        if error:
            sys.exit()
        jukebox.play_tone("c", 0.1)
        time.sleep(0.5)
        jukebox.play_tone("c'", 0.05)
        time.sleep(0.5)
        jukebox.play_tone("c'", 0.05)
        time.sleep(0.5)
        jukebox.play_tone("c'", 0.05)
        time_action += intervall
        gap = time_action - time.time()
        time.sleep(max(0, gap))
    jukebox.stop()

def led(intervall) -> None:
    global error
    try:
        time_action = time.time()
        for i in range(2):
            jukebox.change_color(ev3.LED_RED)
            time.sleep(2)
            raise Exception('Something happened')
            jukebox.change_color(ev3.LED_GREEN)
            time_action += intervall
            gap = time_action - time.time()
            time.sleep(max(0, gap))
    except Exception:
        error = True
        raise

threading.Thread(target=led, args=(4,)).start()
tone(2)
      

the output:

09:05:21.509790 Sent 0x|08:00|2A:00|80|00:00|82:1B:02|
09:05:21.510744 Sent 0x|0D:00|2B:00|80|00:00|94:01:01:82:83:00:81:64|
09:05:22.012858 Sent 0x|0D:00|2C:00|80|00:00|94:01:01:82:06:01:81:32|
09:05:22.515052 Sent 0x|0D:00|2D:00|80|00:00|94:01:01:82:06:01:81:32|
09:05:23.017140 Sent 0x|0D:00|2E:00|80|00:00|94:01:01:82:06:01:81:32|
09:05:23.511133 Sent 0x|0D:00|2F:00|80|00:00|94:01:01:82:83:00:81:64|
Exception in thread Thread-1:
Traceback (most recent call last):
  File "/usr/lib/python3.4/threading.py", line 920, in _bootstrap_inner
    self.run()
  File "/usr/lib/python3.4/threading.py", line 868, in run
    self._target(*self._args, **self._kwargs)
  File "./test_01.py", line 31, in led
    raise Exception('Something happened')
Exception: Something happened

09:05:24.013019 Sent 0x|0D:00|30:00|80|00:00|94:01:01:82:06:01:81:32|
09:05:24.516006 Sent 0x|0D:00|31:00|80|00:00|94:01:01:82:06:01:81:32|
09:05:25.018296 Sent 0x|0D:00|32:00|80|00:00|94:01:01:82:06:01:81:32|
      

Multiplying the code in function tone would prevent the last three beats. Fact is, that function tone has to ask if an error occured, there is no automatic mechanism. This is asynchronous communication.

Coordinating actions of unknown duration

Sometimes, you have a number of parallel actions, but there is no clear responsibility for the timing. You want your program to wait until all of the actions are finished. This also can be solved with multithreading. A Thread object has a join method, which waits until the thread is finished:

#!/usr/bin/env python3

import ev3, ev3_sound, time, datetime, threading

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
jukebox.verbosity = 1

def tone() -> None:
    global jukebox
    jukebox.play_tone("c'", 0.3)
    time.sleep(1)

def led() -> None:
    global jukebox
    jukebox.change_color(ev3.LED_RED)
    time.sleep(2)
    jukebox.change_color(ev3.LED_GREEN)

t1 = threading.Thread(target=tone)
t2 = threading.Thread(target=led)
t1.start()
t2.start()
t1.join()
t2.join()
now = datetime.datetime.now().strftime('%H:%M:%S.%f')
print(now, "all done")
      

The output:

11:19:37.801899 Sent 0x|0E:00|2A:00|80|00:00|94:01:01:82:06:01:82:2C:01|
11:19:37.803362 Sent 0x|08:00|2B:00|80|00:00|82:1B:02|
11:19:39.808145 Sent 0x|08:00|2C:00|80|00:00|82:1B:01|
11:19:39.808878 all done
      

Events

Events allow that one thread signals an event and other threads wait on it. Here is an example:

#!/usr/bin/env python3

import ev3, ev3_sound, threading, time

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
jukebox.verbosity = 1

lock = threading.Lock()

def task1():
    jukebox.play_tone("c", 0.1)
    lock.acquire()
    lock.release()
    jukebox.play_tone("c", 0.1)
    
def task2():
    jukebox.change_color(ev3.LED_RED)
    lock.acquire()
    lock.release()
    jukebox.change_color(ev3.LED_GREEN)

lock.acquire()
threading.Thread(target=task1).start()
threading.Thread(target=task2).start()
time.sleep(5)
lock.release()
      

The base thread and the threads of task1 and task2 all use the same Lock object lock. The event is signaled by the base threads call lock.release(). This allows both tasks to continue their work.

This is a very common case and the above presented code is hard to read. It is a common practice to use class Event instead, which is syntactic sugar, but reads easier. We change the program:

#!/usr/bin/env python3

import ev3, ev3_sound, threading, time

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
jukebox.verbosity = 1

event = threading.Event()

def task1():
    jukebox.play_tone("c", 0.1)
    event.wait()
    jukebox.play_tone("c", 0.1)
    
def task2():
    jukebox.change_color(ev3.LED_RED)
    event.wait()
    jukebox.change_color(ev3.LED_GREEN)

threading.Thread(target=task1).start()
threading.Thread(target=task2).start()
time.sleep(5)
event.set()
      

Both versions produce the same output:

09:47:08.032849 Sent 0x|0D:00|2A:00|80|00:00|94:01:01:82:83:00:81:64|
09:47:08.034327 Sent 0x|08:00|2B:00|80|00:00|82:1B:02|
09:47:13.040731 Sent 0x|0D:00|2C:00|80|00:00|94:01:01:82:83:00:81:64|
09:47:13.041566 Sent 0x|08:00|2D:00|80|00:00|82:1B:01|
      

Timers

There is a special subclass of Thread, that allows to start a thread after some waiting time. Here an example:

#!/usr/bin/env python3

import ev3, ev3_sound, threading, time

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
jukebox.verbosity = 1

def task():
    jukebox.play_tone("c", 0.1)

task()
threading.Timer(5, task).start()
      

The ouput:

09:56:16.543277 Sent 0x|0D:00|2A:00|80|00:00|94:01:01:82:83:00:81:64|
09:56:21.545055 Sent 0x|0D:00|2B:00|80|00:00|94:01:01:82:83:00:81:64|
      

When a Timer is still waiting, it can be canceled:

#!/usr/bin/env python3

import ev3, ev3_sound, threading, time

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
jukebox.verbosity = 1

def task():
    jukebox.play_tone("c", 0.1)

task()
timer = threading.Timer(5, task)
timer.start()
time.sleep(2)
timer.cancel()
      

Its output:

09:59:56.989692 Sent 0x|0D:00|2A:00|80|00:00|94:01:01:82:83:00:81:64|
      

The second call of task never took place, because the Timer was cancelled while it was waiting.

Interruptable Sleeping

We use class Condition to code an interruptable sleeper:

#!/usr/bin/env python3

import threading, time, datetime
lock = threading.Lock()
cond = threading.Condition(lock)

def task():
    lock.acquire()
    now = datetime.datetime.now().strftime('%H:%M:%S.%f')
    print(now, "task started")
    cond.wait(3)
    now = datetime.datetime.now().strftime('%H:%M:%S.%f')
    print(now, "task ended")
    lock.release()
    

threading.Thread(target=task).start()
time.sleep(1)
lock.acquire()
now = datetime.datetime.now().strftime('%H:%M:%S.%f')
print(now, "notify task1")
cond.notify()
time.sleep(1)
now = datetime.datetime.now().strftime('%H:%M:%S.%f')
print(now, "lock will be released")
lock.release()
      

The output:

10:10:32.073769 task started
10:10:33.075199 notify task1
10:10:34.076750 lock will be released
10:10:34.077305 task ended
      

Remarks:

• The base thread interrupts the sleeping of function task, which runs in its own thread. This is possible because both threads use the same Condition object.
• Every condition is bound to a lock.
• Method wait implicitly releases the lock, which allows another thread to acquire it.
• The base thread calls method notify which wakes up the waiting thread task.
• When notified, task tries to acquire the lock.
• Our example, where the base thread holds the lock for another sec. after notifying task is unusual but demonstrates the role of the lock.
• If no notification takes place, cond.wait(3) waits for three sec. and meanwhile releases the lock.

Modify class EV3

It needs some modifications of class EV3 to prepare it for parallel execution of multiple tasks.

Locking

For the moment, the message counter is the only class attribute. It is the common identity of a direct command and its reply. If we use multiple instances of class EV3 parallel, we need a locking mechanism when changing class attributes.

class EV3:
    _msg_cnt = 41
    _lock = threading.Lock()
      

and we modify method _complete_direct_cmd:

    def _complete_direct_cmd(self, ops:bytes,
                             local_mem:int,
                             global_mem:int) -> bytes:
        if global_mem > 0  or self._sync_mode == SYNC:
            cmd_type = _DIRECT_COMMAND_REPLY
        else:
            cmd_type = _DIRECT_COMMAND_NO_REPLY
        self._lock.acquire()
        if self._msg_cnt < 65535:
            self._msg_cnt += 1
        else:
            self._msg_cnt = 1
        msg_cnt = self._msg_cnt
        self._lock.release()
        return b''.join([
            struct.pack('<hh', len(ops) + 5, msg_cnt),
            cmd_type,
            struct.pack('<h', local_mem * 1024 + global_mem),
            ops
        ])
 

This guaranties, that the message counters are distinct until they are reused after 65.535 direct commands.

Foreign replies

Sometimes it may happen, that two direct commands (both with reply) do not hold the sequence: send cmd_1, receive reply_1, send cmd_2, receive reply_2. Instead we see the sequence: send cmd_1, send cmd_2, receive reply_1, receive reply_2. We don't prevent that because we have independent parallel tasks and we want the communication as fast as possible. This says task_1 may send cmd_1, but get reply_2:

12:15:23.903970 Sent 0x|15:00|1C:02|00|08:00|99:1C:00:13:07:01:01:60:99:1C:00:10:07:00:01:64|
12:15:23.910924 Sent 0x|0E:00|1D:02|00|04:00|99:1C:00:00:81:21:00:01:60|
12:15:23.953569 Recv 0x|0B:00|1C:02|02|98:70:00:00:4A:78:00:00|
12:15:23.954865 Recv 0x|07:00|1D:02|02|0D:00:00:00|
      

We solve this problem with a dictionary of foreign replies and add another class attribute to class EV3:

class EV3:
    _msg_cnt = 41
    _lock = threading.Lock()
    _foreign = {}
      

We add two protected methods:

    def _put_foreign_reply(self, counter: bytes, reply: bytes) -> None:
        if counter in self._foreign:
            raise ValueError('reply with counter ' + counter + ' already exists')
        else:
            self._foreign[counter] = reply

    def _get_foreign_reply(self, counter: bytes) -> bytes:
        if counter in self._foreign:
            reply = self._foreign[counter]
            del self._foreign[counter]
            return reply
        else:
            return None
      

The first adds a reply to the dictionary with its counter as key. The second looks, if the dictionary contains a reply with a given key. If so, it returns the reply and deletes it from the dictionary. We add some code to method wait_for_reply:

    def wait_for_reply(self, counter: bytes) -> bytes:
        self._lock.acquire()
        reply = self._get_foreign_reply(counter)
        if reply:
            self._lock.release()
            if reply[4:5] != _DIRECT_REPLY:
                raise DirCmdError(
                    "direct command {:02X}:{:02X} replied error".format(
                        reply[2],
                        reply[3]
                    )
                )
            return reply
        while True:
            if self._protocol in [BLUETOOTH, WIFI]:
                reply = self._socket.recv(1024)
            else:
                reply = bytes(self._device.read(EP_IN, 1024, 0))
            len_data = struct.unpack('<H', reply[:2])[0] + 2
            reply_counter = reply[2:4]
            if self._verbosity >= 1:
                ...
            if counter != reply_counter:
                self._put_foreign_reply(reply_counter, reply[:len_data])
            else:
                self._lock.release()
                if reply[4:5] != _DIRECT_REPLY:
                    raise DirCmdError(
                        "direct command {:02X}:{:02X} replied error".format(
                            reply[2],
                            reply[3]
                        )
                    )
                return reply[:len_data]
      

The logic:

• It first looks, if the reply is already in the dictionary. If so, it does not communicate with the EV3 device.
• If not, it reads reply for reply until it gets the one it looks for. All foreign replies are put into the dictionary.
• The standard situation is an empty dictionary.
• The locking guaranties an exclusive access to the dictionary.

The dancing robot

This lesson ends with a program, that combines three independent actions:

#!/usr/bin/env python3

import ev3, ev3_sound, ev3_vehicle, threading, time

jukebox = ev3_sound.Jukebox(protocol=ev3.BLUETOOTH, host='00:16:53:42:2B:99')
jukebox.volume = 5
vehicle = ev3_vehicle.TwoWheelVehicle(0.02128, 0.1175, ev3_obj=jukebox)

def drive(song):
    time_action = time.time()
    if "upbeat" in song:
        duration = 60 * song["upbeat"] / song["tempo"]
        time_action += duration
        gap = time_action - time.time()
        time.sleep(max(0, gap))
    duration = 2 * 60 * song["beats_per_bar"] / song["tempo"]
    while driving:
        vehicle.drive_turn(speed, 0.2)
        time_action += duration
        gap = time_action - time.time()
        time.sleep(max(0, gap))
        if not driving: break
        vehicle.drive_turn(-speed, -0.2)
        time_action += duration
        gap = time_action - time.time()
        time.sleep(max(0, gap))
    vehicle.stop()

song = ev3_sound.HAPPY_BIRTHDAY
speed = 30
driving = True
threading.Thread(
    target=drive,
    args=(song,)
).start()
jukebox.play_song(song)
driving = False
      

These are three parallel actions, tones, colors, movements. All of them have their own timing. The coordination results from the ratios of the timings, which all are determined by the rhythm of the music. Movement changes every second bar, color changes per bar and the tones fit into bars.

Conclusion

This lesson layed the foundations of multitasking. We coded a sublass of EV3: Jukebox, which uses multithreading. We have done a sightseeing tour, that showed us a number of aspects, we need to take into account. We were no passive visitors, no we wrote little programs and got familiar with multitasking and multithreading. We learned, that the control of time is an important aspect.

We modified class EV3 to prepare it for multiple concurrent tasks. This needed a locking mechanism and some common resources.

Next lesson we will realize some tools, that help to organize and handle multiple tasks. Here is a first specification:

• We want to start, stop and continue tasks.
• We want an easy API for repeated and periodic tasks with exact timing.
• We want to organize tasks as chains of tasks.
• We need some help for locking and error handling.