Source code for toio.scanner.ble

# -*- coding: utf-8 -*-
# ************************************************************
#
#     ble.py
#
#     Copyright 2022 Sony Interactive Entertainment Inc.
#
# ************************************************************
"""
BLE scanner (including specific functions for each supported OS)

Scan toio Core Cubes with internal BLE interface.
"""

import functools
import platform

from typing_extensions import Any, List, NamedTuple, Optional, Set

from ..device_interface import (
    DEFAULT_SCAN_TIMEOUT,
    CubeInfo,
    ScannerInterface,
    ScanStrategy,
    SortKey,
)
from ..device_interface.ble import BaseBleScanner
from ..logger import get_toio_logger

logger = get_toio_logger(__name__)




class PlatformParam(NamedTuple):
    platform: str = "UNKNOWN"
    unsupported_result: Any = None





def async_platform_specified(param: PlatformParam = PlatformParam("UNKNOWN", None)):
    def platform_specified_wrapper(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            if platform.system() == param.platform:
                result = await func(*args, **kwargs)
                return result
            else:
                logger.warning("platform '%s' is not supported", platform.system())
                return param.unsupported_result

        return wrapper

    return platform_specified_wrapper





class UniversalBleScanner(ScannerInterface):
    async def _scan(
        self,
        num: Optional[int] = None,
        cube_id: Optional[Set[str]] = None,
        address: Optional[Set[str]] = None,
        sort: SortKey = None,
        timeout: float = DEFAULT_SCAN_TIMEOUT,
        strategy: ScanStrategy = "best",
    ) -> List[CubeInfo]:
        scanner = BaseBleScanner()
        return await scanner._scan(
            num=num,
            cube_id=cube_id,
            address=address,
            sort=sort,
            timeout=timeout,
            strategy=strategy,
        )



    async def scan(  # type: ignore
        self,
        num: int,
        sort: SortKey = "rssi",
        timeout: float = DEFAULT_SCAN_TIMEOUT,
        strategy: ScanStrategy = "best",
    ) -> List[CubeInfo]:
        """Scan the specified number of toio Core Cubes.

        With strategy "best", the scan collects cubes until timeout and returns
        the top results after sorting. With strategy "quick", the scan returns
        as soon as the specified number of cubes are found.

        Args:
            num (int): Number of cubes to be found.
            sort (SortKey, optional): Key to sort results. Defaults to "rssi".
            timeout (float, optional): Scan timeout. Defaults to DEFAULT_SCAN_TIMEOUT.
            strategy (ScanStrategy, optional): Scan strategy. Defaults to "best".

        Returns:
            List[CubeInfo]: List of found cubes.
        """
        return await self._scan(num=num, sort=sort, timeout=timeout, strategy=strategy)




    async def scan_best(
        self, num: int, sort: SortKey = "rssi", timeout: float = DEFAULT_SCAN_TIMEOUT
    ) -> List[CubeInfo]:
        """Scan cubes until timeout and return the best candidates after sorting."""
        return await self.scan(num=num, sort=sort, timeout=timeout, strategy="best")




    async def scan_quick(
        self, num: int, sort: SortKey = "rssi", timeout: float = DEFAULT_SCAN_TIMEOUT
    ) -> List[CubeInfo]:
        """Scan cubes and return as soon as the specified number are found."""
        return await self.scan(num=num, sort=sort, timeout=timeout, strategy="quick")




    async def scan_with_id(
        self,
        cube_id: Set[str],
        sort: SortKey = "rssi",
        timeout: float = DEFAULT_SCAN_TIMEOUT,
    ) -> List[CubeInfo]:
        """Scan toio Core Cubes with specified id.

        The scan is terminated by a timeout.
        In the case of a timeout, the number of elements in the returned list
        is the number of cubes found at the time of the timeout.

        Args:
            cube_id (set[str]): Set of cube id to be found.
            sort (SortKey, optional): Key to sort results. Defaults to "rssi".
            timeout (float, optional): Scan timeout. Defaults to DEFAULT_SCAN_TIMEOUT.

        Returns:
            List[CubeInfo]: List of found cubes.
        """
        return await self._scan(cube_id=cube_id, sort=sort, timeout=timeout)




    async def scan_with_address(
        self,
        address: Set[str],
        sort: SortKey = "rssi",
        timeout: float = DEFAULT_SCAN_TIMEOUT,
    ) -> List[CubeInfo]:
        """Scan toio Core Cubes with specified BLE address.

        The scan is terminated by a timeout.
        In the case of a timeout, the number of elements in the returned list
        is the number of cubes found at the time of the timeout.

        Args:
            address (set[str]): Set of BLE address to be found.
            sort (SortKey, optional): Key to sort results. Defaults to "rssi".
            timeout (float, optional): Scan timeout. Defaults to DEFAULT_SCAN_TIMEOUT.

        Returns:
            List[CubeInfo]: List of found cubes.
        """
        return await self._scan(address=address, sort=sort, timeout=timeout)




    @async_platform_specified(PlatformParam("Windows", []))
    async def scan_registered_cubes(
        self, num: int, sort: SortKey = "rssi", timeout: float = DEFAULT_SCAN_TIMEOUT
    ) -> List[CubeInfo]:
        """Scan toio Core Cubes registered with Windows

        This function only works on Windows platform.
        On the other platform, this function always returns empty list.

        Even if `num` is greater than the number of registered cubes,
        the maximum size of the list returned by this function is the number of registered cubes.

        Args:
            num (int): Number of cubes to be found.
            sort (SortKey, optional): Key to sort results. Defaults to "rssi".
            timeout (float, optional): Scan timeout. Defaults to DEFAULT_SCAN_TIMEOUT.

        Returns:
            List[CubeInfo]: List of found cubes.
        """
        from ..scanner.platform.windows_ble import get_registered_cubes

        registered_cubes = get_registered_cubes()
        addresses = {x.address.upper() for x in registered_cubes}
        found = await self.scan_with_address(addresses, sort, timeout)
        if len(found) > num:
            return found[:num]
        else:
            return found




    @async_platform_specified(PlatformParam("Windows", []))
    async def scan_registered_cubes_with_id(
        self,
        cube_id: Set[str],
        sort: SortKey = "rssi",
        timeout: float = DEFAULT_SCAN_TIMEOUT,
    ) -> List[CubeInfo]:
        """Scan toio Core Cube specified by the cube_id registered with Windows

        This function only works on Windows platform.
        On the other platform, this function always returns empty list.

        Args:
            cube_id (set[str]): Set of cube id to be found.
            sort (SortKey, optional): Key to sort results. Defaults to "rssi".
            timeout (float, optional): Scan timeout. Defaults to DEFAULT_SCAN_TIMEOUT.

        Returns:
            List[CubeInfo]: List of found cubes.
        """
        from ..scanner.platform.windows_ble import get_registered_cubes

        registered_cubes = get_registered_cubes()
        address_list = set([])
        for cube in registered_cubes:
            for id_str in cube_id:
                if id_str in cube.name:
                    address_list.add(cube.address.upper())
        if len(address_list) == 0:
            return []
        else:
            found = await self.scan_with_address(address_list, sort, timeout)
            return found