Ship the portal FastAPI controller with presets, sequences, and live audio.

Move the app under src/, add ring-spanning patterns with shared fonts and
beat-driven motion, and wire global audio settings into the web UI.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
2026-09-09 00:59:03 +12:00
co-authored by Cursor
parent 5094c7bcee
commit 3d444d1cb5
127 changed files with 10318 additions and 62632 deletions
+81
View File
@@ -0,0 +1,81 @@
#!/usr/bin/env python3
"""Run LED matrix animations on Pico panel(s) over UDP."""
from __future__ import annotations
import argparse
import sys
from leds.panel_udp import (
add_panel_args,
animation_playlist,
format_panel_targets,
panel_targets_from_args,
parse_pins_arg,
run_panel_animation_loop,
run_panel_playlist,
)
def main() -> int:
from leds.animations import ANIMATIONS
names = list(ANIMATIONS.keys())
parser = argparse.ArgumentParser(description="Animations for Pico panel(s) over UDP")
parser.add_argument(
"animation",
nargs="?",
choices=names + ["all"],
default="all",
)
add_panel_args(parser)
parser.add_argument("--duration", type=float, default=12.0)
parser.add_argument("--fps", type=float, default=None)
parser.add_argument("--once", action="store_true")
args = parser.parse_args()
targets = panel_targets_from_args(args)
playlist = animation_playlist(args.animation)
print(
f"Panel animations -> {format_panel_targets(targets, args.port)} "
f"brightness={args.brightness:g}"
)
print(f"Playlist: {', '.join(playlist)}")
try:
if len(playlist) == 1 and args.animation != "all":
run_panel_animation_loop(
targets,
args.port,
playlist[0],
fps=args.fps,
brightness=args.brightness,
panel_id=args.panel_id,
sync_send=args.sync_send,
pin=args.pin,
pins=parse_pins_arg(args),
)
else:
run_panel_playlist(
targets,
args.port,
playlist,
duration=args.duration,
fps_override=args.fps,
brightness=args.brightness,
panel_id=args.panel_id,
loop=not args.once,
sync_send=args.sync_send,
pin=args.pin,
pins=parse_pins_arg(args),
)
except KeyboardInterrupt:
print("\nStopped.")
except OSError as exc:
print(f"Network error: {exc}", file=sys.stderr)
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())
+62
View File
@@ -0,0 +1,62 @@
#!/usr/bin/env python3
"""Full-panel R / G / B flash for Pico panel(s)."""
from __future__ import annotations
import argparse
import socket
import sys
import time
from leds.colors import BLUE, GREEN, OFF, RED, WHITE
from leds.panel_udp import (
add_panel_args,
apply_pin_arg,
format_panel_targets,
panel_targets_from_args,
send_solid_to_targets,
)
COLORS = (
("RED", RED),
("GREEN", GREEN),
("BLUE", BLUE),
("WHITE", WHITE),
("OFF", OFF),
)
def main() -> int:
parser = argparse.ArgumentParser(description="R/G/B color test for Pico panel(s)")
add_panel_args(parser)
parser.add_argument("--pause", type=float, default=1.5)
args = parser.parse_args()
targets = panel_targets_from_args(args)
has_udp = any(not t.local for t in targets)
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) if has_udp else None
targets = apply_pin_arg(args, sock, targets)
print(
f"Color test -> {format_panel_targets(targets, args.port)} "
f"brightness={args.brightness:g}\n"
)
try:
for label, color in COLORS:
print(f" {label}")
send_solid_to_targets(sock, targets, args.port, color, args.brightness, args.panel_id)
time.sleep(args.pause)
except KeyboardInterrupt:
print("\nStopped.")
except OSError as exc:
print(f"Network error: {exc}", file=sys.stderr)
return 1
finally:
send_solid_to_targets(sock, targets, args.port, (0, 0, 0), 1.0, args.panel_id)
if sock is not None:
sock.close()
return 0
if __name__ == "__main__":
raise SystemExit(main())
+42
View File
@@ -0,0 +1,42 @@
#!/usr/bin/env python3
"""Cycle full-panel red / green / blue on Pico panel(s)."""
from __future__ import annotations
import argparse
import sys
import time
from leds.colors import RGB_CYCLE
from leds.panel_udp import PanelClient, add_panel_args, format_panel_targets, panel_targets_from_args
def main() -> int:
parser = argparse.ArgumentParser(description="RGB cycle on Pico panel(s)")
add_panel_args(parser)
parser.add_argument("--pause", type=float, default=1.0)
args = parser.parse_args()
targets = panel_targets_from_args(args)
print(
f"RGB cycle -> {format_panel_targets(targets, args.port)} "
f"brightness={args.brightness:g} (Ctrl+C to stop)"
)
try:
with PanelClient.from_args(args) as client:
while True:
for label, color in RGB_CYCLE:
print(f" {label}")
client.fill(color)
time.sleep(args.pause)
except KeyboardInterrupt:
print("\nStopped.")
except OSError as exc:
print(f"Network error: {exc}", file=sys.stderr)
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())
+369
View File
@@ -0,0 +1,369 @@
#!/usr/bin/env python3
"""Positioning and sync test for the hexagonal portal panels.
Runs a sequence of patterns on all panels (or one via --panel-index):
identify - panel index digit on each face (check physical mapping)
corners - red/green/blue/yellow corners (check orientation)
crosshair - center row + column
columns - sweep the same column index on every panel together
rows - sweep rows together
flash - simultaneous white flashes (check sync)
pulse - in-sync brightness pulse on all panels
width - sweep columns near right edge (find 38 vs 39 vs 40)
all - full sequence (default)
"""
from __future__ import annotations
import argparse
import math
import socket
import sys
import time
from leds.panel_udp import (
HeadlessMatrix,
PanelMatrixBundle,
PanelTarget,
add_panel_args,
apply_pin_arg,
format_panel_targets,
panel_targets_from_args,
push_panel_frames,
)
from leds.array_config import panel_label
from leds.text import draw_text_centered
PANEL_COLORS = (
(255, 80, 80),
(80, 255, 80),
(80, 160, 255),
(255, 200, 80),
(200, 80, 255),
)
CORNER_COLORS = {
"top-left": (255, 0, 0),
"top-right": (0, 255, 0),
"bottom-left": (0, 0, 255),
"bottom-right": (255, 255, 0),
}
def _matrices(targets: list[PanelTarget], brightness: float) -> PanelMatrixBundle:
return PanelMatrixBundle(targets, brightness)
def _push(
sock: socket.socket | None,
bundle: PanelMatrixBundle,
port: int,
panel_id: int | None,
) -> None:
"""Send all panel frames in a tight burst for best sync."""
push_panel_frames(sock, bundle.matrices, port, panel_id)
def _clear_all(
sock: socket.socket | None,
targets: list[PanelTarget],
port: int,
brightness: float,
panel_id: int | None,
) -> None:
with PanelMatrixBundle(targets, brightness) as bundle:
for _, matrix in bundle.matrices:
matrix.clear()
_push(sock, bundle, port, panel_id)
def test_identify(
sock: socket.socket | None,
targets: list[PanelTarget],
port: int,
brightness: float,
panel_id: int | None,
pause: float,
) -> None:
print("Test: identify — each panel shows its index (0–4)")
with _matrices(targets, brightness) as bundle:
for target, matrix in bundle.matrices:
matrix.clear()
color = PANEL_COLORS[target.index % len(PANEL_COLORS)]
draw_text_centered(matrix, str(target.index), color)
print(f" {panel_label(target.index)} -> {target.label()}")
_push(sock, bundle, port, panel_id)
time.sleep(pause)
def test_corners(
sock: socket.socket | None,
targets: list[PanelTarget],
port: int,
brightness: float,
panel_id: int | None,
pause: float,
) -> None:
print("Test: corners — TL=red TR=green BL=blue BR=yellow (all panels)")
with _matrices(targets, brightness) as bundle:
for target, matrix in bundle.matrices:
w, h = target.width, target.height
matrix.clear()
points = {
"top-left": (0, 0),
"top-right": (w - 1, 0),
"bottom-left": (0, h - 1),
"bottom-right": (w - 1, h - 1),
}
for name, (x, y) in points.items():
matrix[x, y] = CORNER_COLORS[name]
_push(sock, bundle, port, panel_id)
time.sleep(pause)
def test_crosshair(
sock: socket.socket | None,
targets: list[PanelTarget],
port: int,
brightness: float,
panel_id: int | None,
pause: float,
) -> None:
print("Test: crosshair — center row + column (white)")
with _matrices(targets, brightness) as bundle:
for target, matrix in bundle.matrices:
w, h = target.width, target.height
matrix.clear()
mid_y = h // 2
mid_x = w // 2
for x in range(w):
matrix[x, mid_y] = (255, 255, 255)
for y in range(h):
matrix[mid_x, y] = (200, 200, 200)
_push(sock, bundle, port, panel_id)
time.sleep(pause)
def test_columns(
sock: socket.socket | None,
targets: list[PanelTarget],
port: int,
brightness: float,
panel_id: int | None,
pause: float,
) -> None:
max_width = max(t.width for t in targets)
print(f"Test: column sweep — x=0..{max_width - 1} on all panels together")
with _matrices(targets, brightness) as bundle:
for x in range(max_width):
for target, matrix in bundle.matrices:
matrix.clear()
if x >= target.width:
continue
hue = int(255 * x / max(max_width - 1, 1))
for y in range(target.height):
matrix[x, y] = (hue, 80, 255 - hue)
_push(sock, bundle, port, panel_id)
time.sleep(pause)
_clear_all(sock, targets, port, brightness, panel_id)
def test_rows(
sock: socket.socket | None,
targets: list[PanelTarget],
port: int,
brightness: float,
panel_id: int | None,
pause: float,
) -> None:
height = targets[0].height if targets else 9
print(f"Test: row sweep — y=0..{height - 1} on all panels together")
with _matrices(targets, brightness) as bundle:
for y in range(height):
for target, matrix in bundle.matrices:
matrix.clear()
for x in range(target.width):
matrix[x, y] = (255, 255, 255)
_push(sock, bundle, port, panel_id)
time.sleep(pause)
_clear_all(sock, targets, port, brightness, panel_id)
def test_flash(
sock: socket.socket | None,
targets: list[PanelTarget],
port: int,
brightness: float,
panel_id: int | None,
pause: float,
count: int,
) -> None:
print(f"Test: sync flash — {count} white flashes on all panels")
for n in range(count):
with _matrices(targets, brightness) as bundle:
for _, matrix in bundle.matrices:
matrix.fill((255, 255, 255))
_push(sock, bundle, port, panel_id)
time.sleep(pause)
_clear_all(sock, targets, port, brightness, panel_id)
time.sleep(pause / 2)
print(f" flash {n + 1}/{count}")
def test_pulse(
sock: socket.socket | None,
targets: list[PanelTarget],
port: int,
brightness: float,
panel_id: int | None,
seconds: float,
fps: float,
) -> None:
print(f"Test: sync pulse — {seconds:.0f}s solid red fade (all panels)")
frames = max(int(seconds * fps), 1)
with _matrices(targets, brightness) as bundle:
for frame in range(frames):
t = frame / frames
level = 0.15 + 0.85 * (0.5 - 0.5 * math.cos(t * 6.28318))
color = (int(255 * level), 0, 0)
for _, matrix in bundle.matrices:
matrix.fill(color)
_push(sock, bundle, port, panel_id)
time.sleep(1.0 / fps)
_clear_all(sock, targets, port, brightness, panel_id)
def test_chain_ends(
sock: socket.socket | None,
targets: list[PanelTarget],
port: int,
brightness: float,
panel_id: int | None,
pause: float,
) -> None:
print("Test: chain ends — LED 0=red, last LED=green (per panel)")
with _matrices(targets, brightness) as bundle:
for target, matrix in bundle.matrices:
matrix.clear()
matrix[0, 0] = (255, 0, 0)
last = target.pixels - 1
x = last % target.width
y = last // target.width
matrix[x, y] = (0, 255, 0)
print(f" panel {target.index}: index 0 and {last} ({x},{y})")
_push(sock, bundle, port, panel_id)
time.sleep(pause)
def test_width(
sock: socket.socket | None,
targets: list[PanelTarget],
port: int,
brightness: float,
panel_id: int | None,
pause: float,
col_start: int,
col_end: int,
) -> None:
udp_targets = [t for t in targets if not t.local]
if not udp_targets:
print("Test: width — skipped (no UDP Pico panels in target list)")
return
probe_w = col_end + 1
probe_leds = probe_w * udp_targets[0].height
print("Test: width — one full column at a time (right-edge probe, UDP panels only)")
print(f" Probing columns {col_start}–{col_end} (sending {probe_w}×9 = {probe_leds} LEDs)")
print(" The last column that lights a real LED → width = column + 1")
print(" (column 37 lit → width 38; column 38 lit → width 39)\n")
for x in range(col_start, col_end + 1):
burst: list[tuple[PanelTarget, HeadlessMatrix]] = []
for target in udp_targets:
matrix = HeadlessMatrix(probe_w, target.height, brightness=brightness)
matrix.clear()
for y in range(target.height):
matrix[x, y] = (255, 255, 255)
probe_target = PanelTarget(target.index, target.host, probe_w, target.height)
burst.append((probe_target, matrix))
push_panel_frames(sock, burst, port, panel_id)
print(f" column {x} → width {x + 1} if this is the last real column")
time.sleep(pause)
_clear_all(sock, targets, port, brightness, panel_id)
TESTS = {
"identify": test_identify,
"corners": test_corners,
"crosshair": test_crosshair,
"columns": test_columns,
"rows": test_rows,
"flash": test_flash,
"pulse": test_pulse,
"ends": test_chain_ends,
"width": test_width,
}
ALL_ORDER = ("identify", "corners", "ends", "crosshair", "columns", "rows", "flash", "pulse")
def main() -> int:
parser = argparse.ArgumentParser(description="Multi-panel positioning and sync test")
add_panel_args(parser)
parser.add_argument(
"--test",
choices=[*TESTS.keys(), "all"],
default="all",
help="which test to run (default: full sequence)",
)
parser.add_argument("--pause", type=float, default=1.2, help="hold time per static pattern")
parser.add_argument("--column-pause", type=float, default=0.12, help="delay per column in sweep")
parser.add_argument("--row-pause", type=float, default=0.25, help="delay per row in sweep")
parser.add_argument("--flashes", type=int, default=3, help="count for sync flash test")
parser.add_argument("--pulse-seconds", type=float, default=4.0)
parser.add_argument("--fps", type=float, default=30.0)
parser.add_argument("--col-start", type=int, default=32, help="width test: first column")
parser.add_argument("--col-end", type=int, default=42, help="width test: last column")
args = parser.parse_args()
targets = panel_targets_from_args(args)
has_udp = any(not t.local for t in targets)
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) if has_udp else None
targets = apply_pin_arg(args, sock, targets)
print(
f"Panel sync test -> {format_panel_targets(targets, args.port)} "
f"brightness={args.brightness:g}"
)
print("Ctrl+C to stop\n")
tests = ALL_ORDER if args.test == "all" else (args.test,)
try:
for name in tests:
fn = TESTS[name]
if name == "flash":
fn(sock, targets, args.port, args.brightness, args.panel_id, args.pause, args.flashes)
elif name == "pulse":
fn(sock, targets, args.port, args.brightness, args.panel_id, args.pulse_seconds, args.fps)
elif name == "columns":
fn(sock, targets, args.port, args.brightness, args.panel_id, args.column_pause)
elif name == "rows":
fn(sock, targets, args.port, args.brightness, args.panel_id, args.row_pause)
elif name == "width":
fn(sock, targets, args.port, args.brightness, args.panel_id, args.pause, args.col_start, args.col_end)
else:
fn(sock, targets, args.port, args.brightness, args.panel_id, args.pause)
print()
print("Done.")
except KeyboardInterrupt:
print("\nStopped.")
except OSError as exc:
print(f"Network error: {exc}", file=sys.stderr)
return 1
finally:
_clear_all(sock, targets, args.port, args.brightness, args.panel_id)
if sock is not None:
sock.close()
return 0
if __name__ == "__main__":
raise SystemExit(main())
+148
View File
@@ -0,0 +1,148 @@
#!/usr/bin/env python3
"""Step-by-step drive test for Pico panel firmware over UDP."""
from __future__ import annotations
import argparse
import sys
import time
from leds.array_config import MATRIX_HEIGHT, MATRIX_PIXELS, MATRIX_WIDTH, matrix_pixel_index
from leds.panel_udp import PanelClient, add_panel_args, format_panel_targets, panel_targets_from_args
def test_flash(client: PanelClient, pause: float) -> None:
print("Test: RGB flash (red, green, blue, off)")
for name, color in (("red", (255, 0, 0)), ("green", (0, 255, 0)), ("blue", (0, 0, 255))):
print(f" {name}")
client.fill(color)
time.sleep(pause)
client.clear()
time.sleep(pause / 2)
def test_corners(client: PanelClient, pause: float) -> None:
target = client.targets[0]
width, height = target.width, target.height
print("Test: matrix corners (left → right rows)")
corners = [
("top-left", matrix_pixel_index(0, 0, width, height), (255, 0, 0)),
("top-right", matrix_pixel_index(width - 1, 0, width, height), (0, 255, 0)),
("bottom-left", matrix_pixel_index(0, height - 1, width, height), (0, 0, 255)),
("bottom-right", matrix_pixel_index(width - 1, height - 1, width, height), (255, 255, 0)),
]
for label, index, color in corners:
if index >= client.pixel_count:
print(f" skip {label} (index {index})")
continue
print(f" {label} -> LED {index}")
client.set_indices([index], color)
time.sleep(pause)
client.clear()
time.sleep(pause / 2)
def test_rows(client: PanelClient, pause: float) -> None:
target = client.targets[0]
width, height = target.width, target.height
print("Test: one row at a time (white)")
width = min(width, client.pixel_count)
rows = min(height, (client.pixel_count + width - 1) // width)
for y in range(rows):
indices = [matrix_pixel_index(x, y, width, height) for x in range(width)]
indices = [i for i in indices if i < client.pixel_count]
print(f" row {y} ({len(indices)} LEDs)")
client.set_indices(indices, (255, 255, 255))
time.sleep(pause)
client.clear()
time.sleep(pause / 2)
def test_chase(client: PanelClient, pause: float, steps: int | None) -> None:
count = steps if steps is not None else min(client.pixel_count, 60)
print(f"Test: chase ({count} steps, cyan)")
for i in range(count):
client.set_indices([i % client.pixel_count], (0, 255, 255))
time.sleep(pause)
client.clear()
time.sleep(pause / 2)
def test_first_last(client: PanelClient, pause: float) -> None:
print("Test: first and last LED in chain")
last = client.pixel_count - 1
print(f" LED 0 = red, LED {last} = green")
pixels = [(0, 0, 0)] * client.pixel_count
pixels[0] = (255, 0, 0)
pixels[last] = (0, 255, 0)
client.send_pixels(pixels)
time.sleep(pause)
client.clear()
time.sleep(pause / 2)
TESTS = {
"flash": lambda c, p, _s: test_flash(c, p),
"first-last": lambda c, p, _s: test_first_last(c, p),
"corners": lambda c, p, _s: test_corners(c, p),
"rows": lambda c, p, _s: test_rows(c, p),
"chase": test_chase,
}
def run_all(client: PanelClient, pause: float, chase_steps: int | None) -> None:
test_flash(client, pause)
test_first_last(client, pause)
if client.pixel_count >= client.targets[0].width:
test_corners(client, pause)
test_rows(client, max(pause * 0.6, 0.15))
test_chase(client, max(pause * 0.25, 0.05), chase_steps)
client.clear()
print("Done.")
def main() -> int:
parser = argparse.ArgumentParser(description="Drive-test Pico panel LEDs over UDP")
add_panel_args(parser)
parser.add_argument("--pause", type=float, default=0.8)
parser.add_argument("--test", choices=[*TESTS.keys(), "all"], default="all")
parser.add_argument("--chase-steps", type=int, default=None)
parser.add_argument("--color", default=None, metavar="R,G,B")
args = parser.parse_args()
if args.count is not None and args.count < 1:
print("Error: --count must be at least 1", file=sys.stderr)
return 1
targets = panel_targets_from_args(args)
print(
f"Panel drive test -> {format_panel_targets(targets, args.port)} "
f"brightness={args.brightness:g}"
)
try:
with PanelClient.from_args(args) as client:
if args.color:
parts = [int(p.strip()) for p in args.color.split(",")]
if len(parts) != 3:
raise SystemExit("--color must be R,G,B")
client.fill((parts[0], parts[1], parts[2]))
print(f"Sent solid color {tuple(parts)}")
return 0
if args.test == "all":
run_all(client, args.pause, args.chase_steps)
else:
TESTS[args.test](client, args.pause, args.chase_steps)
client.clear()
print("Done.")
except KeyboardInterrupt:
print("\nStopped.")
except OSError as exc:
print(f"Network error: {exc}", file=sys.stderr)
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())
+209
View File
@@ -0,0 +1,209 @@
#!/usr/bin/env python3
"""Display text on Pico panel(s) over UDP."""
from __future__ import annotations
import argparse
import socket
import subprocess
import sys
import time
from leds.panel_udp import (
PanelMatrixBundle,
add_panel_args,
apply_pin_arg,
format_panel_targets,
panel_targets_from_args,
push_panel_frames,
)
from leds.text import (
blit_mask,
blit_mask_gradient,
draw_font_text_centered,
draw_text,
draw_text_centered,
render_font_mask,
)
DEFAULT_NAME = "JIMMY"
def _default_name() -> str:
try:
out = subprocess.run(
["git", "config", "user.name"],
capture_output=True,
text=True,
check=False,
)
name = out.stdout.strip()
if name:
return name.upper()
except OSError:
pass
return DEFAULT_NAME
def main() -> int:
parser = argparse.ArgumentParser(description="Display text on LED panel(s)")
parser.add_argument("text", nargs="?", default=None)
add_panel_args(parser)
parser.add_argument("--color", default="255,200,80", metavar="R,G,B")
parser.add_argument("--hold", type=float, default=0.0, help="0 = show until Ctrl+C")
parser.add_argument("--scroll", action="store_true")
parser.add_argument(
"--cursive",
action="store_true",
help="Italic script style (keeps letter case; auto-scrolls if too wide)",
)
parser.add_argument(
"--font-size",
type=int,
default=None,
help="Font point size (default: panel height + 3)",
)
parser.add_argument(
"--letter-spacing",
type=int,
default=0,
help="Pixels between cursive letters (default: 0)",
)
parser.add_argument(
"--speed",
type=float,
default=None,
help="Scroll delay in seconds (default: 0.15 cursive, 0.08 bitmap)",
)
parser.add_argument(
"--rainbow",
action="store_true",
help="Rainbow gradient across cursive text",
)
args = parser.parse_args()
if args.cursive:
text = args.text or "Allyssia"
else:
text = (args.text or _default_name()).upper()
parts = [int(p.strip()) for p in args.color.split(",")]
if len(parts) != 3:
print("Error: --color must be R,G,B", file=sys.stderr)
return 1
color = (parts[0], parts[1], parts[2])
targets = panel_targets_from_args(args)
has_udp = any(not t.local for t in targets)
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) if has_udp else None
targets = apply_pin_arg(args, sock, targets)
style = "cursive" if args.cursive else "bitmap"
cursive_masks: dict[int, tuple[int, list[tuple[int, int]]]] = {}
if args.cursive:
try:
for target in targets:
width, _, ons = render_font_mask(
text,
height=target.height,
size=args.font_size,
letter_spacing=args.letter_spacing,
)
cursive_masks[target.index] = (width, ons)
except (ImportError, FileNotFoundError, OSError) as exc:
print(f"Error: {exc}", file=sys.stderr)
return 1
# Auto-scroll when the rendered text is wider than any target panel.
do_scroll = args.scroll
if args.cursive and not do_scroll:
if any(
cursive_masks[t.index][0] > t.width for t in targets if t.index in cursive_masks
):
do_scroll = True
scroll_delay = args.speed
if scroll_delay is None:
scroll_delay = 0.15 if args.cursive else 0.08
print(
f"Displaying '{text}' ({style}"
f"{', scrolling' if do_scroll else ''})"
f" on {format_panel_targets(targets, args.port)} "
f"brightness={args.brightness:g}"
)
try:
with PanelMatrixBundle(targets, args.brightness) as bundle:
if do_scroll:
start = time.monotonic()
offsets = {
target.index: matrix.width for target, matrix in bundle.matrices
}
while True:
phase = (time.monotonic() - start) * 0.55
for target, matrix in bundle.matrices:
matrix.clear()
if args.cursive:
tw, ons = cursive_masks[target.index]
if args.rainbow:
blit_mask_gradient(
matrix,
ons,
offsets[target.index],
0,
width=tw,
phase=phase,
cycles=2.5,
)
else:
blit_mask(matrix, ons, offsets[target.index], 0, color)
wrap = tw
else:
draw_text(matrix, text, offsets[target.index], 1, color)
wrap = len(text) * 6
offsets[target.index] -= 1
if offsets[target.index] < -wrap:
offsets[target.index] = matrix.width
push_panel_frames(sock, bundle.matrices, args.port, args.panel_id)
time.sleep(scroll_delay)
if args.hold > 0 and time.monotonic() - start >= args.hold:
break
else:
for target, matrix in bundle.matrices:
matrix.clear()
if args.cursive:
if args.rainbow:
tw, ons = cursive_masks[target.index]
x = max(0, (matrix.width - tw) // 2)
blit_mask_gradient(matrix, ons, x, 0, width=tw, phase=0.0)
else:
draw_font_text_centered(
matrix,
text,
color,
size=args.font_size,
letter_spacing=args.letter_spacing,
)
else:
draw_text_centered(matrix, text, color)
push_panel_frames(sock, bundle.matrices, args.port, args.panel_id)
if args.hold <= 0:
print("Showing (Ctrl+C to clear)")
while True:
time.sleep(1)
else:
time.sleep(args.hold)
except KeyboardInterrupt:
print("\nClearing.")
finally:
with PanelMatrixBundle(targets, args.brightness) as bundle:
for _, matrix in bundle.matrices:
matrix.clear()
push_panel_frames(sock, bundle.matrices, args.port, args.panel_id)
if sock is not None:
sock.close()
return 0
if __name__ == "__main__":
raise SystemExit(main())
+53
View File
@@ -0,0 +1,53 @@
#!/usr/bin/env python3
"""Send one animation to Pico panel(s) over UDP."""
from __future__ import annotations
import argparse
import sys
from leds.animations import ANIMATIONS, DEFAULT_FPS
from leds.panel_udp import (
add_panel_args,
format_panel_targets,
panel_targets_from_args,
parse_pins_arg,
run_panel_animation_loop,
)
def main() -> int:
parser = argparse.ArgumentParser(description="UDP animation sender for Pico panel(s)")
add_panel_args(parser)
parser.add_argument("--animation", default="rainbow", choices=sorted(ANIMATIONS))
parser.add_argument("--fps", type=float, default=None)
args = parser.parse_args()
targets = panel_targets_from_args(args)
fps = args.fps or DEFAULT_FPS.get(args.animation, 30)
print(
f"Sending {args.animation} to {format_panel_targets(targets, args.port)} "
f"@ {fps:.0f} fps brightness={args.brightness:g}"
)
try:
run_panel_animation_loop(
targets,
args.port,
args.animation,
fps=args.fps,
brightness=args.brightness,
panel_id=args.panel_id,
pin=args.pin,
pins=parse_pins_arg(args),
)
except KeyboardInterrupt:
print("\nStopped.")
except OSError as exc:
print(f"Network error: {exc}", file=sys.stderr)
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())
+69
View File
@@ -0,0 +1,69 @@
#!/usr/bin/env python3
"""Print or install Pi 5 ws2812-pio overlay lines for two strips."""
from __future__ import annotations
import argparse
import sys
from pathlib import Path
CONFIG = Path("/boot/firmware/config.txt")
MARKER = "# portal ws2812 strips"
def overlay_lines(count0: int, count1: int, gpio0: int, gpio1: int) -> list[str]:
return [
MARKER,
f"dtoverlay=ws2812-pio,gpio={gpio0},num_leds={count0}",
f"dtoverlay=ws2812-pio,gpio={gpio1},num_leds={count1}",
]
def main() -> int:
parser = argparse.ArgumentParser(
description="Configure Pi 5 ws2812-pio overlays for two LED strips"
)
parser.add_argument("-n", type=int, nargs=2, default=[10, 10], metavar=("STRIP0", "STRIP1"))
parser.add_argument("--gpio0", type=int, default=18, help="strip 0 data pin (default 18)")
parser.add_argument("--gpio1", type=int, default=13, help="strip 1 data pin (default 13)")
parser.add_argument(
"--install",
action="store_true",
help="append overlay lines to /boot/firmware/config.txt (requires sudo)",
)
args = parser.parse_args()
lines = overlay_lines(args.n[0], args.n[1], args.gpio0, args.gpio1)
print("Add these lines to /boot/firmware/config.txt under [all], then reboot:\n")
for line in lines:
print(line)
print(
f"\nWiring: strip 0 DIN → GPIO {args.gpio0}, strip 1 DIN → GPIO {args.gpio1}, "
"GND → Pi GND, 5V from external supply for long strips."
)
print("\nAfter reboot you should see /dev/leds0 and /dev/leds1.")
if not args.install:
return 0
if not CONFIG.exists():
print(f"Error: {CONFIG} not found", file=sys.stderr)
return 1
text = CONFIG.read_text()
if MARKER in text:
print("\nOverlay lines already present in config.txt.")
return 0
with CONFIG.open("a") as f:
f.write("\n")
for line in lines:
f.write(line + "\n")
print(f"\nAppended to {CONFIG}. Reboot to apply.")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+250
View File
@@ -0,0 +1,250 @@
"""Pattern plugin smoke tests."""
from __future__ import annotations
import sys
from pathlib import Path
SRC = Path(__file__).resolve().parent.parent / "src"
if str(SRC) not in sys.path:
sys.path.insert(0, str(SRC))
from leds.panel_udp import HeadlessMatrix
from leds.pattern import get_registry, merge_params
NEW_PATTERN_IDS = (
"aurora",
"meteor",
"embers",
"tide",
"vortex",
"serpent",
"lattice",
"eclipse",
)
def test_registry_loads_bundled_patterns() -> None:
ids = get_registry().ids()
assert "rainbow" in ids
assert "scanner" in ids
for pattern_id in NEW_PATTERN_IDS:
assert pattern_id in ids, pattern_id
assert len(ids) >= 18
def test_scanner_draws_pixels() -> None:
cls = get_registry()["scanner"]
inst = cls()
params = merge_params(cls, {"colors": {"beam": "#00ff00"}})
matrix = HeadlessMatrix(20, 9, brightness=1.0)
inst.draw(matrix, 0.5, params)
assert any(matrix.rgb_bytes())
def test_ancient_text_patterns_draw() -> None:
registry = get_registry()
for pattern_id in (
"inscription",
"runes",
"glyphs",
"oracle",
"procession",
"rite",
"braziers",
"constellation",
"veil",
*NEW_PATTERN_IDS,
):
assert pattern_id in registry
cls = registry[pattern_id]
matrix = HeadlessMatrix(45, 9, brightness=1.0)
cls().draw(matrix, 1.25, merge_params(cls))
assert max(matrix.rgb_bytes()) > 40, pattern_id
def test_portal_ring_spans_panels() -> None:
from leds.array_config import PANEL_COUNT, panel_layout
from leds.portal_span import inject_portal, portal_ring_width
cls = get_registry()["procession"]
inst = cls()
params = merge_params(cls)
lit_global = set()
for index in range(PANEL_COUNT):
width, height = panel_layout(index)
matrix = HeadlessMatrix(width, height, brightness=1.0)
inject_portal(params, index)
inst.draw(matrix, 2.5, params)
rgb = matrix.rgb_bytes()
offset = params["portal"]["x_offset"]
for x in range(width):
# Middle row brightness sample.
i = ((height // 2) * width + x) * 3
if rgb[i] + rgb[i + 1] + rgb[i + 2] > 30:
lit_global.add(offset + x)
# Spanning pattern should light pixels across more than one panel's width.
assert len(lit_global) > 50
assert params["portal"]["ring_width"] == portal_ring_width()
def test_ancient_patterns_have_no_preset_bpm() -> None:
for pattern_id in (
"inscription",
"runes",
"glyphs",
"oracle",
"rite",
"veil",
"procession",
"braziers",
"constellation",
"aurora",
"vortex",
"serpent",
"lattice",
"eclipse",
"tide",
"embers",
):
names = {p["name"] for p in get_registry()[pattern_id].schema()["params"]}
assert "bpm" not in names, pattern_id
def test_resolve_bpm_uses_settings_fallback() -> None:
from leds.bpm_limits import resolve_bpm
params = merge_params(get_registry()["rite"])
params["beat"] = {"active": False, "fallback_bpm": 140}
assert resolve_bpm(params, fallback=120) == 140
params["extras"]["beat_detect"] = True
params["beat"] = {"active": True, "bpm": 128, "fallback_bpm": 140}
assert resolve_bpm(params, fallback=120) == 128
params["extras"]["beat_detect"] = False
assert resolve_bpm(params, fallback=120) == 140
def test_no_synthetic_pulse_when_beat_detect_off() -> None:
from leds.bpm_limits import resolve_beat_wave, resolve_pulse
params = merge_params(get_registry()["rite"])
assert params["extras"]["beat_detect"] is False
params["beat"] = {"active": True, "bpm": 128, "pulse": 0.9, "fallback_bpm": 120}
assert resolve_pulse(params, 0.25, fallback_bpm=120) == 0.0
assert resolve_beat_wave(params, 0.25, fallback_bpm=120) == 0.5
def test_patterns_prefer_live_beat_pulse() -> None:
from leds.bpm_limits import resolve_pulse
cls = get_registry()["rite"]
params = merge_params(cls)
params["extras"]["beat_detect"] = True
params["beat"] = {
"active": True,
"bpm": 128,
"pulse": 0.9,
"beat_seq": 12,
"kick": 0.8,
"bass": 0.6,
"energy": 0.7,
"flux": 0.2,
"hat": 0.1,
"snare": 0.3,
"level": 0.4,
}
assert resolve_pulse(params, 0.0, fallback_bpm=120) == 0.9
matrix = HeadlessMatrix(45, 9, brightness=1.0)
cls().draw(matrix, 1.0, params)
assert max(matrix.rgb_bytes()) > 40
def test_inscription_accepts_custom_text() -> None:
cls = get_registry()["inscription"]
schema = cls.schema()
text_param = next(p for p in schema["params"] if p["name"] == "text")
assert text_param["type"] == "text"
params = merge_params(cls, {"extras": {"text": "ave"}})
assert params["extras"]["text"] == "ave"
matrix = HeadlessMatrix(45, 9, brightness=1.0)
cls().draw(matrix, 0.0, params)
assert max(matrix.rgb_bytes()) > 40
def test_oracle_splits_custom_words() -> None:
from patterns.oracle import _words
assert _words("lux / nox") == ("LUX", "NOX")
cls = get_registry()["oracle"]
params = merge_params(cls, {"extras": {"text": "LUX", "beats_per_word": 3.0}})
assert params["extras"]["text"] == "LUX"
assert params["extras"]["beats_per_word"] == 3.0
matrix = HeadlessMatrix(45, 9, brightness=1.0)
cls().draw(matrix, 0.5, params)
assert max(matrix.rgb_bytes()) > 40
def test_preset_normalize_keeps_text() -> None:
from leds.presets import _normalize
preset = _normalize(
{
"name": "Hello",
"pattern": "inscription",
"brightness": 0.4,
"speed": 1.0,
"colors": {"ink": "#e0b44a", "stone": "#2a1c10"},
"extras": {"text": "HELLO"},
}
)
assert preset["extras"]["text"] == "HELLO"
def test_new_pattern_presets_normalize() -> None:
from leds.presets import _normalize
preset = _normalize(
{
"name": "Vortex",
"pattern": "vortex",
"brightness": 0.42,
"speed": 1.0,
"colors": {"core": "#6a20ff", "rim": "#f0c060", "void": "#05030a"},
"extras": {"twist": 0.75, "bg_brightness": 0.4},
}
)
assert preset["pattern"] == "vortex"
assert preset["extras"]["twist"] == 0.75
def test_text_and_background_brightness() -> None:
cls = get_registry()["inscription"]
names = {p["name"] for p in cls.schema()["params"]}
assert "text_brightness" in names
assert "bg_brightness" in names
bright = merge_params(cls, {"extras": {"text": "AVE", "text_brightness": 1.0, "bg_brightness": 0.0}})
dimmed = merge_params(cls, {"extras": {"text": "AVE", "text_brightness": 0.0, "bg_brightness": 0.0}})
a = HeadlessMatrix(45, 9, brightness=1.0)
b = HeadlessMatrix(45, 9, brightness=1.0)
cls().draw(a, 0.0, bright)
cls().draw(b, 0.0, dimmed)
assert sum(a.rgb_bytes()) > sum(b.rgb_bytes())
if __name__ == "__main__":
test_registry_loads_bundled_patterns()
test_scanner_draws_pixels()
test_ancient_text_patterns_draw()
test_portal_ring_spans_panels()
test_ancient_patterns_have_no_preset_bpm()
test_resolve_bpm_uses_settings_fallback()
test_no_synthetic_pulse_when_beat_detect_off()
test_patterns_prefer_live_beat_pulse()
test_inscription_accepts_custom_text()
test_oracle_splits_custom_words()
test_preset_normalize_keeps_text()
test_text_and_background_brightness()
test_new_pattern_presets_normalize()
print("ok")