#!/usr/bin/env python3 # SPDX-License-Identifier: GPL-2.0-only # udt-ledstrip: light a Tuya Wi-Fi LED strip in the wallpaper accent. # Copyright (C) 2026 Danilo M. # Licensed under the GNU General Public License v2 only. """Run the Hyprland border's gradient along every Tuya addressable strip. Usage: udt-ledstrip play the gradient, leaving the strip on or off udt-ledstrip on|off switch it, the gradient with it when on udt-ledstrip sleep breathe the accent, soft and dim, while asleep udt-ledstrip --selftest The colours are the three udt-accent wrote for the window border, the accent between its two hue neighbours, played as the strip's "flow" scene: a gradient moving along the strip. udt-accent starts this detached on every run, like udt-keyboard. hypr-theme's autostart runs `on` at login and `off` on Hyprland's shutdown event, so the strip follows the session. `sleep` is sent before a suspend and `on` after it, so a sleeping desktop breathes instead of going dark. The device id and local key come from ~/.config/udt/tuya/devices.json, written once by `python -m tinytuya wizard` against a Tuya IoT cloud project; after that everything is local. Re-pairing a light in the app changes its key, so the wizard has to run again. The address is cached in ~/.cache/udt, because `off` runs as the session ends and cannot wait for discovery; when the cached one fails, the light's own broadcast on UDP 6667 finds it again within seconds. Speaks Tuya protocol 3.5 itself (AES-GCM, session key negotiated per connection) with the `cryptography` module, rather than depending on tinytuya. Without the device list, or with the light unplugged, this does nothing. """ import base64 import colorsys import hashlib import hmac import json import os import re import socket import struct import sys import time from pathlib import Path DEVICES = Path.home() / ".config/udt/tuya/devices.json" BORDER = Path.home() / ".cache/udt/udt-border.lua" ADDRESSES = Path.home() / ".cache/udt/ledstrip.json" # Every Tuya device encrypts its discovery broadcast with this published key. UDP_KEY = hashlib.md5(b"yGAdlopoPVldABfn").digest() DISCOVERY_TIMEOUT = 20 # Calibration, all 0..100. LEDs render the schemes' pastel accents as near # white, so saturation is floored: 100 plays every hue pure, lower lets a # pale accent stay paler than a vivid one. Brightness is fixed rather than # read back from the strip, so it overrides whatever the app last set. SATURATION_FLOOR = 100 BRIGHTNESS = 100 SPEED = 50 # Sleep: the accent alone, desaturated and at half brightness, breathing slowly. SLEEP_SATURATION = 60 SLEEP_BRIGHTNESS = 50 SLEEP_SPEED = 20 SESS_START, SESS_FINISH, CONTROL = 3, 5, 13 # DP 51, dreamlight_scene_mode, as the app writes it (read off the strip's # status pushes; Tuya does not document it): version 1, scene id (0 for a # custom one), mode, speed twice, flags, two zero bytes, brightness, then # per colour a two byte hue and a saturation byte. Mode 0x0a with flags # 0xe0 is the app's moving gradient; mode 0x03 with flags 0 is its breathing. FLOW, FLOW_FLAGS = 0x0A, 0xE0 BREATH, BREATH_FLAGS = 0x03, 0x00 def scene(colours, mode=FLOW, flags=FLOW_FLAGS, speed=SPEED, brightness=BRIGHTNESS, saturation=None): """DP 51 bytes playing #rrggbb colours, base64 for the wire. A fixed saturation replaces each colour's own, floored one.""" body = bytearray([1, 0, mode, speed, speed, flags, 0, 0, brightness]) for c in colours: h, s, _ = colorsys.rgb_to_hsv(*(int(c[i:i + 2], 16) / 255 for i in (1, 3, 5))) body += (round(h * 360) % 360).to_bytes(2, "big") body.append(saturation or max(round(s * 100), SATURATION_FLOOR)) return base64.b64encode(body).decode() def pack(key, seq, cmd, payload): from cryptography.hazmat.primitives.ciphers.aead import AESGCM iv = os.urandom(12) header = struct.pack(">IHIII", 0x6699, 0, seq, cmd, 12 + len(payload) + 16) return header + iv + AESGCM(key).encrypt(iv, payload, header[4:]) + b"\0\0\x99\x66" def unpack(key, msg): """(cmd, plaintext) of one 0x6699 frame.""" from cryptography.hazmat.primitives.ciphers.aead import AESGCM _, _, _, cmd, length = struct.unpack(">IHIII", msg[:18]) body = msg[18:18 + length] return cmd, AESGCM(key).decrypt(body[:12], body[12:], msg[4:18]) def recv_frame(sock): def exact(n): buf = b"" while len(buf) < n: chunk = sock.recv(n - len(buf)) if not chunk: raise ConnectionError("light closed the connection") buf += chunk return buf header = exact(18) return header + exact(struct.unpack(">I", header[14:18])[0] + 4) def discover(ids): """{id: ip} for the wanted ids heard broadcasting within the timeout.""" sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) sock.bind(("", 6667)) found, deadline = {}, time.monotonic() + DISCOVERY_TIMEOUT while set(found) != ids and (left := deadline - time.monotonic()) > 0: sock.settimeout(left) try: data, _ = sock.recvfrom(4096) # Device-sent payloads start with a four byte return code. info = json.loads(unpack(UDP_KEY, data)[1][4:]) except Exception: # timeout, or not a 3.5 frame (InvalidTag) continue if info.get("gwId") in ids: found[info["gwId"]] = info["ip"] return found def set_dps(ip, key, dps): real_key = key.encode() with socket.create_connection((ip, 6668), timeout=5) as sock: local_nonce = os.urandom(16) sock.sendall(pack(real_key, 1, SESS_START, local_nonce)) reply = unpack(real_key, recv_frame(sock))[1][-48:] remote_nonce, mac = reply[:16], reply[16:] if not hmac.compare_digest(mac, hmac.digest(real_key, local_nonce, "sha256")): raise ValueError("handshake failed: local key is wrong or stale") sock.sendall(pack(real_key, 2, SESS_FINISH, hmac.digest(real_key, remote_nonce, "sha256"))) from cryptography.hazmat.primitives.ciphers.aead import AESGCM mixed = bytes(a ^ b for a, b in zip(local_nonce, remote_nonce)) session = AESGCM(real_key).encrypt(local_nonce[:12], mixed, None)[:16] body = json.dumps({"protocol": 5, "t": int(time.time()), "data": {"dps": dps}}, separators=(",", ":")) sock.sendall(pack(session, 3, CONTROL, b"3.5" + b"\0" * 12 + body.encode())) unpack(session, recv_frame(sock)) # the ack; raises if it is garbage def selftest(): raw = base64.b64decode(scene(["#ff0000", "#0000ff"])) assert raw[:9] == bytes([1, 0, FLOW, SPEED, SPEED, FLOW_FLAGS, 0, 0, BRIGHTNESS]) assert raw[9:] == bytes([0, 0, 100, 0, 240, 100]) # As the app wrote a breathing scene: 01 00 03 32 32 00 00 00 32, hue, sat. raw = base64.b64decode(scene(["#00ff00"], BREATH, BREATH_FLAGS, 50, 50, 60)) assert raw == bytes([1, 0, 3, 50, 50, 0, 0, 0, 50, 0, 120, 60]) key = b"0123456789abcdef" assert unpack(key, pack(key, 7, CONTROL, b"payload")) == (CONTROL, b"payload") print("ok") def main(): if sys.argv[1:] == ["--selftest"]: return selftest() if sys.argv[1:] not in ([], ["on"], ["off"], ["sleep"]): sys.exit(__doc__.split("\n\n")[1]) try: devices = json.loads(DEVICES.read_text()) except (OSError, ValueError): return # Lights only: anything else the Tuya account links has no colour DP. # The cloud spec omits DP 51, so this matches on the DP 24 it does list. lights = {d["id"]: d["key"] for d in devices if d.get("mapping", {}).get("24", {}).get("code") == "colour_data"} border = BORDER.read_text() if BORDER.exists() else "" colours = ["#" + h for h in re.findall(r"rgb\(([0-9a-fA-F]{6})\)", border)] if not lights: return if sys.argv[1:] == ["off"]: dps = {"20": False} elif sys.argv[1:] == ["sleep"] and colours: # The border's middle colour is the accent itself. dps = {"20": True, "21": "scene", "51": scene([colours[len(colours) // 2]], BREATH, BREATH_FLAGS, SLEEP_SPEED, SLEEP_BRIGHTNESS, SLEEP_SATURATION)} elif colours: dps = {"21": "scene", "51": scene(colours)} if sys.argv[1:] == ["on"]: dps["20"] = True elif sys.argv[1:] == ["on"]: dps = {"20": True} else: return try: known = json.loads(ADDRESSES.read_text()) except (OSError, ValueError): known = {} pending = set(lights) for dev_id in pending & set(known): try: set_dps(known[dev_id], lights[dev_id], dps) pending.discard(dev_id) except Exception: pass # moved, or unplugged; discovery below settles which if not pending: return for dev_id, ip in discover(pending).items(): try: set_dps(ip, lights[dev_id], dps) known[dev_id] = ip except Exception as e: print(f"udt-ledstrip: {dev_id} at {ip}: {e}", file=sys.stderr) ADDRESSES.parent.mkdir(parents=True, exist_ok=True) ADDRESSES.write_text(json.dumps(known)) if __name__ == "__main__": main()