Network Layer¶

async def create_device(self) -> Device | None:
    """Create appropriate device instance based on product capabilities.

    Queries the device for its product ID and uses the product registry
    to instantiate the appropriate device class (Device, Light, HevLight,
    InfraredLight, MultiZoneLight, MatrixLight, or CeilingLight) based on
    the product capabilities.

    This is the single source of truth for device type detection and
    instantiation across the library.

    Returns:
        Device instance of the appropriate type

    Raises:
        LifxDeviceNotFoundError: If device doesn't respond
        LifxTimeoutError: If device query times out
        LifxProtocolError: If device returns invalid data

    Example:
        ```python
        async for discovered in discover_devices():
            device = await discovered.create_device()
            if device is None:
                continue  # unsupported product or transient failure
            print(f"Created {type(device).__name__}: {await device.get_label()}")
        ```
    """
    from lifx.devices.base import Device
    from lifx.devices.detection import get_device_class_for_product
    from lifx.exceptions import LifxUnsupportedDeviceError

    kwargs = {
        "serial": self.serial,
        "ip": self.ip,
        "port": self.port,
        "timeout": self.timeout,
        "max_retries": self.max_retries,
    }

    # Create temporary device to query version
    temp_device = Device(**kwargs)

    try:
        await temp_device.ensure_capabilities()

        if temp_device.capabilities and temp_device.version:
            device_class = get_device_class_for_product(
                temp_device.version.product,
                temp_device.capabilities,
            )
            device = device_class(**kwargs)

            # Capability detection already fetched and derived this metadata.
            # Preserve it on the correctly typed instance so callers do not
            # immediately repeat the same network work.
            device.adopt_cached_metadata(temp_device)
            return device

    except LifxUnsupportedDeviceError:
        return None

    except Exception:
        return None

    finally:
        # Always close the temporary device connection
        await temp_device.connection.close()

    return None

def __hash__(self) -> int:
    """Hash based on serial number for deduplication."""
    return hash(self.serial)

def __eq__(self, other: object) -> bool:
    """Equality based on serial number."""
    if not isinstance(other, DiscoveredDevice):
        return False
    return self.serial == other.serial

def __hash__(self) -> int:
    """Hash based on serial number for deduplication."""
    return hash(self.serial)

def __eq__(self, other: object) -> bool:
    """Equality based on serial number."""
    if not isinstance(other, LifxServiceRecord):
        return False
    return self.serial == other.serial

async def open(self) -> None:
    """Open the UDP socket."""
    if self._protocol is not None:
        _LOGGER.debug(
            {
                "class": "UdpTransport",
                "method": "open",
                "action": "already_open",
                "ip_address": self._ip_address,
                "port": self._port,
            }
        )
        return

    try:
        loop = asyncio.get_running_loop()

        _LOGGER.debug(
            {
                "class": "UdpTransport",
                "method": "open",
                "action": "opening_socket",
                "ip_address": self._ip_address,
                "port": self._port,
                "broadcast": self._broadcast,
            }
        )

        # Create protocol
        protocol = _UdpProtocol(
            on_endpoint_lost=self._endpoint_lost, peer=self._peer
        )
        self._protocol = protocol

        # Create datagram endpoint
        self._transport, _ = await loop.create_datagram_endpoint(
            lambda: protocol,
            local_addr=(self._ip_address, self._port),
            reuse_port=bool(hasattr(socket, "SO_REUSEPORT")),
            family=socket.AF_INET,
        )

        # Get actual port assigned
        actual_port = self._transport.get_extra_info("sockname")[1]
        _LOGGER.debug(
            {
                "class": "UdpTransport",
                "method": "open",
                "action": "socket_opened",
                "assigned_port": actual_port,
                "broadcast": self._broadcast,
            }
        )

        # Enable broadcast if requested
        if self._broadcast:
            sock = self._transport.get_extra_info("socket")
            if sock:
                sock.setsockopt(
                    socket.SOL_SOCKET,
                    socket.SO_BROADCAST,
                    1,
                )
                _LOGGER.debug(
                    {
                        "class": "UdpTransport",
                        "method": "open",
                        "action": "broadcast_enabled",
                    }
                )

    except OSError as e:
        _LOGGER.debug(
            self._log(
                method="open",
                action="failed",
                bind_address=self._ip_address,
                bind_port=self._port,
                reason=str(e),
            )
        )
        raise LifxNetworkError(f"Failed to open UDP socket: {e}") from e

async def send(self, data: bytes, address: tuple[str, int]) -> None:
    """Send data to a specific address.

    Args:
        data: Bytes to send
        address: Tuple of (host, port)

    Raises:
        NetworkError: If socket is not open or send fails
    """
    if self._transport is None or self._protocol is None:
        raise LifxNetworkError("Socket not open")

    try:
        self._transport.sendto(data, address)
    except OSError as e:
        _LOGGER.debug(
            self._log(
                method="send",
                action="failed",
                destination_ip=address[0],
                destination_port=address[1],
                packet_size=len(data),
                reason=str(e),
            )
        )
        raise LifxNetworkError(f"Failed to send data: {e}") from e

async def receive(self, timeout: float = 2.0) -> tuple[bytes, tuple[str, int]]:
    """Receive data from socket with size validation.

    Args:
        timeout: Timeout in seconds

    Returns:
        Tuple of (data, address) where address is (host, port)

    Raises:
        LifxTimeoutError: If no data received within timeout
        NetworkError: If socket is not open or receive fails
        ProtocolError: If packet size is invalid
    """
    if self._protocol is None:
        raise LifxNetworkError("Socket not open")

    try:
        data, addr = await asyncio.wait_for(
            self._protocol.queue.get(), timeout=timeout
        )
    except TIMEOUT_ERRORS as e:
        raise LifxTimeoutError(f"No data received within {timeout}s") from e
    except OSError as e:
        _LOGGER.error(self._log(method="receive", action="failed", reason=str(e)))
        raise LifxNetworkError(f"Failed to receive data: {e}") from e

    # Validate packet size
    if len(data) > MAX_PACKET_SIZE:
        _LOGGER.error(
            self._log(
                method="receive",
                action="packet_too_large",
                sender_ip=addr[0],
                sender_port=addr[1],
                packet_size=len(data),
                max_size=MAX_PACKET_SIZE,
            )
        )
        raise LifxProtocolError(
            f"Packet too big: {len(data)} bytes > {MAX_PACKET_SIZE} bytes"
        )

    if len(data) < MIN_PACKET_SIZE:
        _LOGGER.error(
            self._log(
                method="receive",
                action="packet_too_small",
                sender_ip=addr[0],
                sender_port=addr[1],
                packet_size=len(data),
                min_size=MIN_PACKET_SIZE,
            )
        )
        raise LifxProtocolError(
            f"Packet too small: {len(data)} bytes < {MIN_PACKET_SIZE} bytes"
        )

    return data, addr

async def receive_many(
    self, timeout: float = 5.0, max_packets: int | None = None
) -> list[tuple[bytes, tuple[str, int]]]:
    """Receive multiple packets within timeout period.

    Args:
        timeout: Total timeout in seconds
        max_packets: Maximum number of packets to receive (None for unlimited)

    Returns:
        List of (data, address) tuples

    Raises:
        NetworkError: If socket is not open

    .. deprecated:: 5.5.0
        :meth:`receive_many` is deprecated and will be removed in v6.0.
        Use :meth:`receive` in a loop, or the public discovery API in
        :mod:`lifx.api` (e.g. :func:`~lifx.api.discover`), for
        multi-response collection.
    """
    warnings.warn(
        "UdpTransport.receive_many is deprecated and will be removed in "
        "v6.0. Use receive() in a loop, or the public discovery API in "
        "lifx.api, for multi-response collection.",
        DeprecationWarning,
        stacklevel=2,
    )
    if self._protocol is None:
        raise LifxNetworkError("Socket not open")

    packets: list[tuple[bytes, tuple[str, int]]] = []
    deadline = time.monotonic() + timeout

    while True:
        if max_packets is not None and len(packets) >= max_packets:
            break

        remaining = deadline - time.monotonic()
        if remaining <= 0:
            break

        try:
            data, addr = await asyncio.wait_for(
                self._protocol.queue.get(), timeout=remaining
            )

            # Validate packet size
            if len(data) > MAX_PACKET_SIZE:
                # Drop oversized packet to prevent memory exhaustion DoS
                continue

            if len(data) < MIN_PACKET_SIZE:
                # Drop undersized packet (header is 36 bytes)
                continue

            packets.append((data, addr))
        except TIMEOUT_ERRORS:
            # Timeout is expected - return what we collected
            break
        except OSError:
            # Ignore individual receive errors
            break

    return packets

async def close(self) -> None:
    """Close the UDP socket."""
    if self._transport is not None:
        _LOGGER.debug(
            {
                "class": "UdpTransport",
                "method": "close",
                "action": "closing",
            }
        )
        self._transport.close()
        self._transport = None
        self._protocol = None
        _LOGGER.debug(
            {
                "class": "UdpTransport",
                "method": "close",
                "action": "closed",
            }
        )

async def __aenter__(self) -> Self:
    """Enter async context manager."""
    # Don't open connection here - it will open lazily on first request
    return self