# Source code for bleak.backends.p4android.scanner

    import sys
    from typing import TYPE_CHECKING

    if TYPE_CHECKING:
        if sys.platform != "android":
            assert False, "This backend is only available on Android"

    import asyncio
    import logging
    import warnings
    from typing import Literal, Optional

    from android.broadcast import BroadcastReceiver
    from android.permissions import Permission, request_permissions
    from jnius import cast, java_method

    from bleak._compat import override
    from bleak._compat import timeout as async_timeout
    from bleak.backends.p4android import defs, utils
    from bleak.backends.scanner import (
        AdvertisementData,
        AdvertisementDataCallback,
        BaseBleakScanner,
    )
    from bleak.exc import BleakError

    logger = logging.getLogger(__name__)

    class BleakScannerP4Android(BaseBleakScanner):
        """
        The python-for-android Bleak BLE Scanner.

        Args:
            detection_callback:
                Optional function that will be called each time a device is
                discovered or advertising data has changed.
            service_uuids:
                Optional list of service UUIDs to filter on. Only advertisements
                containing this advertising data will be received. Specifying this
                also enables scanning while the screen is off on Android.
            scanning_mode:
                Set to ``"passive"`` to avoid the ``"active"`` scanning mode.
        """

        __scanner = None

        def __init__(
            self,
            detection_callback: Optional[AdvertisementDataCallback],
            service_uuids: Optional[list[str]],
            scanning_mode: Literal["active", "passive"],
            **kwargs,
        ):
            super().__init__(detection_callback, service_uuids)

            if scanning_mode == "passive":
                self.__scan_mode = defs.ScanSettings.SCAN_MODE_OPPORTUNISTIC
            else:
                self.__scan_mode = defs.ScanSettings.SCAN_MODE_LOW_LATENCY

            self.__javascanner = None
            self.__callback = None

        @override
        async def start(self) -> None:
            if BleakScannerP4Android.__scanner is not None:
                raise BleakError("A BleakScanner is already scanning on this adapter.")

            logger.debug("Starting BTLE scan")

            loop = asyncio.get_running_loop()

            if self.__javascanner is None:
                if self.__callback is None:
                    self.__callback = _PythonScanCallback(self, loop)

                permission_acknowledged = loop.create_future()

                def handle_permissions(permissions, grantResults):
                    if any(grantResults):
                        loop.call_soon_threadsafe(
                            permission_acknowledged.set_result, grantResults
                        )
                    else:
                        loop.call_soon_threadsafe(
                            permission_acknowledged.set_exception(
                                BleakError("User denied access to " + str(permissions))
                            )
                        )

                request_permissions(
                    [
                        Permission.ACCESS_FINE_LOCATION,
                        Permission.ACCESS_COARSE_LOCATION,
                        "android.permission.ACCESS_BACKGROUND_LOCATION",
                    ],
                    handle_permissions,
                )
                await permission_acknowledged

                adapter = defs.BluetoothAdapter.getDefaultAdapter()
                if adapter is None:
                    raise BleakError("Bluetooth is not supported on this hardware platform")
                if adapter.getState() != defs.BluetoothAdapter.STATE_ON:
                    raise BleakError("Bluetooth is not turned on")

                self.__javascanner = adapter.getBluetoothLeScanner()

            BleakScannerP4Android.__scanner = self

            filters = cast("java.util.List", defs.List())
            if self._service_uuids:
                for uuid in self._service_uuids:
                    filters.add(
                        defs.ScanFilterBuilder()
                        .setServiceUuid(defs.ParcelUuid.fromString(uuid))
                        .build()
                    )

            scanfuture = self.__callback.perform_and_wait(
                dispatchApi=self.__javascanner.startScan,
                dispatchParams=(
                    filters,
                    defs.ScanSettingsBuilder()
                    .setScanMode(self.__scan_mode)
                    .setReportDelay(0)
                    .setPhy(defs.ScanSettings.PHY_LE_ALL_SUPPORTED)
                    .setNumOfMatches(defs.ScanSettings.MATCH_NUM_MAX_ADVERTISEMENT)
                    .setMatchMode(defs.ScanSettings.MATCH_MODE_AGGRESSIVE)
                    .setCallbackType(defs.ScanSettings.CALLBACK_TYPE_ALL_MATCHES)
                    .build(),
                    self.__callback.java,
                ),
                resultApi="onScan",
                return_indicates_status=False,
            )
            self.__javascanner.flushPendingScanResults(self.__callback.java)

            try:
                async with async_timeout(0.2):
                    await scanfuture
            except asyncio.exceptions.TimeoutError:
                pass
            except BleakError as bleakerror:
                await self.stop()
                if bleakerror.args != (
                    "onScan",
                    "SCAN_FAILED_APPLICATION_REGISTRATION_FAILED",
                ):
                    raise bleakerror
                else:
                    # there might be a clearer solution to this if android source and vendor
                    # documentation are reviewed for the meaning of the error
                    # https://stackoverflow.com/questions/27516399/solution-for-ble-scans-scan-failed-application-registration-failed
                    warnings.warn(
                        "BT API gave SCAN_FAILED_APPLICATION_REGISTRATION_FAILED.  Resetting adapter."
                    )

                    def handlerWaitingForState(state, stateFuture):
                        def handleAdapterStateChanged(context, intent):
                            adapter_state = intent.getIntExtra(
                                defs.BluetoothAdapter.EXTRA_STATE,
                                defs.BluetoothAdapter.STATE_ERROR,
                            )
                            if adapter_state == defs.BluetoothAdapter.STATE_ERROR:
                                loop.call_soon_threadsafe(
                                    stateOffFuture.set_exception,
                                    BleakError(f"Unexpected adapter state {adapter_state}"),
                                )
                            elif adapter_state == state:
                                loop.call_soon_threadsafe(
                                    stateFuture.set_result, adapter_state
                                )

                        return handleAdapterStateChanged

                    logger.info(
                        "disabling bluetooth adapter to handle SCAN_FAILED_APPLICATION_REGSTRATION_FAILED ..."
                    )
                    stateOffFuture = loop.create_future()
                    receiver = BroadcastReceiver(
                        handlerWaitingForState(
                            defs.BluetoothAdapter.STATE_OFF, stateOffFuture
                        ),
                        actions=[defs.BluetoothAdapter.ACTION_STATE_CHANGED],
                    )
                    receiver.start()
                    try:
                        adapter.disable()
                        await stateOffFuture
                    finally:
                        receiver.stop()

                    logger.info("re-enabling bluetooth adapter ...")
                    stateOnFuture = loop.create_future()
                    receiver = BroadcastReceiver(
                        handlerWaitingForState(
                            defs.BluetoothAdapter.STATE_ON, stateOnFuture
                        ),
                        actions=[defs.BluetoothAdapter.ACTION_STATE_CHANGED],
                    )
                    receiver.start()
                    try:
                        adapter.enable()
                        await stateOnFuture
                    finally:
                        receiver.stop()
                    logger.debug("restarting scan ...")

                    return await self.start()

        @override
        async def stop(self) -> None:
            if self.__javascanner is not None:
                logger.debug("Stopping BTLE scan")
                self.__javascanner.stopScan(self.__callback.java)
                BleakScannerP4Android.__scanner = None
                self.__javascanner = None
            else:
                logger.debug("BTLE scan already stopped")

        def _handle_scan_result(self, result) -> None:
            native_device = result.getDevice()
            record = result.getScanRecord()

            service_uuids = record.getServiceUuids()
            if service_uuids is not None:
                service_uuids = [service_uuid.toString() for service_uuid in service_uuids]

            if not self.is_allowed_uuid(service_uuids):
                return

            manufacturer_data = record.getManufacturerSpecificData()
            manufacturer_data = {
                manufacturer_data.keyAt(index): bytes(manufacturer_data.valueAt(index))
                for index in range(manufacturer_data.size())
            }

            service_data = {
                entry.getKey().toString(): bytes(entry.getValue())
                for entry in record.getServiceData().entrySet()
            }
            tx_power = record.getTxPowerLevel()

            # change "not present" value to None to match other backends
            if tx_power == -2147483648:  # Integer#MIN_VALUE
                tx_power = None

            advertisement = AdvertisementData(
                local_name=record.getDeviceName(),
                manufacturer_data=manufacturer_data,
                service_data=service_data,
                service_uuids=service_uuids,
                tx_power=tx_power,
                rssi=result.getRssi(),
                platform_data=(result,),
            )

            device = self.create_or_update_device(
                native_device.getAddress(),
                native_device.getAddress(),
                native_device.getName(),
                native_device,
                advertisement,
            )

            self.call_detection_callbacks(device, advertisement)

    class _PythonScanCallback(utils.AsyncJavaCallbacks):
        __javainterfaces__ = ["com.github.hbldh.bleak.PythonScanCallback$Interface"]

        def __init__(self, scanner: BleakScannerP4Android, loop: asyncio.AbstractEventLoop):
            super().__init__(loop)
            self._scanner = scanner
            self.java = defs.PythonScanCallback(self)

        def result_state(self, status_str, name, *data):
            self._loop.call_soon_threadsafe(
                self._result_state_unthreadsafe, status_str, name, data
            )

        @java_method("(I)V")
        def onScanFailed(self, errorCode):
            self.result_state(defs.ScanFailed(errorCode).name, "onScan")

        @java_method("(Landroid/bluetooth/le/ScanResult;)V")
        def onScanResult(self, result):
            self._loop.call_soon_threadsafe(self._scanner._handle_scan_result, result)

            if "onScan" not in self.states:
                self.result_state(None, "onScan", result)