executing a class in the background so another class can still receive inputs

[siliconchris]

Hi all,

Before I start, let me say that I am fairly if not to say completely new to micropython and also only have rather limited knowledge of multithreading or parallel execution in python as well. I have a Raspberry PiCo which shall control a model of the millennium falcon I am building.

The PiCo is supposed to turn on LEDs (like the engine and cockpit lights or hall way lights) and motors (like the radio dish and the laser turrets) as per infrared remote. Turning these elements on and off works perfectly fine - I can turn the lights and motors on and off. My issue is with driving the laser turret servo. This is a rather lengthy "function" that is positioning the servo in a semi random pattern with semi random pauses and a lengthy pause at the end of the cycle - currently in the code the pattern is static, but I will make it semi random in the future.

Now, what I would like to achieve is this: While this lengthy function is running, I would like to still be able to turn on and off other lights or the motor for the radio dish. Unfortunately I have no idea how to achieve this.

My lengthy function is within it's own class executed from within the class that executes all functions as per code received from the IR class - see codes below.

Can someone telll me, how to initialise this class in the background, so that the initial class can continue receiving commands from the remote?

Kind regards,

Christian

My main.py:

from sys import platform

from local_IR_GET import IR_GET
from local_ISSUE_CMD import ISSUE_CMD
import setup as setup


if __name__ == '__main__':
    # check if this is a raspberry pico
    if platform == setup.SUPPORTED_PLATFORM:
        command_and_control = ISSUE_CMD()
        
        while(True):
            ir_receiver = IR_GET(setup.ir_receive)
            if setup.LOG_LEVEL >= setup.INFO:
                print('Waiting for IR data...')

            data = str(ir_receiver.acquire())
            if setup.LOG_LEVEL >= setup.DEBUG:
                print('received code is ' + str(data))
            
            what_to_do = command_and_control.getCommandFromCode(data)
            if what_to_do >= 0 and what_to_do <= 4:
                if setup.LOG_LEVEL >= setup.INFO:
                    command_and_control.printWhatToDo(int(what_to_do))
                
                command_and_control.executeCommand(int(what_to_do))
         
            else:
                if setup.LOG_LEVEL >= setup.ERROR:
                    print('something went wrong identifying the command "' +str(what_to_do)+ '" from code')
    else:
        if setup.LOG_LEVEL >= setup.ERROR:
            print('Error wrong architecture. Expected rp2, but detected ' + str(platform))

local_ISSUE_CMD.py with the two classes executing the functions:

from machine import Pin, PWM
from time import sleep
import random

import setup as setup
from local_PIN_CONTROL import PIN_CONTROL


class LASER_TURRET():
    last_turret_start_pos = setup.TURRET_POSITION_ZERO
    last_turret_stop_pos = setup.TURRET_POSITION_180DEGREE
    turret_movement_pattern = []
    
    def __init__(self):
        if setup.LOG_LEVEL >= setup.DEBUG:
            print('initilizing the local laser turret class')
        self.createMovementPattern()
        self.setCenterPos()
        
        movement_pattern = self.getMovemementPattern()
        i=0
        while i < len(movement_pattern):
            if setup.LOG_LEVEL >= setup.DEBUG:
                print()
                print('Get Ready, Round ' + str(i+1) + ' - FIGHT!!!')
            new_movement_pattern = movement_pattern[i]            
            self.moveTurret(new_movement_pattern)
            
            # at the end we have another wait time anyways, so no need to have this interim wait time after the lasst call
            if i+1 < len(movement_pattern):
                wait_time = random.randrange(1000, 5000, 50)
                if setup.LOG_LEVEL >= setup.TRACE:
                    print('waiting ' + str(wait_time/1000) + ' seconds between fighting rounds')
                sleep(wait_time/1000)
            i=i+1

        self.setCenterPos()
        
    def createMovementPattern(self):
        # this one shall create a movement pattern for the turret for a semi erratic yet realistic style
        if setup.LOG_LEVEL >= setup.DEBUG:
            print('creating a new movement pattern')
        self.turret_movement_pattern = (
                        [
                            (4200,8000,46),
                            (8000,9000,30),
                            (9000,1400,50),
                            (1400,2000,35),
                            (2000,4000,45),
                            (4000,6550,35),
                            (6550,7050,45),
                            (7050,9000,50)
                            ],
                        [
                            (9000,3400,-25),
                            (3400,4300,35),
                            (4300,1250,-20),
                            (1250,2300,45),
                            (2300,3200,10),
                            (3200,2350,-25),
                            (2350,7500,45),
                            (7500,4550,-35)
                            ],
                        [
                            (4550,4700,20),
                            (4700,5350,25),
                            (5350,6700,50),
                            (6700,4550,-40),
                            (4550,5000,10),
                            (5000,3050,-35),
                            (3050,1200,-50),
                            (1200,2300,50)
                            ]
                        )
        if setup.LOG_LEVEL >= setup.TRACE:
            i=0
            while i < len(self.turret_movement_pattern):
                print('     pattern ' +str(i) + '=' + str(self.turret_movement_pattern[i]))
                i=i+1
        
    def getMovemementPattern(self):
        if setup.LOG_LEVEL >= setup.CRAZY:
            print('returning movment pattern as ' + str(self.turret_movement_pattern))
        return(self.turret_movement_pattern)
        
    def moveTurret(self, movement_array):
        if setup.LOG_LEVEL >= setup.CRAZY:
            print('received this movement array: ' + str(movement_array))
        
        i = 0
        while i < len(movement_array):
            if setup.LOG_LEVEL >= setup.TRACE:
                print('working on: ' + str(movement_array[i]))
            
            for turret_start,turret_stop,turret_steps in [movement_array[i]]:
                self.setServoCycle(turret_start, turret_stop, turret_steps)
                wait_time = random.randrange(50, 2000, 50)
                if setup.LOG_LEVEL >= setup.TRACE:
                    print('waiting ' + str(wait_time/1000) + ' seconds')
                sleep(wait_time/1000)
            i=i+1
        
    def setServoCycle(self, turret_start=setup.TURRET_POSITION_ZERO, turret_stop=setup.TURRET_POSITION_180DEGREE, turret_steps=setup.TURRET_NO_OF_STEPS, turret_speed=0.01):
        if setup.LOG_LEVEL >= setup.DEBUG:
            print('moving the turret', end='')
            if setup.LOG_LEVEL >= setup.TRACE:
                print(' from ' + str(turret_start) + ' to ' + str(turret_stop) + ' in steps of ' + str(turret_steps) + ' with a speed of ' + str(turret_speed))
            print()
        for pos in range(turret_start,turret_stop,turret_steps):
            if setup.WITH_MOTOR == True:
                setup.laser_turret_drive.duty_u16(pos)
            sleep(turret_speed)
    
    def setCenterPos(self):
        if setup.LOG_LEVEL >= setup.DEBUG:
            print('moving the turret to center', end='')
            if setup.LOG_LEVEL >= setup.TRACE:
                print(' ', setup.TURRET_POSITION_CENTER)
            print()
        if setup.WITH_MOTOR == True:
            setup.laser_turret_drive.duty_u16(setup.TURRET_POSITION_CENTER)

class ISSUE_CMD():
    laser_turret = 0
    def __init__(self):
        if setup.LOG_LEVEL >= setup.DEBUG:
            print('local command class called')
        
    def getCommandState(self, cmd):
        return(setup.commandState[cmd].value())
    
    def toggleCommandState(self, cmd):
        return(setup.commandState[cmd].toggle())
    
    def getCommandName(self, cmd):
        return(setup.commandTexts[cmd])
    
    def getCommandFromCode(self, decodedval):
        i = 0
        for e in setup.commandCodes:
            if setup.LOG_LEVEL >= setup.TRACE:
                print('checking if ' + str(decodedval) + ' is ' + str(e))
            if str(decodedval) == str(e):
                if setup.LOG_LEVEL >= setup.DEBUG:
                    print('identified code ' + str(e) + ' matches command no ' + str(i))
                return(i)
            i = i + 1
    
    def executeCommand(self, cmd):
        if setup.LOG_LEVEL >= setup.DEBUG:
            print('turning ' + str(setup.commandTexts[cmd]), end='')
            if self.getCommandState(cmd) == 0:
                print(' on')
            else:
                print(' off')
        self.toggleCommandState(cmd)
        
        # we also need to drive the laser turret
        if cmd == setup.LASER_TURRET_POS_IN_ARRAY:
            if setup.LOG_LEVEL >= setup.DEBUG:
                print('    also need to drive the laser turret motor')
            
            # initialize the laser turret class and start a new movement pattern
            self.laser_turret = LASER_TURRET()
            # afterwords turn the lights off again
            if setup.LOG_LEVEL >= setup.TRACE:
                print('waiting for ' + str(setup.LASER_LED_OFF_AFTER_SEC) + ' seconds until turning off the lights')
            sleep(setup.LASER_LED_OFF_AFTER_SEC)
            self.toggleCommandState(cmd)
        
    def printWhatToDo(self, cmd):
        print('action is to turn ' + self.getCommandName(int(cmd)) + ' ', end='')
        if self.getCommandState(int(cmd)) == 0:
            print(' on')
        else:
            print(' off')

my setup.py that holds all static values and defines the PINs - just for the sake of completeness:

from machine import Pin, PWM

# this whole thing currently only works on the Raspberry PiCo
SUPPORTED_PLATFORM        = 'rp2'

# if set to true, the motor will actually be driven.
# If not, only the status and possition will be printed
WITH_MOTOR                = True

# after the laser turret turned, how long shall the lights in the laser turret cabin stay turned on
LASER_LED_OFF_AFTER_SEC   = const(10)
# how many times should the turret be moved
TURRET_NUMBER_OF_ROUNDS   = const(3)
TURRET_NO_OF_STEPS        = const(50)
TURRET_POSITION_CENTER    = const(4000)
TURRET_POSITION_ZERO      = const(1000)
TURRET_POSITION_180DEGREE = const(9000)


# debugging constants
CRAZY                     = const(4)
TRACE                     = const(3)
DEBUG                     = const(2)
INFO                      = const(1)
ERROR                     = const(0)
LOG_LEVEL                 = TRACE
START_ANIMATION           = False

COCKPIT_LIGHT_CODE        = '4127785389' # cockpit
LASER_TURRET_CODE         = '4161208749' # lasers
RADIO_DISH_CODE           = '4010803629' # engines
HALL_LIGHTS_CODE          = '4144497069' # R2D2
ENGINE_LIGHT_CODE         = '4261478829' # Wing

COCKPIT_LIGHT_TEXT        = 'Cockpit Lights' # cockpit
LASER_TURRET_TEXT         = 'Laser Turret' # lasers
RADIO_DISH_TEXT           = 'Radio Dish' # engines
HALL_LIGHTS_TEXT          = 'Hall Lights' # R2D2
ENGINE_LIGHT_TEXT         = 'Engine' # Wing

# this array holds all functions of the falcon - it's used for easier accessing the pins
COCKPIT_POS_IN_ARRAY      = const(0)
LASER_TURRET_POS_IN_ARRAY = const(1)
RADIO_DISH_POS_IN_ARRAY   = const(2)
HALL_LIGHTS_POS_IN_ARRAY  = const(3)
ENGINE_POS_IN_ARRAY       = const(4)

# PIN Definitins
IR_INPUT_PIN              = const(0)
COCKPIT_LED_PIN           = const(11)
LASER_TURRET_LED_PIN      = const(12)
LASER_TURRET_MOTOR_PIN    = const(28)
ENGINE_PIN                = const(13)
RADIO_DISH_MOTOR_PIN      = const(14)
HALL_LIGHTS_PIN           = const(15)

# Laser Turret motor
laser_turret_drive = PWM(Pin(LASER_TURRET_MOTOR_PIN))
laser_turret_drive.freq(50)

# Setup of the Pins
ir_receive =       Pin(IR_INPUT_PIN, Pin.IN)
cockpit_led =      Pin(COCKPIT_LED_PIN,      Pin.OUT, value=0)
laser_turret_led = Pin(LASER_TURRET_LED_PIN, Pin.OUT, value=0)
engine_led =       Pin(ENGINE_PIN,           Pin.OUT, value=0)
radio_dish_drive = Pin(RADIO_DISH_MOTOR_PIN, Pin.OUT, value=0)
hall_led =         Pin(HALL_LIGHTS_PIN,      Pin.OUT, value=0)

# and finally the arrays to hold the commands and pin definitins
commandCodes = [COCKPIT_LIGHT_CODE, LASER_TURRET_CODE, RADIO_DISH_CODE, HALL_LIGHTS_CODE, ENGINE_LIGHT_CODE]
commandTexts = [COCKPIT_LIGHT_TEXT, LASER_TURRET_TEXT, RADIO_DISH_TEXT, HALL_LIGHTS_TEXT, ENGINE_LIGHT_TEXT]
commandState = [cockpit_led, laser_turret_led, radio_dish_drive, hall_led, engine_led]

my IR class to retrieve the codes from the remote:

from utime import sleep_ms, ticks_us, ticks_diff
from ir_rx import IR_RX

import setup as setup

class IR_GET(IR_RX):
    def __init__(self, pin, nedges=100, twait=100):
        if setup.LOG_LEVEL >= setup.TRACE:
            print('initializing local IR_GET class with pin=' + str(pin) + '' + ' nedges=' + str(nedges) + ' twait=' + str(twait))
        super().__init__(pin, nedges, twait, lambda *_ : None)
        self._extended = True
        self._addr = 0
        self._leader = 4000
        self.data = None
    
    def decode(self, _):
        if setup.LOG_LEVEL >= setup.TRACE:
            print('decode called with: ' + str(_))
        try:
            if self.edge > 68:
                raise RuntimeError(self.OVERRUN)
            width = ticks_diff(self._times[1], self._times[0])
            if width < self._leader:  # 9ms leading mark for all valid data
                raise RuntimeError(self.BADSTART)
            width = ticks_diff(self._times[2], self._times[1])
            if width > 3000:  # 4.5ms space for normal data
                if self.edge < 68:  # Haven't received the correct number of edges
                    if setup.LOG_LEVEL >= setup.ERROR:
                        print('    Error Have not received the correct number of edges - '+str(self.edge))
                    raise RuntimeError(self.BADBLOCK)
                # Time spaces only (marks are always 562.5µs)
                # Space is 1.6875ms (1) or 562.5µs (0)
                # Skip last bit which is always 1
                val = 0
                for edge in range(3, 68 - 2, 2):
                    val >>= 1
                    if ticks_diff(self._times[edge + 1], self._times[edge]) > 1120:
                        val |= 0x80000000
            elif width > 1700: # 2.5ms space for a repeat code. Should have exactly 4 edges.
                raise RuntimeError(self.REPEAT if self.edge == 4 else self.BADREP)  # Treat REPEAT as error.
            else:
                raise RuntimeError(self.BADSTART)
            
            if setup.LOG_LEVEL >= setup.CRAZY:
                print('received val ' + str(val))
            self.data = val
            
            addr = val & 0xff  # 8 bit addr
            cmd = (val >> 16) & 0xff
            if cmd != (val >> 24) ^ 0xff:
                if setup.LOG_LEVEL >= setup.ERROR:
                    print('    Error cmd '+str(cmd)+' != '+str((val >> 24) ^ 0xff))
                raise RuntimeError(self.BADDATA)
            if addr != ((val >> 8) ^ 0xff) & 0xff:  # 8 bit addr doesn't match check
                if not self._extended:
                    if setup.LOG_LEVEL >= setup.ERROR:
                        print('    Error Bad Address ' + str(addr))
                    raise RuntimeError(self.BADADDR)
                addr |= val & 0xff00  # pass assumed 16 bit address to callback
            self._addr = addr
        except RuntimeError as e:
            cmd = e.args[0]
            addr = self._addr if cmd == self.REPEAT else 0  # REPEAT uses last address
        
        # Set up for new data burst and run user callback
        self.do_callback(cmd, addr, 0, self.REPEAT)
    
    def acquire(self):
        while self.data is None:
            sleep_ms(5)
        self.close()
        return(self.data)

init.py from the original IR receiver code:

# ir_rx __init__.py Decoder for IR remote control using synchronous code
# IR_RX abstract base class for IR receivers.

# Author: Peter Hinch
# Copyright Peter Hinch 2020-2021 Released under the MIT license

from machine import Timer, Pin
from array import array
from utime import ticks_us

# Save RAM
# from micropython import alloc_emergency_exception_buf
# alloc_emergency_exception_buf(100)


# On 1st edge start a block timer. While the timer is running, record the time
# of each edge. When the timer times out decode the data. Duration must exceed
# the worst case block transmission time, but be less than the interval between
# a block start and a repeat code start (~108ms depending on protocol)
class IR_RX():
    # Result/error codes
    # Repeat button code
    REPEAT = const(-1)
    # Error codes
    BADSTART = const(-2)
    BADBLOCK = const(-3)
    BADREP = const(-4)
    OVERRUN = const(-5)
    BADDATA = const(-6)
    BADADDR = const(-7)

    def __init__(self, pin, nedges, tblock, callback, *args):  # Optional args for callback
        self._pin = pin
        self._nedges = nedges
        self._tblock = tblock
        self.callback = callback
        self.args = args
        self._errf = lambda _ : None
        self.verbose = True

        self._times = array('i',  (0 for _ in range(nedges + 1)))  # +1 for overrun
        pin.irq(handler = self._cb_pin, trigger = (Pin.IRQ_FALLING | Pin.IRQ_RISING))
        self.edge = 0
        self.tim = Timer(-1)  # Sofware timer
        self.cb = self.decode
            
    # Pin interrupt. Save time of each edge for later decode.
    def _cb_pin(self, line):
        t = ticks_us()
        # On overrun ignore pulses until software timer times out
        if self.edge <= self._nedges:  # Allow 1 extra pulse to record overrun
            if not self.edge:  # First edge received
                self.tim.init(period=self._tblock , mode=Timer.ONE_SHOT, callback=self.cb)
            self._times[self.edge] = t
            self.edge += 1

    def do_callback(self, cmd, addr, ext, thresh=0):
        self.edge = 0
        if cmd >= thresh:
            self.callback(cmd, addr, ext, *self.args)
        else:
            self._errf(cmd)

    def error_function(self, func):
        self._errf = func

    def close(self):
        self._pin.irq(handler = None)
        self.tim.deinit()

[dpgeorge]

This sounds like a case where using uasyncio would be ideal.

You can create independent tasks for each thing that needs to run concurrently.

[dpgeorge]

See this excellent tutorial: https://github.com/peterhinch/micropython-async/blob/master/v3/docs/TUTORIAL.md

[siliconchris]

Hello and many thanks for your answer,

I have tried to adopt uasyncio in my program. For easier reading, I have omitted the previous issue_cmd class and simply made those functions of the main program. So now I only have the class laser_turret which resides in her own file and is the only thing that incorporates uasyncio - you can see the code below. The class can now run by it's own as well, all that is needed is the setup.py and laser_turret_class.py. However, when I try to run it, I get this error:

Traceback (most recent call last):
  File "<stdin>", line 194, in <module>
  File "<stdin>", line 23, in run
  File "uasyncio/core.py", line 1, in run
  File "uasyncio/core.py", line 1, in run_until_complete
  File "uasyncio/core.py", line 1, in run_until_complete
  File "<stdin>", line 43, in main
  File "uasyncio/funcs.py", line 1, in gather
  File "uasyncio/funcs.py", line 1, in <listcomp>
  File "uasyncio/core.py", line 1, in _promote_to_task
  File "uasyncio/core.py", line 1, in create_task
TypeError: coroutine expected

I am pretty sure I am doing something completely stupid here, but this is the original

async def main(rex):
    bar_task = asyncio.create_task(bar(70))  # Note args here
    tasks = []
    tasks.append(barking(21))
    tasks.append(asyncio.wait_for(foo(10), 7))
    asyncio.create_task(do_cancel(bar_task))
    try:
        res = await asyncio.gather(*tasks, return_exceptions=rex)
    except asyncio.TimeoutError:
        print('foo timed out.')
        res = 'No result'
    print('Result: ', res)

and this is, how I adopted it:

async def main(self):
        movement_pattern = self.getMovemementPattern()
        movement_task = asyncio.create_task(self.executeMovementPattern(movement_pattern))
        
        tasks = []
        tasks.append(setup.commandState[self.cmd].toggle())
        tasks.append(asyncio.sleep(5))
        tasks.append(setup.commandState[self.cmd].toggle())
        
        asyncio.create_task(self.do_cancel(movement_task))
        if setup.LOG_LEVEL >= setup.CRAZY:
            print(tasks)
            
        if setup.LOG_LEVEL >= setup.TRACE:
            gc.collect()
            print(str(gc.mem_free()) + ' free mem, starting async main task with ' + str(tasks))   
        try:
            await asyncio.gather(*tasks)
        except asyncio.TimeoutError:
            print('Laser timed out.')

the complete output is this:

initilizing and executing the local laser turret class
creating a new movement pattern
     pattern 0=[(4200, 8000, 46), (8000, 8700, 20), (8700, 1400, 50)]
     pattern 1=[(1400, 3400, 20), (3400, 4300, 35), (4300, 1250, -20)]
     pattern 2=[(1250, 4700, 40), (4700, 5350, 25), (5350, 6700, 35)]
moving the turret to center  4000


Get Ready, Round 1 - FIGHT!!!
Round 1 working on: (4200, 8000, 46)
moving the turret from 4200 to 8000 in steps of 46 with a speed of 0.01

<Task>
164976 free mem, starting async main task with [None, <SingletonGenerator object at 2000c240>, None]
Traceback (most recent call last):
  File "<stdin>", line 194, in <module>
  File "<stdin>", line 23, in run
  File "uasyncio/core.py", line 1, in run
  File "uasyncio/core.py", line 1, in run_until_complete
  File "uasyncio/core.py", line 1, in run_until_complete
  File "<stdin>", line 43, in main
  File "uasyncio/funcs.py", line 1, in gather
  File "uasyncio/funcs.py", line 1, in <listcomp>
  File "uasyncio/core.py", line 1, in _promote_to_task
  File "uasyncio/core.py", line 1, in create_task
TypeError: coroutine expected

Ans as said, my class looks like this now:

from machine import Pin, PWM
#from time import sleep
import random
import uasyncio as asyncio

import setup as setup
if setup.LOG_LEVEL >= setup.TRACE:
    import gc

class LASER_TURRET():
    last_turret_start_pos = setup.TURRET_POSITION_ZERO
    last_turret_stop_pos = setup.TURRET_POSITION_180DEGREE
    turret_movement_pattern = []
    cmd = setup.LASER_TURRET_POS_IN_ARRAY
    
    def __init__(self):
        if setup.LOG_LEVEL >= setup.DEBUG:
            print('initilizing and executing the local laser turret class')
        self.createMovementPattern()
        self.setCenterPos()
    
    def run(self):
        asyncio.run(self.main())
        
    #def main(self):
    async def main(self):
        movement_pattern = self.getMovemementPattern()
        movement_task = asyncio.create_task(self.executeMovementPattern(movement_pattern))
        
        tasks = []
        tasks.append(setup.commandState[self.cmd].toggle())
        tasks.append(asyncio.sleep(5))
        tasks.append(setup.commandState[self.cmd].toggle())
        
        asyncio.create_task(self.do_cancel(movement_task))
        if setup.LOG_LEVEL >= setup.CRAZY:
            print(tasks)
            
        if setup.LOG_LEVEL >= setup.TRACE:
            gc.collect()
            print(str(gc.mem_free()) + ' free mem, starting async main task with ' + str(tasks))   
        try:
            await asyncio.gather(*tasks)
        except asyncio.TimeoutError:
            print('Laser timed out.')
       
    async def do_cancel(self, task):
        await asyncio.sleep(5)
        if setup.LOG_LEVEL >= setup.INFO:
            print('About to cancel task')
        task.cancel()
    
    #async def executeMovementPattern(self):
    def executeMovementPattern(self, movement_pattern):
        i=0
        setup.commandState[self.cmd].toggle()
        try:
            # loop through the 3 big movement patterns
            while i < len(movement_pattern):
                if setup.LOG_LEVEL >= setup.DEBUG:
                    print()
                    print('Get Ready, Round ' + str(i+1) + ' - FIGHT!!!')
                new_movement_pattern = movement_pattern[i]            
                if setup.LOG_LEVEL >= setup.CRAZY:
                    print('received this movement array: ' + str(new_movement_pattern))
                
                # this loop walks through each inner movement pattern array
                ii = 0
                while ii < len(new_movement_pattern):
                    if setup.LOG_LEVEL >= setup.TRACE:
                        print('Round '+str(i+1)+' working on: ' + str(new_movement_pattern[ii]))
                    
                    for turret_start,turret_stop,turret_steps in [new_movement_pattern[ii]]:
                        self.setServoCycle(turret_start, turret_stop, turret_steps)
                        wait_time = random.randrange(50, 2000, 50)
                        if ii+1 < len(new_movement_pattern):
                            if setup.LOG_LEVEL >= setup.TRACE:
                                print('Round '+str(i+1)+' waiting ' + str(wait_time/1000) + ' seconds')
                            #sleep(wait_time/1000)
                            await asyncio.sleep(wait_time/1000)
                    ii=ii+1
                
                # at the end we have another wait time anyways, so no need to have this interim wait time after the lasst call
                if i+1 < len(movement_pattern):
                    wait_time = random.randrange(1000, 5000, 50)
                    if setup.LOG_LEVEL >= setup.TRACE:
                        print('waiting ' + str(wait_time/1000) + ' seconds between fighting rounds ' + str(i+1) + ' and ' + str(i+2))
                    #sleep(wait_time/1000)
                    await asyncio.sleep(wait_time/1000)
                i=i+1
        except asyncio.CancelledError:
            if setup.LOG_LEVEL >= setup.ERROR:
                print('Trapped executeMovementPattern timeout.')
            raise
        
        # afterwords turn the lights off again
        if setup.LOG_LEVEL >= setup.TRACE:
            print('waiting for ' + str(setup.LASER_LED_OFF_AFTER_SEC) + ' seconds until turning off the lights')
        #sleep(setup.LASER_LED_OFF_AFTER_SEC)
        await asyncio.sleep(setup.LASER_LED_OFF_AFTER_SEC)
        setup.commandState[self.cmd].toggle()
    
    def setServoCycle(self, turret_start=setup.TURRET_POSITION_ZERO, turret_stop=setup.TURRET_POSITION_180DEGREE, turret_steps=setup.TURRET_NO_OF_STEPS, turret_speed=0.01):
        if setup.LOG_LEVEL >= setup.DEBUG:
            print('moving the turret', end='')
            if setup.LOG_LEVEL >= setup.TRACE:
                print(' from ' + str(turret_start) + ' to ' + str(turret_stop) + ' in steps of ' + str(turret_steps) + ' with a speed of ' + str(turret_speed))
            print()
        for pos in range(turret_start,turret_stop,turret_steps):
            if setup.WITH_MOTOR == True:
                setup.laser_turret_drive.duty_u16(pos)
            sleep(turret_speed)
    
    def setCenterPos(self):
        if setup.LOG_LEVEL >= setup.DEBUG:
            print('moving the turret to center', end='')
            if setup.LOG_LEVEL >= setup.TRACE:
                print(' ', setup.TURRET_POSITION_CENTER)
            print()
        if setup.WITH_MOTOR == True:
            setup.laser_turret_drive.duty_u16(setup.TURRET_POSITION_CENTER)
    
    def createMovementPattern(self):
        # this one shall create a movement pattern for the turret for a semi erratic yet realistic style
        if setup.LOG_LEVEL >= setup.DEBUG:
            print('creating a new movement pattern')
            if setup.LASER_TURRET_PATTERN_DEMO == True:
                self.turret_movement_pattern = (
                            [
                                (4200,8000,46),
                                (8000,8700,20),
                                (8700,1400,50)
                                ],
                            [
                                (1400,3400,20),
                                (3400,4300,35),
                                (4300,1250,-20)
                                ],
                            [
                                (1250,4700,40),
                                (4700,5350,25),
                                (5350,6700,35)
                                ]
                            )
            else:
                self.turret_movement_pattern = (
                            [
                                (4200,8000,46),
                                (8000,8700,20),
                                (8700,1400,50),
                                (1400,2000,35),
                                (2000,4000,45),
                                (4000,6550,35),
                                (6550,7050,45),
                                (7050,8700,50),
                                (8700,4500,50)
                                ],
                            [
                                (4500,3400,-20),
                                (3400,4300,35),
                                (4300,1250,-20),
                                (1250,2300,35),
                                (2300,2900,10),
                                (2900,3200,10),
                                (3200,2050,-25),
                                (2050,7500,45),
                                (7500,4550,-35)
                                ],
                            [
                                (4550,4700,20),
                                (4700,5350,25),
                                (5350,6700,50),
                                (6700,4550,-40),
                                (4550,5000,10),
                                (5000,1200,-35),
                                (1200,3050,20),
                                (3050,5000,20),
                                (5000,4000,-20)
                                ]
                            )
        if setup.LOG_LEVEL >= setup.TRACE:
            i=0
            while i < len(self.turret_movement_pattern):
                print('     pattern ' +str(i) + '=' + str(self.turret_movement_pattern[i]))
                i=i+1
    
    def getMovemementPattern(self):
        if setup.LOG_LEVEL >= setup.CRAZY:
            print('returning movment pattern as ' + str(self.turret_movement_pattern))
        return(self.turret_movement_pattern)

if __name__ == '__main__':
    laser_turret = LASER_TURRET()
    laser_turret.run()

[dpgeorge]

Your await asyncio.gather(*tasks) must pass a list of tasks. But it looks like tasks is a list of non-tasks. In particular, you can't add asyncio.sleep(5) to that list, it won't work.

You probably need to refactor your code more to be based around asyncio. Eg:

async def move_turret(x):
    x.toggle()
    await asyncio.sleep(5)
    x.toggle()

async def main():
    turret_task = asyncio.create_task(move_turret(x))
    await turret_task

asyncio.run(main())