Animation API Reference¶

async def prepare(self, *, enable_thread: bool = False) -> Animator:
    """Query the device for its geometry, once, and get ready to draw.

    A light's Animator learns its tile layout or zone count from the
    device the first time it is prepared; later calls make no query.
    Every call also tells the device that frames are about to bypass its
    own colour methods, so a Ceiling or Mirror component call reads the
    device instead of undoing the animation. A single light's Animator is
    ready without this, but preparing it is harmless.

    A light evidenced as Thread, by its own replies or an mDNS record, is
    refused unless the caller passes ``enable_thread=True``: a Thread mesh
    is not built for a steady stream of frames. A light not yet heard
    from is not refused.

    Args:
        enable_thread: Stream to a light evidenced as Thread anyway.
            Off by default.

    Returns:
        This Animator, so `animator = await device.animator.prepare()`
        reads naturally.

    Raises:
        ValueError: If a matrix light reports no tiles, or a multizone
            light does not support the extended multizone protocol.
        LifxUnsupportedCommandError: If the light is evidenced as Thread
            and ``enable_thread`` is False

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

        while running:
            animator.send_frame(frame)
            await asyncio.sleep(1 / 30)
        ```
    """
    device = self._device
    if device is None:
        return self
    device._refuse_thread_frames("Animator.prepare()", enable_thread=enable_thread)
    if self._geometry is None:
        framebuffer, packet_generator = await device._query_animation_geometry()
        # A concurrent prepare() may have finished first; keep its templates.
        if self._geometry is None:
            self._install(framebuffer, packet_generator)

    # Frames go straight out over UDP rather than through set64(), so the
    # device never hears about them. Tell it now: a component light
    # (Ceiling, Mirror) forgets the tile it last wrote, so its next
    # component call reads the device instead of undoing the animation.
    device._zones_changed()
    return self

@classmethod
async def for_matrix(
    cls,
    device: MatrixLight,
    duration_ms: int = 0,
    *,
    enable_thread: bool = False,
) -> Animator:
    """Return the device's Animator, prepared for a MatrixLight.

    Deprecated: use `device.animator` and `await device.animator.prepare()`.
    Every call returns the same Animator, the one the device owns, so a
    library effect and this caller share one writer and one ack gate.

    Args:
        device: MatrixLight device (must be connected)
        duration_ms: Transition duration in milliseconds (default 0 for
            instant), applied to frames sent through `send_frame()`.
        enable_thread: Stream to a light evidenced as Thread anyway;
            see `prepare()`. Off by default.

    Returns:
        The device's Animator

    Raises:
        LifxUnsupportedCommandError: If the device is evidenced as Thread
            and ``enable_thread`` is False
        ValueError: If the device reports no tiles
    """
    _warn_deprecated("for_matrix")
    return await cls._borrow(device, duration_ms, enable_thread)

@classmethod
async def for_multizone(
    cls,
    device: MultiZoneLight,
    duration_ms: int = 0,
    *,
    enable_thread: bool = False,
) -> Animator:
    """Return the device's Animator, prepared for a MultiZoneLight.

    Deprecated: use `device.animator` and `await device.animator.prepare()`.
    Every call returns the same Animator, the one the device owns. Only
    devices with extended multizone capability are supported.

    Args:
        device: MultiZoneLight device (must be connected and support
               extended multizone protocol)
        duration_ms: Transition duration in milliseconds (default 0 for
            instant), applied to frames sent through `send_frame()`.
        enable_thread: Stream to a light evidenced as Thread anyway;
            see `prepare()`. Off by default.

    Returns:
        The device's Animator

    Raises:
        LifxUnsupportedCommandError: If the device is evidenced as Thread
            and ``enable_thread`` is False
        ValueError: If device doesn't support extended multizone
    """
    _warn_deprecated("for_multizone")
    return await cls._borrow(device, duration_ms, enable_thread)

@classmethod
def for_light(
    cls,
    device: Light,
    duration_ms: int = 0,
    *,
    enable_thread: bool = False,
) -> Animator:
    """Return the device's Animator for a single Light device.

    Deprecated: use `device.animator`. Every call returns the same
    Animator, the one the device owns. A single light's Animator needs no
    device query, so this stays synchronous.

    Args:
        device: Light device (must have ip and serial set)
        duration_ms: Transition duration in milliseconds (default 0 for
            instant), applied to frames sent through `send_frame()`.
        enable_thread: Stream to a light evidenced as Thread anyway;
            see `prepare()`. Off by default.

    Returns:
        The device's Animator

    Raises:
        LifxUnsupportedCommandError: If the device is evidenced as Thread
            and ``enable_thread`` is False
        RuntimeError: If the device is a matrix or multizone light whose
            Animator has not been prepared yet; await
            `device.animator.prepare()` instead.
    """
    _warn_deprecated("for_light")
    device._refuse_thread_frames(
        "Animator.for_light()", enable_thread=enable_thread
    )
    animator = device.animator
    animator._require_geometry()
    animator.duration_ms = duration_ms
    return animator

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.
        LifxNetworkError: If the destination is invalid or the UDP socket
            cannot be created, or a frame datagram cannot be sent.
        RuntimeError: If the Animator has not been prepared yet.
    """
    geometry = self._require_geometry()
    return self._send(geometry.framebuffer, hsbk, self._duration_ms)

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._addr = 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 whose reported
    # orientation and tile positions are applied. Other matrix devices
    # (Ceiling, Luna, Candle, Path, etc.) are fixed panels whose
    # accelerometer readings are not used (a Luna reports RotatedLeft when
    # standing upright).
    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

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 zones

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

    for tmpl in templates:
        # 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]
            ),
        )