Files
portal/leds/backends/spi.py
Jimmy d5cab2efdf Add multi-panel Pico UDP firmware and Python LED control.
Pico panels get static IPs on 10.1.1.10–14 with per-panel LED counts, Makefile deploy targets, and Python examples for animations, sync tests, and direct Pi SPI control.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-28 22:46:08 +12:00

82 lines
2.2 KiB
Python

"""WS2812 over SPI — works on Pi 5, 4, and Zero when SPI is enabled."""
from __future__ import annotations
from array import array
try:
from spidev import SpiDev
except ImportError as exc: # pragma: no cover - import guard
SpiDev = None
_SPI_IMPORT_ERROR = exc
else:
_SPI_IMPORT_ERROR = None
# WS2812 bit patterns via SPI, from https://github.com/mattaw/ws2812_spi_python
_LED_ZERO = 0b1100_0000
_LED_ONE = 0b1111_1100
_PREAMBLE = 42
class SpiBackend:
def __init__(
self,
led_count: int,
*,
spi_bus: int = 0,
spi_device: int = 0,
max_speed_hz: int = 6_500_000,
) -> None:
if SpiDev is None:
raise ImportError(
"spidev is required for the SPI backend; install portal requirements"
) from _SPI_IMPORT_ERROR
self._led_count = led_count
self._device = SpiDev()
self._device.open(spi_bus, spi_device)
self._device.max_speed_hz = max_speed_hz
self._device.mode = 0b00
self._device.lsbfirst = False
frame_len = _PREAMBLE + led_count * 24
self._frame = array("B", [0] * frame_len)
self._clear_frame = array("B", [0] * frame_len)
for i in range(_PREAMBLE, frame_len):
self._clear_frame[i] = _LED_ZERO
@property
def led_count(self) -> int:
return self._led_count
def write_grb(self, grb_bytes: bytes) -> None:
if len(grb_bytes) != self._led_count * 3:
raise ValueError(
f"expected {self._led_count * 3} bytes, got {len(grb_bytes)}"
)
out = self._frame
pos = _PREAMBLE
for value in grb_bytes:
for bit in (128, 64, 32, 16, 8, 4, 2, 1):
out[pos] = _LED_ONE if value & bit else _LED_ZERO
pos += 1
# writebytes2 handles large frames (400+ LEDs); writebytes is capped at 4 KiB.
self._device.writebytes2(out)
def clear(self) -> None:
self._device.writebytes2(self._clear_frame)
def close(self) -> None:
try:
self.clear()
finally:
self._device.close()
def __enter__(self) -> SpiBackend:
return self
def __exit__(self, *_exc) -> None:
self.close()