portal/leds/panel_udp.py

"""UDP helpers for Pico panel firmware and Pi-direct panels."""

from __future__ import annotations

import argparse
import asyncio
import os
import socket
import subprocess
import time
from dataclasses import dataclass
from typing import Any, Iterable, Sequence, Tuple

from leds.animations import ANIMATIONS, DEFAULT_FPS
from leds.array_config import (
    LOCAL_PANEL_INDICES,
    MATRIX_HEIGHT,
    MATRIX_PIXELS,
    MATRIX_WIDTH,
    PANEL_COUNT,
    PANEL_HOST_BY_INDEX,
    PANEL_STRIP_BY_INDEX,
    PANEL_WS2812_PIN_BY_INDEX,
    PI_BRIDGE_PANEL_INDICES,
    matrix_pixel_index,
    panel_layout,
    panel_pixel_count,
)
from leds.colors import dim

RGB = Tuple[int, int, int]

PANEL_IP_NET = "10.1.1"
PANEL_IP_BASE = 10
PANEL_MAX_FPS = 15.0
PANEL_SEND_TIMEOUT_S = 0.05

DEFAULT_PORT = int(os.environ.get("PORTAL_PANEL_PORT", "50007"))


def panel_ip(panel_id: int) -> str:
    return f"{PANEL_IP_NET}.{PANEL_IP_BASE + panel_id}"


DEFAULT_HOST = panel_ip(0)
DEFAULT_PANEL_HOSTS = [panel_ip(i) for i in range(PANEL_COUNT)]


@dataclass(frozen=True)
class PanelTarget:
    index: int
    host: str
    width: int
    height: int
    local: bool = False
    strip: int = 0
    pin: int | None = None

    @property
    def pixels(self) -> int:
        return self.width * self.height

    def label(self) -> str:
        if self.local:
            return f"panel {self.index} Pi-direct ({self.width}×{self.height})"
        strip = f" strip {self.strip}" if self.strip else ""
        pin = f" GP{self.pin}" if self.pin is not None else ""
        if self.index >= 0:
            return f"{self.host} ({self.width}×{self.height}, panel {self.index}{strip}{pin})"
        return f"{self.host} ({self.width}×{self.height}{strip}{pin})"


def _make_panel_target(index: int, width: int, height: int) -> PanelTarget:
    local = index in LOCAL_PANEL_INDICES
    if local:
        host = "local"
    elif index in PANEL_HOST_BY_INDEX:
        host = PANEL_HOST_BY_INDEX[index]
    elif index in PI_BRIDGE_PANEL_INDICES:
        host = "127.0.0.1"  # panel_spi_bridge on this Pi — no 10.1.1.12 on lo
    else:
        host = panel_ip(index)
    strip = PANEL_STRIP_BY_INDEX.get(index, 0)
    pin = PANEL_WS2812_PIN_BY_INDEX.get(index)
    return PanelTarget(index, host, width, height, local=local, strip=strip, pin=pin)


def panel_hosts(panel_index: int | None = None, count: int = PANEL_COUNT) -> list[str]:
    return [t.host for t in resolve_panel_targets(panel_index=panel_index, panels=count)]


def resolve_panel_targets(
    host: str | None = None,
    panel_index: int | None = None,
    panels: int = PANEL_COUNT,
    *,
    width: int | None = None,
    height: int | None = None,
) -> list[PanelTarget]:
    if panel_index is not None:
        w, h = panel_layout(panel_index)
        if width is not None:
            w = width
        if height is not None:
            h = height
        return [_make_panel_target(panel_index, w, h)]
    if host:
        w = width if width is not None else MATRIX_WIDTH
        h = height if height is not None else MATRIX_HEIGHT
        return [PanelTarget(-1, host, w, h)]
    return [
        _make_panel_target(i, *panel_layout(i))
        for i in range(panels)
    ]


def format_panel_targets(targets: Sequence[PanelTarget], port: int) -> str:
    if len(targets) == 1:
        t = targets[0]
        suffix = "" if t.local else f":{port}"
        return f"{t.label()}{suffix}"
    layouts = ", ".join(f"{t.width}×{t.height}" for t in targets)
    udp = [t for t in targets if not t.local]
    if udp:
        return (
            f"{len(targets)} panels ({udp[0].host} … {udp[-1].host}):{port} [{layouts}]"
        )
    return f"{len(targets)} Pi-direct panels [{layouts}]"


def host_is_reachable(host: str) -> bool:
    """Return False for hosts that would block UDP sends (ARP INCOMPLETE/FAILED).

    A host may have neigh entries on more than one interface (e.g. eth0 REACHABLE
    and wlan0 FAILED). Treat it as reachable if *any* entry is usable.
    """
    if host in ("127.0.0.1", "localhost", "::1"):
        return True
    try:
        result = subprocess.run(
            ["ip", "neigh", "show", host],
            capture_output=True,
            text=True,
            timeout=1,
            check=False,
        )
        lines = [ln for ln in result.stdout.splitlines() if ln.strip()]
        if lines:
            if any(
                "FAILED" not in ln and "INCOMPLETE" not in ln
                for ln in lines
            ):
                return True
    except (OSError, subprocess.SubprocessError):
        pass
    try:
        result = subprocess.run(
            ["ping", "-c", "1", "-W", "1", host],
            capture_output=True,
            timeout=2,
            check=False,
        )
        return result.returncode == 0
    except (OSError, subprocess.SubprocessError):
        return False


def filter_reachable_targets(targets: Sequence[PanelTarget]) -> list[PanelTarget]:
    """Drop offline UDP panels so sendto() does not stall the animation loop on ARP."""
    reachable: list[PanelTarget] = []
    for target in targets:
        if target.local:
            reachable.append(target)
        elif host_is_reachable(target.host):
            reachable.append(target)
        else:
            print(f"Skipping offline panel {target.index} ({target.host})")
    if not reachable:
        raise OSError("No reachable panels on the network")
    return reachable


def add_panel_network_args(parser: argparse.ArgumentParser) -> None:
    parser.add_argument(
        "--host",
        default=None,
        help="Single panel IP override (not needed; default uses 10.1.1.10–14)",
    )
    parser.add_argument(
        "--panel-index",
        type=int,
        default=None,
        choices=range(PANEL_COUNT),
        metavar="N",
        help=f"Drive only panel N (0–{PANEL_COUNT - 1}); default all {PANEL_COUNT}",
    )
    parser.add_argument(
        "--panels",
        type=int,
        default=PANEL_COUNT,
        help=f"Number of panels when driving all (default {PANEL_COUNT})",
    )
    parser.add_argument("--port", type=int, default=DEFAULT_PORT)
    parser.add_argument("--panel-id", type=int, default=None, help="Prefix UDP frames with panel id")
    parser.add_argument(
        "--pin",
        type=int,
        default=None,
        metavar="GP",
        help="Set strip 0 WS2812 GPIO over UDP before sending frames",
    )
    parser.add_argument(
        "--pins",
        default=None,
        metavar="GP[:LEDS],...",
        help="Bind strips 0..n-1, e.g. 28,27 or 28:405,27:351 (per-pin length)",
    )
    parser.add_argument(
        "--include-offline",
        action="store_true",
        help="Send to all configured panels even if unreachable (ARP stalls hit background thread only)",
    )
    parser.add_argument(
        "--sync-send",
        action="store_true",
        help="Send UDP on the animation thread (blocks; not recommended)",
    )


def add_panel_args(parser: argparse.ArgumentParser) -> None:
    add_panel_network_args(parser)
    parser.add_argument(
        "--width",
        type=int,
        default=None,
        help="Override panel width (default from leds/array_config.py per panel)",
    )
    parser.add_argument(
        "--height",
        type=int,
        default=None,
        help=f"Override panel height (default {MATRIX_HEIGHT})",
    )
    parser.add_argument(
        "--count",
        type=int,
        default=None,
        help="LEDs per frame (default: panel width × height)",
    )
    parser.add_argument("--brightness", type=float, default=0.35)


def panel_targets_from_args(args: argparse.Namespace) -> list[PanelTarget]:
    targets = resolve_panel_targets(
        args.host,
        args.panel_index,
        args.panels,
        width=getattr(args, "width", None),
        height=getattr(args, "height", None),
    )
    if getattr(args, "include_offline", False):
        return targets
    if args.panel_index is not None or args.host:
        return targets
    return filter_reachable_targets(targets)


def panel_count_from_args(args: argparse.Namespace) -> int:
    if args.count is not None:
        return args.count
    if args.panel_index is not None:
        return panel_pixel_count(args.panel_index)
    if args.width is not None:
        h = args.height if args.height is not None else MATRIX_HEIGHT
        return args.width * h
    return MATRIX_PIXELS


class HeadlessMatrix:
    """Matrix buffer for UDP senders — no local SPI/PIO."""

    def __init__(
        self,
        width: int = MATRIX_WIDTH,
        height: int = MATRIX_HEIGHT,
        *,
        brightness: float = 0.25,
    ) -> None:
        self.width = width
        self.height = height
        self.brightness = brightness
        self._pixels: list[RGB] = [(0, 0, 0)] * (width * height)

    def _index_at(self, x: int, y: int) -> int:
        return matrix_pixel_index(x, y, self.width, self.height)

    def __len__(self) -> int:
        return len(self._pixels)

    def __getitem__(self, pos: Tuple[int, int]) -> RGB:
        return self._pixels[self._index_at(*pos)]

    def __setitem__(self, pos: Tuple[int, int], color: RGB) -> None:
        self._pixels[self._index_at(*pos)] = color

    def fill(self, color: RGB) -> None:
        self._pixels = [color] * len(self._pixels)

    def clear(self) -> None:
        self.fill((0, 0, 0))

    def set_index(self, index: int, color: RGB) -> None:
        if 0 <= index < len(self._pixels):
            self._pixels[index] = color

    def rgb_bytes(self) -> bytes:
        out = bytearray(len(self._pixels) * 3)
        for i, (r, g, b) in enumerate(self._pixels):
            r, g, b = dim((r, g, b), self.brightness)
            base = i * 3
            out[base], out[base + 1], out[base + 2] = r, g, b
        return bytes(out)


class PanelMatrixBundle:
    """Drawing surfaces for UDP (headless) and Pi-direct (LedMatrix) panels."""

    def __init__(
        self,
        targets: Sequence[PanelTarget],
        brightness: float,
    ) -> None:
        from leds.config import panel_strip_config
        from leds.matrix import LedMatrix

        self.targets = list(targets)
        self.matrices: list[tuple[PanelTarget, Any]] = []
        self._local: list[LedMatrix] = []
        for target in self.targets:
            if target.local:
                matrix = LedMatrix(
                    target.width,
                    target.height,
                    config0=panel_strip_config(
                        target.index, target.pixels, brightness=brightness
                    ),
                )
                self._local.append(matrix)
                self.matrices.append((target, matrix))
            else:
                self.matrices.append(
                    (target, HeadlessMatrix(target.width, target.height, brightness=brightness))
                )

    def close(self) -> None:
        for matrix in self._local:
            matrix.close()

    def __enter__(self) -> PanelMatrixBundle:
        return self

    def __exit__(self, *_exc) -> None:
        self.close()


def push_panel_frame(
    sock: socket.socket | None,
    target: PanelTarget,
    matrix: Any,
    port: int,
    panel_id: int | None,
) -> None:
    if target.local:
        matrix.show()
    else:
        if sock is None:
            raise ValueError("UDP panel requires a socket")
        send_frame(
            sock,
            target.host,
            port,
            matrix.rgb_bytes(),
            panel_id,
            strip=target.strip,
        )


def push_panel_frames(
    sock: socket.socket | None,
    matrices: Sequence[tuple[PanelTarget, Any]],
    port: int,
    panel_id: int | None,
) -> None:
    for target, matrix in matrices:
        push_panel_frame(sock, target, matrix, port, panel_id)


def solid_frame_bytes(
    color: RGB,
    count: int = MATRIX_PIXELS,
    brightness: float = 1.0,
) -> bytes:
    r, g, b = dim(color, brightness)
    return bytes((r, g, b)) * count


def pixels_to_frame_bytes(
    pixels: Iterable[RGB],
    brightness: float = 1.0,
) -> bytes:
    out = bytearray()
    for r, g, b in pixels:
        r, g, b = dim((r, g, b), brightness)
        out.extend((r, g, b))
    return bytes(out)


def encode_frame(rgb: bytes, panel_id: int | None = None, *, strip: int = 0) -> bytes:
    """Prefix strip index (or explicit panel_id) so multi-strip Picos route correctly."""
    prefix = panel_id if panel_id is not None else strip
    if prefix is None:
        return rgb
    # Always prefix when strip!=0; strip 0 may omit for tiny payloads but
    # prefixing is harmless and keeps multi-strip hosts consistent.
    if panel_id is None and strip == 0:
        return rgb
    return bytes([prefix & 0xFF]) + rgb


def encode_pin_command(pin: int, strip: int = 0, length: int = 0) -> bytes:
    """UDP PIN command.

    strip 0, no length → 5 bytes ``PIN\\0`` + gpio.
    Otherwise → 8 bytes ``PIN\\0`` + strip + gpio + len_lo + len_hi (little-endian).
    """
    if not 0 <= pin <= 29:
        raise ValueError(f"WS2812 pin must be 0–29, got {pin}")
    if not 0 <= strip <= 255:
        raise ValueError(f"strip must be 0–255, got {strip}")
    if length < 0 or length > 0xFFFF:
        raise ValueError(f"length must be 0–65535, got {length}")
    if strip == 0 and length == 0:
        return b"PIN\0" + bytes([pin & 0xFF])
    return b"PIN\0" + bytes(
        [strip & 0xFF, pin & 0xFF, length & 0xFF, (length >> 8) & 0xFF]
    )


def send_ws2812_pin(
    sock: socket.socket,
    host: str,
    port: int,
    pin: int,
    strip: int = 0,
    length: int = 0,
) -> None:
    sock.sendto(encode_pin_command(pin, strip, length), (host, port))


def send_ws2812_pin_to_targets(
    sock: socket.socket | None,
    targets: Sequence[PanelTarget],
    port: int,
    pin: int,
    strip: int = 0,
    length: int = 0,
) -> None:
    if sock is None:
        return
    for target in targets:
        if target.local:
            continue
        send_ws2812_pin(sock, target.host, port, pin, strip, length)
        extra = f" leds={length}" if length else ""
        print(f"WS2812 strip {strip} GP{pin}{extra} -> {target.host}:{port}")


def configure_target_pins(
    sock: socket.socket | None,
    targets: Sequence[PanelTarget],
    port: int,
) -> None:
    """Send PIN for each target that has pin set (length = panel pixel count)."""
    if sock is None:
        return
    seen: set[tuple[str, int, int]] = set()
    for target in targets:
        if target.local or target.pin is None:
            continue
        key = (target.host, target.strip, target.pin)
        if key in seen:
            continue
        seen.add(key)
        length = target.pixels
        send_ws2812_pin(sock, target.host, port, target.pin, target.strip, length)
        print(
            f"WS2812 strip {target.strip} GP{target.pin} leds={length} "
            f"-> {target.host}:{port}"
        )


def setup_panel_pins(
    sock: socket.socket | None,
    targets: Sequence[PanelTarget],
    port: int,
    *,
    pin: int | None = None,
    pins: Sequence[tuple[int, int | None]] | None = None,
) -> None:
    """Apply array_config pins, then optional CLI overrides.

    ``pins`` entries are ``(gpio, length|None)`` — length None means unset.
    """
    configure_target_pins(sock, targets, port)
    if sock is None:
        return
    if pins:
        hosts = sorted({t.host for t in targets if not t.local})
        for host in hosts:
            for strip, (gp, length) in enumerate(pins):
                leds = length or 0
                send_ws2812_pin(sock, host, port, gp, strip, leds)
                extra = f" leds={leds}" if leds else ""
                print(f"WS2812 strip {strip} GP{gp}{extra} -> {host}:{port}")
        return
    if pin is not None:
        send_ws2812_pin_to_targets(sock, targets, port, pin, strip=0)


def parse_pins_arg(args: argparse.Namespace) -> list[tuple[int, int | None]] | None:
    """Parse ``--pins 28,27`` or ``--pins 28:405,27:351`` → [(gpio, leds|None), ...]."""
    raw = getattr(args, "pins", None)
    if not raw:
        return None
    out: list[tuple[int, int | None]] = []
    for part in raw.split(","):
        part = part.strip()
        if not part:
            continue
        if ":" in part:
            gp_s, led_s = part.split(":", 1)
            out.append((int(gp_s.strip()), int(led_s.strip())))
        else:
            out.append((int(part), None))
    return out or None


def expand_targets_for_pins(
    targets: Sequence[PanelTarget],
    pins: Sequence[tuple[int, int | None]],
) -> list[PanelTarget]:
    """One PanelTarget per strip so frames reach every GPIO from ``--pins``."""
    out: list[PanelTarget] = []
    for target in targets:
        if target.local:
            out.append(target)
            continue
        for strip, (gp, length) in enumerate(pins):
            if length and length > 0 and target.height > 0 and length % target.height == 0:
                width = length // target.height
                height = target.height
            elif length and length > 0:
                width, height = length, 1
            else:
                width, height = target.width, target.height
            out.append(
                PanelTarget(
                    target.index,
                    target.host,
                    width,
                    height,
                    local=False,
                    strip=strip,
                    pin=gp,
                )
            )
    return out


def apply_pin_arg(
    args: argparse.Namespace,
    sock: socket.socket | None,
    targets: Sequence[PanelTarget],
) -> list[PanelTarget]:
    """Send PIN commands; return targets expanded for ``--pins`` multi-strip."""
    pins = parse_pins_arg(args)
    pin = getattr(args, "pin", None)
    setup_panel_pins(
        sock,
        targets,
        args.port,
        pin=pin,
        pins=pins,
    )
    if pins:
        return expand_targets_for_pins(targets, pins)
    if pin is not None:
        return [
            PanelTarget(
                t.index,
                t.host,
                t.width,
                t.height,
                local=t.local,
                strip=t.strip,
                pin=pin if not t.local else t.pin,
            )
            for t in targets
        ]
    return list(targets)


def send_frame(
    sock: socket.socket,
    host: str,
    port: int,
    rgb: bytes,
    panel_id: int | None,
    *,
    strip: int = 0,
) -> None:
    # Firmware expects logical RGB; converts to GRB on the wire.
    sock.sendto(encode_frame(rgb, panel_id, strip=strip), (host, port))


async def send_frames_async(
    sock: socket.socket | None,
    matrices: Sequence[tuple[PanelTarget, Any]],
    port: int,
    panel_id: int | None,
    *,
    sync_send: bool = False,
    send_timeout: float = PANEL_SEND_TIMEOUT_S,
) -> None:
    """Send one frame to every panel (UDP async, Pi-direct via show())."""
    if sync_send:
        push_panel_frames(sock, matrices, port, panel_id)
        return
    tasks: list[asyncio.Task[None]] = []
    for target, matrix in matrices:
        if target.local:
            matrix.show()
        else:
            tasks.append(
                asyncio.create_task(
                    asyncio.to_thread(
                        send_frame,
                        sock,
                        target.host,
                        port,
                        matrix.rgb_bytes(),
                        panel_id,
                        strip=target.strip,
                    )
                )
            )
    if not tasks:
        return
    _, pending = await asyncio.wait(tasks, timeout=send_timeout)
    for task in pending:
        task.cancel()


def send_solid_to_targets(
    sock: socket.socket | None,
    targets: Sequence[PanelTarget],
    port: int,
    color: RGB,
    brightness: float,
    panel_id: int | None,
) -> None:
    with PanelMatrixBundle(targets, brightness) as bundle:
        for _, matrix in bundle.matrices:
            matrix.fill(color)
        push_panel_frames(sock, bundle.matrices, port, panel_id)


class PanelClient:
    """Send frames to one or more portal panel firmware targets."""

    def __init__(
        self,
        targets: PanelTarget | Sequence[PanelTarget] | str,
        port: int = DEFAULT_PORT,
        *,
        pixel_count: int | None = None,
        panel_id: int | None = None,
        brightness: float = 0.35,
    ) -> None:
        if isinstance(targets, PanelTarget):
            self.targets = [targets]
        elif isinstance(targets, str):
            count = pixel_count or MATRIX_PIXELS
            w = MATRIX_WIDTH
            h = count // w if count % w == 0 else MATRIX_HEIGHT
            self.targets = [PanelTarget(-1, targets, w, h)]
        else:
            self.targets = list(targets)
        self.port = port
        self.panel_id = panel_id
        self.brightness = brightness
        self._has_local = any(t.local for t in self.targets)
        self._bundle: PanelMatrixBundle | None = None
        if self._has_local:
            self._bundle = PanelMatrixBundle(self.targets, brightness)
        self._sock = (
            socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            if any(not t.local for t in self.targets)
            else None
        )

    @classmethod
    def from_args(cls, args: argparse.Namespace) -> PanelClient:
        client = cls(
            panel_targets_from_args(args),
            args.port,
            pixel_count=panel_count_from_args(args),
            panel_id=args.panel_id,
            brightness=args.brightness,
        )
        client.targets = apply_pin_arg(args, client._sock, client.targets)
        return client

    @property
    def pixel_count(self) -> int:
        if len(self.targets) != 1:
            raise ValueError("pixel_count requires a single panel target")
        return self.targets[0].pixels

    @property
    def hosts(self) -> list[str]:
        return [t.host for t in self.targets]

    def close(self) -> None:
        if self._sock is not None:
            self._sock.close()
        if self._bundle is not None:
            self._bundle.close()
            self._bundle = None

    def __enter__(self) -> PanelClient:
        return self

    def __exit__(self, *_exc) -> None:
        self.clear()
        self.close()

    def send_bytes(self, rgb: bytes) -> None:
        if len(self.targets) != 1:
            raise ValueError("send_bytes requires a single panel target")
        target = self.targets[0]
        if target.local:
            if self._bundle is None:
                raise RuntimeError("local panel bundle not initialized")
            _, matrix = self._bundle.matrices[0]
            if len(rgb) != target.pixels * 3:
                raise ValueError(f"expected {target.pixels * 3} bytes, got {len(rgb)}")
            for i in range(target.pixels):
                base = i * 3
                matrix.set_pixel(
                    i % target.width,
                    i // target.width,
                    (rgb[base], rgb[base + 1], rgb[base + 2]),
                )
            matrix.show()
            return
        if self._sock is None:
            raise RuntimeError("UDP socket not initialized")
        if len(rgb) != target.pixels * 3:
            raise ValueError(f"expected {target.pixels * 3} bytes, got {len(rgb)}")
        send_frame(
            self._sock,
            target.host,
            self.port,
            rgb,
            self.panel_id,
            strip=target.strip,
        )

    def fill(self, color: RGB = (0, 0, 0)) -> None:
        if self._bundle is not None:
            for _, matrix in self._bundle.matrices:
                matrix.fill(color)
            push_panel_frames(self._sock, self._bundle.matrices, self.port, self.panel_id)
            return
        if self._sock is None:
            raise RuntimeError("UDP socket not initialized")
        for target in self.targets:
            send_frame(
                self._sock,
                target.host,
                self.port,
                solid_frame_bytes(color, target.pixels, self.brightness),
                self.panel_id,
                strip=target.strip,
            )

    def send_pixels(self, pixels: list[RGB]) -> None:
        if len(self.targets) != 1:
            raise ValueError("send_pixels requires a single panel target")
        if len(pixels) != self.pixel_count:
            raise ValueError(f"expected {self.pixel_count} pixels, got {len(pixels)}")
        self.send_bytes(pixels_to_frame_bytes(pixels, self.brightness))

    def set_indices(self, indices: Iterable[int], color: RGB) -> None:
        pixels = [(0, 0, 0)] * self.pixel_count
        for i in indices:
            if 0 <= i < self.pixel_count:
                pixels[i] = color
        self.send_pixels(pixels)

    def clear(self) -> None:
        self.fill((0, 0, 0))


def _panel_matrices(
    targets: Sequence[PanelTarget],
    brightness: float,
) -> PanelMatrixBundle:
    return PanelMatrixBundle(targets, brightness)


async def run_panel_playlist_async(
    targets: Sequence[PanelTarget],
    port: int,
    playlist: list[str],
    *,
    duration: float = 12.0,
    fps_override: float | None = None,
    brightness: float = 0.25,
    panel_id: int | None = None,
    loop: bool = True,
    sync_send: bool = False,
    pin: int | None = None,
    pins: Sequence[tuple[int, int | None]] | None = None,
) -> None:
    bundle = _panel_matrices(targets, brightness)
    sock = (
        socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        if any(not t.local for t in targets)
        else None
    )
    try:
        setup_panel_pins(sock, targets, port, pin=pin, pins=pins)
        while True:
            for name in playlist:
                draw = ANIMATIONS[name]
                fps = min(fps_override or DEFAULT_FPS.get(name, 30), PANEL_MAX_FPS)
                delay = 1.0 / fps
                frames = max(int(duration * fps), 1)
                print(f"  {name} ({fps:.0f} fps, {frames} frames)")
                for frame in range(frames):
                    t0 = time.monotonic()
                    for _, matrix in bundle.matrices:
                        draw(matrix, frame)
                    await send_frames_async(
                        sock, bundle.matrices, port, panel_id, sync_send=sync_send
                    )
                    elapsed = time.monotonic() - t0
                    if elapsed < delay:
                        await asyncio.sleep(delay - elapsed)
            if not loop:
                break
            print("  --- loop ---")
        for _, matrix in bundle.matrices:
            matrix.clear()
        await send_frames_async(sock, bundle.matrices, port, panel_id, sync_send=sync_send)
    finally:
        if sock is not None:
            sock.close()
        bundle.close()


def run_panel_playlist(
    targets: Sequence[PanelTarget],
    port: int,
    playlist: list[str],
    *,
    duration: float = 12.0,
    fps_override: float | None = None,
    brightness: float = 0.25,
    panel_id: int | None = None,
    loop: bool = True,
    sync_send: bool = False,
    pin: int | None = None,
    pins: Sequence[tuple[int, int | None]] | None = None,
) -> None:
    asyncio.run(
        run_panel_playlist_async(
            targets,
            port,
            playlist,
            duration=duration,
            fps_override=fps_override,
            brightness=brightness,
            panel_id=panel_id,
            loop=loop,
            sync_send=sync_send,
            pin=pin,
            pins=pins,
        )
    )


async def run_panel_animation_loop_async(
    targets: Sequence[PanelTarget],
    port: int,
    animation: str,
    *,
    fps: float | None = None,
    brightness: float = 0.25,
    panel_id: int | None = None,
    sync_send: bool = False,
    pin: int | None = None,
    pins: Sequence[tuple[int, int | None]] | None = None,
) -> None:
    draw = ANIMATIONS[animation]
    rate = min(fps or DEFAULT_FPS.get(animation, 30), PANEL_MAX_FPS)
    interval = 1.0 / rate
    bundle = _panel_matrices(targets, brightness)
    sock = (
        socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        if any(not t.local for t in targets)
        else None
    )
    frame = 0
    try:
        setup_panel_pins(
            sock, targets, port, pin=pin, pins=pins
        )
        while True:
            t0 = time.monotonic()
            for _, matrix in bundle.matrices:
                draw(matrix, frame)
            await send_frames_async(sock, bundle.matrices, port, panel_id, sync_send=sync_send)
            frame += 1
            elapsed = time.monotonic() - t0
            if elapsed < interval:
                await asyncio.sleep(interval - elapsed)
    finally:
        for _, matrix in bundle.matrices:
            matrix.clear()
        await send_frames_async(sock, bundle.matrices, port, panel_id, sync_send=sync_send)
        if sock is not None:
            sock.close()
        bundle.close()


def run_panel_animation_loop(
    targets: Sequence[PanelTarget],
    port: int,
    animation: str,
    *,
    fps: float | None = None,
    brightness: float = 0.25,
    panel_id: int | None = None,
    sync_send: bool = False,
    pin: int | None = None,
    pins: Sequence[tuple[int, int | None]] | None = None,
) -> None:
    asyncio.run(
        run_panel_animation_loop_async(
            targets,
            port,
            animation,
            fps=fps,
            brightness=brightness,
            panel_id=panel_id,
            sync_send=sync_send,
            pin=pin,
            pins=pins,
        )
    )


def animation_playlist(animation: str) -> list[str]:
    names = list(ANIMATIONS.keys())
    if animation == "all":
        # Skip solid — 1 fps looks frozen on UDP panels; start with motion.
        return [n for n in names if n != "solid"]
    return [animation]