Lesson 9 - Stop and Continue

Introduction

We are on the way, to write classes, that encapsulate an unlimited number of (related) actions into tasks objects. All tasks own the same easy handling which is independent of the type of actions. Moving a robot, reading sensor values, playing tones etc., all these procedures can be task objects with the methods: start, stop, cont (continue) and join:

     |  cont(self, gap:float=None) -> 'Task'
     |      continues a stopped task (must be a root task)
     |      
     |      Keyword Arguments:
     |      gap: sets the waiting time before the next action occurs (in seconds)
     |  
     |  join(self) -> None
     |      joins the thread of the task
     |  
     |  start(self, gap:float=0) -> 'Task'
     |      starts execution of task (finished or stopped tasks may be started again)
     |      
     |      Keyword Arguments:
     |      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
      

This makes the usage of task objects simple and allows to think about their dependencies instead of fighting with technical details. Task objects use multithreading. This allows to start parallel tasks or even construct tasks, which inside operate parallel actions. The creative part is the construction of task objects and this can be done behind the scenes. The outside world handles task objects, which are ready to use. The subclasses of class EV3 will produce them. This says, at the end, our class TwoWheelVehicle will return tasks, that drive the vehicle. As a consequence the vehicles movements can be stopped and continued and we can run other tasks parallel while a movement takes place.

But yet we are not users of task objects, we have to code them! Usage will be simple, coding will be not. This lesson has three topics:

• Handling exceptions
• Stopping tasks and preparing continuation
• Continuing tasks

Again we use class Jukebox for tests.

Handling Exceptions

One of the open topics from last lesson is the error handling in task objects. We know from lesson 7, that throwing an exception will not influence foreign threads. In programs with multithreading, we need a central place of information about exceptions. If all our threads ask regularly if there has been an exception somewhere, they can react on it. We use a very simple mechanism, where the reaction is an exit.

The central place is a class, what else:

class ExceptionHandler:
    def __init__(self):
        self._exc = False

    def put(self, exc: Exception):
        self._exc = True

    def fire(self):
        if self._exc: sys.exit(1)
      

Its API:

    class ExceptionHandler(builtins.object)
     |  Handles Exceptions of task objects
     |  If anywhere an exceptions occured and was put to the ExceptionHandler,
     |  any thread that uses the same instance of ExceptionHandler exits,
     |  when it calls its method fire
     |  
     |  Methods defined here:
     |  
     |  __init__(self)
     |      Initialize self.  See help(type(self)) for accurate signature.
     |  
     |  fire(self)
     |      fires sys.exit(1) if an exception occured, else does nothing
     |  
     |  put(self, exc:Exception)
     |      informs, that an exception occured
     |      
     |      Arguments:
     |      exc: Exception, ignored, but subclasses may distinguish
      

Method put has a parameter exc, which is never used. This is for future subclasses, which may distinguish between different types of exceptions.

How do we use this class? Let's look at an example:

    def start(self) -> None:
        self._root._exc.fire()
        self._root._lock.acquire()
        try:
            assert self._root is self, 'only root tasks can be started'
            assert self._state in [
                STATE_INIT,
                STATE_STOPPED,
                STATE_FINISHED
            ], "can't start from state " + self._state
        except Exception as exc:
            self._root._exc.put(exc)
            self._root._lock.release()
            raise
        ...
      

Some remarks:

• Attribute _exc holds the ExceptionHandler of the task.
• method fire is called in an unlocked state. If the task would be locked, when the thread exits, this blocked other threads.
• Raising the exception also occurs in an unlocked state.
• The first thing, method start does, is asking if somewhere an exception occured. If so, its thread exits.
• If one of the assertions throws an AssertionException, this is put to the tasks ExceptionHandler, then the exception is raised.

We formulate some rules:

• Before locking (acquire the Lock object), always call method fire of the tasks ExceptionHandler.
• Before raising an exception, it must be put to the tasks ExceptionHandler.
• All raising of exceptions must occur in an unlocked state.
• Use one ExceptionHandler for all your tasks. If not, this needs strong arguments.

As a consequence we add a class attribute to class Task:

class Task:
    _exc_default = ExceptionHandler()
      

This will be the default of all instances of class Task

We add some code to the constructor of Task objects:

    def __init__(self, action: typing.Callable, **kwargs):
        self._action = action
        self._args = kwargs.pop('args', ())
        self._kwargs = kwargs.pop('kwargs', {})
        self._join = kwargs.pop('join', False)
        self._duration = kwargs.pop('duration', None)
        self._num = kwargs.pop('num', 0)
        self._next = None
        self._root = self
        self._time_end = None
        self._netto_time = False
        self._cnt = 0
        # the following are root only attributes
        self._state = STATE_INIT
        self._thread = None
        self._lock = threading.Lock()
        self._cond = threading.Condition(self._lock)
        self._last = None
        self._activity = ACTIVITY_NONE
        self._time_action = None
        self._contained = []
        self._exc = kwargs.pop('exc', self._exc_default)
        self._exc.fire()
        assert not kwargs, 'unknown keyword arguments: ' + str(kwargs.keys())
      

The number of keyword arguments has grown to a limit, where I prefer to use a keyworded variable length argument list **kwargs.

The subclasses Periodic and Repeated also use **kwargs. F.i. class Periodic:

class Periodic(Task):
    def __init__(self, intervall: float, action: typing.Callable, **kwargs):
        self._intervall = intervall
        self._netto_time = kwargs.pop('netto_time', False)
        assert not kwargs['join'], "no keyword argument 'join' for instances of class Periodic"
        if hasattr(action, '__self__') and \
           isinstance(action.__self__, Task) and \
           action.__name__ == "start":
            kwargs.update({'join': True})
        else:
            kwargs.update({'join': False})
        super().__init__(action, **kwargs)
        assert isinstance(self._intervall, numbers.Number), 'intervall must be a number' + intervall
        assert self._intervall >= 0, 'intervall must be positive'
        assert isinstance(self._netto_time, bool), 'netto_time must be a bool value'
      

Class Sleep:

class Sleep(Task):
    def __init__(self, seconds: float, exc: ExceptionHandler=None):
        if not exc:
            exc = self._exc_default
        super().__init__(self._do_nothing, duration=seconds, exc=exc)
      

Stopping Tasks

We code more logic than needed for stopping. We also prepare continuation because we don't like to go twice trough the code.

States

We already know some states from the last lesson:

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

Let's talk about their meaning:

• INIT: The initial state of a task. When a task is constructed, but never started, its state is INIT. After a tasks start, it will never return to state INIT.
• STARTED: The call of method start changes the tasks state from INIT or FINISHED to state STARTED. While the whole regular execution of the task, it stays in state STARTED. When the tasks last action is finished, the state changes from STARTED to FINISHED.
• FINISHED: The final state of a task. When a task finished regularly and was not stopped, its state is FINISHED. A task in state FINISHED can be started again.

We extend these states:

STATE_INIT = 'INIT'
STATE_TO_START = 'TO_START'
STATE_STARTED = 'STARTED'
STATE_TO_STOP = 'TO_STOP'
STATE_STOPPED = 'STOPPED'
STATE_TO_CONTINUE = 'TO_CONTINUE'
STATE_FINISHED = 'FINISHED'
      

The meaning of the additional states:

• TO_START: We set it if method start was called with an argument gap, which schedules the starting for the future. In the meantime the state is TO_START. State TO_START signals, there is no execution or sleeping in progress.
• TO_STOP: Method stop already was called (this changed the state STARTED → TO_STOP) and the task or its contained tasks have not yet ended their last action. If a call of start follows, while the old execution still is in progress, the series of states is: STARTED → TO_STOP → TO_START → STARTED.
• STOPPED: The task ended an action and red the state TO_STOP. This prevents the execution of the next action or sleeping and the state changes TO_STOP → STOPPED.
• TO_CONTINUE: Method cont was called, and it waits to execute the next action. State TO_CONTINUE signals (like INIT, STOPPED, FINISHED or TO_START), there is no execution or sleeping in progress.

While an action is executed, there can be a series of method-calls, f.i. stop, start, stop. Only the first of them results in a change of the state: STARTED → TO_STOP. We need another criterion to identify the situations after the second or third call of a method. This is the existence of the new thread (_thread_start or _thread_cont). The second call of start) creates the new thread _thread_start, the next call of stop prevents the new thread from executing an action.

The combination of the state and the existence of new threads makes the full understanding of the situation:

• TO_STOP and _thread_start != None: methods stop and start were called while the last action was executed.
• TO_STOP and _thread_cont != None: methods stop and cont were called while the last action was executed.
• TO_STOP and _thread_start == None and _thread_cont == None: the last call was method stop and the last action still executes.

Responsibilities for state transitions

We look, which parts of our tasks are responsible for changes of the state (without method cont).

• Method start and its followers _start2 and _start3:
  • From [INIT, STOPPED, FINISHED] to STARTED: if called with gap == 0.
  • From [INIT, STOPPED, FINISHED] to TO_START: if called with gap > 0.
  • Unchanged state TO_STOP: the old thread still is executing. A new thread is created and started, but the state remains TO_STOP.
  • From TO_START to STARTED: when the old thread ended and gap is over.
• Method stop:
  • From STARTED to TO_STOP: never changes directly from STARTED to STOPPED.
  • From TO_START to STOPPED: TO_START signals, that _execute was not yet called.
  • From TO_CONTINUE to STOPPED: the old thread came to its end, the new thread did not yet call _execute.
• Method _execute:
  • From STARTED to FINISHED: This is the regular case.
  • From TO_STOP to FINISHED: The actual action ended and there was no next one and no final sleeping.
  • From TO_STOP to TO_START: The actual action ended and a new thread _thread_start already was started from method start.
  • From TO_STOP to TO_CONTINUE: The actual action ended and a new thread _thread_cont already was started from method cont.
  • From TO_STOP to STOPPED: The actual action ended and there were no new threads.

Additional Attributes

We add attributes to class Task:

class Task:
    _exc_default = ExceptionHandler()
    _contained_register = {}
    
    def __init__(self, action: typing.Callable, **kwargs):
        ...
        # the following are root only attributes
        ...
        self._action_stop = kwargs.pop('action_stop', None)
        self._args_stop = kwargs.pop('args_stop', ())
        self._kwargs_stop = kwargs.pop('kwargs_stop', {})
        self._action_cont = kwargs.pop('action_cont', None)
        self._args_cont = kwargs.pop('args_cont', ())
        self._kwargs_cont = kwargs.pop('kwargs_cont', {})
        self._thread_start = None
        self._thread_cont = None
        self._actual = None
        self._cont_join = None
        self._time_called_stop = None
        self._restart = False
      

Their meaning:

• _action_stop: The default stopping only prevents the next action from execution. Often it additionally needs a stopping action, f.i. stopping the sound or a movement.
• _args_stop: The stopping action may have positional arguments.
• _kwargs_stop: The stopping action may have keyword arguments.
• _action_cont: like _action_stop, f.i. when _action_stop ends a movement, _action_cont restarts it.
• _args_cont: positional arguments of _action_cont.
• _kwargs_cont: keyword arguments of _action_cont.
• _thread_start: We already discussed it. The old thread _thread may be in state TO_STOP when method start is called. Then attribute _thread_start holds the new thread as long as the old one also is needed.
• _thread_cont: like thread_start, but for continuation.
• _actual: holds the actual link in the chain. Continuation needs to know it.
• _cont_join: If the task was stopped while joining a contained task, this attribute holds the task to join.
• _time_called_stop: Continuation has to continue at the correct time. It takes value _time_action and adds the time distance between the call of the method stop and the time when it could start executing the next action.
• _restart: If there was a sequence of method calls, while the state was TO_STOP, f.i. start, stop, cont, the last continuation needs to know, that is has to restart instead of continue.

Method stop

We add method stop:

    def stop(self) -> None:
        self._root._exc.fire()
        self._root._lock.acquire()
        try:
            assert self is self._root, 'only root tasks can be stopped'
            assert self._state in [
                STATE_TO_START,
                STATE_STARTED,
                STATE_TO_STOP,
                STATE_TO_CONTINUE,
                STATE_FINISHED
            ], "can't stop from state: " + self._state
            assert self._state != STATE_TO_STOP or self._thread_start or self._thread_cont, \
                "stopping is already in progress"
        except Exception as exc:
            self._root._exc.put(exc)
            self._root._lock.release()
            raise
        if self._state == STATE_FINISHED:
            self._lock.release()
            return
        if self._time_called_stop is None:
            self._time_called_stop = time.time()
        if self._activity is ACTIVITY_SLEEP:
            self._cond.notify()
        not_stopped = []
        for task in self._contained:
            if not task in self._contained_register or \
               not self._contained_register[task] is self:
                continue
            task.lock.acquire()
            if task._state in [STATE_STARTED, STATE_TO_START, STATE_TO_CONTINUE]:
                not_stopped.append(task)
            elif task._state == STATE_TO_STOP and (task._thread_start or task._thread_cont):
                not_stopped.append(task)
            task.lock.release()
        for task in not_stopped:
            task.stop()
        if self._state == STATE_STARTED:
            self._state = STATE_TO_STOP
        elif self._thread_start:
            self._thread_start = None
            if self._state == STATE_TO_START:
                self._state = STATE_STOPPED
        else:
            self._thread_cont = None
            if self._state == STATE_TO_CONTINUE:
                self._state = STATE_STOPPED
        self._lock.release()
      

Step by step we go through this code:

• If the task already is finished, it silently does nothing:

  
          if self._state == STATE_FINISHED:
              self._lock.release()
              return
     

• The time of the first call of stop is needed for the correct timing in method cont:

  
          if self._time_called_stop is None:
              self._time_called_stop = time.time()
     

• If the task is sleeping, we interrupt the sleeping:

  
          if self._activity is ACTIVITY_SLEEP:
              self._cond.notify()
     

• Stopping all contained tasks:

  
          not_stopped = []
          for task in self._contained:
              if not task in self._contained_register or \
                 not self._contained_register[task] is self:
                  continue
              task.lock.acquire()
              if task._state in [STATE_STARTED, STATE_TO_START, STATE_TO_CONTINUE]:
                  not_stopped.append(task)
              elif task._state == STATE_TO_STOP and (task._thread_start or task._thread_cont):
                  not_stopped.append(task)
              task.lock.release()
          for task in not_stopped:
              task.stop()
     

  This works recursive and stops all direct or indirect children tasks.
• Changing state STARTED → TO_STOP:

  
          if self._state == STATE_STARTED:
              self._state = STATE_TO_STOP
     

• If there was a call of method start, its new thread looses its reference:

  
          elif self._thread_start:
              self._thread_start = None
     

  This signals method start that the thread has to end before it reaches state STARTED
• If there was a call of method cont, its new thread looses its reference:

  
          else:
              self._thread_cont = None
              if self._state == STATE_TO_CONTINUE:
                  self._state = STATE_STOPPED
     

  This signals method cont that the thread has to end before it reaches state STARTED

Method start

Method start has to handle the situation, when it finds a task in state TO_STOP and it got a keyword argument gap:

    def start(self, gap: float=0) -> 'Task':
        self._root._exc.fire()
        self._root._lock.acquire()
        try:
            assert isinstance(gap, numbers.Number), 'gap needs to be a number'
            assert gap >= 0, 'gap needs to be positive'
            assert self._root is self, 'only root tasks can be started'
            assert self._state in [
                STATE_INIT,
                STATE_TO_STOP,
                STATE_STOPPED,
                STATE_FINISHED
            ], "can't start from state " + self._state
            assert self._thread_start is None, "starting is already in progress"
            assert self._thread_cont is None, "continuation is already in progress"
        except Exception as exc:
            self._root._exc.put(exc)
            self._root._lock.release()
            raise
        if self._state == STATE_TO_STOP or gap > 0:
            if self._state == STATE_TO_STOP:
                self._restart = True
            else:
                self._state = STATE_TO_START
            if gap:
                self._thread_start = threading.Thread(
                    target=self._start2,
                    args=(time.time() + gap,)
                )
            else:
                self._thread_start = threading.Thread(target=self._start2)
            self._thread_start.start()
        else:
            self._start3()
            self._thread = threading.Thread(target=self._execute)
            self._thread.start()
        return self
      

Method _start2

    def _start2(self, time_action: float=None) -> None:
        if self._state == STATE_TO_STOP:
            self._lock.release()
            self._thread.join()
            self._exc.fire()
            self._lock.acquire()
        if not threading.current_thread() is self._thread_start:
            self._lock.release()
            return
        if time_action:
            gap = time_action - time.time()
            if gap > 0:
                self._activity = ACTIVITY_SLEEP
                self._cond.wait(gap)
                self._activity = ACTIVITY_NONE
                if not threading.current_thread() is self._thread_start:
                    self._lock.release()
                    return
        self._thread = self._thread_start
        self._thread_start = None
        self._start3()
        self._execute()
      

Annotations:

• First it joins the old thread.
• Then it tests if there was a call of method stop. If so, it returns without changing the state or executing something.
• If method start was called with keyword argument gap, it waits until its time has come. Then it tests if meanwhile there was a call of stop.
• Its thread becomes the thread of the task and it executes its actions.

Method _start3

These are a few lines of code, we need twice (in methods start and _start2):

    def _start3(self) -> None:
        self._state = STATE_STARTED
        self._restart = False
        self._time_called_stop = None
        self._actual = self
        self._cnt = 0
        self._time_action = time.time()
        if self._duration != None:
            self._time_end = self._time_action + self._duration
      

Method join

Joining needs to join all threads, the old and the new ones:

    def join(self) -> None:
        try:
            assert self._root is self, "only root tasks can be joined"
            assert self._state != STATE_INIT, "can't join tasks in state " + str(self._state)
        except Exception as exc:
            self._root._exc.put(exc)
            raise
        self._exc.fire()
        try: self._thread_start.join()
        except Exception: pass
        try: self._thread_cont.join()
        except Exception: pass
        try: self._thread.join()
        except Exception: pass
      

Method _execute

While a tasks action is executed or while it's sleeping, it releases its lock. This allows to execute method stop, which changes the state STARTED → TO_STOP. Method _execute needs to handle state TO_STOP and it needs to react as fast as possible:

    def _execute(self) -> None:
        while True:
            if self._root._state != STATE_STARTED:
                self._final(outstand=True)
                return
            try:
                gap = self._wrapper()
            except Exception as exc:
                self._exc.put(exc)
                raise
            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:
                if self._root._state != STATE_STARTED:
                    self._final(outstand=True)
                    return
                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
            gap = self._root._time_action - time.time()
            if self._root._state == STATE_STARTED and gap > 0:
                self._root._activity = ACTIVITY_SLEEP
                self._root._cond.wait(gap)
                self._root._activity = ACTIVITY_NONE
            if self._root._state == STATE_STARTED:
                self._time_end = None
            elif not self is self._root:
                self._root._time_end = self._time_end
                self._time_end = None
        else:
            self._root._time_action = time.time()
        if self._next:
            self._root._actual = self._next
            self._next._cnt = 0
            self._root._time_end = None
            if self._next._duration != None:
                self._next._time_end = self._root._time_action + self._next._duration
            self._next._execute()
        else:
            self._final()
      

The last task may have an argument duration. If we stop it while its last sleeping, the root task will end with attribute _time_end but not _actual. This signals: the last sleeping was not finished.

Method _wrapper1

    def _wrapper1(self) -> None:
        if hasattr(self._action, '__self__') and \
           isinstance(self._action.__self__, Task) and \
           self._action.__name__ in ["start", "cont", "join"]:
            task = self._action.__self__
            name = self._action.__name__
            if (self._join or name is "join"):
                self._root._cont_join = task
            if name in ["start", "cont"]:
                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__ in ["start", "cont"] or \
           self._action.__name__ == "start" and self._join:
            self._root._activity = ACTIVITY_BUSY
            self._root._lock.release()
            self._root._exc.fire()
      

Remarks:

• Method cont is not time consuming, there is no need for releasing and aquiring the lock.
• Attribute _cont_join needs to be set if we join a task.
• If the action is the continuation of a task, we have to actualize the roots attribute _contained and the class attribute _contained_register.

Method wrapper2

The corresponding logic after the execution of an action:

    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__ in ["start", "cont"] or \
           self._action.__name__ == "start" and self._join:
            self._root._exc.fire()
            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", "stop", "cont", "join"]:
            task = self._action.__self__
            name = self._action.__name__
            state = task.state
            if self._root._cont_join and \
               (self._root._state == STATE_STARTED or \
                state == STATE_FINISHED):
                self._root._cont_join = None
            if (state == STATE_FINISHED or name == "stop") and \
               task in self._root._contained:
                self._root._contained.remove(task)
            if name == "stop" and \
               task in self._contained_register:
                self._contained_register.pop(task)
      

Method _final

    def _final(self, outstand=False) -> None:
        self._root._contained = self._join_contained()
        if self._root._state == STATE_STARTED:
            self._root._state = STATE_FINISHED
        elif self._root._state == STATE_TO_STOP:
            if not self._next and \
               not self._root._contained and \
               not self._root._time_end and \
               not outstand:
                self._root._state = STATE_FINISHED
            elif self._root._action_stop:
                self._root._action_stop(
                    *self._root._args_stop,
                    **self._root._kwargs_stop
                )
        if self._root._state == STATE_FINISHED:
            if self._root in self._contained_register:
                self._contained_register.pop(self._root)
            self._root._thread_cont = None
            self._root._actual = None
            self._root._time_action = None
        else:
            if not self._next and not outstand:
                self._root._actual = None
                self._root._time_action = None
            if self._root._thread_start:
                self._root._actual = None
                self._root._time_action = None
                self._root._state = STATE_TO_START
            elif self._root._thread_cont:
                self._root._state = STATE_TO_CONTINUE
            else:
                self._root._state = STATE_STOPPED
        if self._root._time_action and self._root._time_action < time.time():
            self._root._time_action = None
        self._root._lock.release()
      

Remarks:

• There is a lot of logic for changing the state.
• The stopping action may be called.
• The new attributes need to be set.

These were all modifications of the code.

Class Jukebox

In the lessons 7 and 8 we used class Jukebox to demonstrate multithreading and the chances of the task concept. Now we modify method sound and prepare it to stop appropriate then we do the same with method song.

Modifying method sound

We change method sound:

    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 repeat and not duration:
            return task.Task(
                self.send_direct_cmd,
                args=(ops,)
            )
        elif not repeat and duration:
            t_inner = task.Task(
                self.send_direct_cmd,
                args=(ops,),
                duration=duration,
                action_stop=self.stop
            )
            return task.Task(t_inner.start, join=True)
        elif repeat and not duration:
            t_inner = task.Task(
                self.send_direct_cmd,
                args=(ops,),
                duration=9999999,
                action_stop=self.stop,
                action_cont=self.send_direct_cmd,
                args_cont=(ops,)
            )
            return task.Task(t_inner.start, join=True)
        elif repeat and duration:
            t_inner = task.Task(
                self.send_direct_cmd,
                args=(ops,),
                duration=duration,
                action_stop=self.stop,
                action_cont=self.send_direct_cmd,
                args_cont=(ops,)
            )
            return task.Task(t_inner.start, join=True)
      

There is no direct command to continue a sound file. If you compare with our first version from lesson 8:

• not repeat and not duration: this is unchanged, we implement no stopping because the task is not time consuming. This type of calling method sound should be used for short sound signals.
• not repeat and duration: it stops the sound, but in case of continuation it will wait silently. Restarting does not fit the intended timing.
• repeat and not duration: a task with endless duration has actions for stopping and continuation. It needs a final stop to end the playing of the sound file.
• repeat and duration: this is straight foreward but later we will see, that it is not perfect.

Tests of method sound

not repeat and not duration

We stop the task directly after its start, this will change the state STARTED → TO_STOP. The task will wait until the action is finished, then change the state TO_STOP → FINISHED. We run the following program:

#!/usr/bin/env python3

import ev3, ev3_sound, time

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

t = jukebox.sound('./ui/DownloadSucces')
print("state:", t.state)
t.start()
print("state:", t.state)
t.stop()
print("state:", t.state)
time_stop = time.time()
t.join()
time_join = time.time()
print("state: {}, duration of stopping: {}, time_action: {}".format(
    t.state,
    time_join - time_stop,
    t.time_action
))
      

and get this output:

state: INIT
07:32:35.683207 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|
state: STARTED
state: TO_STOP
state: FINISHED, duration of stopping: 0.0002593994140625, time_action: None
      

Indeed, the call of method stop returned immediately and set the state to TO_STOP. A very short time of less than 1 thousandth sec. later, the task was finished. The sequence of states was INIT → STARTED → TO_STOP → FINISHED.

not repeat and duration

This uses a contained task. We print data from both tasks. Of special interest is attribute _time_end of the inner task. It holds the rest of the original duration.

#!/usr/bin/env python3

import ev3, ev3_sound, time

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

t = jukebox.sound('./ui/DownloadSucces', duration=1)
print("state:", t.state)
t.start()
print("state: {}, state of contained task: {}".format(t.state, t._cont_join.state))
t.stop()
print("state: {}, state of contained task: {}".format(t.state, t._cont_join.state))
time_stop = time.time()
t.join()
time_join = time.time()
print("duration of stopping: {}, state: {}, time_action: {}".format(
    time_join - time_stop,
    t.state,
    t.time_action
))
print("Contained task:")
print("state: {}, actual: {}, time_end: {}".format(
    t._cont_join.state,
    t._cont_join._actual,
    t._cont_join._time_end - time.time()
))
      

The output:

state: INIT
07:55:56.404427 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|
state: STARTED, state of contained task: STARTED
state: TO_STOP, state of contained task: TO_STOP
07:55:56.405752 Sent 0x|07:00|2B:00|80|00:00|94:00|
duration of stopping: 0.0015096664428710938, state: STOPPED, time_action: None
Contained task:
state: STOPPED, actual: None, time_end: 0.9963183403015137
      

The program sent two direct commands. The first started the playing, the second stopped it. The stopping needed a bit more time, it had to stop the contained task too. Both tasks ended in state STOPPED and _time_end holds the rest of the original duration. When calling method cont, the task will silently wait because there is no actual action (attribute actual is None) and no action_cont.

repeat and not duration

We start an unlimited repeated playing of a sound file and stop it after three seconds:

#!/usr/bin/env python3

import ev3, ev3_sound, time

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

t = jukebox.sound('./ui/DownloadSucces', repeat=True)
t.start()
time.sleep(3)
t.stop()
t.join()
print("Contained task:")
print("state: {}, actual: {}, time_end: {}".format(
    t._cont_join.state,
    t._cont_join._actual,
    t._cont_join._time_end - time.time()
))
print(t._cont_join._action_cont)
      

The output:

08:31:22.900092 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|
08:31:25.903642 Sent 0x|07:00|2B:00|80|00:00|94:00|
Contained task:
state: STOPPED, actual: None, time_end: 999999995.9944854
<bound method EV3.send_direct_cmd of <ev3_sound.Jukebox object at 0x7f61d5be19b0>>
      

The rest of the duration is very long now and the inner task it will not silently continue because attribute _action_cont is set.

repeat and duration

We don't test the stopping but we come back to it, when we test the continuation.

#!/usr/bin/env python3

import ev3, ev3_sound, time

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

t = jukebox.sound('./ui/DownloadSucces', repeat=True, duration=3)
t.start().join()
print("state:", t.state)
      

The output:

08:41:59.082772 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|
08:42:02.083546 Sent 0x|07:00|2B:00|80|00:00|94:00|
state: FINISHED
      

It seems to work.

Modifying method song

Method song returns a Task object with two contained tasks, one for colors, one for tones. We managed to stop the colors, when the sequence of tones ends, but we didn't like our solution. The new method stop allows to make colors and tones independent and group the colors around the tones:

task.concat(
    task.Task(colors.start),
    task.Task(tones.start, join=True),
    task.Task(colors.stop)
)
      

Our modifications:

• We remove method play_song. Calling jukebox.song(ev3_sound.HAPPY_BIRTHDAY).start() does the job as well as jukebox.play_song(ev3_sound.HAPPY_BIRTHDAY).
• We remove attribute _plays.
• We simplify method stop:

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

• We modify the task factory song:

  
      def song(self, song: dict) -> task.Task:
          tones = task.concat(
              task.Task(
                  self._init_tone,
                  action_stop=self.stop
              ),
              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,
                  action_stop=self.change_color,
                  args_stop=(ev3.LED_GREEN,)
              ),
              colors
          )
          return task.concat(
              task.Task(colors.start),
              task.Task(tones.start, join=True),
              task.Task(colors.stop)
          )
     

All four members of the task family, Task, Repeated, Periodic and Sleep are in use. Method song returns a Task object which can be combined with other task objects.

Testing method song

We test it with this program:

#!/usr/bin/env python3

import ev3, ev3_sound, task, time

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

It plays Happy Birthday for 9 sec., then it stops for 2 sec. (the color changes to green), then ist starts again and plays Happy Birthday, which is followed by the trias. The handling of contained task seems to be correct!

Tasks with long-lasting actions

Stopping

We code an action that lasts two sec. The stopping is called, when the action was executed since one sec. and needs another sec. to finish:

#!/usr/bin/env python3

import task, time

def first_action():
    print("first action begin")
    time.sleep(2)
    print("first action end")

t = task.concat(
    task.Task(first_action),
    task.Task(print, args=("second action",))
)
t.start()
print("task started, state:", t.state)
time.sleep(1)
t.stop()
time_stop = time.time()
print("task stopped, state:", t.state)
t.join()
time_join = time.time()
print("duration of joining:", time_join - time_stop)
print("task joined, state:", t.state)
print("time_action:", t.time_action)
      

We expect that the state TO_STOP lasts one sec. because the stopping algorithm has to wait until the actual action is finished. Then the state will change TO_STOP → STOPPED. The output:

first action begin
task started, state: STARTED
task stopped, state: TO_STOP
first action end
duration of joining: 1.0017707347869873
task joined, state: STOPPED
time_action: None
      

Indeed, the stopping needed about one sec. The value None of property time_action says: no limitation, the next action can be started as fast as possible.

Restarting

We call a series of methods start and stop, while the stopping is in execution:

#!/usr/bin/env python3

import task, time

def first_action():
    print("first action begin")
    time.sleep(2)
    print("first action end")

t = task.concat(
    task.Task(first_action),
    task.Task(print, args=("second action",))
)
t.start()
time.sleep(1)
t.stop()
print("task stopped, state:", t.state)
t.start()
print("task started, state:", t.state, " thread_start:", t._thread_start)
t.stop()
print("task stopped, state:", t.state, " thread_start:", t._thread_start)
      

The output:

first action begin
task stopped, state: TO_STOP
task started, state: TO_STOP  thread_start: <Thread(Thread-2, started 139977212032768)>
task stopped, state: TO_STOP  thread_start: None
first action end
      

The first call of method stop changed the state STARTED → TO_STOP. From then on it needs one sec. until the action ends. In this time the program calls another start and another stop. Method start creates and starts thread _thread_start, but it does not execute anything because its reference disappeares while it is waiting for the end of the old thread.

Continue a stopped task

We prepared continuation when we realized method stop. Now we add method cont.

Method cont

Like method start, cont starts a thread and returns immediately:

    def cont(self, gap: float=None) -> 'Task':
        self._exc.fire()
        self._lock.acquire()
        try:
            assert self is self._root, 'only root tasks can be continued'
            assert gap is None or isinstance(gap, numbers.Number), 'gap needs to be a number'
            assert gap is None or gap >= 0, 'gap needs to be positive'
            assert self._state in [
                STATE_STOPPED,
                STATE_TO_STOP,
                STATE_FINISHED
            ], "can't continue from state: {} (task: {})".format(
                self._state,
                self
            )
            assert self._thread_start is None, "starting is already in progress"
            assert self._thread_cont is None, "continuation is already in progress"
        except Exception as exc:
            self._root._exc.put(exc)
            self._root._lock.release()
            raise
        if self._state == STATE_FINISHED:
            self._lock.release()
            return self
        if gap is None:
            self._thread_cont = threading.Thread(target=self._cont2)
        else:
            self._thread_cont = threading.Thread(
                target=self._cont2,
                kwargs={"time_cont": time.time() + gap}
            )
        self._thread_cont.start()
        return self
      

Annotations:

• If the task already is finished, it does nothing and returns silently.
• After starting the new thread, it returns without releasing the lock.
• Unlike method start its thread never calls method _execute. Continuation has to handle contained tasks and only method _cont2 knows how to do that.

Method _cont2

Method _cont2 runs in the thread, that was started from method cont. The state of the task is either STOPPED or TO_STOP.

    def _cont2(self, time_cont: float=None, time_delta: float=None) -> None:
        if self._state == STATE_STOPPED:
            self._state = STATE_TO_CONTINUE
        elif self._state == STATE_TO_STOP:
            self._lock.release()
            self._thread.join()
            self._exc.fire()
            self._lock.acquire()
        if not threading.current_thread() is self._thread_cont:
            self._lock.release()
            return
        if time_cont:
            gap = time_cont - time.time()
            if gap > 0:
                self._activity = ACTIVITY_SLEEP
                self._cond.wait(gap)
                self._activity = ACTIVITY_NONE
                if not threading.current_thread() is self._thread_cont:
                    self._lock.release()
                    return
        if self._restart:
            self._restart = False
            self._actual = self
            self._contained = []
            self._time_action = time.time()
            if self._duration:
                self._time_end = self._time_action + self._duration
            else:
                self._time_end = None
        else:
            if self._action_cont:
                self._action_cont(*self._args_cont, **self._kwargs_cont)
            if not time_cont and not time_delta:
                time_delta = time.time() - self._time_called_stop
            elif not time_delta:
                next_time_action = self.time_action_no_lock
                if next_time_action:
                    time_delta = time.time() - next_time_action
                elif self._time_end:
                    time_delta = time.time() - self._time_called_stop
                else:
                    time_delta = -1
            if self._actual:
                if self._time_action:
                    self._time_action += time_delta
                if self._actual._time_end:
                    self._actual._time_end += time_delta
            elif self._time_end:
                self._time_end += time_delta
        self._state = STATE_STARTED
        self._time_called_stop = None
        self._thread = self._thread_cont
        self._thread_cont = None
        if self._contained:
            for task in self._contained:
                if task._state is STATE_FINISHED:
                    continue
                if not task in self._contained_register or \
                   self._contained_register[task] != self:
                    continue
                task._lock.acquire()
                task._thread_cont = threading.Thread(
                    target=task._cont2,
                    kwargs={'time_cont': time_cont, 'time_delta': time_delta}
                )
                task._thread_cont.start()
            if self._cont_join:
                self._activity = ACTIVITY_JOIN
                self._lock.release()
                self._cont_join.join()
                self._exc.fire()
                self._lock.acquire()
                self._activity = ACTIVITY_NONE
                if self._state != STATE_STARTED:
                    self._final()
                    return
        if self._actual:
            if self._time_action:
                gap = self._time_action - time.time()
                if gap > 0:
                    self._activity = ACTIVITY_SLEEP
                    self._cond.wait(gap)
                    self._activity = ACTIVITY_NONE
                    if self._state != STATE_STARTED:
                        self._final()
                        return
            self._actual._execute()
        else:
            if self._time_end:
                gap = self._time_end  - time.time()
                if gap > 0:
                    self._activity = ACTIVITY_SLEEP
                    self._cond.wait(gap)
                    self._activity = ACTIVITY_NONE
                    if self._state != STATE_STARTED:
                        self._final()
                        return
            self._time_end = None
            self._final()
      

Step by step we go through this code:

• If the tasks old thread still is executing an action, we wait until it ends.

  
          if self._state == STATE_STOPPED:
              self._state = STATE_TO_CONTINUE
          elif self._state == STATE_TO_STOP:
              self._lock.release()
              self._thread.join()
              self._exc.fire()
              self._lock.acquire()
     

  Joining is time consuming, meanwhile the lock is released.
• While joining, there could be a call of method stop. If so, it sets _thread_cont = None and another call of cont would creat a new _thread_cont. We control both:

  
          if not threading.current_thread() is self._thread_cont:
              self._lock.release()
              return
     

• If the method was called with argument gap. We wait this time:

  
          if time_cont:
              gap = time_cont - time.time()
              if gap > 0:
                  self._activity = ACTIVITY_SLEEP
                  self._cond.wait(gap)
                  self._activity = ACTIVITY_NONE
                  if not threading.current_thread() is self._thread_cont:
                      self._lock.release()
                      return
     

• If there was a call of method start between the first stop and the actual cont, we have to continue with the first link in the chain (the root task) and ignore all contained tasks.

  
          if self._restart:
              self._restart = False
              self._actual = self
              self._contained = []
              self._time_action = time.time()
              if self._duration:
                  self._time_end = self._time_action + self._duration
              else:
                  self._time_end = None
     

• When there is a special action for continuation, it is called:

  
          else:
              if self._action_cont:
                  self._action_cont(*self._args_cont, **self._kwargs_cont)
     

• We have to correct all timing by a time shift. This may be the time between the calls of stop and cont or argument gap made it.

  
              if not time_cont and not time_delta:
                  time_delta = time.time() - self._time_called_stop
              elif not time_delta:
                  next_time_action = self.time_action_no_lock
                  if next_time_action:
                      time_delta = time.time() - next_time_action
                  elif self._time_end:
                      time_delta = time.time() - self._time_called_stop
                  else:
                      time_delta = -1
     

  If the task is a contained task, time_delta came in as an argument. Value -1 says, it's not needed, but set. This prevents the contained tasks from repeating the calculation.
• The correction of the timing:

  
              if self._actual:
                  if self._time_action:
                      self._time_action += time_delta
                  if self._actual._time_end:
                      self._actual._time_end += time_delta
              elif self._time_end:
                  self._time_end += time_delta
     

  Recursion will do the same shift in all contained tasks so that the original synchronisation still will be conserved.
• The state changes to STARTED and the continuation thread becomes the thread of the task:

  
          self._state = STATE_STARTED
          self._time_called_stop = None
          self._thread = self._thread_cont
          self._thread_cont = None
     

• All contained tasks have to continue. Their synchronization must be reconstructed. This is done by calling their method _cont2 with keyword argument time_delta:

  
          if self._contained:
              for task in self._contained:
                  if task._state is STATE_FINISHED:
                      continue
                  if not task in self._contained_register or \
                     self._contained_register[task] != self:
                      continue
                  task._lock.acquire()
                  task._thread_cont = threading.Thread(
                      target=task._cont2,
                      kwargs={'time_cont': time_cont, 'time_delta': time_delta}
                  )
                  task._thread_cont.start()
     

• The task may have been stopped, while joining a contained task. If so, the joining must occur before the tasks next action is executed.

  
              if self._cont_join:
                  self._activity = ACTIVITY_JOIN
                  self._lock.release()
                  self._cont_join.join()
                  self._exc.fire()
                  self._lock.acquire()
                  self._activity = ACTIVITY_NONE
                  if self._state != STATE_STARTED:
                      self._final()
                      return
     

• The execution of the next action may need another sleeping. If there is no more action, the task has to join its contained tasks. This is done by method _final.

  
          if self._actual:
              if self._time_action:
                  gap = self._time_action - time.time()
                  if gap > 0:
                      self._activity = ACTIVITY_SLEEP
                      self._cond.wait(gap)
                      self._activity = ACTIVITY_NONE
                      if self._state != STATE_STARTED:
                          self._final()
                          return
              self._actual._execute()
          else:
              if self._time_end:
                  gap = self._time_end  - time.time()
                  if gap > 0:
                      self._activity = ACTIVITY_SLEEP
                      self._cond.wait(gap)
                      self._activity = ACTIVITY_NONE
                      if self._state != STATE_STARTED:
                          self._final()
                          return
              self._time_end = None
              self._final()
     

Continuing tasks with long-lasting actions

We test the continuation with this program:

#!/usr/bin/env python3

import task, time, datetime

def action(txt):
    now = datetime.datetime.now().strftime('%H:%M:%S.%f')
    print(now, txt, "begin")
    time.sleep(2)
    now = datetime.datetime.now().strftime('%H:%M:%S.%f')
    print(now, txt, "end")

t = task.concat(
    task.Task(action, args=("first action",)),
    task.Task(action, args=("last action",))
)
now = datetime.datetime.now().strftime('%H:%M:%S.%f')
print(now, "task created, state:", t.state)
t.start()
now = datetime.datetime.now().strftime('%H:%M:%S.%f')
print(now, "task started, state:", t.state)
time.sleep(1)
t.stop()
now = datetime.datetime.now().strftime('%H:%M:%S.%f')
print(now, "task stopped, state:", t.state)
t.cont(gap=2)
now = datetime.datetime.now().strftime('%H:%M:%S.%f')
print(now, "task continued, state:", t.state)
time.sleep(1)
now = datetime.datetime.now().strftime('%H:%M:%S.%f')
print(now, "waited 1 sec., state:", t.state)
time.sleep(1)
now = datetime.datetime.now().strftime('%H:%M:%S.%f')
print(now, "waited 1 sec., state:", t.state)
t.join()
now = datetime.datetime.now().strftime('%H:%M:%S.%f')
print(now, "task joined, state:", t.state)
      

The output:

18:54:17.673259 task created, state: INIT
18:54:17.674380 first action begin
18:54:17.674556 task started, state: STARTED
18:54:18.675986 task stopped, state: TO_STOP
18:54:18.677085 task continued, state: TO_STOP
18:54:19.676891 first action end
18:54:19.678387 waited 1 sec., state: TO_CONTINUE
18:54:20.677659 last action begin
18:54:20.679900 waited 1 sec., state: STARTED
18:54:22.680142 last action end
18:54:22.680710 task joined, state: FINISHED
      

Looks good, both actions are executed once, the second action begins two sec. after the call of cont, the states are INIT → STARTED → TO_STOP → TO_CONTINUE → STARTED → FINISHED.

Stopping and continuing a song

Again we use class Jukebox to test contained tasks:

#!/usr/bin/env python3

import ev3, ev3_sound, time

jukebox = ev3_sound.Jukebox(
    protocol=ev3.BLUETOOTH,
    host='00:16:53:42:2B:99'
)
jukebox.verbosity = 1
t_song = jukebox.song(ev3_sound.TRIAS)
t_song.start()
time.sleep(1)
t_song.stop()
t_song.cont(2)
      

The output:

20:30:17.552945 Sent 0x|08:00|2A:00|80|00:00|82:1B:03|
20:30:17.557831 Sent 0x|0C:00|2B:00|80|00:00|94:01:01:82:06:01:00|
20:30:18.309089 Sent 0x|0C:00|2C:00|80|00:00|94:01:01:82:4A:01:00|
20:30:18.558362 Sent 0x|08:00|2D:00|80|00:00|82:1B:01|
20:30:18.559895 Sent 0x|07:00|2E:00|80|00:00|94:00|
20:30:20.560389 Sent 0x|0C:00|2F:00|80|00:00|94:01:01:82:88:01:00|
20:30:21.305400 Sent 0x|08:00|30:00|80|00:00|82:1B:05|
20:30:21.310159 Sent 0x|0C:00|31:00|80|00:00|94:01:01:82:0B:02:00|
20:30:23.555294 Sent 0x|08:00|32:00|80|00:00|82:1B:03|
20:30:23.559971 Sent 0x|07:00|33:00|80|00:00|94:00|
20:30:23.561446 Sent 0x|08:00|34:00|80|00:00|82:1B:01|
      

Synchronisation of contained tasks works correct, tones and light are synchrone in the first and the second part of the song.

Stopping and continuing a repeated sound file

Situations, where the algorithm doesn't work as we intent, help to understand the mechanism. One learns more by errors than by anything else. What's our problem? There is no direct command to continue the playing of a sound file! We can stop and restart but not continue. To continue a repeated sound file with a fixed duration, we start it again but with a modified timing (shorter duration).

We test the stopping and continuation:

#!/usr/bin/env python3

import ev3, ev3_sound, time

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)
t.start()
time.sleep(2)
t.stop()
time.sleep(2)
t.cont()
      

The output:

20:46:58.965333 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|
20:47:00.967754 Sent 0x|07:00|2B:00|80|00:00|94:00|
20:47:02.970920 Sent 0x|1D:00|2C: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|
20:47:05.968598 Sent 0x|07:00|2D:00|80|00:00|94:00|
      

Looks great!

The problem of the next action

We modify the program slightly and call method cont with argument gap:

#!/usr/bin/env python3

import ev3, ev3_sound, time

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)
t.start()
time.sleep(2)
t.stop()
t.cont(2)
 

The output:

20:51:15.311826 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|
20:51:17.314737 Sent 0x|07:00|2B:00|80|00:00|94:00|
20:51:19.315760 Sent 0x|1D:00|2C: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|
20:51:19.317129 Sent 0x|07:00|2D:00|80|00:00|94:00|
 

The stopping occurs immediately after the continuation, why? Calling method cont with argument gap says: wait, then continue with the next action. The next action is stop! The callers intention is, that the repeated playing is the action and she does not know, that the playing of sound files can't be continued. We tricksed when we restarted and named it continuation. This trick fails under some special conditions.

The solution: subclassing Task

The solution of the problem lies in subclassing. We modify the continuation and prevent the immediate execution of the next action. The new version of method sound:

    def sound(self, path: str, duration: float=None, repeat: bool=False) -> task.Task:
        ...
        elif repeat:
            class _Task(task.Task):
                def _final(self, **kwargs):
                    super()._final(**kwargs)
                    if self._root._time_action:
                        self._root._time_rest = self._root._time_action - time.time()
                        self._root._time_action -= self._root._time_rest
                def _cont2(self, **kwargs):
                    self._time_action += self._time_rest
                    super()._cont2(**kwargs)

            t_inner = task.concat(
                _Task(
                    self.send_direct_cmd,
                    args=(ops,),
                    duration=duration,
                    action_stop=self.stop,
                    action_cont=self.send_direct_cmd,
                    args_cont=(ops,)
                ),
                _Task(self.stop)
            )
            return task.Task(t_inner.start, join=True)
        ...
      

We shift _time_action backwards from the stopping to the sound continuation. This is what the outer world expects. When continuing, we shift in reverse direction and reconstruct the old situation. If in the meantime somebody asks for property time_action, he gets the shifted value. This will change the behaviour of method cont when argument gap is set. We again start the program. Its output:

21:01:58.678197 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|
21:02:00.680940 Sent 0x|07:00|2B:00|80|00:00|94:00|
21:02:02.681971 Sent 0x|1D:00|2C: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|
21:02:05.682649 Sent 0x|07:00|2D:00|80|00:00|94:00|
      

Well done!

Conclusion

Stopping and continuing tasks is a complex mechanism. This lesson was hard stuff! We had to manage all the timing and synchronization, a lot of technical details. When we look from the outside on the tasks, we don't see all this details and realize the benefits:

• It provides modularity. We can combine small tasks to medium tasks and medium tasks to complex tasks.
• It helps to manage high complexity because the compexity of the outside API doesn't grow.
• It helps for good design. Starting, stopping and continuation seems very natural.
• It hides the locking and the multithreading mechanism and the synchronization of contained tasks.

We also see the drawbacks:

• Stopping and continuing need a special design of the tasks. A simple wrapping of a callable into a task object often does not fit our needs. This complicates the coding of the robots actions.
• Simple things become at least medium complex. Tasks are an abstraction layer, that doesn't keep simple things simple.

I would be glad to get your feedback! Our next lesson will be about class TwoWheelVehicle. We will train it to stop immediately and continue appropriate.