add filters module - scanning filters wrappers #25

able/android/dispatcher.py

@@ -97,11 +97,12 @@ def _request_runtime_permissions(self):
     @require_bluetooth_enabled
     @require_runtime_permissions
     def start_scan(self, filters=None, settings=None):
-        if not filters:
-            filters = ArrayList()
+        filters_array = ArrayList()
+        for f in filters:
+            filters_array.add(f.build())
         if not settings:
             settings = ScanSettingsBuilder().build()
-        self._ble.startScan(self.enable_ble_code, filters, settings)
+        self._ble.startScan(self.enable_ble_code, filters_array, settings)
 
     def stop_scan(self):
         self._ble.stopScan()

able/dispatcher.py

@@ -1,9 +1,10 @@
-from typing import Optional
+from typing import List, Optional
 
 from kivy.event import EventDispatcher
 from kivy.logger import Logger
 
 from able import WriteType
+from able.filters import Filter
 from able.queue import BLEQueue, ble_task, ble_task_done
 from able.utils import force_convertible_to_java_array
 
@@ -104,13 +105,15 @@ def set_queue_timeout(self, timeout):
         self.queue_timeout = timeout
         self.queue.set_timeout(timeout)
 
-    def start_scan(self, filters=None, settings=None):
+    def start_scan(self, filters: Optional[List[Filter]]=None, settings=None):
         """Start a scan for devices.
         Ask for runtime permission to access location.
         Start a system activity that allows the user to turn on Bluetooth,
         if Bluetooth is not enabled.
         The status of the scan start are reported with
         :func:`scan_started <on_scan_started>` event.
+
+        :param filters: list of filters to restrict scan results
         """
         pass
 

able/filters.py

@@ -0,0 +1,222 @@
+"""BLE scanning filters,
+wrappers for Java https://developer.android.com/reference/android/bluetooth/le/ScanFilter.Builder
+"""
+from abc import abstractmethod
+from dataclasses import dataclass, field
+from typing import List, Union
+import uuid
+
+from jnius import autoclass
+
+ParcelUuid = autoclass('android.os.ParcelUuid')
+BluetoothAdapter = autoclass('android.bluetooth.BluetoothAdapter')
+ScanFilter = autoclass('android.bluetooth.le.ScanFilter')
+ScanFilterBuilder = autoclass('android.bluetooth.le.ScanFilter$Builder')
+
+
+@dataclass
+class Filter:
+
+    def __post_init__(self):
+        self.filters = [self]
+
+    def __and__(self, other):
+        self.filters.extend(other.filters)
+        return self
+
+    def build(self):
+        builder = ScanFilterBuilder()
+        for scan_filter in self.filters:
+            scan_filter.filter(builder)
+        return builder.build()
+
+    @abstractmethod
+    def filter(self, builder):
+        pass
+
+
+class EmptyFilter(Filter):
+
+    def filter(self, builder):
+        pass
+
+
+@dataclass
+class DeviceAddressFilter(Filter):
+    """Set filter on device address.
+    Uses Java method `ScanFilter.Builder.setDeviceAddress`.
+
+    :param address: Address in the format of "01:02:03:AB:CD:EF"
+
+    >>> DeviceAddressFilter("01:02:03:AB:CD:EF")
+    DeviceAddressFilter(address='01:02:03:AB:CD:EF')
+    """
+    address: str
+
+    def __post_init__(self):
+        super().__post_init__()
+        if not BluetoothAdapter.checkBluetoothAddress(str(self.address)):
+            raise ValueError(f"{self.address} is not a valid Bluetooth address")
+
+    def filter(self, builder):
+        builder.setDeviceAddress(str(self.address))
+
+
+@dataclass
+class DeviceNameFilter(Filter):
+    """Set filter on device name.
+    Uses Java method `ScanFilter.Builder.setDeviceName`.
+
+    :param name: Device name
+    """
+    name: str
+
+    def filter(self, builder):
+        builder.setDeviceName(str(self.name))
+
+
+@dataclass
+class ManufacturerDataFilter(Filter):
+    """Set filter on manufacture data.
+    Uses Java method `ScanFilter.Builder.setManufacturerData`.
+
+    :param id: Manufacturer ID
+    :param data: Manufacturer specific data
+    :param mask: bit mask for partial filtration of the `data`. For any bit in the mask,
+                 set it to 1 if it needs to match the one in manufacturer data,
+                 otherwise set it to 0 to ignore that bit.
+
+    # Filter by just ID, ignoring the data:
+    >>> ManufacturerDataFilter(0x0AD0, [])
+    ManufacturerDataFilter(id=2768, data=[], mask=None)
+
+    >>> ManufacturerDataFilter(0x0AD0, [0x2, 0x15, 0x8d])
+    ManufacturerDataFilter(id=2768, data=[2, 21, 141], mask=None)
+
+    # With mask set to ignore the second data byte:
+    >>> ManufacturerDataFilter(0x0AD0, [0x2, 0, 0x8d], [0xff, 0, 0xff])
+    ManufacturerDataFilter(id=2768, data=[2, 0, 141], mask=[255, 0, 255])
+
+    >>> ManufacturerDataFilter(0x0AD0, [0x2, 21, 0x8d], [0xff])
+    Traceback (most recent call last):
+    ValueError: mask is shorter than the data
+    """
+    id: int
+    data: Union[list, tuple, bytes, bytearray]
+    mask: List[int] = field(default_factory=lambda: None)
+
+    def __post_init__(self):
+        super().__post_init__()
+        if self.mask and len(self.mask) < len(self.data):
+            raise ValueError('mask is shorter than the data')
+
+    def filter(self, builder):
+        if self.mask:
+            builder.setManufacturerData(self.id, self.data, self.mask)
+        else:
+            builder.setManufacturerData(self.id, self.data)
+
+
+@dataclass
+class ServiceDataFilter(Filter):
+    """Set filter on service data.
+    Uses Java method `ScanFilter.Builder.setServiceData`.
+
+    :param uid: UUID of the service in the format of
+                "0000180f-0000-1000-8000-00805f9b34fb"
+    :param data: service data
+    :param mask: bit mask for partial filtration of the `data`. For any bit in the mask,
+                 set it to 1 if it needs to match the one in service data,
+                 otherwise set it to 0 to ignore that bit.
+
+    >>> ServiceDataFilter("0000180f-0000-1000-8000-00805f9b34fb", [])
+    ServiceDataFilter(uid='0000180f-0000-1000-8000-00805f9b34fb', data=[], mask=None)
+
+    # With mask set to ignore the first data byte:
+    >>> ServiceDataFilter("0000180f-0000-1000-8000-00805f9b34fb", [0, 0x11], [0, 0xff])
+    ServiceDataFilter(uid='0000180f-0000-1000-8000-00805f9b34fb', data=[0, 17], mask=[0, 255])
+
+    >>> ServiceDataFilter("0000180f", [])
+    Traceback (most recent call last):
+    ValueError: badly formed hexadecimal UUID string
+
+    >>> ServiceDataFilter("0000180f-0000-1000-8000-00805f9b34fb", [0x12, 0x34], [0xff])
+    Traceback (most recent call last):
+    ValueError: mask is shorter than the data
+    """
+    uid: str
+    data: Union[list, tuple, bytes, bytearray]
+    mask: List[int] = field(default_factory=lambda: None)
+
+    def __post_init__(self):
+        super().__post_init__()
+        # validate UUID value
+        uuid.UUID(self.uid)
+        if self.mask and len(self.mask) < len(self.data):
+            raise ValueError('mask is shorter than the data')
+
+    def filter(self, builder):
+        uid = ParcelUuid.fromString(self.uid)
+        if self.mask:
+            builder.setServiceData(uid, self.data, self.mask)
+        else:
+            builder.setServiceData(uid, self.data)
+
+
+@dataclass
+class ServiceSolicitationFilter(Filter):
+    """Set filter on service solicitation uuid.
+    Uses Java method `ScanFilter.Builder.setServiceSolicitation`.
+
+    :param uid: UUID of the service in the format of
+                "0000180f-0000-1000-8000-00805f9b34fb"
+    """
+    uid: str
+
+    def filter(self, builder):
+        uid = ParcelUuid.fromString(self.uid)
+        builder.setServiceSolicitation(uid)
+
+
+@dataclass
+class ServiceUUIDFilter(Filter):
+    """Set filter on service uuid.
+    Uses Java method `ScanFilter.Builder.setServiceUuid`.
+
+    :param uid: UUID of the service in the format of
+                "0000180f-0000-1000-8000-00805f9b34fb"
+    :mask: bit mask for partial filtration of the UUID, in the format of
+           "ffffffff-0000-0000-0000-ffffffffffff". Set any bit in the mask
+           to 1 to indicate a match is needed for the bit in `uid`,
+           and 0 to ignore that bit.
+
+    >>> ServiceUUIDFilter('16fe0d00-c111-11e3-b8c8-0002a5d5c51b')
+    ServiceUUIDFilter(uid='16fe0d00-c111-11e3-b8c8-0002a5d5c51b', mask=None)
+
+    >>> ServiceUUIDFilter(
+    ... '16fe0d00-c111-11e3-b8c8-0002a5d5c51b',
+    ... 'ffffffff-0000-0000-0000-000000000000'
+    ... )  #doctest: +ELLIPSIS
+    ServiceUUIDFilter(uid='16fe0d00-...', mask='ffffffff-...')
+
+    >>> ServiceUUIDFilter('123')
+    Traceback (most recent call last):
+    ValueError: badly formed hexadecimal UUID string
+    """
+    uid: str
+    mask: str = None
+
+    def __post_init__(self):
+        super().__post_init__()
+        # validate UUID values
+        uuid.UUID(self.uid)
+        if self.mask:
+            uuid.UUID(self.mask)
+
+    def filter(self, builder):
+        uid = ParcelUuid.fromString(self.uid)
+        if self.mask:
+            mask = ParcelUuid.fromString(self.mask)
+            builder.setServiceUuid(uid, mask)
+        else:
+            builder.setServiceUuid(uid)

tests/notebooks/test_scan_filters.expected

@@ -1,8 +1,8 @@
 [[setup]]
 = Setup
 
-[[test-device-is-found-with-filter-by-name]]
-= Test device is found with filter by name
+[[test-device-is-found-with-scan-filters-set]]
+= Test device is found with scan filters set
 
 
 ----
@@ -16,3 +16,57 @@
 ----
 []
 ----
+
+[[test-scan-filter-mathes]]
+= Test scan filter mathes
+
+
+----
+EmptyFilter() True
+EmptyFilter() True
+EmptyFilter() True
+----
+
+
+----
+DeviceAddressFilter(address='AA:AA:AA:AA:AA:AA') True
+DeviceAddressFilter(address='AA:AA:AA:AA:AA:AB') False
+AA is not a valid Bluetooth address
+----
+
+
+----
+DeviceNameFilter(name='KivyBLETest') True
+DeviceNameFilter(name='KivyBLETes') False
+----
+
+
+----
+ManufacturerDataFilter(id=76, data=[], mask=None) False
+ManufacturerDataFilter(id=76, data=[], mask=None) True
+ManufacturerDataFilter(id=76, data=[2, 21, 141, 166, 131, 214, 229], mask=None) True
+ManufacturerDataFilter(id=76, data=[2, 21, 141, 166, 131, 214, 170], mask=None) False
+ManufacturerDataFilter(id=76, data=[2, 21, 141, 166, 131, 214, 170], mask=[255, 255, 255, 255, 255, 255, 0]) True
+ManufacturerDataFilter(id=76, data=[2, 21, 141, 166, 131, 214, 170], mask=[255, 255, 255, 255, 255, 255, 255]) False
+ManufacturerDataFilter(id=76, data=[2, 0, 141, 166, 131], mask=[255, 0, 255, 255, 255]) True
+ManufacturerDataFilter(id=76, data=b'\x02\x15', mask=None) True
+ManufacturerDataFilter(id=76, data=b'\x02\x16', mask=None) False
+----
+
+
+----
+ServiceDataFilter(uid='0000180f-0000-1000-8000-00805f9b34fb', data=[], mask=None) True
+ServiceDataFilter(uid='0000180f-0000-1000-8000-00805f9b34fc', data=[], mask=None) False
+ServiceDataFilter(uid='0000180f-0000-1000-8000-00805f9b34fb', data=[34], mask=None) True
+ServiceDataFilter(uid='0000180f-0000-1000-8000-00805f9b34fb', data=[33], mask=None) False
+ServiceDataFilter(uid='0000180f-0000-1000-8000-00805f9b34fb', data=[33], mask=[240]) True
+ServiceDataFilter(uid='0000180f-0000-1000-8000-00805f9b34fb', data=[33], mask=[15]) False
+----
+
+
+----
+ServiceUUIDFilter(uid='16fe0d00-c111-11e3-b8c8-0002a5d5c51B', mask=None) True
+ServiceUUIDFilter(uid='16fe0d00-c111-11e3-b8c8-0002a5d5c51C', mask=None) False
+ServiceUUIDFilter(uid='16fe0d00-c111-11e3-b8c8-0002a5d5c51C', mask='ffffffff-ffff-ffff-ffff-ffffffffffff') False
+ServiceUUIDFilter(uid='16fe0d00-c111-11e3-b8c8-0002a5d5c51C', mask='ffffffff-ffff-ffff-ffff-fffffffffff0') True
+----