From a0ba21d29a2d2b142dbda7328b94db7576a0f930 Mon Sep 17 00:00:00 2001 From: TungLV-GOG1071 Date: Tue, 27 Oct 2026 01:04:47 +0700 Subject: [PATCH] feat: drive WLED Wi-Fi adapters over WebSocket When the configured port is a host/IP instead of a /dev path, frames are sent as per-LED JSON over the WLED WebSocket (/ws) rather than serial. The vendor WLED firmware has every realtime UDP port closed (verified on hardware), so UDP DRGB/DDP/E1.31/Art-Net cannot work; the WebSocket is the only realtime path that does. - device.py: WebSocket transport alongside serial; effect that is running before connect is restored on close - cli.py: 'skydimo info' shows the link type - pyproject.toml, install.sh: add websockets>=12 dependency - README.md: document Wi-Fi setup; led_count must match the strip's real LED count - tests/test_core.py: unit tests for the transport Co-Authored-By: Claude Sonnet 5.5 --- README.md | 15 +++++++ install.sh | 10 ++--- pyproject.toml | 1 + src/skydimo/cli.py | 4 +- src/skydimo/device.py | 102 +++++++++++++++++++++++++++++++++++++++++- tests/test_core.py | 41 +++++++++++++++++ 6 files changed, 164 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index e18132e..2c543f6 100644 --- a/README.md +++ b/README.md @@ -180,6 +180,21 @@ toggle DTR on open; pyserial and Hyperion do not. ~/.local/share/applications/skydimo.desktop ``` +## Wi-Fi (WLED) adapters + +The Wi-Fi variant runs WLED. Point `port` at its address and frames go to its +WebSocket as per-LED colours (about 18 fps on an ESP8266), with no serial +handshake: + +``` +skydimo set port=192.168.1.38 led_count=144 # count: http:///json/info +``` + +The realtime UDP ports (DRGB, DDP, E1.31, Art-Net) are closed on the vendor +firmware, which is why WebSocket is used. `color_order` is ignored (WLED has its +own setting). When the engine stops, the effect that was running before it +connected is restored. + ## Troubleshooting **No adapter found** — check `lsusb` for `1a86:7523`. Pin it explicitly with diff --git a/install.sh b/install.sh index f37cd2d..a2ea3d2 100755 --- a/install.sh +++ b/install.sh @@ -355,23 +355,23 @@ fi # ---------------------------------------------------------------- verify ---- step "Verifying" -if PYTHONNOUSERSITE=1 "$VENV/bin/python" -c 'import skydimo, numpy, serial' 2>/dev/null; then +if PYTHONNOUSERSITE=1 "$VENV/bin/python" -c 'import skydimo, numpy, serial, websockets' 2>/dev/null; then ok "package imports cleanly" else die "the installed package does not import -- see the output above" fi if ! PYTHONNOUSERSITE=1 "$VENV/bin/python" - >/dev/null 2>&1 <<'EOF' -import pathlib, sys, numpy, serial +import pathlib, sys, numpy, serial, websockets venv = pathlib.Path(sys.prefix).resolve() -for m in (numpy, serial): +for m in (numpy, serial, websockets): if venv not in pathlib.Path(m.__file__).resolve().parents: raise SystemExit(1) EOF then - warn "numpy/pyserial are not inside the venv; reinstalling them there" + warn "numpy/pyserial/websockets are not inside the venv; reinstalling them there" "$VENV/bin/python" -m pip install --quiet --upgrade --force-reinstall \ - --no-deps numpy pyserial || die "could not install numpy/pyserial" + --no-deps numpy pyserial websockets || die "could not install numpy/pyserial/websockets" ok "dependencies reinstalled inside the venv" else ok "dependencies live inside the venv" diff --git a/pyproject.toml b/pyproject.toml index c2bae1e..295b938 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,6 +23,7 @@ classifiers = [ dependencies = [ "numpy>=1.20", "pyserial>=3.5", + "websockets>=12", ] [project.urls] diff --git a/src/skydimo/cli.py b/src/skydimo/cli.py index 3548f45..41b73e4 100644 --- a/src/skydimo/cli.py +++ b/src/skydimo/cli.py @@ -78,7 +78,7 @@ def __exit__(self, *exc): def cmd_info(args, cfg): from . import service - from .device import BAUD, find_port, Skydimo + from .device import BAUD, find_port, is_network, Skydimo try: port = find_port() if cfg["port"] == "auto" else cfg["port"] @@ -105,7 +105,7 @@ def cmd_info(args, cfg): lay = cfg.get("layout") print(f"device id {dev_id}") - print(f"baud {BAUD}") + print(f"link {'WebSocket (WLED)' if is_network(port) else f'{BAUD} baud serial'}") print(f"led count {cfg['led_count']}") print(f"colour ord {cfg['color_order']}") print(f"zones {len(cfg.get('zones') or [])}") diff --git a/src/skydimo/device.py b/src/skydimo/device.py index a39bb23..edbfa60 100644 --- a/src/skydimo/device.py +++ b/src/skydimo/device.py @@ -32,16 +32,32 @@ f2 = (f2 + f1) % 255 ext = (ext + (b ^ i)) % 255 i = byte index, & 0xFF f3 = ext, or 0xAA when ext == 0x41 + +Network transport (WLED over Wi-Fi) +----------------------------------- +The Wi-Fi variant of the adapter runs WLED. When `port` is a host or IP instead +of a /dev path ("192.168.1.38") frames go to WLED's WebSocket (`/ws`) as JSON +per-LED colours, and none of the serial handshake applies: + + {"tt": 0, "v": false, "seg": [{"id": 0, "i": [start, "RRGGBB", ...]}]} + +The stock realtime UDP ports (DRGB/DDP/E1.31/Art-Net) are closed on this +firmware, so UDP is not an option. WLED owns the colour order, so `color_order` +is ignored. The strip keeps its last picture until told otherwise, so the +effect running before we connected is restored on close. """ from __future__ import annotations import glob +import json import os import time +import urllib.request import serial from serial.tools import list_ports +from websockets.sync.client import connect as ws_connect BAUD = 1_000_000 VID_CH340 = 0x1A86 @@ -52,6 +68,8 @@ HDR_MAGIC1 = 0x6E # 'n' HDR_PLAIN = 0x68 # 'h' -- no white-channel calibration HDR_WHITE = 0x54 # 'T' -- white-channel calibration appended + +WS_CHUNK = 192 # LEDs per message; keeps each JSON small for the ESP8266's heap HEADER_SIZE = 6 COLOR_ORDERS = { @@ -68,6 +86,28 @@ class HandshakeError(RuntimeError): """The adapter did not complete the r/m handshake.""" +def is_network(port: str) -> bool: + """True when `port` names a host rather than a serial device path.""" + return not port.startswith("/") + + +def target_host(port: str) -> str: + """Strip an optional scheme from "[ws://|http://|wled://]host[:port]".""" + return port.split("://", 1)[-1].rstrip("/") + + +def wled_json(host: str, path: str, body: dict | None = None, timeout: float = 3.0): + """GET (or POST `body` to) a WLED JSON endpoint. Raises OSError if unreachable.""" + req = urllib.request.Request(f"http://{host}/json/{path}") + if body is not None: + req = urllib.request.Request( + req.full_url, json.dumps(body).encode(), + {"Content-Type": "application/json"}, + ) + with urllib.request.urlopen(req, timeout=timeout) as r: + return json.load(r) + + def find_port() -> str: """Locate the adapter's serial port (CH340 bridge).""" # The udev rule names the adapter, so this never has to guess. @@ -112,6 +152,10 @@ def __init__( self.attempts = attempts self.device_id: bytes = b"" self.s: serial.Serial | None = None + self._ws = None + self._host = "" + self.net_info: dict = {} + self._net_state: dict = {} self._header = self._build_header() if open_now: self.open() @@ -120,6 +164,8 @@ def __init__( def open(self) -> None: """Open the port, reset the MCU, and complete the handshake.""" + if is_network(self.port): + return self._open_network() last = None for attempt in range(1, self.attempts + 1): try: @@ -164,6 +210,32 @@ def _open_once(self) -> None: f'adapter did not acknowledge data mode; "m" returned {ack!r}' ) + def _open_network(self) -> None: + self._host = target_host(self.port) + try: + self.net_info = wled_json(self._host, "info") + self._net_state = wled_json(self._host, "state") + self._ws = ws_connect( + f"ws://{self._host}/ws", open_timeout=5, max_size=None, + ) + except (OSError, ValueError, TimeoutError) as exc: + raise HandshakeError(f"Could not reach WLED at {self._host}: {exc}") from exc + self.device_id = ( + f"WLED {self.net_info.get('ver', '?')} " + f"{self.net_info.get('name', '')}".strip().encode() + ) + + def _restore_network(self) -> None: + """Put back the effect that was running before we took over.""" + seg = (self._net_state.get("seg") or [{}])[0] + keys = ("fx", "sx", "ix", "pal", "col") + if not all(k in seg for k in keys): + return + wled_json(self._host, "state", { + "seg": [{"id": seg.get("id", 0), **{k: seg[k] for k in keys}}], + "transition": self._net_state.get("transition", 7), + }) + def _command(self, cmd: bytes) -> bytes: assert self.s is not None self.s.write(cmd) @@ -174,6 +246,12 @@ def _command(self, cmd: bytes) -> bytes: return reply def _quiet_close(self) -> None: + if self._ws is not None: + try: + self._ws.close() + except Exception: + pass + self._ws = None if self.s is not None: try: self.s.close() @@ -182,7 +260,13 @@ def _quiet_close(self) -> None: self.s = None def close(self, blank: bool = True) -> None: - if self.s is not None and blank: + if self._ws is not None: + if blank: + try: + self._restore_network() + except Exception: + pass + elif self.s is not None and blank: try: self.fill(0, 0, 0) time.sleep(0.05) @@ -225,6 +309,8 @@ def write_raw(self, rgb: bytes | bytearray) -> None: raise ValueError( f"expected {self.leds * 3} bytes ({self.leds} LEDs), got {len(rgb)}" ) + if self._ws is not None: + return self._send_ws(rgb) if self.s is None: raise RuntimeError("device is not open") body = bytearray(rgb) @@ -234,9 +320,21 @@ def write_raw(self, rgb: bytes | bytearray) -> None: self.s.write(self._header + bytes(body)) self.s.flush() + def _send_ws(self, rgb: bytes | bytearray) -> None: + assert self._ws is not None + hexes = [rgb[i:i + 3].hex() for i in range(0, len(rgb), 3)] + for start in range(0, len(hexes), WS_CHUNK): + self._ws.send(json.dumps( + {"tt": 0, "v": False, + "seg": [{"id": 0, "i": [start, *hexes[start:start + WS_CHUNK]]}]}, + separators=(",", ":"), + )) + def write(self, pixels) -> None: """Send one frame from an iterable of (r, g, b) triples.""" - i0, i1, i2 = COLOR_ORDERS[self.color_order] + # WLED applies its own colour order, so send plain RGB to it. + order = "rgb" if self._ws is not None else self.color_order + i0, i1, i2 = COLOR_ORDERS[order] body = bytearray() for px in pixels: body += bytes((px[i0] & 0xFF, px[i1] & 0xFF, px[i2] & 0xFF)) diff --git a/tests/test_core.py b/tests/test_core.py index a950062..f7efa5e 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -5,6 +5,7 @@ `run()` so its sampling helpers stay usable here. """ +import json import unittest import numpy as np @@ -21,10 +22,50 @@ def bare_device(leds, color_order="rgb", white_cal=None): d.color_order = color_order d.white_cal = tuple(white_cal) if white_cal else None d.s = None + d._ws = None d._header = d._build_header() return d +class TestNetworkTransport(unittest.TestCase): + def _sent(self, leds, pixel=(1, 2, 3)): + sent = [] + + class WS: + def send(self, msg): + sent.append(json.loads(msg)) + + d = bare_device(leds, color_order="bgr") + d._ws = WS() + d.write([pixel] * leds) + return sent + + def test_target_host(self): + from skydimo.device import is_network, target_host + self.assertEqual(target_host("192.168.1.38"), "192.168.1.38") + self.assertEqual(target_host("ws://192.168.1.38:80/"), "192.168.1.38:80") + self.assertTrue(is_network("192.168.1.38")) + self.assertFalse(is_network("/dev/ttyUSB0")) + + def test_single_message_for_small_strip(self): + (msg,) = self._sent(144) + seg = msg["seg"][0] + self.assertEqual(seg["i"][0], 0) + self.assertEqual(seg["i"][1:], ["010203"] * 144) + self.assertFalse(msg["v"]) + + def test_color_order_is_ignored(self): + # WLED owns the colour order; a "bgr" config must not swap channels + (msg,) = self._sent(2, pixel=(255, 0, 0)) + self.assertEqual(msg["seg"][0]["i"][1:], ["ff0000"] * 2) + + def test_large_strip_is_chunked_with_start_offsets(self): + sent = self._sent(400) + starts = [m["seg"][0]["i"][0] for m in sent] + self.assertEqual(starts, [0, 192, 384]) + self.assertEqual(sum(len(m["seg"][0]["i"]) - 1 for m in sent), 400) + + class TestFrameFormat(unittest.TestCase): def test_header_plain(self): # 3 LEDs, no white calibration -> 'T','n','h', count-1 big-endian, xor check