Device Classes¶

def is_fresh(self, max_age: float = 5.0) -> bool:
    """Check if state is fresh (recently updated).

    Args:
        max_age: Maximum age in seconds (default: 5.0)

    Returns:
        True if state age is less than max_age
    """
    return self.age < max_age

def __post_init__(self, host_firmware: FirmwareInfo | None) -> None:
    """Calculate RSSI from signal and classify the RSSI unit."""
    if self.signal is None:
        # Signal was never fetched, so there is no RSSI to derive.
        self.rssi = None
    elif self.signal > 0:
        self.rssi = int(floor(10 * log10(self.signal) + 0.5))
    else:
        # Minimum RSSI value if signal is ever less than or equal to zero
        self.rssi = -100

    if host_firmware is None:
        self.rssi_unit = None
    elif (
        host_firmware.version_major,
        host_firmware.version_minor,
    ) <= self._RSSI_DB_FW_BOUNDARY:
        self.rssi_unit = "dB"
    else:
        self.rssi_unit = "dBm"

def adopt_cached_metadata(self, source: Device) -> None:
    """Adopt metadata already fetched by a temporary device instance.

    Device factories use a base ``Device`` to identify the correct concrete
    class. This transfers only registry-derived or firmware-keyed metadata,
    which the library treats as read-only; live state and connection
    lifecycle remain owned by the new instance.
    """
    self._version = source._version
    self._host_firmware = source._host_firmware
    self._capabilities = source._capabilities
    self._mac_address = source._mac_address
    self._mac_address_firmware = source._mac_address_firmware

@classmethod
async def from_ip(
    cls,
    ip: str,
    port: int = LIFX_UDP_PORT,
    serial: str | None = None,
    timeout: float = DEFAULT_REQUEST_TIMEOUT,
    max_retries: int = DEFAULT_MAX_RETRIES,
    *,
    fetch_wifi_info: bool = False,
    fetch_ambient_light: bool = False,
) -> Self:
    """Create and return an instance for the given IP address.

    This is a convenience class method for connecting to a known device
    by IP address. The returned instance can be used as a context manager.

    Args:
        ip: IP address of the device
        port: Port number (default LIFX_UDP_PORT)
        serial: Serial number as 12-digit hex string
        timeout: Request timeout for this device instance
        max_retries: Maximum number of retry attempts
        fetch_wifi_info: Query WiFi signal strength during state initialization
        fetch_ambient_light: Query the ambient light sensor during state
            initialization (lights only)

    Returns:
        Device instance ready to use with async context manager

    Example:
        ```python
        async with await Device.from_ip(ip="192.168.1.100") as device:
            label = await device.get_label()
        ```
    """
    if serial is None:
        temp_conn = DeviceConnection(
            serial="000000000000",
            ip=ip,
            port=port,
            timeout=timeout,
            max_retries=max_retries,
        )
        try:
            response = await temp_conn.request(
                packets.Device.GetService(), timeout=timeout
            )
            if response and isinstance(response, packets.Device.StateService):
                if temp_conn.serial and temp_conn.serial != "000000000000":
                    return cls(
                        serial=temp_conn.serial,
                        ip=ip,
                        port=port,
                        timeout=timeout,
                        max_retries=max_retries,
                        fetch_wifi_info=fetch_wifi_info,
                        fetch_ambient_light=fetch_ambient_light,
                    )
        finally:
            # Always close the temporary connection to prevent resource leaks
            await temp_conn.close()
    else:
        return cls(
            serial=serial,
            ip=ip,
            port=port,
            timeout=timeout,
            max_retries=max_retries,
            fetch_wifi_info=fetch_wifi_info,
            fetch_ambient_light=fetch_ambient_light,
        )

    raise LifxDeviceNotFoundError()

@classmethod
async def connect(
    cls,
    ip: str,
    serial: str | None = None,
    port: int = LIFX_UDP_PORT,
    timeout: float = DEFAULT_REQUEST_TIMEOUT,
    max_retries: int = DEFAULT_MAX_RETRIES,
    *,
    fetch_wifi_info: bool = False,
    fetch_ambient_light: bool = False,
) -> Light | HevLight | InfraredLight | MultiZoneLight | MatrixLight | CeilingLight:
    """Create a device instance with the correct type for the given IP.

    This factory method queries the device to determine its product type
    and returns the appropriate subclass (Light, MatrixLight, etc.).

    State is NOT initialized on return. Use ``async with`` to enter the
    context manager, which calls ``__aenter__()`` → ``_initialize_state()``
    and guarantees that ``device.state`` is non-None.

    Args:
        ip: IP address of the device
        serial: Optional serial number (12-digit hex, with or without colons).
                If None, queries device to get serial.
        port: Port number (default LIFX_UDP_PORT)
        timeout: Request timeout for this device instance
        max_retries: Maximum number of retry attempts
        fetch_wifi_info: Query WiFi signal strength during state initialization
        fetch_ambient_light: Query the ambient light sensor during state
            initialization (lights only)

    Returns:
        Device instance of the correct subclass, ready to use with
        ``async with`` for state initialization and automatic cleanup.

    Raises:
        LifxDeviceNotFoundError: If device cannot be found or contacted
        LifxTimeoutError: If device does not respond
        ValueError: If serial format is invalid

    Example:
        ```python
        # Connect by IP (serial auto-detected)
        device = await Device.connect(ip="192.168.1.100")
        async with device:
            # State is initialized here by __aenter__()
            print(f"{device.state.model}: {device.state.label}")
            if device.state.is_on:
                print("Device is on")

        # Connect with known serial
        device = await Device.connect(ip="192.168.1.100", serial="d073d5123456")
        async with device:
            await device.set_power(True)
        ```
    """
    # Step 1: Get serial if not provided
    if serial is None:
        temp_conn = DeviceConnection(
            serial="000000000000",
            ip=ip,
            port=port,
            timeout=timeout,
            max_retries=max_retries,
        )
        try:
            response = await temp_conn.request(
                packets.Device.GetService(), timeout=timeout
            )
            if response and isinstance(response, packets.Device.StateService):
                if temp_conn.serial and temp_conn.serial != "000000000000":
                    serial = temp_conn.serial
                else:
                    raise LifxDeviceNotFoundError(
                        "Could not determine device serial"
                    )
            else:
                raise LifxDeviceNotFoundError("No response from device")
        finally:
            await temp_conn.close()

    # Step 2: Normalize serial (accept with or without colons)
    serial = serial.replace(":", "")

    # Step 3: Create temporary device to get product info
    temp_device = cls(
        serial=serial,
        ip=ip,
        port=port,
        timeout=timeout,
        max_retries=max_retries,
    )

    try:
        # Get version to determine product
        version = await temp_device.get_version()
        product_info = get_product(version.product)

        if product_info is None:
            raise LifxDeviceNotFoundError(f"Unknown product ID: {version.product}")

        # Step 4: Determine correct device class based on capabilities
        from lifx.devices.detection import (
            get_device_class_for_product,
        )

        device_class = get_device_class_for_product(version.product, product_info)

        # Step 5: Create instance of correct device class
        device = device_class(
            serial=serial,
            ip=ip,
            port=port,
            timeout=timeout,
            max_retries=max_retries,
            fetch_wifi_info=fetch_wifi_info,
            fetch_ambient_light=fetch_ambient_light,
        )

        # Type system note: device._state is guaranteed non-None after
        # _initialize_state().
        # Each subclass overrides _state to be non-optional
        return device

    finally:
        # Clean up temporary device
        await temp_device.connection.close()

async def get_mac_address(self) -> str:
    """Calculate and return the MAC address for this device.

    The MAC usually equals the serial. Firmware with
    ``version_major == 3 and version_minor >= 70`` is the exception: its MAC
    is the serial with the final octet incremented by one (wrapping at 256).

    The bound is deliberately narrower than "major version 3" and comes
    from LIFX. Devices on earlier 3.x firmware report a MAC identical to
    their serial -- consistent with LIFX Tiles on 3.50, whose real ARP MAC
    matched the serial while a major-only rule predicted serial + 1.

    The result is memoised against the firmware it was derived from. Because
    the rule now turns on the minor version, an ordinary in-place update
    across the 3.70 boundary changes the correct answer -- so a cache keyed
    on nothing but "already computed" would hand a long-lived caller a MAC
    the device no longer has.
    """
    firmware = (
        self._host_firmware
        if self._host_firmware is not None
        else await self.get_host_firmware()
    )
    firmware_key = (firmware.version_major, firmware.version_minor)

    if self._mac_address is None or self._mac_address_firmware != firmware_key:
        direct_mac, offset_mac = mac_candidates_for_serial(self.serial)
        if (
            firmware.version_major == MAC_OFFSET_FIRMWARE_MAJOR
            and firmware.version_minor >= MAC_OFFSET_FIRMWARE_MIN_MINOR
        ):
            self._mac_address = offset_mac
        else:
            self._mac_address = direct_mac
        self._mac_address_firmware = firmware_key

    return self._mac_address

async def ensure_capabilities(self) -> None:
    """Ensure device capabilities are populated.

    Fetches device version and firmware to determine product capabilities.
    This is a no-op if capabilities have already been loaded (e.g. via
    the async context manager).
    """
    await self._ensure_capabilities()

async def get_label(self) -> str:
    """Get device label/name.

    Always fetches from device. Use the `label` property to access stored value.

    Returns:
        Device label as string (max 32 bytes UTF-8)

    Raises:
        LifxDeviceNotFoundError: If device is not connected
        LifxTimeoutError: If device does not respond
        LifxProtocolError: If response is invalid
        LifxUnsupportedCommandError: If device doesn't support this command

    Example:
        ```python
        label = await device.get_label()
        print(f"Device name: {label}")

        # Or use cached value
        if device.label:
            print(f"Cached label: {device.label}")
        ```
    """
    # Request automatically unpacks and decodes label
    state = await self.connection.request(packets.Device.GetLabel())
    self._raise_if_unhandled(state)

    # Store label
    label_value = state.label
    self._label = label_value
    # Update state if it exists
    if self._state is not None:
        self._state.label = label_value
        self._state.last_updated = time.time()

    _LOGGER.debug(
        {
            "class": "Device",
            "method": "get_label",
            "action": "query",
            "reply": {"label": label_value},
        }
    )
    return label_value

async def set_label(self, label: str) -> None:
    """Set device label/name.

    Args:
        label: New device label (max 32 bytes UTF-8)

    Raises:
        ValueError: If label is too long
        LifxDeviceNotFoundError: If device is not connected
        LifxTimeoutError: If device does not respond
        LifxUnsupportedCommandError: If device doesn't support this command

    Example:
        ```python
        # Set label
        await device.set_label("Living Room Light")
        ```
    """
    # Encode and pad to 32 bytes
    label_bytes = label.encode("utf-8")
    if len(label_bytes) > 32:
        raise ValueError(f"Label too long: {len(label_bytes)} bytes (max 32)")

    # Pad with zeros
    label_bytes = label_bytes.ljust(32, b"\x00")

    # Request automatically handles acknowledgement
    result = await self.connection.request(
        packets.Device.SetLabel(label=label_bytes),
    )
    self._raise_if_unhandled(result)

    if result:
        self._label = label

        if self._state is not None:
            self._state.label = label
            await self._schedule_refresh()

    _LOGGER.debug(
        {
            "class": "Device",
            "method": "set_label",
            "action": "change",
            "values": {"label": label},
        }
    )

async def get_power(self) -> int:
    """Get device power state.

    Always fetches from device.

    Returns:
        Power level as integer (0 for off, 65535 for on)

    Raises:
        LifxDeviceNotFoundError: If device is not connected
        LifxTimeoutError: If device does not respond
        LifxProtocolError: If response is invalid
        LifxUnsupportedCommandError: If device doesn't support this command

    Example:
        ```python
        level = await device.get_power()
        print(f"Power: {'ON' if level > 0 else 'OFF'}")
        ```
    """
    # Request automatically unpacks response
    state = await self.connection.request(packets.Device.GetPower())
    self._raise_if_unhandled(state)

    # Power level is uint16 (0 or 65535)
    power_level = state.level
    # Update state if it exists
    if self._state is not None:
        self._state.power = power_level
        self._state.last_updated = time.time()

    _LOGGER.debug(
        {
            "class": "Device",
            "method": "get_power",
            "action": "query",
            "reply": {"level": power_level},
        }
    )
    return power_level