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
        devices = await discover_devices()
        for discovered in devices:
            device = await discovered.create_device()
            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,
            )
            return device_class(**kwargs)

    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

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()
        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(
            {
                "class": "UdpTransport",
                "method": "open",
                "action": "failed",
                "ip_address": self._ip_address,
                "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(
            {
                "class": "UdpTransport",
                "method": "send",
                "action": "failed",
                "destination": address,
                "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(
            {
                "class": "UdpTransport",
                "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(
            {
                "class": "UdpTransport",
                "method": "receive",
                "action": "packet_too_large",
                "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(
            {
                "class": "UdpTransport",
                "method": "receive",
                "action": "packet_too_small",
                "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

async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: object,
) -> None:
    """Exit async context manager and close connection."""
    await self.close()

async def open(self) -> None:
    """Open connection to device.

    Opens the UDP transport for sending and receiving packets.
    Called automatically on first request if not already open.
    """
    if self._is_open:
        return

    # Prevent concurrent open() calls (see __init__ for why this is a
    # poll loop rather than an asyncio.Lock)
    if self._is_opening:
        # Another task is already opening, wait for it
        while self._is_opening:
            await asyncio.sleep(0.001)
        return

    self._is_opening = True
    try:
        # Double-check after setting flag
        if self._is_open:  # pragma: no cover
            return

        # Create shutdown event for receiver task
        self._receiver_shutdown = asyncio.Event()

        # Open transport
        self._transport = UdpTransport(port=0, broadcast=False)
        await self._transport.open()
        self._is_open = True

        # Start background receiver task
        self._receiver_task = asyncio.create_task(self._background_receiver())

        _LOGGER.debug(
            {
                "class": "DeviceConnection",
                "method": "open",
                "serial": self.serial,
                "ip": self.ip,
                "port": self.port,
            }
        )
    finally:
        self._is_opening = False

async def close(self) -> None:
    """Close connection to device."""
    if not self._is_open:
        return

    self._is_open = False

    # Signal shutdown to receiver task
    if self._receiver_shutdown:
        self._receiver_shutdown.set()

    # Wait for receiver to stop (with timeout)
    if self._receiver_task:
        try:
            await asyncio.wait_for(
                self._receiver_task, timeout=_RECEIVER_SHUTDOWN_TIMEOUT
            )
        except TIMEOUT_ERRORS:
            self._receiver_task.cancel()
            try:
                await self._receiver_task
            except asyncio.CancelledError:
                pass

    # Cancel all pending request queues
    for queue in self._pending_requests.values():
        # Drain queue
        while not queue.empty():
            try:
                queue.get_nowait()
            except asyncio.QueueEmpty:
                break
    self._pending_requests.clear()

    # Close transport
    if self._transport is not None:
        await self._transport.close()

    _LOGGER.debug(
        {
            "class": "DeviceConnection",
            "method": "close",
            "serial": self.serial,
            "ip": self.ip,
        }
    )
    self._transport = None

async def send_packet(
    self,
    packet: Any,
    source: int | None = None,
    sequence: int = 0,
    ack_required: bool = False,
    res_required: bool = False,
) -> None:
    """Send a packet to the device.

    Args:
        packet: Packet dataclass instance
        source: Client source identifier (optional, allocated if None)
        sequence: Sequence number (default: 0)
        ack_required: Request acknowledgement
        res_required: Request response

    Raises:
        ConnectionError: If connection is not open or send fails
    """
    if not self._is_open or self._transport is None:
        raise LifxConnectionError("Connection not open")

    # Allocate source if not provided
    if source is None:
        source = allocate_source()

    message = create_message(
        packet=packet,
        source=source,
        sequence=sequence,
        target=self._send_target,
        ack_required=ack_required,
        res_required=res_required,
    )

    # Send to device
    await self._transport.send(message, (self.ip, self.port))

async def receive_packet(self, timeout: float = 0.5) -> tuple[LifxHeader, bytes]:
    """Receive a packet from the device.

    Note:
        This method does not validate the source IP address. Validation is instead
        performed using the LIFX protocol's built-in target field (serial number)
        and sequence number matching in request_stream() and request_ack_stream().
        This approach is more reliable in complex network configurations (NAT,
        multiple interfaces, bridges, etc.) while maintaining security through
        proper protocol-level validation.

    Args:
        timeout: Timeout in seconds

    Returns:
        Tuple of (header, payload)

    Raises:
        ConnectionError: If connection is not open
        TimeoutError: If no response within timeout
    """
    if not self._is_open or self._transport is None:
        raise LifxConnectionError("Connection not open")

    # Receive message - source address not validated here
    # Validation occurs via target field and sequence number matching
    data, _addr = await self._transport.receive(timeout=timeout)

    # Parse and return message
    return parse_message(data)