Multithreaded Async Commands for Multiple Devices

[jasonacox]

Request from Alfonso:

Hi Jason, very nice package. I’m not a python programmer and I was be able to play with it only following the examples. It works very nice. The only problem is that I cannot find a way to turn on 12 lights at the same time with a routine. Is there any multithread function to do that in the package?

There are no built-in multithreaded functions. There have been suggestions to add asyncio (Python 3) and I suspect that will happen at some point. In the meantime, you can have python spawn threads or you can use the “no wait” send-only functionality of TinyTuya functions like so:

import tinytuya

lights = [{'id': 'xxxxxxxxxxxxxxxxxxID1', 'ip': '10.1.1.2', 'key': '1234567890123abc'}, {'id': 'xxxxxxxxxxxxxxxxxxID2', 'ip': '10.1.1.2', 'key': '1234567890123abc'}]

# Loop through all devices
for light in lights:
    DEVICEID = light["id"]
    DEVICEIP = light["ip"]
    DEVICEKEY = light["key"]
    # Connect
    print("Turning on %s..." % DEVICEID)
    d = tinytuya.BulbDevice(DEVICEID, DEVICEIP, DEVICEKEY)
    # Turn it on
    d.turn_on(nowait=True)

[foncci]

I tried 3 ways I found on a page like this https://superfastpython.com/multiprocessing-pool-for-loop/ but I always get some kind of error inside the library that I never got if I run it synchronously. Any idea? Thank in advance.
This is the code

import tinytuya
import time
import ClassRutinas
import concurrent.futures

MyDevices = [
    ("LP01", "xxx", "192.168.6.216", "www"),
    ("LP02", "yyy", "192.168.6.215", "zzz"),
]

# Cycle through the Rainbow
rainbow = {"red": [255, 0, 0], "orange": [255, 127, 0], "yellow": [255, 200, 0],
           "green": [0, 255, 0], "blue": [0, 0, 255], "indigo": [46, 43, 95],
           "violet": [139, 0, 255]}

ArrLuces = []
for x in MyDevices:
    ArrLuces.append(ClassRutinas.Rutinas())
    if x[0] == "LP01":
        ArrLuces[-1].Connect(x[1], x[2], x[3])
    if x[0] == "LP02":
        ArrLuces[-1].Connect(x[1], x[2], x[3])

with concurrent.futures.ProcessPoolExecutor() as executor:
    p1 = executor.submit(target=ArrLuces[0].Flash())
    p2 = executor.submit(target=ArrLuces[1].Flash())



# File ClassRutinas.py
import tinytuya

cBrightStep = 200
cBrightStepMin = 25
cBrightStepMax = 1000


class Rutinas:
    def Connect(self, DriverID, IPAddr, LocalKey):
        self.LP = tinytuya.BulbDevice(DriverID, IPAddr, LocalKey)
        self.LP.set_version(3.3)  # -1 Es el ultimo elemento de la lista
        # Optional: Keep socket open for multiple commands
        self.LP.set_socketPersistent(True)

    def RGBDim(self):
        self.LP.turn_on()

        self.LP.set_colour(0, 0, 0, nowait=False)
        self.LP.set_brightness(cBrightStepMin, nowait=False)
        for x in range(3):
            if x == 0:
                self.LP.set_colour(255, 0, 0, nowait=False)
            if x == 1:
                self.LP.set_colour(0, 255, 0, nowait=False)
            if x == 2:
                self.LP.set_colour(0, 0, 255, nowait=False)

            for x in range(1, 1010, cBrightStep):
                if x > 1000:
                    x = 1000
                self.LP.set_brightness(x, nowait=False)
            for x in range(1010, 1, -1*cBrightStep):
                if x < cBrightStepMin:
                    x = cBrightStepMin
                self.LP.set_brightness(x, nowait=False)

        self.LP.set_colour(0, 0, 0, nowait=False)
        self.LP.set_brightness(0, nowait=False)
        self.LP.turn_off()

    def Flash(self):
        self.LP.turn_on()

        self.LP.set_brightness(cBrightStepMin, nowait=False)
        for x in range(10):
            if x % 2 == 0:
                self.LP.set_brightness(cBrightStepMax, nowait=False)
            else:
                self.LP.set_brightness(cBrightStepMin, nowait=False)

        self.LP.set_colour(0,0,0, nowait=False)
        self.LP.set_brightness(0, nowait=False)
        self.LP.turn_off()

[uzlonewolf]

My scanner rewrite basically does this for status polling; what you need to do is set the sockets as non-blocking, dump them all into select.select(), and then do recv processing once you have enough data. To handle that last part I basically wrote a wrapper class that handled timeouts and parsed the received data when select() fired on ready-to-read.

# <create wrapper instances for all devices here>

    while True:
        if client:
            read_socks = [client, clients]
        else:
            read_socks = []

        write_socks = []
        all_socks = {}
        remove = []

        <some timeout/remove code snipped>

        for dev in devicelist:
            if do_timeout and dev.hard_time_limit < time.time():
                dev.stop()
            if dev.remove:
                remove.append(dev)
                continue
            else:
                if do_timeout and dev.timeo < time.time():
                    dev.timeout()

            if not dev.sock:
                continue

            if dev.read:
                read_socks.append(dev.sock)

            if dev.write:
                write_socks.append(dev.sock)

            all_socks[dev.sock] = dev

        for dev in remove:
            devicelist.remove(dev)

        try:
            if len(write_socks) > 0:
                rd, wr, _ = select.select( read_socks, write_socks, [], 2 )
            else:
                rd, _, _ = select.select( read_socks, [], [], 2 )
                wr = []
        except KeyboardInterrupt as err:
            log.debug("Keyboard Interrupt - Exiting")
            if verbose: print("\n**User Break**")
            sys.exit()

        # these sockets are now writable, just connected, or failed
        for sock in wr:
            all_socks[sock].write_data()

        # these sockets are now have data waiting to be read
        for sock in rd:
            all_socks[sock].read_data()

If you use a loopback socket to pass messages back and forth, each "half" of the program will be single-threaded and will avoid sharing Python's Global Interpreter Lock. I'll see if I can find time to write a fully usable module, probably some time next week.