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_cig_setup.py <config-file>' 42 '<transport-spec-for-device-1> <transport-spec-for-device-2>' 43 ) 44 print( 45 'example: run_cig_setup.py device1.json' 46 'tcp-client:127.0.0.1:6402 tcp-client:127.0.0.1:6402' 47 ) 48 return 49 50 print('<<< connecting to HCI...') 51 hci_transports = await asyncio.gather( 52 open_transport_or_link(sys.argv[2]), open_transport_or_link(sys.argv[3]) 53 ) 54 print('<<< connected') 55 56 devices = [ 57 Device.from_config_file_with_hci( 58 sys.argv[1], hci_transport.source, hci_transport.sink 59 ) 60 for hci_transport in hci_transports 61 ] 62 63 sirk = secrets.token_bytes(16) 64 65 for i, device in enumerate(devices): 66 device.random_address = Address(secrets.token_bytes(6)) 67 await device.power_on() 68 csis = CoordinatedSetIdentificationService( 69 set_identity_resolving_key=sirk, 70 set_identity_resolving_key_type=SirkType.PLAINTEXT, 71 coordinated_set_size=2, 72 ) 73 device.add_service(CommonAudioServiceService(csis)) 74 advertising_data = ( 75 bytes( 76 AdvertisingData( 77 [ 78 ( 79 AdvertisingData.COMPLETE_LOCAL_NAME, 80 bytes(f'Bumble LE Audio-{i}', 'utf-8'), 81 ), 82 ( 83 AdvertisingData.FLAGS, 84 bytes( 85 [ 86 AdvertisingData.LE_GENERAL_DISCOVERABLE_MODE_FLAG 87 | AdvertisingData.BR_EDR_HOST_FLAG 88 | AdvertisingData.BR_EDR_CONTROLLER_FLAG 89 ] 90 ), 91 ), 92 ( 93 AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS, 94 bytes(CoordinatedSetIdentificationService.UUID), 95 ), 96 ] 97 ) 98 ) 99 + csis.get_advertising_data() 100 ) 101 await device.create_advertising_set(advertising_data=advertising_data) 102 103 await asyncio.gather( 104 *[hci_transport.source.terminated for hci_transport in hci_transports] 105 ) 106 107 108# ----------------------------------------------------------------------------- 109logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper()) 110asyncio.run(main()) 111