lianli-wireless/lianli_poc.py

"""Proof of concept: talk to a Lian Li UNI FAN wireless kit (SLV3 TX/RX dongles).

Protocol taken from github.com/sgtaziz/lian-li-linux (MIT). Close L-Connect first.

    python lianli_poc.py scan            show what the receiver reports
    python lianli_poc.py color ff0000    set every bound fan to one colour
    python lianli_poc.py demo            cycle a few colours, verifying each one
"""
import os
import subprocess
import sys
import time

import libusb_package
import usb.core
import usb.util

TX_ID = (0x0416, 0x8040)
RX_ID = (0x0416, 0x8041)
EP_OUT, EP_IN = 0x01, 0x81

USB_CMD_SEND_RF = 0x10
USB_CMD_GET_MAC = 0x11
RF_SELECT = 0x12
RF_SET_RGB = 0x20
RF_DATA_SIZE = 240
RF_CHUNK_SIZE = 60
CHUNK_BYTES = 220
HEADER_REPEATS = 4

TUZC = os.path.join(os.path.dirname(os.path.abspath(__file__)), "tuzc.exe")

# fan-type byte -> (name, LEDs per fan); subset of lian-li-linux fan_type.rs
FAN_TYPES = [
    (range(20, 23), "SL V3 Wireless", 40),
    (range(23, 27), "SL V3 Wireless LCD", 40),
    (range(28, 32), "TL V2 Wireless", 26),
    (range(36, 40), "SL-INF Wireless", 44),
    (range(43, 51), "SL-INF V3 Wireless", 44),
    (range(51, 59), "TL V3 Wireless", 26),
    (range(59, 63), "SL V4 Wireless", 52),
]


def fan_type(byte):
    for ids, name, leds in FAN_TYPES:
        if byte in ids:
            return name, leds
    return f"unknown (0x{byte:02x})", None


def mac_str(mac):
    return ":".join(f"{b:02x}" for b in mac)


class Dongle:
    def __init__(self, ids, name):
        backend = libusb_package.get_libusb1_backend()
        self.dev = usb.core.find(idVendor=ids[0], idProduct=ids[1], backend=backend)
        if self.dev is None:
            sys.exit(f"{name} dongle {ids[0]:04x}:{ids[1]:04x} not found")
        try:
            self.dev.set_configuration()
        except usb.core.USBError:
            pass  # already configured
        usb.util.claim_interface(self.dev, 0)
        self.name = name

    def write(self, data, timeout=5000):
        return self.dev.write(EP_OUT, bytes(data), timeout)

    def read(self, size=64, timeout=100):
        try:
            return bytes(self.dev.read(EP_IN, size, timeout))
        except usb.core.USBTimeoutError:
            return b""

    def read_until_silence(self, first_timeout=100, chunk_timeout=10, limit=26 * 512):
        out, timeout = b"", first_timeout
        while len(out) < limit:
            chunk = self.read(64, timeout)
            if not chunk:
                break
            out += chunk
            timeout = chunk_timeout
        return out

    def flush(self):
        while self.read(64, 5):
            pass

    def close(self):
        usb.util.release_interface(self.dev, 0)
        usb.util.dispose_resources(self.dev)


def find_master(tx):
    """Ask the TX dongle for its MAC; it only answers on its own RF channel."""
    channels = [8] + [c for c in range(2, 39, 2) if c != 8] + list(range(1, 40, 2))
    for channel in channels:
        tx.flush()
        tx.write(bytes([USB_CMD_GET_MAC, channel]) + bytes(62))
        r = tx.read(64, 500)
        if (len(r) >= 13 and r[0] == USB_CMD_GET_MAC and any(r[1:7])
                and int.from_bytes(r[7:11], "big") > 1):
            return r[1:7], channel, int.from_bytes(r[11:13], "big")
    sys.exit("TX dongle did not answer on any channel")


def get_devices(rx):
    """GetDev: the RX dongle lists every wireless device it can hear."""
    pages, records = 1, []
    for _ in range(2):  # second pass if more than 10 devices were reported
        rx.flush()
        rx.write(bytes([USB_CMD_SEND_RF, pages]) + bytes(62))
        r = rx.read_until_silence(limit=pages * 512)
        if len(r) < 4 or r[0] != USB_CMD_SEND_RF:
            sys.exit(f"bad GetDev reply: {r[:16].hex()}")
        total = r[1]
        if (total + 9) // 10 <= pages:
            break
        pages = (total + 9) // 10
    count = min(total, pages * 10)
    for i in range(count):
        rec = r[4 + i * 42: 4 + (i + 1) * 42]
        if len(rec) < 42 or rec[41] != 0x1C or rec[18] == 0xFF:
            continue  # 0xFF records are master dongles, not fans
        raw_fans = rec[19]
        fan_bytes = list(rec[24:28])
        type_byte = next((b for b in fan_bytes if b), 0)
        records.append({
            "mac": rec[0:6],
            "master": rec[6:12],
            "channel": rec[12],
            "rx": rec[13],
            "device_type": rec[18],
            "fans": min(raw_fans - 10 if raw_fans >= 10 else raw_fans, 4),
            "right_attach": raw_fans >= 10,
            "fan_type": type_byte,
            "effect": rec[20:24],
            "rpm": [((rec[28 + 2 * k] & 0x0F) << 8) | rec[29 + 2 * k] for k in range(4)],
            "pwm": list(rec[36:40]),
            "mb_sync": bool(rec[28] & 0x40),
        })
    return records


def tinyuz(raw):
    p = subprocess.run([TUZC], input=raw, capture_output=True)
    if p.returncode:
        sys.exit(f"tuzc failed with exit code {p.returncode}")
    return p.stdout


def build_upload(colors, interval_ms=5000):
    """One static frame, packed like WirelessRgbUpload in rgb.rs."""
    raw = b"".join(bytes(c) for c in colors)
    compressed = tinyuz(raw)
    hundredths = interval_ms * 160
    ticks, fraction = hundredths // 100, hundredths % 100
    frame_count, led_count = 1, len(colors)
    h = 0x811C9DC5
    tail = (ticks.to_bytes(2, "big") + bytes([fraction]) + frame_count.to_bytes(2, "big")
            + bytes([led_count]) + bytes(2) + bytes(2) + bytes(1))
    for b in compressed + tail:
        h = ((h ^ b) * 0x01000193) & 0xFFFFFFFF
    effect = bytearray(h.to_bytes(4, "big"))
    effect[0] |= 0x80
    return {"compressed": compressed, "ticks": ticks, "fraction": fraction,
            "frames": frame_count, "leds": led_count, "effect": bytes(effect)}


def rgb_packet(up, mac, master, index):
    p = bytearray(RF_DATA_SIZE)
    n_data = (len(up["compressed"]) + CHUNK_BYTES - 1) // CHUNK_BYTES
    p[0], p[1] = RF_SELECT, RF_SET_RGB
    p[2:8], p[8:14], p[14:18] = mac, master, up["effect"]
    p[18], p[19] = index, n_data + 1
    if index == 0:
        p[20:24] = len(up["compressed"]).to_bytes(4, "big")
        p[25:27] = up["frames"].to_bytes(2, "big")
        p[27] = up["leds"]
        p[32:34] = up["ticks"].to_bytes(2, "big")
        p[34] = up["fraction"]
    else:
        data = up["compressed"][(index - 1) * CHUNK_BYTES: index * CHUNK_BYTES]
        p[20:20 + len(data)] = data
    return bytes(p)


def send_rf(tx, channel, rx_slot, rf):
    for i in range(RF_DATA_SIZE // RF_CHUNK_SIZE):
        pkt = bytes([USB_CMD_SEND_RF, i, channel, rx_slot]) + rf[i * RF_CHUNK_SIZE:(i + 1) * RF_CHUNK_SIZE]
        tx.write(pkt)
        time.sleep(0.001)


def set_color(tx, master, master_ch, dev, rgb):
    name, leds = fan_type(dev["fan_type"])
    if leds is None:
        print(f"  skip {mac_str(dev['mac'])}: {name}")
        return None
    up = build_upload([rgb] * (leds * dev["fans"]))
    channel = dev["channel"] if 1 <= dev["channel"] <= 39 else master_ch
    n_packets = (len(up["compressed"]) + CHUNK_BYTES - 1) // CHUNK_BYTES + 1
    for index in range(n_packets):
        pkt = rgb_packet(up, dev["mac"], master, index)
        for _ in range(HEADER_REPEATS if index == 0 else 1):
            send_rf(tx, channel, dev["rx"], pkt)
            time.sleep(0.02)
        time.sleep(0.01)
    return up["effect"]


def show(devices, master):
    for d in devices:
        name, leds = fan_type(d["fan_type"])
        ours = "yes" if d["master"] == master else f"NO (master {mac_str(d['master'])})"
        print(f"  device {mac_str(d['mac'])}  {name} x{d['fans']}")
        print(f"    bound to this TX: {ours}   channel {d['channel']}  rx slot {d['rx']}")
        print(f"    running effect id: {d['effect'].hex()}   motherboard sync: {d['mb_sync']}")
        print(f"    rpm {d['rpm'][:max(d['fans'], 1)]}   pwm {d['pwm'][:max(d['fans'], 1)]}")


def wait_for_effect(rx, mac, effect, seconds=3.0):
    end = time.time() + seconds
    while time.time() < end:
        for d in get_devices(rx):
            if d["mac"] == mac and d["effect"] == effect:
                return True
        time.sleep(0.3)
    return False


def main():
    cmd = sys.argv[1] if len(sys.argv) > 1 else "scan"
    tx, rx = Dongle(TX_ID, "TX"), Dongle(RX_ID, "RX")
    try:
        master, master_ch, fw = find_master(tx)
        print(f"TX master {mac_str(master)}  channel {master_ch}  firmware {fw}")
        devices = get_devices(rx)
        print(f"RX hears {len(devices)} device(s):")
        show(devices, master)
        targets = [d for d in devices if d["master"] == master]

        if cmd == "color":
            hexrgb = sys.argv[2].lstrip("#")
            colors = [tuple(bytes.fromhex(hexrgb))]
        elif cmd == "demo":
            colors = [(255, 0, 0), (0, 255, 0), (0, 0, 255), (255, 0, 160), (255, 255, 255)]
        else:
            return

        for rgb in colors:
            print(f"\nsetting #{bytes(rgb).hex()}")
            for d in targets:
                effect = set_color(tx, master, master_ch, d, rgb)
                if effect:
                    ok = wait_for_effect(rx, d["mac"], effect)
                    print(f"  {mac_str(d['mac'])}: effect {effect.hex()} "
                          f"{'confirmed by fan' if ok else 'NOT confirmed'}")
            if cmd == "demo":
                time.sleep(2)
    finally:
        tx.close()
        rx.close()


if __name__ == "__main__":
    main()