commit 35240e5dfc9945fa0a9bb6b0e72aab44a5bdc35a Author: MangaValk Date: Mon Oct 26 00:12:23 2026 +0100 Add Lian Li SL-INF wireless tools and SignalRGB plugin Proof-of-concept tools for the UNI FAN SL-INF Wireless kit (SLV3 TX/RX dongles 0416:8040/8041): a Python scan/colour script, tinyuz helpers, a JavaScript tinyuz encoder, the SignalRGB plugin and tests. Co-Authored-By: Claude Opus 5.5 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..63dd645 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +__pycache__/ +*.o +build-src/tinyuz/ +build-src/HDiffPatch/ diff --git a/LianLi_Wireless_SLINF_PoC.js b/LianLi_Wireless_SLINF_PoC.js new file mode 100644 index 0000000..4eeabc0 --- /dev/null +++ b/LianLi_Wireless_SLINF_PoC.js @@ -0,0 +1,346 @@ +// Proof of concept: Lian Li UNI FAN SL-INF Wireless through the SLV3 TX dongle. +// Protocol from github.com/sgtaziz/lian-li-linux (MIT). Every colour change is a +// full RF upload, so the fan is updated at most a few times per second. +export function Name() { return "Lian Li SL-INF Wireless (PoC)"; } +export function VendorId() { return 0x0416; } +export function ProductId() { return 0x8040; } +export function Publisher() { return "PoC"; } +export function Size() { return [7, 7]; } +export function DefaultPosition() { return [0, 0]; } +export function DefaultScale() { return 8.0; } +export function Type() { return "rawusb"; } +export function DeviceType() { return "lightingcontroller"; } +/* global +shutdownColor:readonly +LightingMode:readonly +forcedColor:readonly +updateInterval:readonly +*/ +export function ControllableParameters() { + return [ + {"property":"shutdownColor", "group":"lighting", "label":"Shutdown Color", description: "This color is applied to the device when the System, or SignalRGB is shutting down", "min":"0", "max":"360", "type":"color", "default":"#8000ff"}, + {"property":"LightingMode", "group":"lighting", "label":"Lighting Mode", description: "Determines where the device's RGB comes from. Canvas will pull from the active Effect, while Forced will override it to a specific color", "type":"combobox", "values":["Canvas", "Forced"], "default":"Canvas"}, + {"property":"forcedColor", "group":"lighting", "label":"Forced Color", description: "The color used when 'Forced' Lighting Mode is enabled", "min":"0", "max":"360", "type":"color", "default":"#009bde"}, + {"property":"updateInterval", "group":"", "label":"Update Interval (ms)", description: "Minimum time between RF uploads", "step":"50", "type":"number", "min":"100", "max":"2000", "default":"200"}, + ]; +} + +// Bound fan, as reported by `lianli_poc.py scan`. +const FAN_MAC = [0xe8, 0xb5, 0x62, 0x62, 0x32, 0xe1]; +const FAN_RX_SLOT = 1; +const FAN_COUNT = 1; +const LEDS_PER_FAN = 44; + +const USB_CMD_SEND_RF = 0x10; +const USB_CMD_GET_MAC = 0x11; +const RF_SELECT = 0x12; +const RF_SET_RGB = 0x20; +const RF_DATA_SIZE = 240; +const RF_CHUNK_SIZE = 60; +const CHUNK_BYTES = 220; +const HEADER_REPEATS = 2; +const STATIC_INTERVAL_MS = 5000; + +// Per fan, in wire order: center ring (8), inner A (10), outer A (8), inner B (10), outer B (8). +// Sides A and B get the same canvas positions. +const vLedNames = []; +const vLedPositions = []; +const ledIndexToPosition = []; + +function ring(count, radius, phase) { + const out = []; + for (let i = 0; i < count; i++) { + const a = phase + (2 * Math.PI * i) / count; + out.push([Math.round(3 + radius * Math.cos(a)), Math.round(3 + radius * Math.sin(a))]); + } + return out; +} + +function buildLayout() { + const center = ring(8, 1.2, 0); + const inner = ring(10, 2.1, 0); + const outer = ring(8, 3, Math.PI / 8); + const segments = [["Center", center], ["Inner A", inner], ["Outer A", outer], ["Inner B", inner], ["Outer B", outer]]; + for (const [label, positions] of segments) { + positions.forEach((pos, i) => { + ledIndexToPosition.push(pos); + if (!label.endsWith("B")) { + vLedNames.push(`${label} ${i + 1}`); + vLedPositions.push(pos); + } + }); + } +} +buildLayout(); + +export function LedNames() { return vLedNames; } +export function LedPositions() { return vLedPositions; } + +let masterMac = null; +let masterChannel = 8; +let lastColors = null; +let lastSend = 0; + +export function Initialize() { + findMaster(); + if (masterMac) { + device.log(`TX master ${masterMac.map(b => b.toString(16).padStart(2, "0")).join(":")} channel ${masterChannel}`); + } else { + device.log("TX master did not answer; is L-Connect or the Python script using the dongle?"); + } +} + +export function Render() { + if (!masterMac) { + return; + } + sendColors(currentColors(), false); +} + +export function Shutdown(SystemSuspending) { + if (!masterMac) { + return; + } + const color = hexToRgb(SystemSuspending ? "#000000" : shutdownColor); + sendColors(new Array(LEDS_PER_FAN * FAN_COUNT).fill(color), true); +} + +function currentColors() { + if (LightingMode === "Forced") { + return new Array(LEDS_PER_FAN * FAN_COUNT).fill(hexToRgb(forcedColor)); + } + return ledIndexToPosition.map(([x, y]) => device.color(x, y)); +} + +function sendColors(colors, force) { + const now = Date.now(); + if (!force && now - lastSend < updateInterval) { + return; + } + if (!force && lastColors && !changedEnough(lastColors, colors)) { + return; + } + lastColors = colors; + lastSend = now; + uploadFrame(colors); +} + +function changedEnough(a, b) { + for (let i = 0; i < a.length; i++) { + if (Math.abs(a[i][0] - b[i][0]) + Math.abs(a[i][1] - b[i][1]) + Math.abs(a[i][2] - b[i][2]) > 6) { + return true; + } + } + return false; +} + +function findMaster() { + const channels = [8]; + for (let c = 2; c <= 38; c += 2) { if (c !== 8) { channels.push(c); } } + for (let c = 1; c <= 39; c += 2) { channels.push(c); } + + for (const channel of channels) { + flushReads(); + const cmd = new Array(64).fill(0); + cmd[0] = USB_CMD_GET_MAC; + cmd[1] = channel; + device.bulk_transfer(0x01, cmd, 64, 1000); + const r = readPacket(500); + if (r.length >= 13 && r[0] === USB_CMD_GET_MAC && r.slice(1, 7).some(b => b !== 0)) { + masterMac = r.slice(1, 7); + masterChannel = channel; + return; + } + } +} + +function readPacket(timeout) { + try { + const raw = device.bulk_transfer(0x81, new Array(64).fill(0), 64, timeout); + const data = (raw && raw.data !== undefined) ? raw.data : raw; + return Array.isArray(data) ? data : (data ? Array.from(data) : []); + } catch (_) { + return []; + } +} + +function flushReads() { + for (let i = 0; i < 16 && readPacket(5).length > 0; i++) { /* drain */ } +} + +function uploadFrame(colors) { + const raw = []; + for (const c of colors) { + raw.push(c[0] & 0xff, c[1] & 0xff, c[2] & 0xff); + } + const compressed = tuzCompress(raw); + const hundredths = STATIC_INTERVAL_MS * 160; + const ticks = Math.floor(hundredths / 100); + const fraction = hundredths % 100; + const ledCount = colors.length; + + let hash = 0x811c9dc5; + const tail = [ticks >> 8, ticks & 0xff, fraction, 0, 1, ledCount, 0, 0, 0, 0, 0]; + for (const b of compressed.concat(tail)) { + hash = Math.imul(hash ^ b, 0x01000193) >>> 0; + } + const effect = [(hash >>> 24) | 0x80, (hash >>> 16) & 0xff, (hash >>> 8) & 0xff, hash & 0xff]; + + const dataPackets = Math.ceil(compressed.length / CHUNK_BYTES); + for (let index = 0; index <= dataPackets; index++) { + const p = new Array(RF_DATA_SIZE).fill(0); + p[0] = RF_SELECT; + p[1] = RF_SET_RGB; + for (let k = 0; k < 6; k++) { + p[2 + k] = FAN_MAC[k]; + p[8 + k] = masterMac[k]; + } + for (let k = 0; k < 4; k++) { + p[14 + k] = effect[k]; + } + p[18] = index; + p[19] = dataPackets + 1; + if (index === 0) { + const n = compressed.length; + p[20] = (n >>> 24) & 0xff; p[21] = (n >>> 16) & 0xff; p[22] = (n >>> 8) & 0xff; p[23] = n & 0xff; + p[25] = 0; p[26] = 1; // one frame + p[27] = ledCount; + p[32] = ticks >> 8; p[33] = ticks & 0xff; + p[34] = fraction; + } else { + const chunk = compressed.slice((index - 1) * CHUNK_BYTES, index * CHUNK_BYTES); + for (let k = 0; k < chunk.length; k++) { + p[20 + k] = chunk[k]; + } + } + const repeats = index === 0 ? HEADER_REPEATS : 1; + for (let r = 0; r < repeats; r++) { + sendRf(p); + device.pause(2); + } + } +} + +function sendRf(rf) { + for (let i = 0; i < RF_DATA_SIZE / RF_CHUNK_SIZE; i++) { + const pkt = [USB_CMD_SEND_RF, i, masterChannel, FAN_RX_SLOT].concat(rf.slice(i * RF_CHUNK_SIZE, (i + 1) * RF_CHUNK_SIZE)); + device.bulk_transfer(0x01, pkt, 64, 1000); + } +} + +function hexToRgb(hex) { + const m = /^#?([a-f\d]{2})([a-f\d]{2})([a-f\d]{2})$/i.exec(hex); + return m ? [parseInt(m[1], 16), parseInt(m[2], 16), parseInt(m[3], 16)] : [0, 0, 0]; +} + +// ---- Minimal tinyuz encoder (greedy LZ77), format of github.com/sisong/tinyuz ---- +const kMinDictMatchLen = 2; +const kBigPosForLen = (1 << 11) + (1 << 9) + (1 << 7) - 1; +const kMaxDist = 4096; +const kCtrlStreamEnd = 3; + +function tuzCompress(data) { + const code = [0, 0, 0, 0]; // dict size, patched at the end + let typesIndex = 0; + let typeCount = 0; + let haveDataBack = false; + let posBack = 1; + let dictSizeMax = 1; + + const outType = (bit) => { + if (typeCount === 0) { + typesIndex = code.length; + code.push(0); + } + code[typesIndex] |= bit << typeCount; + typeCount = (typeCount + 1) & 7; + }; + + const outLen = (v, packBit) => { + let count = 1; + while (v >= 2 ** (count * packBit)) { + v -= 2 ** (count * packBit); + count++; + } + while (count--) { + for (let i = 0; i < packBit; i++) { + outType(Math.floor(v / 2 ** (count * packBit + i)) & 1); + } + outType(count > 0 ? 1 : 0); + } + }; + + const outDictPos = (pos) => { + const big = pos >= 128 ? 1 : 0; + if (big) { + pos -= 128; + } + code.push((pos & 0x7f) | (big << 7)); + if (big) { + outLen(pos >> 7, 2); + } + }; + + const outLiteral = (b) => { + outType(1); + code.push(b); + haveDataBack = true; + }; + + const outDict = (len, dist) => { + outType(0); + const same = haveDataBack && dist === posBack; + let saved = len - kMinDictMatchLen; + if (!same && dist > kBigPosForLen) { + saved--; + } + outLen(saved, 1); + if (haveDataBack) { + outType(same ? 1 : 0); + } + if (!same) { + outDictPos(dist); + } + dictSizeMax = Math.max(dictSizeMax, dist); + haveDataBack = false; + posBack = dist; + }; + + let i = 0; + while (i < data.length) { + let bestLen = 0; + let bestDist = 0; + for (let dist = 1; dist <= Math.min(i, kMaxDist); dist++) { + let len = 0; + while (i + len < data.length && data[i + len] === data[i + len - dist]) { + len++; + } + if (len > bestLen) { + bestLen = len; + bestDist = dist; + } + } + const minLen = bestDist > kBigPosForLen ? 3 : kMinDictMatchLen; + if (bestLen >= minLen && (bestLen > 2 || bestDist < 128)) { + outDict(bestLen, bestDist); + i += bestLen; + } else { + outLiteral(data[i]); + i++; + } + } + + outType(0); + outLen(kCtrlStreamEnd, 1); + if (haveDataBack) { + outType(0); + } + outDictPos(0); + + for (let k = 0; k < 4; k++) { + code[k] = (dictSizeMax >> (8 * k)) & 0xff; + } + return code; +} + +export function ImageUrl() { return "https://assets.signalrgb.com/devices/default/misc/usb-drive-render.png"; } diff --git a/README.md b/README.md new file mode 100644 index 0000000..761705e --- /dev/null +++ b/README.md @@ -0,0 +1,34 @@ +# Lian Li wireless fan tools + +Proof-of-concept tools for the Lian Li UNI FAN SL-INF Wireless kit (SLV3 TX `0416:8040`, RX `0416:8041`). +Protocol from [sgtaziz/lian-li-linux](https://github.com/sgtaziz/lian-li-linux) (MIT); compression is +[tinyuz](https://github.com/sisong/tinyuz) (MIT). + +Only one program can use a dongle at a time: quit SignalRGB (it holds the TX) and L-Connect before +running anything that opens the TX. + +| File | What it does | +|---|---| +| `lianli_poc.py` | `scan` / `color RRGGBB` / `demo`. Needs `pip install --user pyusb libusb-package`. | +| `tuzc.exe` | tinyuz compressor used by `lianli_poc.py` (stdin to stdout, round-trip checked). | +| `tuzd.exe` | reference tinyuz decoder, `tuzd `, used by the tests. | +| `tuz.js` | JavaScript tinyuz encoder (same code is inlined in the plugin). | +| `LianLi_Wireless_SLINF_PoC.js` | SignalRGB plugin source. Installed copy: `Documents\WhirlwindFX\Plugins`. | +| `test/tuz_test.js` | `node test/tuz_test.js`: JS encoder vs reference decoder, no hardware. | +| `test/js_tuz_on_fan.py` | sends JS-compressed frames to the fan (needs TX free). | +| `test/watch_rx.py` | `python test/watch_rx.py [seconds]`: watches the fan's effect id via the RX only; works while SignalRGB runs. | +| `build-src/` | sources for `tuzc.exe` / `tuzd.exe`. | + +The plugin hardcodes the fan MAC `e8:b5:62:62:32:e1` and RX slot 1 (from `lianli_poc.py scan`). + +## Rebuilding tuzc.exe / tuzd.exe + +Clone `https://github.com/sisong/tinyuz` and `https://github.com/sisong/HDiffPatch` into `build-src/`, then with MinGW g++: + +```bash +cd build-src +F="-O2 -DNDEBUG -D_HPATCH_IS_USED_MULTITHREAD=0 -D_IS_USED_MULTITHREAD=0 -IHDiffPatch/libParallel" +gcc $F -c tinyuz/decompress/tuz_dec.c -o tuz_dec.o +g++ $F -static -o ../tuzc.exe tuzc_main.cpp tuz_wrapper.cpp tuz_dec.o tinyuz/compress/tuz_enc.cpp tinyuz/compress/tuz_enc_private/*.cpp HDiffPatch/libHDiffPatch/HDiff/private_diff/libdivsufsort/divsufsort.cpp -x c HDiffPatch/libHDiffPatch/HPatch/patch.c +gcc -O2 -D_IS_RUN_MEM_SAFE_CHECK=1 -o ../tuzd.exe tuzd_main.c tinyuz/decompress/tuz_dec.c +``` diff --git a/build-src/tuz_wrapper.cpp b/build-src/tuz_wrapper.cpp new file mode 100644 index 0000000..c9e0ddc --- /dev/null +++ b/build-src/tuz_wrapper.cpp @@ -0,0 +1,86 @@ +// C wrapper around tinyuz compression for Rust FFI. +// Provides a simple memory-to-memory compression API. +// +// MIT License — wraps tinyuz (https://github.com/sisong/tinyuz) + +#include "tinyuz/compress/tuz_enc.h" +#include +#include + + +struct MemStreamInput { + const unsigned char* data; + size_t size; +}; + +struct MemStreamOutput { + unsigned char* buf; + size_t capacity; + size_t written; +}; + +static hpatch_BOOL mem_read(const hpatch_TStreamInput* stream, + hpatch_StreamPos_t readFromPos, + unsigned char* out_data, + unsigned char* out_data_end) { + auto* ms = (const MemStreamInput*)stream->streamImport; + size_t len = (size_t)(out_data_end - out_data); + if (readFromPos + len > ms->size) return hpatch_FALSE; + memcpy(out_data, ms->data + readFromPos, len); + return hpatch_TRUE; +} + +static hpatch_BOOL mem_write(const hpatch_TStreamOutput* stream, + hpatch_StreamPos_t writeToPos, + const unsigned char* data, + const unsigned char* data_end) { + auto* ms = (MemStreamOutput*)stream->streamImport; + size_t len = (size_t)(data_end - data); + if (writeToPos + len > ms->capacity) return hpatch_FALSE; + memcpy(ms->buf + writeToPos, data, len); + size_t end = (size_t)writeToPos + len; + if (end > ms->written) ms->written = end; + return hpatch_TRUE; +} + + +extern "C" { + +/// Compress `input` using tinyuz with the given dictionary size. +/// +/// Returns the compressed size on success, or 0 on failure. +/// `output` must be at least `tuz_maxCompressedSize(input_len)` bytes. +size_t tuz_compress_mem(const unsigned char* input, size_t input_len, + unsigned char* output, size_t output_capacity, + size_t dict_size) { + MemStreamInput in_mem = { input, input_len }; + MemStreamOutput out_mem = { output, output_capacity, 0 }; + + hpatch_TStreamInput in_stream; + memset(&in_stream, 0, sizeof(in_stream)); + in_stream.streamImport = (void*)&in_mem; + in_stream.streamSize = input_len; + in_stream.read = mem_read; + + hpatch_TStreamOutput out_stream; + memset(&out_stream, 0, sizeof(out_stream)); + out_stream.streamImport = (void*)&out_mem; + out_stream.streamSize = output_capacity; + out_stream.read_writed = nullptr; + out_stream.write = mem_write; + + tuz_TCompressProps props = tuz_kDefaultCompressProps; + props.dictSize = dict_size; + props.threadNum = 1; + + hpatch_StreamPos_t result = tuz_compress(&out_stream, &in_stream, &props); + if (result == 0 || result > output_capacity) return 0; + return (size_t)result; +} + +/// Returns the maximum compressed output size for a given input length. +size_t tuz_max_compressed_size(size_t input_len) { + return (size_t)tuz_maxCompressedSize(input_len); +} + +} // extern "C" diff --git a/build-src/tuzc_main.cpp b/build-src/tuzc_main.cpp new file mode 100644 index 0000000..42aba27 --- /dev/null +++ b/build-src/tuzc_main.cpp @@ -0,0 +1,25 @@ +// tuzc: compress stdin with tinyuz (4 KB dict) to stdout; verifies by round-trip. +#include +#include +#include +#include +#include "tinyuz/decompress/tuz_dec.h" +extern "C" size_t tuz_compress_mem(const unsigned char*, size_t, unsigned char*, size_t, size_t); +extern "C" size_t tuz_max_compressed_size(size_t); +int main() { + _setmode(_fileno(stdin), _O_BINARY); + _setmode(_fileno(stdout), _O_BINARY); + std::vector in; + int c; + while ((c = getchar()) != EOF) in.push_back((unsigned char)c); + if (in.empty()) return 2; + std::vector out(tuz_max_compressed_size(in.size())); + size_t n = tuz_compress_mem(in.data(), in.size(), out.data(), out.size(), 4096); + if (n == 0) return 3; + std::vector back(in.size()); + tuz_size_t back_size = (tuz_size_t)back.size(); + if (tuz_decompress_mem(out.data(), n, back.data(), &back_size) != tuz_STREAM_END + || back_size != in.size() || back != in) return 4; + fwrite(out.data(), 1, n, stdout); + return 0; +} diff --git a/build-src/tuzd_main.c b/build-src/tuzd_main.c new file mode 100644 index 0000000..056d479 --- /dev/null +++ b/build-src/tuzd_main.c @@ -0,0 +1,17 @@ +// tuzd : decompress stdin with the reference tinyuz decoder to stdout. +#include +#include +#include +#include +#include "tinyuz/decompress/tuz_dec.h" +int main(int argc, char** argv) { + static unsigned char in[65536], out[65536]; + _setmode(_fileno(stdin), _O_BINARY); + _setmode(_fileno(stdout), _O_BINARY); + size_t n = fread(in, 1, sizeof(in), stdin); + tuz_size_t size = (tuz_size_t)atoi(argv[1]); + tuz_TResult r = tuz_decompress_mem(in, (tuz_size_t)n, out, &size); + if (r != tuz_STREAM_END) { fprintf(stderr, "decode result %d\n", (int)r); return 1; } + fwrite(out, 1, size, stdout); + return 0; +} diff --git a/lianli_poc.py b/lianli_poc.py new file mode 100644 index 0000000..2b84a1e --- /dev/null +++ b/lianli_poc.py @@ -0,0 +1,270 @@ +"""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() diff --git a/test/js_tuz_on_fan.py b/test/js_tuz_on_fan.py new file mode 100644 index 0000000..ab2e83b --- /dev/null +++ b/test/js_tuz_on_fan.py @@ -0,0 +1,30 @@ +# Send colours compressed by the JS encoder (tuz.js) and check the fan confirms them. +import os, subprocess, sys, time +TOOLS = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..") +sys.path.insert(0, TOOLS) +import lianli_poc as p + +def js_tinyuz(raw): + js = "const {tuzCompress}=require(process.argv[1]);process.stdout.write(Buffer.from(tuzCompress([...Buffer.from(process.argv[2],'hex')])))" + return subprocess.run(["node", "-e", js, TOOLS + "/tuz.js", raw.hex()], capture_output=True, check=True).stdout + +p.tinyuz = js_tinyuz +tx, rx = p.Dongle(p.TX_ID, "TX"), p.Dongle(p.RX_ID, "RX") +try: + master, ch, _ = p.find_master(tx) + dev = [d for d in p.get_devices(rx) if d["master"] == master][0] + # gradient frame exercises dict matches and literals, then back to purple + grad = [(i * 5, 0, 255 - i * 5) for i in range(44)] + for name, colors in [("gradient", grad), ("purple", [(128, 0, 255)] * 44)]: + up = p.build_upload(colors) + # reuse set_color's send path with a custom frame + n = (len(up["compressed"]) + p.CHUNK_BYTES - 1) // p.CHUNK_BYTES + 1 + for i in range(n): + pkt = p.rgb_packet(up, dev["mac"], master, i) + for _ in range(2 if i == 0 else 1): + p.send_rf(tx, ch, dev["rx"], pkt); time.sleep(0.002) + ok = p.wait_for_effect(rx, dev["mac"], up["effect"]) + print(name, len(up["compressed"]), "bytes", "confirmed" if ok else "NOT confirmed") + time.sleep(3) +finally: + tx.close(); rx.close() diff --git a/test/tuz_test.js b/test/tuz_test.js new file mode 100644 index 0000000..3d70151 --- /dev/null +++ b/test/tuz_test.js @@ -0,0 +1,25 @@ +const { tuzCompress } = require("../tuz.js"); +const { execFileSync } = require("child_process"); +let rnd = 12345; +const rand = () => (rnd = (rnd * 1103515245 + 12345) & 0x7fffffff) & 0xff; +const cases = []; +for (const leds of [44, 88, 176]) { + cases.push(Array.from({ length: leds * 3 }, (_, i) => [255, 0, 128][i % 3])); + cases.push(Array.from({ length: leds * 3 }, (_, i) => (i * 7) & 0xff)); + cases.push(Array.from({ length: leds * 3 }, rand)); + cases.push(Array.from({ length: leds * 3 }, (_, i) => (Math.floor(i / 30) * 40) & 0xff)); +} +for (let t = 0; t < 200; t++) { + const n = 1 + (rand() * 7) % 700; + const alph = 1 + rand() % 6; + cases.push(Array.from({ length: n }, () => rand() % alph * 50)); +} +let ok = 0; +for (const c of cases) { + const code = Buffer.from(tuzCompress(c)); + const out = execFileSync(require("path").join(__dirname, "..", "tuzd.exe"), [String(c.length)], { input: code }); + if (Buffer.compare(out, Buffer.from(c)) !== 0) { console.log("MISMATCH len", c.length); process.exit(1); } + ok++; +} +const solid = tuzCompress(Array.from({ length: 132 }, (_, i) => [128, 0, 255][i % 3])); +console.log(`${ok}/${cases.length} round-trips ok; solid 44 LEDs -> ${solid.length} bytes`); diff --git a/test/watch_rx.py b/test/watch_rx.py new file mode 100644 index 0000000..f37399c --- /dev/null +++ b/test/watch_rx.py @@ -0,0 +1,19 @@ +# Watch the fan's effect id via the RX dongle only (TX stays free for SignalRGB). +import os, sys, time +TOOLS = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..") +sys.path.insert(0, TOOLS) +import lianli_poc as p +rx = p.Dongle(p.RX_ID, "RX") +try: + seen = [] + end = time.time() + (float(sys.argv[1]) if len(sys.argv) > 1 else 10) + while time.time() < end: + for d in p.get_devices(rx): + e = d["effect"].hex() + if not seen or seen[-1] != e: + seen.append(e) + print(f"{time.strftime('%H:%M:%S')} effect {e}") + time.sleep(0.2) + print(f"{len(seen)} distinct effect ids") +finally: + rx.close() diff --git a/tuz.js b/tuz.js new file mode 100644 index 0000000..cc07cd3 --- /dev/null +++ b/tuz.js @@ -0,0 +1,115 @@ +// Minimal tinyuz encoder (greedy LZ77), stream format of github.com/sisong/tinyuz. +// Output decodes with the reference tuz_decompress_mem. +const kMinDictMatchLen = 2; +const kBigPosForLen = (1 << 11) + (1 << 9) + (1 << 7) - 1; +const kMaxDist = 4096; +const kCtrlStreamEnd = 3; + +function tuzCompress(data) { + const code = [0, 0, 0, 0]; // dict size, patched at the end + let typesIndex = 0; + let typeCount = 0; + let haveDataBack = false; + let posBack = 1; + let dictSizeMax = 1; + + const outType = (bit) => { + if (typeCount === 0) { + typesIndex = code.length; + code.push(0); + } + code[typesIndex] |= bit << typeCount; + typeCount = (typeCount + 1) & 7; + }; + + const outLen = (v, packBit) => { + let count = 1; + while (v >= 2 ** (count * packBit)) { + v -= 2 ** (count * packBit); + count++; + } + while (count--) { + for (let i = 0; i < packBit; i++) { + outType(Math.floor(v / 2 ** (count * packBit + i)) & 1); + } + outType(count > 0 ? 1 : 0); + } + }; + + const outDictPos = (pos) => { + const big = pos >= 128 ? 1 : 0; + if (big) { + pos -= 128; + } + code.push((pos & 0x7f) | (big << 7)); + if (big) { + outLen(pos >> 7, 2); + } + }; + + const outLiteral = (b) => { + outType(1); + code.push(b); + haveDataBack = true; + }; + + const outDict = (len, dist) => { + outType(0); + const same = haveDataBack && dist === posBack; + let saved = len - kMinDictMatchLen; + if (!same && dist > kBigPosForLen) { + saved--; + } + outLen(saved, 1); + if (haveDataBack) { + outType(same ? 1 : 0); + } + if (!same) { + outDictPos(dist); + } + dictSizeMax = Math.max(dictSizeMax, dist); + haveDataBack = false; + posBack = dist; + }; + + let i = 0; + while (i < data.length) { + let bestLen = 0; + let bestDist = 0; + for (let dist = 1; dist <= Math.min(i, kMaxDist); dist++) { + let len = 0; + while (i + len < data.length && data[i + len] === data[i + len - dist]) { + len++; + } + if (len > bestLen) { + bestLen = len; + bestDist = dist; + } + } + const minLen = bestDist > kBigPosForLen ? 3 : kMinDictMatchLen; + if (bestLen >= minLen && (bestLen > 2 || bestDist < 128)) { + outDict(bestLen, bestDist); + i += bestLen; + } else { + outLiteral(data[i]); + i++; + } + } + + // stream end control code + outType(0); + outLen(kCtrlStreamEnd, 1); + if (haveDataBack) { + outType(0); + } + outDictPos(0); + + for (let k = 0; k < 4; k++) { + code[k] = (dictSizeMax >> (8 * k)) & 0xff; + } + return code; +} + +if (typeof module !== "undefined") { + module.exports = { tuzCompress }; +} diff --git a/tuzc.exe b/tuzc.exe new file mode 100644 index 0000000..8944235 Binary files /dev/null and b/tuzc.exe differ diff --git a/tuzd.exe b/tuzd.exe new file mode 100644 index 0000000..51df251 Binary files /dev/null and b/tuzd.exe differ