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 __post_init__(self) -> None:
    """Derive the approximate RSSI from the link margin."""
    self.rssi = self.link_margin_db - self._RSSI_OFFSET_DB

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
    self._connectivity = source._connectivity
    self._connectivity_metadata = source._connectivity_metadata

    # Discovery observes the transport on the temporary device, whose
    # connection is closed before this instance is built. Without this the
    # authoritative frame address report is discarded on every discovery
    # path and connectivity silently falls back to discovery metadata.
    # An unobserved donor never erases an observation already made here.
    self.connection._adopt_thread_connection(source.connection.thread_connection)

@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 | None = DEFAULT_MAX_RETRIES,
    *,
    fetch_wifi_info: bool = False,
    fetch_thread_info: bool = False,
    fetch_radio_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_thread_info: Query Thread mesh information during state
            initialization
        fetch_radio_info: Query whichever radio matches the device's
            evidenced connectivity 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()
        ```
    """
    # Reject an unusable endpoint before the serial-less leg below builds
    # a DeviceConnection directly; the advisory belongs to construction.
    validate_address(ip)
    validate_port(port)

    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 not response:
                # Nothing answered, so the port is worth naming. A device
                # that answered without a usable serial is not a port
                # problem and raises below without the hint.
                _warn_unanswered_port(ip, port)
            elif 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_thread_info=fetch_thread_info,
                        fetch_radio_info=fetch_radio_info,
                        fetch_ambient_light=fetch_ambient_light,
                    )
            else:
                # The device answered, just not with StateService, so the
                # port is not the problem. Named the same way as connect().
                raise LifxDeviceNotFoundError(
                    f"Unexpected response to GetService: {type(response).__name__}"
                )
        except LifxTimeoutError:
            _warn_unanswered_port(ip, port)
            raise
        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_thread_info=fetch_thread_info,
            fetch_radio_info=fetch_radio_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 | None = DEFAULT_MAX_RETRIES,
    *,
    fetch_wifi_info: bool = False,
    fetch_thread_info: bool = False,
    fetch_radio_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_thread_info: Query Thread mesh information during state
            initialization
        fetch_radio_info: Query whichever radio matches the device's
            evidenced connectivity 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)
        ```
    """
    # Validate before Step 1. The serial-less leg below builds a
    # DeviceConnection directly and never reaches Device.__init__, so
    # without this call an unusable address would cost a full silent
    # request timeout here.
    validate_address(ip)
    validate_port(port)

    # 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"
                    )
            elif not response:
                # Nothing answered, so the port is worth naming
                _warn_unanswered_port(ip, port)
                raise LifxDeviceNotFoundError("No response from device")
            else:
                # The device answered, just not with StateService, so the
                # port is not the problem
                raise LifxDeviceNotFoundError(
                    f"Unexpected response to GetService: {type(response).__name__}"
                )
        except LifxTimeoutError:
            _warn_unanswered_port(ip, port)
            raise
        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_thread_info=fetch_thread_info,
            fetch_radio_info=fetch_radio_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()