import sys
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    if sys.platform != "linux":
        assert False, "This is only available on Linux"


import asyncio
import contextlib
import logging
from typing import AsyncGenerator

from bumble.controller import Controller
from bumble.link import LocalLink
from bumble.transport import open_transport
from bumble.transport.common import Transport
from dbus_fast import BusType, Message, MessageType, Variant
from dbus_fast.aio.message_bus import MessageBus

from bleak._compat import timeout as async_timeout
from bleak.backends.bluezdbus import defs
from bleak.backends.bluezdbus.signals import MatchRules, add_match
from bleak.backends.bluezdbus.utils import assert_reply, get_dbus_authenticator

BLEAK_TEST_MANUFACTURER_ID = 0xB1EA

logger = logging.getLogger(__name__)


@contextlib.asynccontextmanager
async def open_message_bus() -> AsyncGenerator[MessageBus, None]:
    """
    Open a D-Bus message bus connection.
    """
    bus = MessageBus(bus_type=BusType.SYSTEM, auth=get_dbus_authenticator())
    try:
        await bus.connect()
        yield bus
    finally:
        bus.disconnect()
        await bus.wait_for_disconnect()


async def power_on_controller(
    bus: MessageBus, adapter_path: str, timeout: float = 2.0
) -> None:
    """
    Power on a Bluetooth controller via D-Bus.

    It may take some time for the adapter to fully configure itself after
    powering on. So we make multiple attempts until timeout is reached.
    """
    async with async_timeout(timeout):
        while True:
            try:
                reply = await bus.call(
                    Message(
                        destination=defs.BLUEZ_SERVICE,
                        path=adapter_path,
                        interface=defs.PROPERTIES_INTERFACE,
                        member="Set",
                        signature="ssv",
                        body=[defs.ADAPTER_INTERFACE, "Powered", Variant("b", True)],
                    )
                )
                assert_reply(reply)
                return
            except Exception as e:
                logger.debug("Failed to power on adapter at %s: %s", adapter_path, e)
                await asyncio.sleep(0.1)


@contextlib.asynccontextmanager
async def wait_for_new_adapter() -> (
    AsyncGenerator[tuple[MessageBus, asyncio.Future[str]], None]
):
    """
    Connect to D-Bus and wait for a new Bluetooth adapter to be added.

    Yields a Future that will be resolved with the adapter path when
    a new adapter is detected via InterfacesAdded signal.
    """
    async with open_message_bus() as bus:
        loop = asyncio.get_running_loop()
        adapter_path_future: asyncio.Future[str] = loop.create_future()

        def _on_interfaces_added(message: Message):
            if message.message_type != MessageType.SIGNAL:
                return
            if message.member != "InterfacesAdded":
                return

            obj_path, ifaces = message.body
            adapter = ifaces.get(defs.ADAPTER_INTERFACE)
            if not adapter:
                return

            # Check for our test manufacturer ID to identify the adapter
            if adapter["Manufacturer"].value != BLEAK_TEST_MANUFACTURER_ID:
                return

            if not adapter_path_future.done():
                adapter_path_future.set_result(obj_path)

        bus.add_message_handler(_on_interfaces_added)

        try:
            # Subscribe to InterfacesAdded signals
            reply = await add_match(
                bus,
                MatchRules(
                    interface=defs.OBJECT_MANAGER_INTERFACE,
                    member="InterfacesAdded",
                    arg0path="/org/bluez/",
                ),
            )
            assert_reply(reply)

            # Yield the future and bus for the caller to use
            yield bus, adapter_path_future
        finally:
            bus.remove_message_handler(_on_interfaces_added)


def _clear_bit(flags: bytes, bit_pos: int) -> bytes:
    int_flags = int.from_bytes(flags, byteorder="little")
    int_flags &= ~(1 << bit_pos)
    return int_flags.to_bytes(len(flags), byteorder="little")


@contextlib.asynccontextmanager
async def open_bluez_bluetooth_controller_link(
    hci_transport_name: str,
) -> AsyncGenerator[LocalLink, None]:
    """
    Open a local link (virtual RF connection) to a bumble Bluetooth
    controller that is connected to BlueZ.
    """
    async with wait_for_new_adapter() as (bus, adapter_path_future):
        # Open a HCI transport connected to the OS
        async with await open_transport(hci_transport_name) as hci_transport:
            # Local link to create virtual RF connection between multiple Bluetooth Controllers
            link = LocalLink()

            # Bluetooth controller that BlueZ can connect to.
            # (This will register itself to the link.)
            bluez_controller = Controller(
                "BLEAK-TEST-BLUEZ",
                host_source=hci_transport.source,
                host_sink=hci_transport.sink,
                link=link,
            )
            bluez_controller.manufacturer_name = BLEAK_TEST_MANUFACTURER_ID

            # HACK: Work around Bumble missing feature combined with Linux kernel
            # requirement. https://github.com/google/bumble/issues/841
            #
            # According to the Bluetooth spec:
            #
            #   C24: [HCI_LE_Enhanced_Connection_Complete event is] Mandatory if
            #   the LE Controller supports Connection State and either LE Feature (LL
            #   Privacy) or LE Feature (Extended Advertising) is supported, otherwise optional if
            #   the LE Controller supports Connection State, otherwise excluded.
            #
            # And the Linux kernel enforces this in hci_le_create_conn_sync().
            # It will get a timeout if one of these features is enabled and the
            # Enhanced Connection Complete event is not sent.
            #
            # However, Bumble (as of 0.0.220) always sends HCI_LE_Connection_Complete_Event
            # in response to HCI_LE_Create_Connection_Command even when it should
            # be sending HCI_LE_Enhanced_Connection_Complete_Event.
            #
            # For now, we can work around the issue by disabling LL Privacy and
            # Extended Advertising features in the BlueZ controller.
            #
            # Ideally, this should be fixed in Bumble.

            bluez_controller.le_features = _clear_bit(
                bluez_controller.le_features, 6  # LL Privacy
            )
            bluez_controller.le_features = _clear_bit(
                bluez_controller.le_features, 12  # Extended Advertising
            )

            # Wait up to 5 seconds for the new adapter to appear via InterfacesAdded
            adapter_path = await asyncio.wait_for(adapter_path_future, timeout=5.0)
            logger.info(f"New adapter detected at {adapter_path}")

            # Ensure controller is powered on. This also ensures that BlueZ has fully
            # initialized the adapter and it is ready for use.
            await power_on_controller(bus, adapter_path)

            # We dont need the bus anymore. We have done everything needed with it.
            # Bleak will open its own D-Bus connection.
            bus.disconnect()
            await bus.wait_for_disconnect()

            # Yield the local link for use in tests
            yield link


@contextlib.asynccontextmanager
async def open_transport_with_bluez_vhci() -> AsyncGenerator[Transport, None]:
    """
    Create a bumble HCI Transport connected to BlueZ via the vhci driver and connect
    a Bluetooth controller for a peripheral device to it.
    """
    async with open_bluez_bluetooth_controller_link("vhci") as local_link:
        peripheral_controller = Controller("BLEAK-TEST-PERIPHERAL", link=local_link)
        yield Transport(peripheral_controller, peripheral_controller)