1# Copyright 2021-2022 Google LLC
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#      https://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14
15# -----------------------------------------------------------------------------
16# Imports
17# -----------------------------------------------------------------------------
18import asyncio
19import sys
20import os
21import logging
22from bumble.colors import color
23
24from bumble.device import Device
25from bumble.transport import open_transport_or_link
26from bumble.core import BT_BR_EDR_TRANSPORT, BT_L2CAP_PROTOCOL_ID, CommandTimeoutError
27from bumble.sdp import (
28    Client as SDP_Client,
29    SDP_PUBLIC_BROWSE_ROOT,
30    SDP_ALL_ATTRIBUTES_RANGE,
31)
32
33
34# -----------------------------------------------------------------------------
35async def main() -> None:
36    if len(sys.argv) < 3:
37        print(
38            'Usage: run_classic_connect.py <device-config> <transport-spec> '
39            '<bluetooth-addresses..>'
40        )
41        print('example: run_classic_connect.py classic1.json usb:0 E1:CA:72:48:C4:E8')
42        return
43
44    print('<<< connecting to HCI...')
45    async with await open_transport_or_link(sys.argv[2]) as hci_transport:
46        print('<<< connected')
47
48        # Create a device
49        device = Device.from_config_file_with_hci(
50            sys.argv[1], hci_transport.source, hci_transport.sink
51        )
52        device.classic_enabled = True
53        device.le_enabled = False
54        await device.power_on()
55
56        async def connect(target_address):
57            print(f'=== Connecting to {target_address}...')
58            try:
59                connection = await device.connect(
60                    target_address, transport=BT_BR_EDR_TRANSPORT
61                )
62            except CommandTimeoutError:
63                print('!!! Connection timed out')
64                return
65            print(f'=== Connected to {connection.peer_address}!')
66
67            # Connect to the SDP Server
68            sdp_client = SDP_Client(connection)
69            await sdp_client.connect()
70
71            # List all services in the root browse group
72            service_record_handles = await sdp_client.search_services(
73                [SDP_PUBLIC_BROWSE_ROOT]
74            )
75            print(color('\n==================================', 'blue'))
76            print(color('SERVICES:', 'yellow'), service_record_handles)
77
78            # For each service in the root browse group, get all its attributes
79            for service_record_handle in service_record_handles:
80                attributes = await sdp_client.get_attributes(
81                    service_record_handle, [SDP_ALL_ATTRIBUTES_RANGE]
82                )
83                print(
84                    color(f'SERVICE {service_record_handle:04X} attributes:', 'yellow')
85                )
86                for attribute in attributes:
87                    print('  ', attribute.to_string(with_colors=True))
88
89            # Search for services with an L2CAP service attribute
90            search_result = await sdp_client.search_attributes(
91                [BT_L2CAP_PROTOCOL_ID], [SDP_ALL_ATTRIBUTES_RANGE]
92            )
93            print(color('\n==================================', 'blue'))
94            print(color('SEARCH RESULTS:', 'yellow'))
95            for attribute_list in search_result:
96                print(color('SERVICE:', 'green'))
97                print(
98                    '  '
99                    + '\n  '.join(
100                        [
101                            attribute.to_string(with_colors=True)
102                            for attribute in attribute_list
103                        ]
104                    )
105                )
106
107            await sdp_client.disconnect()
108
109        # Connect to a peer
110        target_addresses = sys.argv[3:]
111        await asyncio.wait(
112            [
113                asyncio.create_task(connect(target_address))
114                for target_address in target_addresses
115            ]
116        )
117
118
119# -----------------------------------------------------------------------------
120logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
121asyncio.run(main())
122