Control a BlissLights Sky Lite Evolve (Tuya product jteg9ktrhbxtiicz) locally

evolve.py

#!/usr/bin/env python3
"""
Control a BlissLights Sky Lite Evolve (Tuya product jteg9ktrhbxtiicz) locally.

Protocol version 3.5 -- this matters, see notes at the bottom.

Usage:
    ./evolve.py status
    ./evolve.py on | off
    ./evolve.py color <hue 0-360> <sat 0-100> <val 0-100>
    ./evolve.py laser on|off
    ./evolve.py laser-brightness <0-100>
    ./evolve.py motion on|off
    ./evolve.py speed <0-100>
    ./evolve.py effect scene|laser|colour|manual|music
    ./evolve.py theme cosmos|sky|space|ocean|custom1|custom2|custom3|<base64 blob>
    ./evolve.py timer <minutes>
    ./evolve.py raw <dp> <value>      # value: true/false/int/string
    ./evolve.py monitor               # watch pushed state changes
"""
import base64
import json
import sys

import tinytuya

DEVICE_ID = "<insert device id here>"
IP = "192.168.2.99"
LOCAL_KEY = "<insert local key here>"
VERSION = 3.5

# DP map, from make-all/tuya-local galaxy_projector_light.yaml + live probing
DP_SWITCH = 20          # bool   main power
DP_RGBHSV = 24          # hex    hhhhssssvvvv (h 0-360, s 0-1000, v 10-1000)
DP_TIMER = 26           # int    seconds, 0-86400
DP_EFFECT = 51          # str    scene|laser|colour|manual|music
DP_COLOR_MODE = 52      # bool   False=white, True=hs
DP_LASER = 53           # bool   laser on/off
DP_LASER_BRIGHT = 54    # int    10-1000
DP_THEME = 58           # str    base64: AAA=cosmos AAE=sky AAI=space AAM=ocean
DP_MOTION = 60          # bool   laser movement
DP_SPEED = 62           # int    0-100
DP_NEBULA = 101         # int    0-1000, not in the yaml spec. Accepts any int in
                        # range; a string coerces to 0. Scene mode drives it to
                        # 1000 alongside laser_brightness=50, so it is very
                        # likely the nebula/cloud brightness (unverified visually).

# Built-in themes are a 2-byte big-endian index.
THEMES = {"cosmos": "AAA=", "sky": "AAE=", "space": "AAI=", "ocean": "AAM="}

# Custom themes made in the app are NOT an index -- dp 58 carries the whole scene
# definition inline (46 bytes for this one). Captured off the wire by listening on a
# second connection while selecting the theme in the app; the device pushes dp 58 to
# connected clients on change even though it never appears in a status poll.
# Add more by re-running that capture.
CUSTOM_THEMES = {
    # 4 colours, breathe
    "custom1": "ABQBAg4EAbgD6APoAGQAAAH0A+gAqgDmAAAAAAAAAQBkAQPoAWQDSAGkAAAAAA==",
    # 5 colours, flash -- 54 bytes, proves the blob is variable length
    "custom2": "ABQBARsFAmID6APoAGQAAAH0A+gAqgDmAAAAAAAAAf4CWAIcAQBkAQPoAWQDSAGkAAAAAAJs",
    # 1 colour, static -- 22 bytes, the shortest form; theme id 21 not 20
    "custom3": "ABUBAAEBAAAD6APoAQBkAQPoAWQAAA==",
}

DP_NAMES = {
    20: "switch", 24: "hsv", 26: "timer(s)", 51: "effect", 52: "color_mode",
    53: "laser", 54: "laser_brightness", 58: "theme", 60: "motion",
    62: "speed", 101: "nebula?(101)",
}


def connect():
    d = tinytuya.Device(DEVICE_ID, IP, LOCAL_KEY, version=VERSION)
    d.set_socketTimeout(8)
    d.set_socketPersistent(True)
    return d


def encode_hsv(h, s, v):
    """h 0-360, s 0-100, v 0-100 -> 12-char hex the device wants."""
    return f"{h:04x}{int(s * 10):04x}{max(int(v * 10), 10):04x}"


def decode_hsv(hexstr):
    h = int(hexstr[0:4], 16)
    s = int(hexstr[4:8], 16) / 10
    v = int(hexstr[8:12], 16) / 10
    return h, s, v


# --- custom theme blob codec (dp 58, variable length) ------------------------
# Decoded differentially from single-variable edits made in the app. Layout, where
# n is the colour count:
#   [0:2]              theme id: 0-3 = built-ins, 20+ = custom themes (seen 20, 21)
#   [2]                unidentified, 0x01 in every sample so far
#   [3]                mode (0=static, 1=flash, 2=breathe)
#   [4]                cloud speed
#   [5]                colour count n
#   [6 : 6+6n]         n colours as (r, b, g), 16-bit BE, percent x 10
#   [6+6n : 6+6n+8]    01, cloud_bright(16), 01, laser(16), 01, motor(8)
#   [6+6n+8 : end]     n whites, 16-bit BE, percent x 10
# Total length is 14 + 8n. Note the whites are a separate trailing block, NOT
# inline with each colour, so adding a colour grows the blob in two places.

MODES = {0: "static", 1: "flash", 2: "breathe"}
MAX_COLORS = 8          # the app clamps here


def decode_theme(blob):
    b = base64.b64decode(blob)
    if len(b) < 6:
        raise SystemExit(f"theme blob too short: {len(b)} bytes")
    n = b[5]
    if len(b) != 14 + 8 * n:
        raise SystemExit(
            f"theme blob is {len(b)} bytes, expected {14 + 8 * n} for {n} colours"
        )
    w16 = lambda o: int.from_bytes(b[o:o + 2], "big")
    s = 6 + 6 * n          # settings block
    wo = s + 8             # whites block
    return {
        "marker": w16(0),
        "_unk2": b[2],
        "mode": b[3],
        "cloud_speed": b[4],
        "colors": [
            (w16(6 + c * 6) // 10, w16(8 + c * 6) // 10,
             w16(10 + c * 6) // 10, w16(wo + c * 2) // 10)   # r, b, g, w
            for c in range(n)
        ],
        "_m1": b[s], "cloud_bright": w16(s + 1),
        "_m2": b[s + 3], "laser": w16(s + 4) // 10,          # 0-100 (stored x10)
        "_m3": b[s + 6], "motor": b[s + 7],                  # 0-100 (stored raw)
    }


def encode_theme(t):
    n = len(t["colors"])
    if not 1 <= n <= MAX_COLORS:
        raise SystemExit(f"colour count must be 1-{MAX_COLORS}, got {n}")
    b = bytearray(14 + 8 * n)
    p16 = lambda o, v: b.__setitem__(slice(o, o + 2), int(v).to_bytes(2, "big"))
    s, wo = 6 + 6 * n, 6 + 6 * n + 8
    p16(0, t["marker"])
    b[2] = t["_unk2"]
    b[3] = t["mode"]
    b[4] = t["cloud_speed"]
    b[5] = n
    for c, (r, bl, g, w) in enumerate(t["colors"]):
        p16(6 + c * 6, r * 10)
        p16(8 + c * 6, bl * 10)
        p16(10 + c * 6, g * 10)
        p16(wo + c * 2, w * 10)
    b[s] = t["_m1"]
    p16(s + 1, t["cloud_bright"])
    b[s + 3] = t["_m2"]
    p16(s + 4, t["laser"] * 10)
    b[s + 6] = t["_m3"]
    b[s + 7] = t["motor"]
    return base64.b64encode(bytes(b)).decode()


def show_theme(t):
    mode = MODES.get(t["mode"], f"unknown({t['mode']})")
    kind = "built-in index" if t["marker"] < 4 else "custom theme id"
    print(f"  marker       {t['marker']} ({kind})")
    print(f"  mode         {mode}")
    print(f"  cloud_speed  {t['cloud_speed']}")
    print(f"  cloud_bright {t['cloud_bright']}")
    print(f"  laser        {t['laser']}%")
    print(f"  motor        {t['motor']}%")
    print(f"  colors ({len(t['colors'])}, r/b/g/w as %):")
    for i, c in enumerate(t["colors"], 1):
        print(f"    {i}: {c}")
    print(f"  unknown      [2]={t['_unk2']}  markers={t['_m1']},{t['_m2']},{t['_m3']}")


def show(result):
    dps = result.get("dps") if isinstance(result, dict) else None
    if not dps:
        print(json.dumps(result, indent=2))
        return
    for k in sorted(dps, key=int):
        name = DP_NAMES.get(int(k), "?")
        val = dps[k]
        extra = ""
        if int(k) == DP_RGBHSV and isinstance(val, str) and len(val) == 12:
            h, s, v = decode_hsv(val)
            extra = f"  (h={h} s={s:.0f}% v={v:.0f}%)"
        elif int(k) == DP_THEME:
            rev = {v: n for n, v in THEMES.items()}
            extra = f"  ({rev.get(val, '?')})"
        elif int(k) == DP_TIMER and isinstance(val, int):
            extra = f"  ({val // 60} min)"
        print(f"  {k:>4} {name:<18} {val!r}{extra}")


def parse_bool(s):
    if s.lower() in ("on", "true", "1", "yes"):
        return True
    if s.lower() in ("off", "false", "0", "no"):
        return False
    raise SystemExit(f"expected on/off, got {s!r}")


def main(argv):
    if not argv:
        print(__doc__)
        return 1

    cmd, args = argv[0], argv[1:]
    d = connect()

    if cmd == "status":
        show(d.status())

    elif cmd in ("on", "off"):
        show(d.set_value(DP_SWITCH, cmd == "on"))

    elif cmd == "color":
        h, s, v = (int(x) for x in args)
        # colour mode must be on, else the hsv write is ignored
        show(d.set_multiple_values({
            DP_COLOR_MODE: True,
            DP_EFFECT: "colour",
            DP_RGBHSV: encode_hsv(h, s, v),
        }))

    elif cmd == "laser":
        show(d.set_value(DP_LASER, parse_bool(args[0])))

    elif cmd == "laser-brightness":
        show(d.set_value(DP_LASER_BRIGHT, max(int(args[0]) * 10, 10)))

    elif cmd == "motion":
        show(d.set_value(DP_MOTION, parse_bool(args[0])))

    elif cmd == "speed":
        show(d.set_value(DP_SPEED, int(args[0])))

    elif cmd == "effect":
        show(d.set_value(DP_EFFECT, args[0]))

    elif cmd == "theme":
        key = args[0]
        if key.lower() in THEMES:
            val = THEMES[key.lower()]
        elif key.lower() in CUSTOM_THEMES:
            val = CUSTOM_THEMES[key.lower()]
        elif key.endswith("=") or len(key) > 8:
            val = key  # raw base64 blob straight through
        else:
            raise SystemExit(
                f"theme must be one of {', '.join(list(THEMES) + list(CUSTOM_THEMES))}, "
                "or a raw base64 blob"
            )
        # scene data is ignored unless the device is in scene mode
        show(d.set_multiple_values({DP_EFFECT: "scene", DP_THEME: val}))

    elif cmd == "timer":
        show(d.set_value(DP_TIMER, int(args[0]) * 60))

    elif cmd == "raw":
        dp, val = args[0], args[1]
        if val.lower() in ("true", "false"):
            val = val.lower() == "true"
        else:
            try:
                val = int(val)
            except ValueError:
                pass
        show(d.set_value(int(dp), val))

    elif cmd == "theme-decode":
        key = args[0]
        blob = CUSTOM_THEMES.get(key.lower(), key)
        show_theme(decode_theme(blob))
        d.close()
        return 0

    elif cmd == "theme-edit":
        # ./evolve.py theme-edit custom1 laser=50 motor=80 cloud_speed=30 c1=44,100,100,84
        key = args[0]
        t = decode_theme(CUSTOM_THEMES.get(key.lower(), key))
        for kv in args[1:]:
            k, _, v = kv.partition("=")
            if k.startswith("c") and k[1:].isdigit():
                idx = int(k[1:]) - 1
                col = tuple(int(x) for x in v.split(","))
                if len(col) != 4:
                    raise SystemExit(f"{k} needs r,b,g,w (4 values), got {len(col)}")
                if idx == len(t["colors"]):
                    t["colors"].append(col)          # append a new colour
                elif 0 <= idx < len(t["colors"]):
                    t["colors"][idx] = col
                else:
                    raise SystemExit(
                        f"colour {idx + 1} out of range: have {len(t['colors'])}, "
                        f"next appendable is c{len(t['colors']) + 1} (max {MAX_COLORS})"
                    )
            elif k == "mode":
                rev = {name: num for num, name in MODES.items()}
                if v.lower() in rev:
                    t["mode"] = rev[v.lower()]
                elif v.isdigit():
                    t["mode"] = int(v)
                else:
                    raise SystemExit(f"mode must be {'/'.join(rev)} or a number")
            elif k in ("laser", "motor", "cloud_speed", "cloud_bright"):
                t[k] = int(v)
            else:
                raise SystemExit(f"unknown field {k!r}")
        blob = encode_theme(t)
        print(f"  blob: {blob}")
        show_theme(decode_theme(blob))
        show(d.set_multiple_values({DP_EFFECT: "scene", DP_THEME: blob}))

    elif cmd == "monitor":
        d.set_socketTimeout(15)
        print("watching for pushed updates, ^C to stop")
        show(d.status())
        while True:
            data = d.receive()
            if data:
                print("---")
                show(data)
            d.heartbeat(nowait=True)

    else:
        print(__doc__)
        return 1

    d.close()
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))