| # Copyright 2022 Google LLC |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # https://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| import asyncio |
| import bumble.device |
| import grpc |
| import grpc.aio |
| import logging |
| import struct |
| |
| from . import utils |
| from bumble.core import ( |
| BT_BR_EDR_TRANSPORT, |
| BT_LE_TRANSPORT, |
| BT_PERIPHERAL_ROLE, |
| UUID, |
| AdvertisingData, |
| ConnectionError, |
| ) |
| from bumble.device import ( |
| DEVICE_DEFAULT_SCAN_INTERVAL, |
| DEVICE_DEFAULT_SCAN_WINDOW, |
| Advertisement, |
| AdvertisingType, |
| Device, |
| ) |
| from bumble.gatt import Service |
| from bumble.hci import ( |
| HCI_CONNECTION_ALREADY_EXISTS_ERROR, |
| HCI_PAGE_TIMEOUT_ERROR, |
| HCI_REMOTE_USER_TERMINATED_CONNECTION_ERROR, |
| Address, |
| ) |
| from google.protobuf import any_pb2, empty_pb2 # pytype: disable=pyi-error |
| from pandora.host_grpc_aio import HostServicer |
| from pandora.host_pb2 import ( |
| NOT_CONNECTABLE, |
| NOT_DISCOVERABLE, |
| PRIMARY_1M, |
| PRIMARY_CODED, |
| SECONDARY_1M, |
| SECONDARY_2M, |
| SECONDARY_CODED, |
| SECONDARY_NONE, |
| AdvertiseRequest, |
| AdvertiseResponse, |
| Connection, |
| ConnectLERequest, |
| ConnectLEResponse, |
| ConnectRequest, |
| ConnectResponse, |
| DataTypes, |
| DisconnectRequest, |
| InquiryResponse, |
| PrimaryPhy, |
| ReadLocalAddressResponse, |
| ScanningResponse, |
| ScanRequest, |
| SecondaryPhy, |
| SetConnectabilityModeRequest, |
| SetDiscoverabilityModeRequest, |
| WaitConnectionRequest, |
| WaitConnectionResponse, |
| WaitDisconnectionRequest, |
| ) |
| from typing import AsyncGenerator, Dict, List, Optional, Set, Tuple, cast |
| |
| PRIMARY_PHY_MAP: Dict[int, PrimaryPhy] = { |
| # Default value reported by Bumble for legacy Advertising reports. |
| # FIXME(uael): `None` might be a better value, but Bumble need to change accordingly. |
| 0: PRIMARY_1M, |
| 1: PRIMARY_1M, |
| 3: PRIMARY_CODED, |
| } |
| |
| SECONDARY_PHY_MAP: Dict[int, SecondaryPhy] = { |
| 0: SECONDARY_NONE, |
| 1: SECONDARY_1M, |
| 2: SECONDARY_2M, |
| 3: SECONDARY_CODED, |
| } |
| |
| |
| class HostService(HostServicer): |
| grpc_server: grpc.aio.Server |
| device: Device |
| waited_connections: Set[int] |
| |
| def __init__(self, grpc_server: grpc.aio.Server, device: Device) -> None: |
| super().__init__() |
| self.log = utils.BumbleServerLoggerAdapter(logging.getLogger(), {'service_name': 'Host', 'device': device}) |
| self.grpc_server = grpc_server |
| self.device = device |
| self.waited_connections = set() |
| |
| @utils.rpc |
| async def FactoryReset(self, request: empty_pb2.Empty, context: grpc.ServicerContext) -> empty_pb2.Empty: |
| self.log.info('FactoryReset') |
| |
| # delete all bonds |
| if self.device.keystore is not None: |
| await self.device.keystore.delete_all() |
| |
| # trigger gRCP server stop then return |
| asyncio.create_task(self.grpc_server.stop(None)) |
| return empty_pb2.Empty() |
| |
| @utils.rpc |
| async def Reset(self, request: empty_pb2.Empty, context: grpc.ServicerContext) -> empty_pb2.Empty: |
| self.log.info('Reset') |
| |
| # clear service. |
| self.waited_connections.clear() |
| |
| # (re) power device on |
| await self.device.power_on() |
| return empty_pb2.Empty() |
| |
| @utils.rpc |
| async def ReadLocalAddress( |
| self, request: empty_pb2.Empty, context: grpc.ServicerContext |
| ) -> ReadLocalAddressResponse: |
| self.log.info('ReadLocalAddress') |
| return ReadLocalAddressResponse(address=bytes(reversed(bytes(self.device.public_address)))) |
| |
| @utils.rpc |
| async def Connect(self, request: ConnectRequest, context: grpc.ServicerContext) -> ConnectResponse: |
| # Need to reverse bytes order since Bumble Address is using MSB. |
| address = Address(bytes(reversed(request.address)), address_type=Address.PUBLIC_DEVICE_ADDRESS) |
| self.log.info(f"Connect to {address}") |
| |
| try: |
| connection = await self.device.connect(address, transport=BT_BR_EDR_TRANSPORT) |
| except ConnectionError as e: |
| if e.error_code == HCI_PAGE_TIMEOUT_ERROR: |
| self.log.warning(f"Peer not found: {e}") |
| return ConnectResponse(peer_not_found=empty_pb2.Empty()) |
| if e.error_code == HCI_CONNECTION_ALREADY_EXISTS_ERROR: |
| self.log.warning(f"Connection already exists: {e}") |
| return ConnectResponse(connection_already_exists=empty_pb2.Empty()) |
| raise e |
| |
| self.log.info(f"Connect to {address} done (handle={connection.handle})") |
| |
| cookie = any_pb2.Any(value=connection.handle.to_bytes(4, 'big')) |
| return ConnectResponse(connection=Connection(cookie=cookie)) |
| |
| @utils.rpc |
| async def WaitConnection( |
| self, request: WaitConnectionRequest, context: grpc.ServicerContext |
| ) -> WaitConnectionResponse: |
| if not request.address: |
| raise ValueError('Request address field must be set') |
| |
| # Need to reverse bytes order since Bumble Address is using MSB. |
| address = Address(bytes(reversed(request.address)), address_type=Address.PUBLIC_DEVICE_ADDRESS) |
| if address in (Address.NIL, Address.ANY): |
| raise ValueError('Invalid address') |
| |
| self.log.info(f"WaitConnection from {address}...") |
| |
| connection = self.device.find_connection_by_bd_addr(address, transport=BT_BR_EDR_TRANSPORT) |
| if connection and id(connection) in self.waited_connections: |
| # this connection was already returned: wait for a new one. |
| connection = None |
| |
| if not connection: |
| connection = await self.device.accept(address) |
| |
| # save connection has waited and respond. |
| self.waited_connections.add(id(connection)) |
| |
| self.log.info(f"WaitConnection from {address} done (handle={connection.handle})") |
| |
| cookie = any_pb2.Any(value=connection.handle.to_bytes(4, 'big')) |
| return WaitConnectionResponse(connection=Connection(cookie=cookie)) |
| |
| @utils.rpc |
| async def ConnectLE(self, request: ConnectLERequest, context: grpc.ServicerContext) -> ConnectLEResponse: |
| address = utils.address_from_request(request, request.WhichOneof("address")) |
| if address in (Address.NIL, Address.ANY): |
| raise ValueError('Invalid address') |
| |
| self.log.info(f"ConnectLE to {address}...") |
| |
| try: |
| connection = await self.device.connect( |
| address, transport=BT_LE_TRANSPORT, own_address_type=request.own_address_type |
| ) |
| except ConnectionError as e: |
| if e.error_code == HCI_PAGE_TIMEOUT_ERROR: |
| self.log.warning(f"Peer not found: {e}") |
| return ConnectLEResponse(peer_not_found=empty_pb2.Empty()) |
| if e.error_code == HCI_CONNECTION_ALREADY_EXISTS_ERROR: |
| self.log.warning(f"Connection already exists: {e}") |
| return ConnectLEResponse(connection_already_exists=empty_pb2.Empty()) |
| raise e |
| |
| self.log.info(f"ConnectLE to {address} done (handle={connection.handle})") |
| |
| cookie = any_pb2.Any(value=connection.handle.to_bytes(4, 'big')) |
| return ConnectLEResponse(connection=Connection(cookie=cookie)) |
| |
| @utils.rpc |
| async def Disconnect(self, request: DisconnectRequest, context: grpc.ServicerContext) -> empty_pb2.Empty: |
| connection_handle = int.from_bytes(request.connection.cookie.value, 'big') |
| self.log.info(f"Disconnect: {connection_handle}") |
| |
| self.log.info("Disconnecting...") |
| if connection := self.device.lookup_connection(connection_handle): |
| await connection.disconnect(HCI_REMOTE_USER_TERMINATED_CONNECTION_ERROR) |
| self.log.info("Disconnected") |
| |
| return empty_pb2.Empty() |
| |
| @utils.rpc |
| async def WaitDisconnection( |
| self, request: WaitDisconnectionRequest, context: grpc.ServicerContext |
| ) -> empty_pb2.Empty: |
| connection_handle = int.from_bytes(request.connection.cookie.value, 'big') |
| self.log.info(f"WaitDisconnection: {connection_handle}") |
| |
| if connection := self.device.lookup_connection(connection_handle): |
| disconnection_future: asyncio.Future[None] = asyncio.get_running_loop().create_future() |
| |
| def on_disconnection(_: None) -> None: |
| disconnection_future.set_result(None) |
| |
| connection.on('disconnection', on_disconnection) |
| try: |
| await disconnection_future |
| self.log.info("Disconnected") |
| finally: |
| connection.remove_listener('disconnection', on_disconnection) # type: ignore |
| |
| return empty_pb2.Empty() |
| |
| @utils.rpc |
| async def Advertise( |
| self, request: AdvertiseRequest, context: grpc.ServicerContext |
| ) -> AsyncGenerator[AdvertiseResponse, None]: |
| if not request.legacy: |
| raise NotImplementedError("TODO: add support for extended advertising in Bumble") |
| if request.interval: |
| raise NotImplementedError("TODO: add support for `request.interval`") |
| if request.interval_range: |
| raise NotImplementedError("TODO: add support for `request.interval_range`") |
| if request.primary_phy: |
| raise NotImplementedError("TODO: add support for `request.primary_phy`") |
| if request.secondary_phy: |
| raise NotImplementedError("TODO: add support for `request.secondary_phy`") |
| |
| if self.device.is_advertising: |
| raise NotImplementedError('TODO: add support for advertising sets') |
| |
| if data := request.data: |
| self.device.advertising_data = bytes(self.unpack_data_types(data)) |
| |
| if scan_response_data := request.scan_response_data: |
| self.device.scan_response_data = bytes(self.unpack_data_types(scan_response_data)) |
| scannable = True |
| else: |
| scannable = False |
| |
| # Retrieve services data |
| for service in self.device.gatt_server.attributes: |
| if isinstance(service, Service) and (service_data := service.get_advertising_data()): |
| service_uuid = service.uuid.to_hex_str() |
| if ( |
| service_uuid in request.data.incomplete_service_class_uuids16 |
| or service_uuid in request.data.complete_service_class_uuids16 |
| or service_uuid in request.data.incomplete_service_class_uuids32 |
| or service_uuid in request.data.complete_service_class_uuids32 |
| or service_uuid in request.data.incomplete_service_class_uuids128 |
| or service_uuid in request.data.complete_service_class_uuids128 |
| ): |
| self.device.advertising_data += service_data |
| if ( |
| service_uuid in scan_response_data.incomplete_service_class_uuids16 |
| or service_uuid in scan_response_data.complete_service_class_uuids16 |
| or service_uuid in scan_response_data.incomplete_service_class_uuids32 |
| or service_uuid in scan_response_data.complete_service_class_uuids32 |
| or service_uuid in scan_response_data.incomplete_service_class_uuids128 |
| or service_uuid in scan_response_data.complete_service_class_uuids128 |
| ): |
| self.device.scan_response_data += service_data |
| |
| target = None |
| if request.connectable and scannable: |
| advertising_type = AdvertisingType.UNDIRECTED_CONNECTABLE_SCANNABLE |
| elif scannable: |
| advertising_type = AdvertisingType.UNDIRECTED_SCANNABLE |
| else: |
| advertising_type = AdvertisingType.UNDIRECTED |
| else: |
| target = None |
| advertising_type = AdvertisingType.UNDIRECTED |
| |
| if request.target: |
| # Need to reverse bytes order since Bumble Address is using MSB. |
| target_bytes = bytes(reversed(request.target)) |
| if request.target_variant() == "public": |
| target = Address(target_bytes, Address.PUBLIC_DEVICE_ADDRESS) |
| advertising_type = AdvertisingType.DIRECTED_CONNECTABLE_HIGH_DUTY # FIXME: HIGH_DUTY ? |
| else: |
| target = Address(target_bytes, Address.RANDOM_DEVICE_ADDRESS) |
| advertising_type = AdvertisingType.DIRECTED_CONNECTABLE_HIGH_DUTY # FIXME: HIGH_DUTY ? |
| |
| if request.connectable: |
| |
| def on_connection(connection: bumble.device.Connection) -> None: |
| if connection.transport == BT_LE_TRANSPORT and connection.role == BT_PERIPHERAL_ROLE: |
| pending_connection.set_result(connection) |
| |
| self.device.on('connection', on_connection) |
| |
| try: |
| while True: |
| if not self.device.is_advertising: |
| self.log.info('Advertise') |
| await self.device.start_advertising( |
| target=target, advertising_type=advertising_type, own_address_type=request.own_address_type |
| ) |
| |
| if not request.connectable: |
| await asyncio.sleep(1) |
| continue |
| |
| pending_connection: asyncio.Future[ |
| bumble.device.Connection |
| ] = asyncio.get_running_loop().create_future() |
| |
| self.log.info('Wait for LE connection...') |
| connection = await pending_connection |
| |
| self.log.info(f"Advertise: Connected to {connection.peer_address} (handle={connection.handle})") |
| |
| cookie = any_pb2.Any(value=connection.handle.to_bytes(4, 'big')) |
| yield AdvertiseResponse(connection=Connection(cookie=cookie)) |
| |
| # wait a small delay before restarting the advertisement. |
| await asyncio.sleep(1) |
| finally: |
| if request.connectable: |
| self.device.remove_listener('connection', on_connection) # type: ignore |
| |
| try: |
| self.log.info('Stop advertising') |
| await self.device.abort_on('flush', self.device.stop_advertising()) |
| except: |
| pass |
| |
| @utils.rpc |
| async def Scan( |
| self, request: ScanRequest, context: grpc.ServicerContext |
| ) -> AsyncGenerator[ScanningResponse, None]: |
| # TODO: modify `start_scanning` to accept floats instead of int for ms values |
| if request.phys: |
| raise NotImplementedError("TODO: add support for `request.phys`") |
| |
| self.log.info('Scan') |
| |
| scan_queue: asyncio.Queue[Advertisement] = asyncio.Queue() |
| handler = self.device.on('advertisement', scan_queue.put_nowait) |
| await self.device.start_scanning( |
| legacy=request.legacy, |
| active=not request.passive, |
| own_address_type=request.own_address_type, |
| scan_interval=int(request.interval) if request.interval else DEVICE_DEFAULT_SCAN_INTERVAL, |
| scan_window=int(request.window) if request.window else DEVICE_DEFAULT_SCAN_WINDOW, |
| ) |
| |
| try: |
| # TODO: add support for `direct_address` in Bumble |
| # TODO: add support for `periodic_advertising_interval` in Bumble |
| while adv := await scan_queue.get(): |
| sr = ScanningResponse( |
| legacy=adv.is_legacy, |
| connectable=adv.is_connectable, |
| scannable=adv.is_scannable, |
| truncated=adv.is_truncated, |
| sid=adv.sid, |
| primary_phy=PRIMARY_PHY_MAP[adv.primary_phy], |
| secondary_phy=SECONDARY_PHY_MAP[adv.secondary_phy], |
| tx_power=adv.tx_power, |
| rssi=adv.rssi, |
| data=self.pack_data_types(adv.data), |
| ) |
| |
| if adv.address.address_type == Address.PUBLIC_DEVICE_ADDRESS: |
| sr.public = bytes(reversed(bytes(adv.address))) |
| elif adv.address.address_type == Address.RANDOM_DEVICE_ADDRESS: |
| sr.random = bytes(reversed(bytes(adv.address))) |
| elif adv.address.address_type == Address.PUBLIC_IDENTITY_ADDRESS: |
| sr.public_identity = bytes(reversed(bytes(adv.address))) |
| else: |
| sr.random_static_identity = bytes(reversed(bytes(adv.address))) |
| |
| yield sr |
| |
| finally: |
| self.device.remove_listener('advertisement', handler) # type: ignore |
| try: |
| self.log.info('Stop scanning') |
| await self.device.abort_on('flush', self.device.stop_scanning()) |
| except: |
| pass |
| |
| @utils.rpc |
| async def Inquiry( |
| self, request: empty_pb2.Empty, context: grpc.ServicerContext |
| ) -> AsyncGenerator[InquiryResponse, None]: |
| self.log.info('Inquiry') |
| |
| inquiry_queue: asyncio.Queue[Optional[Tuple[Address, int, AdvertisingData, int]]] = asyncio.Queue() |
| complete_handler = self.device.on('inquiry_complete', lambda: inquiry_queue.put_nowait(None)) |
| result_handler = self.device.on( # type: ignore |
| 'inquiry_result', |
| lambda address, class_of_device, eir_data, rssi: inquiry_queue.put_nowait( # type: ignore |
| (address, class_of_device, eir_data, rssi) # type: ignore |
| ), |
| ) |
| |
| await self.device.start_discovery(auto_restart=False) |
| try: |
| while inquiry_result := await inquiry_queue.get(): |
| (address, class_of_device, eir_data, rssi) = inquiry_result |
| # FIXME: if needed, add support for `page_scan_repetition_mode` and `clock_offset` in Bumble |
| yield InquiryResponse( |
| address=bytes(reversed(bytes(address))), |
| class_of_device=class_of_device, |
| rssi=rssi, |
| data=self.pack_data_types(eir_data), |
| ) |
| |
| finally: |
| self.device.remove_listener('inquiry_complete', complete_handler) # type: ignore |
| self.device.remove_listener('inquiry_result', result_handler) # type: ignore |
| try: |
| self.log.info('Stop inquiry') |
| await self.device.abort_on('flush', self.device.stop_discovery()) |
| except: |
| pass |
| |
| @utils.rpc |
| async def SetDiscoverabilityMode( |
| self, request: SetDiscoverabilityModeRequest, context: grpc.ServicerContext |
| ) -> empty_pb2.Empty: |
| self.log.info("SetDiscoverabilityMode") |
| await self.device.set_discoverable(request.mode != NOT_DISCOVERABLE) |
| return empty_pb2.Empty() |
| |
| @utils.rpc |
| async def SetConnectabilityMode( |
| self, request: SetConnectabilityModeRequest, context: grpc.ServicerContext |
| ) -> empty_pb2.Empty: |
| self.log.info("SetConnectabilityMode") |
| await self.device.set_connectable(request.mode != NOT_CONNECTABLE) |
| return empty_pb2.Empty() |
| |
| def unpack_data_types(self, dt: DataTypes) -> AdvertisingData: |
| ad_structures: List[Tuple[int, bytes]] = [] |
| |
| uuids: List[str] |
| datas: Dict[str, bytes] |
| |
| def uuid128_from_str(uuid: str) -> bytes: |
| """Decode a 128-bit uuid encoded as XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX |
| to byte format.""" |
| return bytes(reversed(bytes.fromhex(uuid.replace('-', '')))) |
| |
| def uuid32_from_str(uuid: str) -> bytes: |
| """Decode a 32-bit uuid encoded as XXXXXXXX to byte format.""" |
| return bytes(reversed(bytes.fromhex(uuid))) |
| |
| def uuid16_from_str(uuid: str) -> bytes: |
| """Decode a 16-bit uuid encoded as XXXX to byte format.""" |
| return bytes(reversed(bytes.fromhex(uuid))) |
| |
| if uuids := dt.incomplete_service_class_uuids16: |
| ad_structures.append( |
| ( |
| AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS, |
| b''.join([uuid16_from_str(uuid) for uuid in uuids]), |
| ) |
| ) |
| if uuids := dt.complete_service_class_uuids16: |
| ad_structures.append( |
| ( |
| AdvertisingData.COMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS, |
| b''.join([uuid16_from_str(uuid) for uuid in uuids]), |
| ) |
| ) |
| if uuids := dt.incomplete_service_class_uuids32: |
| ad_structures.append( |
| ( |
| AdvertisingData.INCOMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS, |
| b''.join([uuid32_from_str(uuid) for uuid in uuids]), |
| ) |
| ) |
| if uuids := dt.complete_service_class_uuids32: |
| ad_structures.append( |
| ( |
| AdvertisingData.COMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS, |
| b''.join([uuid32_from_str(uuid) for uuid in uuids]), |
| ) |
| ) |
| if uuids := dt.incomplete_service_class_uuids128: |
| ad_structures.append( |
| ( |
| AdvertisingData.INCOMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS, |
| b''.join([uuid128_from_str(uuid) for uuid in uuids]), |
| ) |
| ) |
| if uuids := dt.complete_service_class_uuids128: |
| ad_structures.append( |
| ( |
| AdvertisingData.COMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS, |
| b''.join([uuid128_from_str(uuid) for uuid in uuids]), |
| ) |
| ) |
| if dt.HasField('include_shortened_local_name'): |
| ad_structures.append((AdvertisingData.SHORTENED_LOCAL_NAME, bytes(self.device.name[:8], 'utf-8'))) |
| elif dt.shortened_local_name: |
| ad_structures.append((AdvertisingData.SHORTENED_LOCAL_NAME, bytes(dt.shortened_local_name, 'utf-8'))) |
| if dt.HasField('include_complete_local_name'): |
| ad_structures.append((AdvertisingData.COMPLETE_LOCAL_NAME, bytes(self.device.name, 'utf-8'))) |
| elif dt.complete_local_name: |
| ad_structures.append((AdvertisingData.COMPLETE_LOCAL_NAME, bytes(dt.complete_local_name, 'utf-8'))) |
| if dt.HasField('include_tx_power_level'): |
| raise ValueError('unsupported data type') |
| elif dt.tx_power_level: |
| ad_structures.append((AdvertisingData.TX_POWER_LEVEL, bytes(struct.pack('<I', dt.tx_power_level)[:1]))) |
| if dt.HasField('include_class_of_device'): |
| ad_structures.append( |
| (AdvertisingData.CLASS_OF_DEVICE, bytes(struct.pack('<I', self.device.class_of_device)[:-1])) |
| ) |
| elif dt.class_of_device: |
| ad_structures.append((AdvertisingData.CLASS_OF_DEVICE, bytes(struct.pack('<I', dt.class_of_device)[:-1]))) |
| if dt.peripheral_connection_interval_min: |
| ad_structures.append( |
| ( |
| AdvertisingData.PERIPHERAL_CONNECTION_INTERVAL_RANGE, |
| bytes( |
| [ |
| *struct.pack('<H', dt.peripheral_connection_interval_min), |
| *struct.pack( |
| '<H', |
| dt.peripheral_connection_interval_max |
| if dt.peripheral_connection_interval_max |
| else dt.peripheral_connection_interval_min, |
| ), |
| ] |
| ), |
| ) |
| ) |
| if uuids := dt.service_solicitation_uuids16: |
| ad_structures.append( |
| ( |
| AdvertisingData.LIST_OF_16_BIT_SERVICE_SOLICITATION_UUIDS, |
| b''.join([uuid16_from_str(uuid) for uuid in uuids]), |
| ) |
| ) |
| if uuids := dt.service_solicitation_uuids32: |
| ad_structures.append( |
| ( |
| AdvertisingData.LIST_OF_32_BIT_SERVICE_SOLICITATION_UUIDS, |
| b''.join([uuid32_from_str(uuid) for uuid in uuids]), |
| ) |
| ) |
| if uuids := dt.service_solicitation_uuids128: |
| ad_structures.append( |
| ( |
| AdvertisingData.LIST_OF_128_BIT_SERVICE_SOLICITATION_UUIDS, |
| b''.join([uuid128_from_str(uuid) for uuid in uuids]), |
| ) |
| ) |
| if datas := dt.service_data_uuid16: |
| ad_structures.extend( |
| [ |
| (AdvertisingData.SERVICE_DATA_16_BIT_UUID, uuid16_from_str(uuid) + data) |
| for uuid, data in datas.items() |
| ] |
| ) |
| if datas := dt.service_data_uuid32: |
| ad_structures.extend( |
| [ |
| (AdvertisingData.SERVICE_DATA_32_BIT_UUID, uuid32_from_str(uuid) + data) |
| for uuid, data in datas.items() |
| ] |
| ) |
| if datas := dt.service_data_uuid128: |
| ad_structures.extend( |
| [ |
| (AdvertisingData.SERVICE_DATA_128_BIT_UUID, uuid128_from_str(uuid) + data) |
| for uuid, data in datas.items() |
| ] |
| ) |
| if dt.appearance: |
| ad_structures.append((AdvertisingData.APPEARANCE, struct.pack('<H', dt.appearance))) |
| if dt.advertising_interval: |
| ad_structures.append((AdvertisingData.ADVERTISING_INTERVAL, struct.pack('<H', dt.advertising_interval))) |
| if dt.uri: |
| ad_structures.append((AdvertisingData.URI, bytes(dt.uri, 'utf-8'))) |
| if dt.le_supported_features: |
| ad_structures.append((AdvertisingData.LE_SUPPORTED_FEATURES, dt.le_supported_features)) |
| if dt.manufacturer_specific_data: |
| ad_structures.append((AdvertisingData.MANUFACTURER_SPECIFIC_DATA, dt.manufacturer_specific_data)) |
| |
| return AdvertisingData(ad_structures) |
| |
| def pack_data_types(self, ad: AdvertisingData) -> DataTypes: |
| dt = DataTypes() |
| uuids: List[UUID] |
| s: str |
| i: int |
| ij: Tuple[int, int] |
| uuid_data: Tuple[UUID, bytes] |
| data: bytes |
| |
| if uuids := cast(List[UUID], ad.get(AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS)): |
| dt.incomplete_service_class_uuids16.extend(list(map(lambda x: x.to_hex_str(), uuids))) |
| if uuids := cast(List[UUID], ad.get(AdvertisingData.COMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS)): |
| dt.complete_service_class_uuids16.extend(list(map(lambda x: x.to_hex_str(), uuids))) |
| if uuids := cast(List[UUID], ad.get(AdvertisingData.INCOMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS)): |
| dt.incomplete_service_class_uuids32.extend(list(map(lambda x: x.to_hex_str(), uuids))) |
| if uuids := cast(List[UUID], ad.get(AdvertisingData.COMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS)): |
| dt.complete_service_class_uuids32.extend(list(map(lambda x: x.to_hex_str(), uuids))) |
| if uuids := cast(List[UUID], ad.get(AdvertisingData.INCOMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS)): |
| dt.incomplete_service_class_uuids128.extend(list(map(lambda x: x.to_hex_str(), uuids))) |
| if uuids := cast(List[UUID], ad.get(AdvertisingData.COMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS)): |
| dt.complete_service_class_uuids128.extend(list(map(lambda x: x.to_hex_str(), uuids))) |
| if s := cast(str, ad.get(AdvertisingData.SHORTENED_LOCAL_NAME)): |
| dt.shortened_local_name = s |
| if s := cast(str, ad.get(AdvertisingData.COMPLETE_LOCAL_NAME)): |
| dt.complete_local_name = s |
| if i := cast(int, ad.get(AdvertisingData.TX_POWER_LEVEL)): |
| dt.tx_power_level = i |
| if i := cast(int, ad.get(AdvertisingData.CLASS_OF_DEVICE)): |
| dt.class_of_device = i |
| if ij := cast(Tuple[int, int], ad.get(AdvertisingData.PERIPHERAL_CONNECTION_INTERVAL_RANGE)): |
| dt.peripheral_connection_interval_min = ij[0] |
| dt.peripheral_connection_interval_max = ij[1] |
| if uuids := cast(List[UUID], ad.get(AdvertisingData.LIST_OF_16_BIT_SERVICE_SOLICITATION_UUIDS)): |
| dt.service_solicitation_uuids16.extend(list(map(lambda x: x.to_hex_str(), uuids))) |
| if uuids := cast(List[UUID], ad.get(AdvertisingData.LIST_OF_32_BIT_SERVICE_SOLICITATION_UUIDS)): |
| dt.service_solicitation_uuids32.extend(list(map(lambda x: x.to_hex_str(), uuids))) |
| if uuids := cast(List[UUID], ad.get(AdvertisingData.LIST_OF_128_BIT_SERVICE_SOLICITATION_UUIDS)): |
| dt.service_solicitation_uuids128.extend(list(map(lambda x: x.to_hex_str(), uuids))) |
| if uuid_data := cast(Tuple[UUID, bytes], ad.get(AdvertisingData.SERVICE_DATA_16_BIT_UUID)): |
| dt.service_data_uuid16[uuid_data[0].to_hex_str()] = uuid_data[1] |
| if uuid_data := cast(Tuple[UUID, bytes], ad.get(AdvertisingData.SERVICE_DATA_32_BIT_UUID)): |
| dt.service_data_uuid32[uuid_data[0].to_hex_str()] = uuid_data[1] |
| if uuid_data := cast(Tuple[UUID, bytes], ad.get(AdvertisingData.SERVICE_DATA_128_BIT_UUID)): |
| dt.service_data_uuid128[uuid_data[0].to_hex_str()] = uuid_data[1] |
| if data := cast(bytes, ad.get(AdvertisingData.PUBLIC_TARGET_ADDRESS, raw=True)): |
| dt.public_target_addresses.extend([data[i * 6 :: i * 6 + 6] for i in range(int(len(data) / 6))]) |
| if data := cast(bytes, ad.get(AdvertisingData.RANDOM_TARGET_ADDRESS, raw=True)): |
| dt.random_target_addresses.extend([data[i * 6 :: i * 6 + 6] for i in range(int(len(data) / 6))]) |
| if i := cast(int, ad.get(AdvertisingData.APPEARANCE)): |
| dt.appearance = i |
| if i := cast(int, ad.get(AdvertisingData.ADVERTISING_INTERVAL)): |
| dt.advertising_interval = i |
| if s := cast(str, ad.get(AdvertisingData.URI)): |
| dt.uri = s |
| if data := cast(bytes, ad.get(AdvertisingData.LE_SUPPORTED_FEATURES, raw=True)): |
| dt.le_supported_features = data |
| if data := cast(bytes, ad.get(AdvertisingData.MANUFACTURER_SPECIFIC_DATA, raw=True)): |
| dt.manufacturer_specific_data = data |
| |
| return dt |
