High-Level API¶

async def discover(
    timeout: float = DISCOVERY_TIMEOUT,
    broadcast_address: str = "255.255.255.255",
    port: int = LIFX_UDP_PORT,
    max_response_time: float = MAX_RESPONSE_TIME,
    idle_timeout_multiplier: float = IDLE_TIMEOUT_MULTIPLIER,
    device_timeout: float = DEFAULT_REQUEST_TIMEOUT,
    max_retries: int = DEFAULT_MAX_RETRIES,
) -> AsyncGenerator[Device, None]:
    """Discover LIFX devices and yield them as they are found.

    Args:
        timeout: Discovery timeout in seconds (default 15.0)
        broadcast_address: Broadcast address to use (default "255.255.255.255")
        port: Port to use (default LIFX_UDP_PORT)
        max_response_time: Max time to wait for responses
        idle_timeout_multiplier: Idle timeout multiplier
        device_timeout: request timeout set on discovered devices
        max_retries: max retries per request set on discovered devices
    Yields:
        Device instances as they are discovered

    Note:
        Internal device-class construction errors are logged and isolated to
        that responder so the remaining discovery sweep can continue.

    Raises:
        ValueError: If ``broadcast_address`` is not a valid IPv4 or IPv6
            literal, including a link-local IPv6 address without a
            syntactically valid zone.
        LifxNetworkError: If the validated destination cannot be resolved to a
            native socket address or the discovery transport fails.

    Example:
        ```python
        # Process devices as they're discovered
        async for device in discover():
            print(f"Found: {device.serial}")
            async with device:
                await device.set_power(True)

        # Or collect all devices first
        devices = []
        async for device in discover():
            devices.append(device)
        ```
    """
    validate_address(broadcast_address)
    validate_port(port)
    deadline = time.monotonic() + max(0.0, timeout)
    observer: _DiscoveryObserver | None = _current_discovery_observer()
    queue: asyncio.Queue[_DiscoveryEvent] = asyncio.Queue()
    fatal_errors: dict[_DiscoverySource, BaseException] = {}

    udp_devices = discover_devices_shared(
        timeout=timeout,
        broadcast_address=broadcast_address,
        port=port,
        max_response_time=max_response_time,
        idle_timeout_multiplier=idle_timeout_multiplier,
        device_timeout=device_timeout,
        max_retries=max_retries,
        _caller_deadline=deadline,
        _observer=observer,
    )

    def mdns_source(
        failure_sink: Callable[[_MdnsFailure], None],
    ) -> AsyncGenerator[Device, None]:
        return _discover_verified_devices_mdns(
            timeout=timeout,
            max_response_time=max_response_time,
            idle_timeout_multiplier=idle_timeout_multiplier,
            device_timeout=device_timeout,
            max_retries=max_retries,
            failure_sink=failure_sink,
            _deadline=deadline,
            _observer=observer,
        )

    tasks = [
        asyncio.create_task(
            _pump_udp_discovery(
                udp_devices,
                queue,
                deadline=deadline,
                fatal_errors=fatal_errors,
            )
        ),
        asyncio.create_task(
            _pump_mdns_discovery(
                mdns_source,
                queue,
                deadline=deadline,
                fatal_errors=fatal_errors,
            )
        ),
    ]
    completed_legs = 0
    seen_serials: set[str] = set()
    try:
        while completed_legs < 2:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                return
            try:
                event = await asyncio.wait_for(queue.get(), timeout=remaining)
            except asyncio.TimeoutError:
                return

            if event.kind == "leg_done":
                completed_legs += 1
                continue
            if event.kind == "absorbed":
                _LOGGER.debug(
                    {
                        "module": "lifx.api",
                        "method": "discover",
                        "action": "mdns_unavailable",
                        "stage": event.stage,
                        "reason": event.reason,
                        "error_type": event.error_type,
                    }
                )
                continue
            if event.kind == "fatal":
                raise fatal_errors[event.source]

            device = event.device
            if device is None:
                raise RuntimeError("discovery device event contained no device")
            serial = Serial.from_string(device.serial).to_string()
            disposition: Literal["winner", "duplicate"]
            if serial in seen_serials:
                disposition = "duplicate"
            else:
                seen_serials.add(serial)
                disposition = "winner"
            _emit_discovery_event(
                observer,
                source=event.source,
                stage=disposition,
                raw_identity=serial,
            )
            if disposition == "winner" and time.monotonic() < deadline:
                yield device
    finally:
        await _cancel_and_reap(tasks)

async def discover_mdns(
    timeout: float = DISCOVERY_TIMEOUT,
    max_response_time: float = MAX_RESPONSE_TIME,
    idle_timeout_multiplier: float = IDLE_TIMEOUT_MULTIPLIER,
    device_timeout: float = DEFAULT_REQUEST_TIMEOUT,
    max_retries: int = DEFAULT_MAX_RETRIES,
) -> AsyncGenerator[Device, None]:
    """Discover LIFX devices via mDNS and yield them as they are found.

    Uses mDNS/DNS-SD discovery with the _lifx._udp.local service type. Device
    type metadata in DNS-SD responses avoids a separate LIFX product query for
    every device.

    The IPv4 multicast query is sent from an ephemeral source port and the
    transport accepts legacy-unicast replies addressed directly to that
    socket. It does not join the multicast group and does not receive unsolicited
    announcements. Each call observes direct traffic delivered to its per-call
    socket during the sweep and has its own non-reusable discovery cache, but it
    does not authenticate or correlate responders with outstanding queries.
    An mDNS result proves an advertisement exists, not that the device is
    currently reachable; see the discovery guide's Limitations section for
    detail, including how discover()'s mDNS candidates verify reachability
    with a correlated device request.

    Each yielded device reports ``connectivity`` from its mDNS TXT record:
    ``Connectivity.THREAD`` only after an exact positive Thread report, and
    ``Connectivity.WIFI`` for every other construction or discovery outcome.
    Once the device answers any request, its own frame address report
    supersedes this in both directions. This descriptive value does not
    authenticate the device or change its routing, retry, or tuning
    behaviour.

    Args:
        timeout: Discovery timeout in seconds (default 15.0)
        max_response_time: Max time to wait for responses
        idle_timeout_multiplier: Idle timeout multiplier
        device_timeout: request timeout set on discovered devices
        max_retries: max retries per request set on discovered devices

    Yields:
        Device instances (Light, MatrixLight, MultiZoneLight, etc.)
        as they are discovered

    Example:
        ```python
        # Process devices as they're discovered
        async for device in discover_mdns():
            print(f"Found: {device.serial}")
            async with device:
                await device.set_power(True)

        # Or collect all devices first
        devices = []
        async for device in discover_mdns():
            devices.append(device)
        ```
    """
    from lifx.network.discovery.mdns.discovery import discover_devices_mdns

    devices = discover_devices_mdns(
        timeout=timeout,
        max_response_time=max_response_time,
        idle_timeout_multiplier=idle_timeout_multiplier,
        device_timeout=device_timeout,
        max_retries=max_retries,
    )
    async with aclosing(devices):
        async for device in devices:
            yield device

async def find_by_serial(
    serial: str,
    timeout: float = DISCOVERY_TIMEOUT,
    broadcast_address: str = "255.255.255.255",
    port: int = LIFX_UDP_PORT,
    max_response_time: float = MAX_RESPONSE_TIME,
    idle_timeout_multiplier: float = IDLE_TIMEOUT_MULTIPLIER,
    device_timeout: float = DEFAULT_REQUEST_TIMEOUT,
    max_retries: int = DEFAULT_MAX_RETRIES,
) -> Device | None:
    """Find a specific device by serial number.

    Args:
        serial: Serial number as hex string (with or without separators)
        timeout: Discovery timeout in seconds (default DISCOVERY_TIMEOUT)
        broadcast_address: Broadcast address to use (default "255.255.255.255")
        port: Port to use (default LIFX_UDP_PORT)
        max_response_time: Max time to wait for responses
        idle_timeout_multiplier: Idle timeout multiplier
        device_timeout: request timeout set on discovered device
        max_retries: max retries per request set on discovered device

    Returns:
        Device instance if found, None otherwise

    Note:
        Internal device-class construction errors are logged and return
        ``None`` rather than escaping this lookup.

    Raises:
        ValueError: If ``broadcast_address`` is not a valid IPv4 or IPv6
            literal, including a link-local IPv6 address without a
            syntactically valid zone.
        LifxNetworkError: If the validated destination cannot be resolved to a
            native socket address or the discovery transport fails.

    Example:
        ```python
        # Find by serial number
        device = await find_by_serial("d073d5123456")
        if device:
            async with device:
                await device.set_power(True)
        ```
    """
    # Serial.from_string intentionally removes ASCII spaces as well as the
    # documented colon/hyphen separators. Preserve that established parsing
    # contract while rejecting malformed text before any source is created.
    try:
        serial_str = Serial.from_string(serial).to_string()
    except ValueError:
        return None

    return await _race_serial_sources(
        serial_str,
        timeout=timeout,
        broadcast_address=broadcast_address,
        port=port,
        max_response_time=max_response_time,
        idle_timeout_multiplier=idle_timeout_multiplier,
        device_timeout=device_timeout,
        max_retries=max_retries,
    )

async def find_by_label(
    label: str,
    exact_match: bool = False,
    timeout: float = DISCOVERY_TIMEOUT,
    broadcast_address: str = "255.255.255.255",
    port: int = LIFX_UDP_PORT,
    max_response_time: float = MAX_RESPONSE_TIME,
    idle_timeout_multiplier: float = IDLE_TIMEOUT_MULTIPLIER,
    device_timeout: float = DEFAULT_REQUEST_TIMEOUT,
    max_retries: int = DEFAULT_MAX_RETRIES,
) -> AsyncGenerator[Device]:
    """Find LIFX devices by label (name).

    Uses a protocol trick by broadcasting GetLabel instead of GetService,
    which returns all device labels in StateLabel responses. This is more
    efficient than querying each device individually.

    Args:
        label: Device label to search for (case-insensitive)
        exact_match: If True, match label exactly and yield at most one device;
                     if False, match substring and yield all matching devices
                     (default False)
        timeout: Discovery timeout in seconds (default DISCOVERY_TIMEOUT)
        broadcast_address: Broadcast address to use (default "255.255.255.255")
        port: Port to use (default LIFX_UDP_PORT)
        max_response_time: Max time to wait for responses
        idle_timeout_multiplier: Idle timeout multiplier
        device_timeout: request timeout set on discovered device(s)
        max_retries: max retries per request set on discovered device(s)

    Yields:
        Matching Device instance(s)

    Note:
        Internal device-class construction errors are logged and isolated to
        that responder so remaining matches can still be yielded.

    Raises:
        ValueError: If ``broadcast_address`` is not a valid IPv4 or IPv6
            literal, including a link-local IPv6 address without a
            syntactically valid zone.
        LifxNetworkError: If the validated destination cannot be resolved to a
            native socket address or the discovery transport fails.

    Example:
        ```python
        # Find all devices with "Living" in the label
        async for device in find_by_label("Living"):
            async with device:
                await device.set_power(True)

        # Find device by exact label match (yields at most one)
        async for device in find_by_label("Living Room", exact_match=True):
            async with device:
                await device.set_power(True)
        ```
    """
    responses = _discover_with_packet(
        packets.Device.GetLabel(),
        timeout=timeout,
        broadcast_address=broadcast_address,
        port=port,
        max_response_time=max_response_time,
        idle_timeout_multiplier=idle_timeout_multiplier,
    )
    async with aclosing(responses):
        async for resp in responses:
            device_label = resp.response_payload.get("label", "")
            matched = False

            if exact_match:
                # Exact match - return first match only
                if device_label.lower() == label.lower():
                    matched = True
            else:
                # Substring match - return all matches
                if label.lower() in device_label.lower():
                    matched = True

            if matched:
                # Create DiscoveredDevice from response
                disc = DiscoveredDevice(
                    serial=resp.serial,
                    ip=resp.ip,
                    port=resp.port,
                    response_time=resp.response_time,
                    timeout=device_timeout,
                    max_retries=max_retries,
                )

                device = await _create_discovered_device(
                    disc,
                    method="find_by_label",
                )
                if device is not None:
                    yield device
                    if exact_match:
                        return

async def find_by_ip(
    ip: str,
    timeout: float = DISCOVERY_TIMEOUT,
    port: int = LIFX_UDP_PORT,
    max_response_time: float = MAX_RESPONSE_TIME,
    idle_timeout_multiplier: float = IDLE_TIMEOUT_MULTIPLIER,
    device_timeout: float = DEFAULT_REQUEST_TIMEOUT,
    max_retries: int = DEFAULT_MAX_RETRIES,
) -> Device | None:
    """Find a LIFX device by IP address.

    Uses targeted discovery by sending the request to the validated IPv4 or IPv6
    literal, which means only that device will respond (if it exists). Zoned IPv6
    link-local literals are supported when they include their interface identifier.
    This is more efficient than broadcasting to all devices and filtering.

    Args:
        ip: Validated IPv4 or IPv6 device address literal, including a zoned
            link-local IPv6 literal
        timeout: Discovery timeout in seconds (default DISCOVERY_TIMEOUT)
        port: Port to use (default LIFX_UDP_PORT)
        max_response_time: Max time to wait for responses
        idle_timeout_multiplier: Idle timeout multiplier
        device_timeout: request timeout set on discovered device
        max_retries: max retries per request set on discovered device

    Returns:
        Device instance if found, None otherwise

    Note:
        Internal device-class construction errors are logged and return
        ``None`` rather than escaping this lookup.

    Raises:
        ValueError: If ``ip`` is not a valid IPv4 or IPv6 literal, including a
            link-local IPv6 address without a syntactically valid zone.
        LifxNetworkError: If the validated destination cannot be resolved to a
            native socket address or the discovery transport fails.

    Example:
        ```python
        # Find device at specific IP
        device = await find_by_ip("192.168.1.100")
        if device:
            async with device:
                print(f"Found: {device.label}")
        ```
    """
    # Validate before any socket exists, so a permanent configuration error
    # such as a zone-less link-local address fails immediately instead of
    # waiting out the discovery timeout.
    validate_address(ip)

    # Use the target IP as the "broadcast" address - only that device will respond
    devices = discover_devices(
        timeout=timeout,
        broadcast_address=ip,  # Protocol trick: send directly to target IP
        port=port,
        max_response_time=max_response_time,
        idle_timeout_multiplier=idle_timeout_multiplier,
        device_timeout=device_timeout,
        max_retries=max_retries,
    )
    async with aclosing(devices):
        async for discovered in devices:
            # Should only get one response (or none)
            return await _create_discovered_device(discovered, method="find_by_ip")

    return None