1# Copyright 2021-2023 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 logging
20import sys
21import os
22import secrets
23
24from bumble.core import AdvertisingData
25from bumble.device import Device
26from bumble.hci import (
27    Address,
28    OwnAddressType,
29    HCI_LE_Set_Extended_Advertising_Parameters_Command,
30)
31from bumble.profiles.cap import CommonAudioServiceService
32from bumble.profiles.csip import CoordinatedSetIdentificationService, SirkType
33
34from bumble.transport import open_transport_or_link
35
36
37# -----------------------------------------------------------------------------
38async def main() -> None:
39    if len(sys.argv) < 3:
40        print(
41            'Usage: run_csis_servers.py <config-file> '
42            '<transport-spec-for-device-1> <transport-spec-for-device-2>'
43        )
44        print('example: run_csis_servers.py device1.json ' 'hci-socket:0 hci-socket:1')
45        return
46
47    print('<<< connecting to HCI...')
48    hci_transports = await asyncio.gather(
49        open_transport_or_link(sys.argv[2]), open_transport_or_link(sys.argv[3])
50    )
51    print('<<< connected')
52
53    devices = [
54        Device.from_config_file_with_hci(
55            sys.argv[1], hci_transport.source, hci_transport.sink
56        )
57        for hci_transport in hci_transports
58    ]
59
60    sirk = secrets.token_bytes(16)
61
62    for i, device in enumerate(devices):
63        device.random_address = Address(secrets.token_bytes(6))
64        await device.power_on()
65        csis = CoordinatedSetIdentificationService(
66            set_identity_resolving_key=sirk,
67            set_identity_resolving_key_type=SirkType.PLAINTEXT,
68            coordinated_set_size=2,
69        )
70        device.add_service(CommonAudioServiceService(csis))
71        advertising_data = (
72            bytes(
73                AdvertisingData(
74                    [
75                        (
76                            AdvertisingData.COMPLETE_LOCAL_NAME,
77                            bytes(f'Bumble LE Audio-{i}', 'utf-8'),
78                        ),
79                        (
80                            AdvertisingData.FLAGS,
81                            bytes(
82                                [
83                                    AdvertisingData.LE_GENERAL_DISCOVERABLE_MODE_FLAG
84                                    | AdvertisingData.BR_EDR_HOST_FLAG
85                                    | AdvertisingData.BR_EDR_CONTROLLER_FLAG
86                                ]
87                            ),
88                        ),
89                        (
90                            AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS,
91                            bytes(CoordinatedSetIdentificationService.UUID),
92                        ),
93                    ]
94                )
95            )
96            + csis.get_advertising_data()
97        )
98        await device.create_advertising_set(advertising_data=advertising_data)
99
100    await asyncio.gather(
101        *[hci_transport.source.terminated for hci_transport in hci_transports]
102    )
103
104
105# -----------------------------------------------------------------------------
106logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
107asyncio.run(main())
108