# Copyright 2021-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. # ----------------------------------------------------------------------------- # Imports # ----------------------------------------------------------------------------- import asyncio import sys import os import logging from bumble.colors import color import bumble.core from bumble.device import Device from bumble.transport import open_transport_or_link from bumble.core import ( BT_HANDSFREE_SERVICE, BT_RFCOMM_PROTOCOL_ID, BT_BR_EDR_TRANSPORT, ) from bumble.rfcomm import Client from bumble.sdp import ( Client as SDP_Client, DataElement, ServiceAttribute, SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID, SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID, SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID, ) from bumble.hfp import HfpProtocol # ----------------------------------------------------------------------------- # pylint: disable-next=too-many-nested-blocks async def list_rfcomm_channels(device, connection): # Connect to the SDP Server sdp_client = SDP_Client(device) await sdp_client.connect(connection) # Search for services that support the Handsfree Profile search_result = await sdp_client.search_attributes( [BT_HANDSFREE_SERVICE], [ SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID, SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID, SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID, ], ) print(color('==================================', 'blue')) print(color('Handsfree Services:', 'yellow')) rfcomm_channels = [] # pylint: disable-next=too-many-nested-blocks for attribute_list in search_result: # Look for the RFCOMM Channel number protocol_descriptor_list = ServiceAttribute.find_attribute_in_list( attribute_list, SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID ) if protocol_descriptor_list: for protocol_descriptor in protocol_descriptor_list.value: if len(protocol_descriptor.value) >= 2: if protocol_descriptor.value[0].value == BT_RFCOMM_PROTOCOL_ID: print(color('SERVICE:', 'green')) print( color(' RFCOMM Channel:', 'cyan'), protocol_descriptor.value[1].value, ) rfcomm_channels.append(protocol_descriptor.value[1].value) # List profiles bluetooth_profile_descriptor_list = ( ServiceAttribute.find_attribute_in_list( attribute_list, SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID, ) ) if bluetooth_profile_descriptor_list: if bluetooth_profile_descriptor_list.value: if ( bluetooth_profile_descriptor_list.value[0].type == DataElement.SEQUENCE ): bluetooth_profile_descriptors = ( bluetooth_profile_descriptor_list.value ) else: # Sometimes, instead of a list of lists, we just # find a list. Fix that bluetooth_profile_descriptors = [ bluetooth_profile_descriptor_list ] print(color(' Profiles:', 'green')) for ( bluetooth_profile_descriptor ) in bluetooth_profile_descriptors: version_major = ( bluetooth_profile_descriptor.value[1].value >> 8 ) version_minor = ( bluetooth_profile_descriptor.value[1].value & 0xFF ) print( ' ' f'{bluetooth_profile_descriptor.value[0].value}' f' - version {version_major}.{version_minor}' ) # List service classes service_class_id_list = ServiceAttribute.find_attribute_in_list( attribute_list, SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID ) if service_class_id_list: if service_class_id_list.value: print(color(' Service Classes:', 'green')) for service_class_id in service_class_id_list.value: print(' ', service_class_id.value) await sdp_client.disconnect() return rfcomm_channels # ----------------------------------------------------------------------------- async def main(): if len(sys.argv) < 4: print( 'Usage: run_hfp_gateway.py ' '' ) print( ' specifying a channel number, or "discover" to list all RFCOMM channels' ) print('example: run_hfp_gateway.py hfp_gateway.json usb:0 E1:CA:72:48:C4:E8') return print('<<< connecting to HCI...') async with await open_transport_or_link(sys.argv[2]) as (hci_source, hci_sink): print('<<< connected') # Create a device device = Device.from_config_file_with_hci(sys.argv[1], hci_source, hci_sink) device.classic_enabled = True await device.power_on() # Connect to a peer target_address = sys.argv[3] print(f'=== Connecting to {target_address}...') connection = await device.connect(target_address, transport=BT_BR_EDR_TRANSPORT) print(f'=== Connected to {connection.peer_address}!') # Get a list of all the Handsfree services (should only be 1) channels = await list_rfcomm_channels(device, connection) if len(channels) == 0: print('!!! no service found') return # Pick the first one channel = channels[0] # Request authentication print('*** Authenticating...') await connection.authenticate() print('*** Authenticated') # Enable encryption print('*** Enabling encryption...') await connection.encrypt() print('*** Encryption on') # Create a client and start it print('@@@ Starting to RFCOMM client...') rfcomm_client = Client(device, connection) rfcomm_mux = await rfcomm_client.start() print('@@@ Started') print(f'### Opening session for channel {channel}...') try: session = await rfcomm_mux.open_dlc(channel) print('### Session open', session) except bumble.core.ConnectionError as error: print(f'### Session open failed: {error}') await rfcomm_mux.disconnect() print('@@@ Disconnected from RFCOMM server') return # Protocol loop (just for testing at this point) protocol = HfpProtocol(session) while True: line = await protocol.next_line() if line.startswith('AT+BRSF='): protocol.send_response_line('+BRSF: 30') protocol.send_response_line('OK') elif line.startswith('AT+CIND=?'): protocol.send_response_line( '+CIND: ("call",(0,1)),("callsetup",(0-3)),("service",(0-1)),' '("signal",(0-5)),("roam",(0,1)),("battchg",(0-5)),' '("callheld",(0-2))' ) protocol.send_response_line('OK') elif line.startswith('AT+CIND?'): protocol.send_response_line('+CIND: 0,0,1,4,1,5,0') protocol.send_response_line('OK') elif line.startswith('AT+CMER='): protocol.send_response_line('OK') elif line.startswith('AT+CHLD=?'): protocol.send_response_line('+CHLD: 0') protocol.send_response_line('OK') elif line.startswith('AT+BTRH?'): protocol.send_response_line('+BTRH: 0') protocol.send_response_line('OK') elif line.startswith('AT+CLIP='): protocol.send_response_line('OK') elif line.startswith('AT+VGS='): protocol.send_response_line('OK') elif line.startswith('AT+BIA='): protocol.send_response_line('OK') elif line.startswith('AT+BVRA='): protocol.send_response_line( '+BVRA: 1,1,12AA,1,1,"Message 1 from Janina"' ) elif line.startswith('AT+XEVENT='): protocol.send_response_line('OK') elif line.startswith('AT+XAPL='): protocol.send_response_line('OK') else: print(color('UNSUPPORTED AT COMMAND', 'red')) protocol.send_response_line('ERROR') await hci_source.wait_for_termination() # ----------------------------------------------------------------------------- logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper()) asyncio.run(main())