1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
|
#!/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. <danix@danix.xyz>
# Licensed under the GNU General Public License v2 only.
"""Run the Hyprland border's gradient along every Tuya addressable strip.
Usage: udt-ledstrip
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.
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 not stored: the light broadcasts
it on UDP 6667 every few seconds, so a DHCP change costs nothing.
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"
# 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
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.
FLOW, FLOW_FLAGS = 0x0A, 0xE0
def scene(colours):
"""DP 51 bytes playing #rrggbb colours as a flow, base64 for the wire."""
body = bytearray([1, 0, FLOW, SPEED, SPEED, FLOW_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(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])
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:]:
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 colours or not lights:
return
dps = {"21": "scene", "51": scene(colours)}
for dev_id, ip in discover(set(lights)).items():
try:
set_dps(ip, lights[dev_id], dps)
except Exception as e:
print(f"udt-ledstrip: {dev_id} at {ip}: {e}", file=sys.stderr)
if __name__ == "__main__":
main()
|
