Are there any examples of people using Tinytuya for a full-fledged smart home GUI?

[RYoder97]

Hi again. Just wanted to see if anyone has successfully created a GUI to control all their smart devices that I could learn from code-wise. I'm still very new to Python after all.

My vision is to make a GUI to control all gadgetry in my home. This includes both Tuya and things that I may want to build on my own using Raspberry Pi (security cameras, battery operated devices with a Pi Pico W, etc.) Tuya is great because brand doesn't matter so long as it's powered by Tuya it should still work with Tinytuya. I'm still having a tad bit of trouble though.

My latest hurdle is statuses. I'm not sure if it's my specific devices (the brands I'm using like Feit and Geeni), but I can't seem to accurately monitor the devices.

I will explain my code here, and try to attach the entire files since I see there is an option for it.

Basically, I start a BottlePy server for the front end of the GUI, and just make requests to the server to execute Python code. Getting statuses is done via a websocket, while requests to set DPS is done through AJAX POST.

I have an infinitely running while loop in a separate thread that is supposed to get the device status of every device. It can take a bit of time, about 70ish milliseconds per device I reckon. About 1.2 full seconds to fully iterate over my 18 devices. This isn't a big deal though, since mainly I just want to get the initial status of each device to give the GUI a starting point.

But when I turn off a device from the front end, it doesn't seem to update on the back end for a few iterations, so maybe 4+ seconds. My GUI will show the light as turned off/on because I obviously update the GUI right away once the device is clicked on, but then my websocket will retrieve data from the back end and will force the GUI to display incorrectly that the device is in the opposite state.

I understand I could do nowait=True, but obviously that wouldn't return any data to send back to the front end to update accordingly.

Could it be my devices? Or in general does it take time for updates to reflect in the DPS system? A sort of cache? What could I do to make this GUI more..natural feeling with this in mind?
Archive.zip

[jasonacox]

The status should return the actual state, it doesn't cache the state. I think the issue is probably your long running loop taking a while to refresh.

My suggestion is that you have a thread maintain state, cache the state for some TTL amount of time (e.g. 5s) and when the TTL expires, have it poll just that device. When you make a change via UI, make a websocket call should invalidate the TTL cache for that device and force an immediate poll so the websock update message back to the browser would match what you changed manually.

I always wanted to do a nice UI as well, but this is as far as I got and it is NOT using websockets: https://github.com/jasonacox/tinytuya/tree/master/server. - Doing it over I would do:

• FastAPI / uvicorn
• Socket.io for all client/browser updates (user updates or refresh from backend)

Thanks for sharing your progress!

[RYoder97]

Thanks for letting me know about your endeavors with a UI. I was actually able to get my existing code to run...just on a regular laptop with Raspberry Pi Desktop OS (not my Chromebook). Go figures..something is odd with how my Chromebook handles Linux and networking. For one, the inability to discover devices using UDP despite turning on port forwarding for the Linux container. (What a pain it's been to manually enter the IP's and Tuya versions for each device in a copy of devices.json. :()

But because of these poor networking capabilities, it seems that communication for something this sensitive is much slower than it should be. Lot of tinytuya network errors on that device too, it seems. The networking chip may just be going bad honestly. But meanwhile this eye sore of a ThinkPad L450 with true Linux seems to excel at connecting to and staying connected to the devices.

But I noticed in your explanation you mentioned that it would use the websock to match what I changed manually...But what about through the Tuya app or better yet, through Google Home? I know you mentioned that you can't use Tinytuya at the same time TuyaSmart is open, but I can't say I've seen any issues running both at once.

But anyways, as the code is now, there's still about a 10s delay from when I turn a bulb off on Google Home OR TuyaSmart. This is because the thread with the while loop takes time to re-iterate through every device. I suppose this isn't a huge deal though, considering that this UI I'm designing is meant to replace Google Home in MOST instances. My ultimate goal is to get this running off a Pi Zero W 2's with touch displays to create a sort of control panel, and put the panels in most of the rooms of our house, typically above/near the room's corresponding light switch. My wife loves the voice control. I do too, but being able to have more intimate control only a few feet away (and something that can control non-Google Home devices like Pi Pico W's, Raspi security cameras, etc.) is more preferred by me.

Overall, now that I'm testing the code on a pure Linux system, the code as-is seems to work as a "source of truth" for device statuses, with a 10 second delay if we want to use a voice command.

[RYoder97]

OK. After doing some research, I found that for loops are faster than infinite while loops in Python.

So I updated the device status thread like so:

def getDeviceStatuses():
    global devices
    global statuses
    
    x = 0
    for device in devices:
        statuses[x] = {"id":device.id, "status":device.status()}
        x += 1
    
    getDeviceStatuses()

And instead of creating a tmp list and populating it at the end of the loop, we update the global statuses during each iteration, using x as an index to tell us which device we are updating. Meanwhile, the websocket continues to constantly push updates to the client side, but now instead of waiting for a while loop to populate a new list, the existing list is just updated.

This has resulted in only a couple seconds of a delay between issuing a voice command to Google, and 1ish second delay between two devices running the front end of the GUI. (The host device and then my phone with Chrome with the GUI web page open).

This is getting really cool, and is really helping me learn Python. I want to try testing this on a Pi 4 eventually, to see if it can serve as a central server for the Python code. And then use Pi Zero W's as "clients". This would be the most cost effective route as opposed to using multiple Pi 3's, 4's, or even 5's to basically just run a web page on boot.

[uzlonewolf]

Starting a thread for every device has always felt like a dirty kludge to me, I much prefer the traditional socket programming method of dumping a bunch of open sockets into select(). Python's threading isn't true multithreading either - the Global Interpreter Lock (GIL) means there will only ever be 1 thread running at a time. This can be good in that you can be sure your program will never use more than 1 CPU core, but bad if you actually want it to use multiple cores.

While I'm not planning on making a full-fledged GUI for it, I've started (but still haven't finished) a program I'm calling Multi-Server. It takes a list of devices, opens persistent connections to them all, and sends async status updates in real time to every connected client as devices are updated. It uses JSON to pass data and commands to/from the clients and currently supports client connections by WebSocket and plain TCP, both TLS-encrypted and plain text, and I plan on adding MQTT support at some point. It's still unfinished in that devices must be online at server start or they won't be seen, and handling device disconnects/reconnects is pretty rough at the moment, but it otherwise works. https://github.com/uzlonewolf/tinytuya/tree/socket-server/multiserver

[jasonacox]

Thanks @uzlonewolf, we should finish that. 😉 Select is more elegant for all the reasons you say but may not be as approachable for new users. Still, here is the same script I wrote above but converted from threading to select:

import select
import time
import tinytuya

# Define the devices to control
config = {
    "TuyaDevices": [
        {
            "Address": "192.168.1.10",
            "Device ID": "00112233445566778899",
            "Local Key": "1234567890123abc",
            "Version": "3.3",
        },
        {
            "Address": "192.168.1.11",
            "Device ID": "10112233445566778899",
            "Local Key": "1234567890123abc",
            "Version": "3.3",
        },
        {
            "Address": "192.168.1.12",
            "Device ID": "20112233445566778899",
            "Local Key": "1234567890123abc",
            "Version": "3.3",
        },
        {
            "Address": "192.168.1.13",
            "Device ID": "30112233445566778899",
            "Local Key": "1234567890123abc",
            "Version": "3.3",
        }
    ]
}

# Create array, devices, that is an array of tinytuya.Device objects
devices = []
for i in config["TuyaDevices"]:
    print(f"Connecting to {i['Device ID']} at {i['Address']} with key {i['Local Key']}")
    d = tinytuya.Device(i["Device ID"], i["Address"], i["Local Key"], version=i["Version"])
    devices.append(d)  # Add the device to the devices array

def getDeviceStatuses():
    global devices
    global statuses

    # Enable persistent socket connection for each device
    for device in devices:
        device.set_socketPersistent(True)
        # Call status() once to get the initial status and connect
        initial_status = device.status()
        device_id = device.id
        index = devices.index(device)
        print(f"INITIAL status from {device_id}: {initial_status}")
        statuses[index] = {"id": device_id, "status": initial_status["dps"]}

    # Create a list of sockets to monitor
    sockets = [device.socket for device in devices]

    # Infinite loop to listen for status updates using select
    while True:
        # Use select to wait for any of the device sockets to have data
        readable, _, _ = select.select(sockets, [], [], 0.1)

        # Process each socket with incoming data
        for sock in readable:
            # Find the corresponding device for this socket
            device = next(device for device in devices if device.socket == sock)
            updated_status = device.receive()

            if updated_status:
                print(f"UPDATE status from {device.id}: {updated_status}")
                index = devices.index(device)
                # We may only get one DPS, so just update that one item
                if "dps" in updated_status:
                    for key in updated_status["dps"]:
                        statuses[index]["status"][key] = updated_status["dps"][key]
                        print(f" - Updated status for {device.id} DPS {key} to {updated_status['dps'][key]}")

        # Add a small delay (optional) to prevent tight looping
        time.sleep(0.1)

# Example usage
statuses = [None] * len(devices)  # Initialize statuses list to hold results for each device

# Start the status listener
getDeviceStatuses()

Uploaded examples here: https://github.com/jasonacox/tinytuya/blob/master/examples/multi-select.py

[jasonacox]

I updated both the threading and select examples and put them in our examples list. After playing with both, I'm sure it is something I'm doing wrong with select, but the threading version is considerably more reliable and consistently detects Tuya device state change messages where select often misses them. I also think that moving to asyncio would likely accomplish the same, but I didn't try that.

[uzlonewolf]

You also need to set the sockets to non-blocking or _send_recv() (called from receive()) will block and wait for data if it receives a data-less packet (like an ack). At some point I may add either a new function or a flag to turn this retrying off for when we're doing stuff like this. Also, since TCP connections are actually streams and not packets, to do it correctly we technically should be collecting received data into a buffer and then parse packets out of that, though device responses are so small it's unlikely to be an issue for us.

[jasonacox]

Great idea. I really think we need to expose this and make it easier to use. As more people move from just "running a script" for some control/gather ad-hoc work to true backend continuous services, the more this becomes needed. In fact, we could make those adjustments, with some updated clean APIs (functions), that would help create (or should I say, finish your) standalone server with GUI. 😉