examples/run_rfcomm_client.py

# 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 bumble.core
import bumble.logging
from bumble.colors import color
from bumble.core import BT_L2CAP_PROTOCOL_ID, BT_RFCOMM_PROTOCOL_ID, PhysicalTransport
from bumble.device import Device
from bumble.rfcomm import Client
from bumble.sdp import (
    SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
    SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
    SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
    DataElement,
    ServiceAttribute,
)
from bumble.sdp import Client as SDP_Client
from bumble.transport import open_transport


# -----------------------------------------------------------------------------
async def list_rfcomm_channels(connection):
    # Connect to the SDP Server
    sdp_client = SDP_Client(connection)
    await sdp_client.connect()

    # Search for services with an L2CAP service attribute
    search_result = await sdp_client.search_attributes(
        [BT_L2CAP_PROTOCOL_ID],
        [
            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('RFCOMM Services:', 'yellow'))
    # 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,
                        )

                        # 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()


# -----------------------------------------------------------------------------
class TcpServerProtocol(asyncio.Protocol):
    def __init__(self, rfcomm_session):
        self.rfcomm_session = rfcomm_session
        self.transport = None

    def connection_made(self, transport):
        peer_name = transport.get_extra_info('peer_name')
        print(f'<<< TCP Server: connection from {peer_name}')
        self.transport = transport
        self.rfcomm_session.sink = self.rfcomm_data_received

    def rfcomm_data_received(self, data):
        print(f'<<< RFCOMM Data: {data.hex()}')
        if self.transport:
            self.transport.write(data)
        else:
            print('!!! no TCP connection, dropping data')

    def data_received(self, data):
        print(f'<<< TCP Server: data received: {len(data)} bytes - {data.hex()}')
        self.rfcomm_session.write(data)


# -----------------------------------------------------------------------------
async def tcp_server(tcp_port, rfcomm_session):
    print(f'$$$ Starting TCP server on port {tcp_port}')

    server = await asyncio.get_running_loop().create_server(
        lambda: TcpServerProtocol(rfcomm_session), '127.0.0.1', tcp_port
    )
    await asyncio.get_running_loop().create_future()

    async with server:
        await server.serve_forever()


# -----------------------------------------------------------------------------
async def main() -> None:
    if len(sys.argv) < 5:
        print(
            'Usage: run_rfcomm_client.py <device-config> <transport-spec> '
            '<bluetooth-address> <channel>|discover [tcp-port]'
        )
        print(
            '  specifying a channel number, or "discover" to list all RFCOMM channels'
        )
        print('example: run_rfcomm_client.py classic1.json usb:0 E1:CA:72:48:C4:E8 8')
        return

    print('<<< connecting to HCI...')
    async with await open_transport(sys.argv[2]) as hci_transport:
        print('<<< connected')

        # Create a device
        device = Device.from_config_file_with_hci(
            sys.argv[1], hci_transport.source, hci_transport.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=PhysicalTransport.BR_EDR
        )
        print(f'=== Connected to {connection.peer_address}!')

        channel_str = sys.argv[4]
        if channel_str == 'discover':
            await list_rfcomm_channels(connection)
            return

        # 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 RFCOMM client...')
        rfcomm_client = Client(connection)
        rfcomm_mux = await rfcomm_client.start()
        print('@@@ Started')

        channel = int(channel_str)
        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

        if len(sys.argv) == 6:
            # A TCP port was specified, start listening
            tcp_port = int(sys.argv[5])
            asyncio.create_task(tcp_server(tcp_port, session))

        await hci_transport.source.terminated


# -----------------------------------------------------------------------------
bumble.logging.setup_basic_logging('DEBUG')
asyncio.run(main())