feat: drive WLED Wi-Fi adapters over WebSocket

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://<ip>/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

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"

pyproject.toml

@@ -23,6 +23,7 @@ classifiers = [
 dependencies = [
   "numpy>=1.20",
   "pyserial>=3.5",
+  "websockets>=12",
 ]
 
 [project.urls]

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 [])}")

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))

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