Network Layer¶

async def create_device(self):
    """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:
        TypeError: If a concrete device constructor no longer accepts the
            shared discovery arguments.
        AttributeError: If an internal device capability or metadata
            contract is broken.

    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()}")
        ```
    """
    # Intentional local imports preserve the documented layer direction.
    # Importing the device layer while this network module initialises
    # creates a cycle when callers reach ``lifx.devices`` before the
    # top-level package has already populated ``lifx.network``.
    from lifx.devices.base import Device
    from lifx.devices.detection import get_device_class_for_product

    try:
        # Create temporary device to query version. Address validation can
        # fail here, before a connection exists, so keep construction
        # inside the same failure boundary as capability detection.
        temp_device = Device(
            serial=self.serial,
            ip=self.ip,
            port=self.port,
            timeout=self.timeout,
            max_retries=self.max_retries,
        )
        construction_task = asyncio.current_task()
        if construction_task is not None:
            self._construction_connections[construction_task] = (
                temp_device.connection
            )

    except ValueError as error:
        _LOGGER.debug(
            {
                "class": "DiscoveredDevice",
                "method": "create_device",
                "action": "invalid_device_address",
                "serial": self.serial,
                "ip": self.ip,
                "reason": str(error),
            }
        )
        return None

    try:
        await temp_device.ensure_capabilities()
    except LifxError as error:
        _LOGGER.debug(
            {
                "class": "DiscoveredDevice",
                "method": "create_device",
                "action": "capability_detection_failed",
                "error_type": type(error).__name__,
            }
        )
        return None
    finally:
        # Always close the temporary device connection
        try:
            await temp_device.connection.close()
        finally:
            if construction_task is not None:
                self._construction_connections.pop(construction_task, None)

    if not temp_device.capabilities or not temp_device.version:
        return None

    try:
        device_class = get_device_class_for_product(
            temp_device.version.product,
            temp_device.capabilities,
        )
    except LifxUnsupportedDeviceError:
        return None

    # Keep typed-device construction outside the transient network failure
    # boundary. A missing constructor pass-through is a programming error,
    # and must fail visibly instead of silently removing that product from
    # discovery.
    device = device_class(
        serial=self.serial,
        ip=self.ip,
        port=self.port,
        timeout=self.timeout,
        max_retries=self.max_retries,
    )

    # 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

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."""
    generation = self._state_generation
    async with self._state_lock:
        if generation != self._state_generation:
            return

        if self.is_open:
            _LOGGER.debug(
                {
                    "class": "UdpTransport",
                    "method": "open",
                    "action": "already_open",
                    "ip_address": self._ip_address,
                    "port": self._port,
                }
            )
            return

        datagram_transport: DatagramTransport | None = None
        raw_socket: socket.socket | None = None
        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
            )

            # Create datagram endpoint. The socket family follows the local
            # bind address, derived by the one shared rule in
            # lifx.network.address, which is what lets this transport reach
            # Thread devices that have no IPv4 address.
            family = family_for(self._ip_address)
            bind_address = sockaddr_for(
                (self._ip_address, self._port), require_routable=False
            )
            if family is socket.AF_INET6:
                raw_socket = _socket_factory(socket.AF_INET6, socket.SOCK_DGRAM)
                raw_socket.setsockopt(
                    socket.IPPROTO_IPV6,
                    socket.IPV6_V6ONLY,
                    1,
                )
                if _SO_REUSEPORT is not None:
                    raw_socket.setsockopt(
                        socket.SOL_SOCKET,
                        _SO_REUSEPORT,
                        1,
                    )
                raw_socket.bind(bind_address)
                raw_socket.setblocking(False)
                datagram_transport, _ = await loop.create_datagram_endpoint(
                    lambda: protocol,
                    sock=raw_socket,
                )
                raw_socket = None  # asyncio owns the descriptor now
            else:
                assert len(bind_address) == 2
                datagram_transport, _ = await loop.create_datagram_endpoint(
                    lambda: protocol,
                    local_addr=bind_address,
                    reuse_port=_SO_REUSEPORT is not None,
                    family=family,
                )

            # Get actual port assigned
            actual_port = datagram_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 = datagram_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",
                        }
                    )

            if generation != self._state_generation:
                datagram_transport.close()
                return

            self._protocol = protocol
            self._transport = datagram_transport
            self._family = family

        except BaseException as e:
            if datagram_transport is not None:
                datagram_transport.close()
            if raw_socket is not None:
                raw_socket.close()
            self._transport = None
            self._protocol = None
            self._family = None

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

async def send(self, data: bytes, address: SocketAddress) -> None:
    """Send data to a specific address.

    The destination's address family is checked against the socket's
    before the datagram is handed to asyncio. A mismatch is a permanent
    configuration error, but the socket reports it as a ``gaierror``,
    an ``OSError`` subclass that :meth:`_UdpProtocol.error_received`
    deliberately swallows along with every genuine peer error. Without
    this check an unreachable-by-construction address is
    indistinguishable from a sleeping device: the caller waits out the
    full retry schedule and is told only that it timed out. Checking
    first costs one parse and names the actual problem.

    This is a pre-send guard and nothing more. The routing of errors
    that arrive from the network is untouched: EHOSTUNREACH, EHOSTDOWN
    and ENETUNREACH still reach ``error_received`` and still leave the
    endpoint alive.

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

    Raises:
        LifxNetworkError: If the socket is not open, the destination's
            address family does not match the socket's, an IPv6 zone
            identifier cannot be resolved, or the send fails.
    """
    if self._transport is None or self._protocol is None or self._family is None:
        raise LifxNetworkError("Socket not open")

    try:
        send_address = sockaddr_for(address)
    except ValueError as error:
        _LOGGER.debug(
            self._log(
                method="send",
                action="invalid_destination",
                destination_ip=address[0],
                destination_port=address[1],
                reason=str(error),
            )
        )
        raise LifxNetworkError(
            f"Invalid destination {address[0]!r}: {error}"
        ) from error

    destination_family = family_for_sockaddr(send_address)
    if destination_family is not self._family:
        _LOGGER.debug(
            self._log(
                method="send",
                action="family_mismatch",
                destination_ip=address[0],
                destination_port=address[1],
                destination_family=destination_family.name,
                socket_family=self._family.name,
            )
        )
        raise LifxNetworkError(
            f"Destination {address[0]} requires {destination_family.name} "
            f"but the socket family is {self._family.name}"
        )

    try:
        self._transport.sendto(data, send_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, SocketAddress]:
    """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
        LifxNetworkError: If socket is not open or receive fails
        LifxProtocolError: 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 close(self) -> None:
    """Close the UDP socket."""
    self._close_immediately()

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.
    """
    generation = self._state_generation
    if self._is_open:
        return

    # Claim the opener role without an await between the check and write.
    # A waiter retries the decision after the current opener finishes: it
    # must never return success solely because a failed opener cleared the
    # flag. The poll remains loop-agnostic for connections reopened under
    # a different event loop.
    while True:
        if generation != self._state_generation:
            return
        if self._is_open:
            return
        if self._is_closing:
            while self._is_closing:
                await asyncio.sleep(0.001)
            continue
        if not self._is_opening:
            self._is_opening = True
            break
        while self._is_opening:
            await asyncio.sleep(0.001)

    transport: UdpTransport | None = None
    try:
        # Double-check after setting flag
        if self._is_open:  # pragma: no cover
            return

        # Open transport, binding the wildcard that matches the device
        # address family: IPv6 for Thread devices, IPv4 otherwise. The
        # shared rule owns the choice, so this method makes no family
        # test of its own.
        try:
            local_ip = wildcard_for(self.ip)
            send_address = sockaddr_for((self.ip, self.port))
        except ValueError as error:
            raise LifxNetworkError(
                f"Invalid destination {self.ip!r}: {error}"
            ) from error
        transport = UdpTransport(
            ip_address=local_ip, port=0, broadcast=False, peer=self._peer
        )
        self._opening_transport = transport
        await transport.open()

        if generation != self._state_generation:
            # close() synchronously takes ownership from
            # _opening_transport before its first await, so it also owns
            # closing this invalidated endpoint.
            return

        self._opening_transport = None
        self._transport = transport
        self._send_address = send_address
        self._receiver_shutdown = asyncio.Event()
        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,
            }
        )
    except BaseException:
        failed_transport = (
            transport if self._opening_transport is transport else None
        )
        self._opening_transport = None
        transport = None
        self._transport = None
        self._receiver_shutdown = None
        if failed_transport is not None:
            try:
                await failed_transport.close()
            except BaseException as cleanup_error:
                _LOGGER.debug(
                    {
                        "class": "DeviceConnection",
                        "method": "open",
                        "action": "failed_transport_cleanup_failed",
                        "serial": self.serial,
                        "ip": self.ip,
                        "port": self.port,
                        "error": str(cleanup_error),
                    }
                )
        raise
    finally:
        self._is_opening = False