Animation API Reference¶

@classmethod
async def for_matrix(
    cls,
    device: MatrixLight,
    duration_ms: int = 0,
) -> Animator:
    """Create an Animator configured for a MatrixLight device.

    Queries the device for tile information, then returns an animator
    that sends frames via direct UDP (no device connection needed
    after creation).

    Args:
        device: MatrixLight device (must be connected)
        duration_ms: Transition duration in milliseconds (default 0 for instant).
                    When non-zero, device smoothly interpolates between frames.

    Returns:
        Configured Animator instance

    Example:
        ```python
        async with await Device.connect("192.168.1.100") as device:
            assert isinstance(device, MatrixLight)
            animator = await Animator.for_matrix(device)

        # Device connection closed, animator still works via UDP
        while running:
            stats = animator.send_frame(frame)
            await asyncio.sleep(1 / 30)  # 30 FPS
        ```
    """
    # Get device info
    ip = device.ip
    serial = Serial.from_string(device.serial)

    # Ensure we have tile chain
    if device.device_chain is None:
        await device.get_device_chain()

    tiles = device.device_chain
    if not tiles:
        raise ValueError("Device has no tiles")

    # Create framebuffer with orientation correction
    framebuffer = await FrameBuffer.for_matrix(device)

    # Create packet generator
    packet_generator = MatrixPacketGenerator(
        tile_count=len(tiles),
        tile_width=tiles[0].width,
        tile_height=tiles[0].height,
        duration_ms=duration_ms,
    )

    return cls(ip, serial, framebuffer, packet_generator, port=device.port)

@classmethod
async def for_multizone(
    cls,
    device: MultiZoneLight,
    duration_ms: int = 0,
) -> Animator:
    """Create an Animator configured for a MultiZoneLight device.

    Only devices with extended multizone capability are supported.
    Queries the device for zone count, then returns an animator
    that sends frames via direct UDP.

    Args:
        device: MultiZoneLight device (must be connected and support
               extended multizone protocol)
        duration_ms: Transition duration in milliseconds (default 0 for instant).
                    When non-zero, device smoothly interpolates between frames.

    Returns:
        Configured Animator instance

    Raises:
        ValueError: If device doesn't support extended multizone

    Example:
        ```python
        async with await Device.connect("192.168.1.100") as device:
            assert isinstance(device, MultiZoneLight)
            animator = await Animator.for_multizone(device)

        # Device connection closed, animator still works via UDP
        while running:
            stats = animator.send_frame(frame)
            await asyncio.sleep(1 / 30)  # 30 FPS
        ```
    """
    # Ensure capabilities are loaded
    if device.capabilities is None:
        await device.ensure_capabilities()

    # Check extended multizone capability
    has_extended = bool(
        device.capabilities and device.capabilities.has_extended_multizone
    )
    if not has_extended:
        raise ValueError(
            "Device does not support extended multizone protocol. "
            "Only extended multizone devices are supported for animation."
        )

    # Get device info
    ip = device.ip
    serial = Serial.from_string(device.serial)

    # Create framebuffer (no orientation for multizone)
    framebuffer = await FrameBuffer.for_multizone(device)

    # Get zone count
    zone_count = await device.get_zone_count()

    # Create packet generator
    packet_generator = MultiZonePacketGenerator(
        zone_count=zone_count, duration_ms=duration_ms
    )

    return cls(ip, serial, framebuffer, packet_generator, port=device.port)

@classmethod
def for_light(
    cls,
    device: Light,
    duration_ms: int = 0,
) -> Animator:
    """Create an Animator configured for a single Light device.

    Unlike the matrix/multizone factories, this does not need to be async
    because single lights don't require any device queries for configuration.

    Args:
        device: Light device (must have ip and serial set)
        duration_ms: Transition duration in milliseconds (default 0 for instant).
                    When non-zero, device smoothly interpolates between frames.

    Returns:
        Configured Animator instance

    Example:
        ```python
        async with await Device.connect("192.168.1.100") as device:
            animator = Animator.for_light(device)

        # Device connection closed, animator still works via UDP
        while running:
            stats = animator.send_frame([(65535, 65535, 65535, 3500)])
            await asyncio.sleep(1 / 30)  # 30 FPS
        ```
    """
    ip = device.ip
    serial = Serial.from_string(device.serial)
    framebuffer = FrameBuffer.for_light(device)
    packet_generator = LightPacketGenerator(duration_ms=duration_ms)

    return cls(ip, serial, framebuffer, packet_generator, port=device.port)

def send_frame(
    self,
    hsbk: list[tuple[int, int, int, int]],
) -> AnimatorStats:
    """Send a frame to the device via direct UDP.

    Applies orientation mapping (for matrix devices), updates colors
    in prebaked packets, and sends them directly via UDP -- unless
    ack-gated flow control drops the frame first. Each call first
    sweeps the animator's own socket for arrived acks; if the device
    is still behind on acknowledgements after the sweep, the frame is
    dropped entirely (no framebuffer work, no packets sent, no
    sequence numbers consumed) and the caller should simply send its
    next frame rather than retry or queue this one (latest-frame-wins).

    This is a synchronous method for minimum overhead. UDP sendto()
    is non-blocking for datagrams, and the ack sweep is a non-blocking
    `recvfrom_into` loop on the same socket -- no event loop is
    required.

    Args:
        hsbk: Protocol-ready HSBK data for all pixels.
              Each tuple is (hue, sat, brightness, kelvin) where
              H/S/B are 0-65535 and K is 1500-9000.

    Returns:
        AnimatorStats with operation statistics. `gated=True` means the
        frame was dropped by flow control.

    Raises:
        ValueError: If hsbk length doesn't match pixel_count. This
            validation always runs, even when the gate is saturated --
            a full gate must never suppress input validation.
    """
    start_time = time.perf_counter()

    if len(hsbk) != self._framebuffer.canvas_size:
        raise ValueError(
            f"HSBK length ({len(hsbk)}) must match "
            f"pixel_count ({self._framebuffer.canvas_size})"
        )

    # Ensure socket exists
    if self._socket is None:
        self._socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self._socket.setblocking(False)

    now = time.monotonic()
    self._ack_gate.sweep(self._socket, self._source, now)
    if self._ack_gate.gated:
        return AnimatorStats(
            packets_sent=0,
            total_time_ms=(time.perf_counter() - start_time) * 1000,
            gated=True,
            acks_outstanding=self._ack_gate.outstanding_count,
        )

    # Apply orientation mapping
    device_data = self._framebuffer.apply(hsbk)

    # Update colors in prebaked templates
    self._packet_generator.update_colors(self._templates, device_data)

    # Send each packet, updating sequence number
    for i, tmpl in enumerate(self._templates):
        tmpl.data[SEQUENCE_OFFSET] = self._sequence
        if i == self._probe_index:
            self._ack_gate.track(self._sequence, now)
        self._sequence = (self._sequence + 1) % 256
        self._socket.sendto(tmpl.data, self._addr)

    end_time = time.perf_counter()

    return AnimatorStats(
        packets_sent=len(self._templates),
        total_time_ms=(end_time - start_time) * 1000,
        acks_outstanding=self._ack_gate.outstanding_count,
    )

def close(self) -> None:
    """Close the UDP socket.

    Call this when done with the animator to free resources. Also
    resets the ack gate, so a fresh animator session (new socket, new
    gate) starts ungated.
    """
    if self._socket is not None:
        self._socket.close()
        self._socket = None
    self._ack_gate.reset()

@classmethod
async def for_matrix(
    cls,
    device: MatrixLight,
) -> FrameBuffer:
    """Create a FrameBuffer configured for a MatrixLight device.

    Automatically determines pixel count from device chain and creates
    appropriate mapping for tile orientations and positions.

    For multi-tile devices (has_chain capability), creates a unified canvas
    based on tile positions (user_x, user_y). Each tile's position determines
    which region of the canvas it displays, allowing animations to span
    across all tiles.

    Args:
        device: MatrixLight device (must be connected)

    Returns:
        Configured FrameBuffer instance

    Example:
        ```python
        async with await Device.connect("192.168.1.100") as matrix:
            assert isinstance(matrix, MatrixLight)
            fb = await FrameBuffer.for_matrix(matrix)
            print(f"Canvas: {fb.canvas_width}x{fb.canvas_height}")
        ```
    """
    # Ensure device chain is loaded
    if device.device_chain is None:
        await device.get_device_chain()

    tiles = device.device_chain
    if not tiles:
        raise ValueError("Device has no tiles")

    # Calculate total device pixels
    pixel_count = sum(t.width * t.height for t in tiles)

    # Ensure capabilities are loaded
    if device.capabilities is None:
        await device.ensure_capabilities()

    # Only build canvas mapping for devices with chain capability.
    # The original LIFX Tile is the only matrix device with accelerometer-based
    # orientation detection and multi-tile positioning. Other matrix devices
    # (Ceiling, Luna, Candle, Path, etc.) have fixed positions.
    if device.capabilities and device.capabilities.has_chain:
        return cls._for_multi_tile(tiles, pixel_count)
    else:
        # Single tile device - simple passthrough
        first_tile = tiles[0]
        return cls(
            pixel_count=pixel_count,
            canvas_width=first_tile.width,
            canvas_height=first_tile.height,
        )

@classmethod
async def for_multizone(
    cls,
    device: MultiZoneLight,
) -> FrameBuffer:
    """Create a FrameBuffer configured for a MultiZoneLight device.

    Automatically determines pixel count from zone count.
    Multizone devices don't need permutation (zones are linear).

    Args:
        device: MultiZoneLight device (must be connected)

    Returns:
        Configured FrameBuffer instance

    Example:
        ```python
        async with await Device.connect("192.168.1.100") as strip:
            assert isinstance(strip, MultiZoneLight)
            fb = await FrameBuffer.for_multizone(strip)
        ```
    """
    # Get zone count (fetches from device if not cached)
    zone_count = await device.get_zone_count()

    return cls(pixel_count=zone_count)

@classmethod
def for_light(cls, _device: Light) -> FrameBuffer:
    """Create a FrameBuffer configured for a single Light device.

    Single lights have exactly 1 pixel, so this is a trivial passthrough.

    Args:
        _device: Light device (unused — single lights always have 1 pixel)

    Returns:
        Configured FrameBuffer instance

    Example:
        ```python
        fb = FrameBuffer.for_light(light)
        assert fb.pixel_count == 1
        ```
    """
    return cls(pixel_count=1, canvas_width=1, canvas_height=1)

def apply(
    self, hsbk: list[tuple[int, int, int, int]]
) -> list[tuple[int, int, int, int]]:
    """Apply orientation mapping to frame data.

    For multi-tile devices, the input is interpreted as a row-major 2D
    canvas of size (canvas_width x canvas_height). Each tile extracts
    its region from the canvas based on its position.

    For single-tile or multizone devices, this is a passthrough.

    Args:
        hsbk: List of protocol-ready HSBK tuples.
              - For multi-tile: length must match canvas_size
              - For single-tile/multizone: length must match pixel_count
              Each tuple is (hue, sat, brightness, kelvin) where
              H/S/B are 0-65535 and K is 1500-9000.

    Returns:
        Remapped HSBK data in device order

    Raises:
        ValueError: If hsbk length doesn't match expected size
    """
    # Multi-tile canvas mode — use pre-computed LUT
    if self._lut is not None:
        expected_size = self._canvas_width * self._canvas_height
        if len(hsbk) != expected_size:
            raise ValueError(
                f"HSBK length ({len(hsbk)}) must match "
                f"canvas_size ({expected_size})"
            )
        return [hsbk[i] for i in self._lut]

    # Single-tile or multizone mode (passthrough)
    if len(hsbk) != self._pixel_count:
        raise ValueError(
            f"HSBK length ({len(hsbk)}) must match "
            f"pixel_count ({self._pixel_count})"
        )

    return list(hsbk)

@abstractmethod
def create_templates(self, source: int, target: bytes) -> list[PacketTemplate]:
    """Create prebaked packet templates.

    Args:
        source: Client source ID for header
        target: 6-byte device serial for header

    Returns:
        List of PacketTemplate with prebaked packets
    """

@abstractmethod
def update_colors(
    self, templates: list[PacketTemplate], hsbk: list[tuple[int, int, int, int]]
) -> None:
    """Update color data in prebaked templates.

    Args:
        templates: Prebaked packet templates
        hsbk: Protocol-ready HSBK data for all pixels
    """

@abstractmethod
def pixel_count(self) -> int:
    """Get the total pixel count this generator expects."""

def pixel_count(self) -> int:
    """Get total pixel count."""
    return self._total_pixels

def create_templates(self, source: int, target: bytes) -> list[PacketTemplate]:
    """Create prebaked packet templates for all tiles.

    Args:
        source: Client source ID
        target: 6-byte device serial

    Returns:
        List of PacketTemplate with complete prebaked packets
    """
    if self._is_large_tile:
        return self._create_large_tile_templates(source, target)
    else:
        return self._create_standard_templates(source, target)

def update_colors(
    self, templates: list[PacketTemplate], hsbk: list[tuple[int, int, int, int]]
) -> None:
    """Update color data in prebaked templates.

    Args:
        templates: Prebaked packet templates
        hsbk: Protocol-ready HSBK data for all pixels

    Raises:
        ValueError: If hsbk has fewer values than the total pixel count
    """
    if len(hsbk) < self._total_pixels:
        raise ValueError(
            f"Expected {self._total_pixels} HSBK values, got {len(hsbk)}"
        )

    for tmpl in templates:
        if tmpl.color_count == 0:
            continue  # Skip CopyFrameBuffer packets

        # Write all HSBK tuples into the packet buffer with a single
        # C-level pack using the precomputed bulk format string
        struct.pack_into(
            tmpl.fmt,
            tmpl.data,
            tmpl.color_offset,
            *chain.from_iterable(
                hsbk[tmpl.hsbk_start : tmpl.hsbk_start + tmpl.color_count]
            ),
        )

def pixel_count(self) -> int:
    """Get total zone count."""
    return self._zone_count

def create_templates(self, source: int, target: bytes) -> list[PacketTemplate]:
    """Create prebaked packet templates for all zones.

    Args:
        source: Client source ID
        target: 6-byte device serial

    Returns:
        List of PacketTemplate with complete prebaked packets
    """
    templates: list[PacketTemplate] = []

    for pkt_idx in range(self._packets_needed):
        zone_start = pkt_idx * self._MAX_ZONES_PER_PACKET
        zone_end = min(zone_start + self._MAX_ZONES_PER_PACKET, self._zone_count)
        zone_count = zone_end - zone_start

        # Build header
        header = _build_header(
            self.SET_EXTENDED_COLOR_ZONES_PKT_TYPE,
            source,
            target,
            self._PAYLOAD_SIZE,
        )

        # Build payload
        payload = bytearray(self._PAYLOAD_SIZE)
        # duration
        struct.pack_into("<I", payload, 0, self._duration_ms)
        # apply = 1 (APPLY)
        payload[4] = 1
        # zone_index
        struct.pack_into("<H", payload, 5, zone_start)
        # colors_count
        payload[7] = zone_count
        # colors filled with black as default
        for i in range(82):
            offset = self._COLORS_OFFSET_IN_PAYLOAD + i * 8
            struct.pack_into("<HHHH", payload, offset, 0, 0, 0, 3500)

        packet = header + payload

        templates.append(
            PacketTemplate(
                data=packet,
                color_offset=HEADER_SIZE + self._COLORS_OFFSET_IN_PAYLOAD,
                color_count=zone_count,
                hsbk_start=zone_start,
                fmt=f"<{'HHHH' * zone_count}",
            )
        )

    return templates