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
22from bumble.device import (
23    Device,
24    Connection,
25    AdvertisingParameters,
26    AdvertisingEventProperties,
27)
28from bumble.hci import (
29    OwnAddressType,
30)
31
32from bumble.transport import open_transport_or_link
33
34
35# -----------------------------------------------------------------------------
36async def main() -> None:
37    if len(sys.argv) < 3:
38        print(
39            'Usage: run_cig_setup.py <config-file> '
40            '<transport-spec-for-device-1> <transport-spec-for-device-2>'
41        )
42        print('example: run_cig_setup.py device1.json hci-socket:0 hci-socket:1')
43        return
44
45    print('<<< connecting to HCI...')
46    hci_transports = await asyncio.gather(
47        open_transport_or_link(sys.argv[2]), open_transport_or_link(sys.argv[3])
48    )
49    print('<<< connected')
50
51    devices = [
52        Device.from_config_file_with_hci(
53            sys.argv[1], hci_transport.source, hci_transport.sink
54        )
55        for hci_transport in hci_transports
56    ]
57
58    devices[0].cis_enabled = True
59    devices[1].cis_enabled = True
60
61    await asyncio.gather(*[device.power_on() for device in devices])
62    advertising_set = await devices[0].create_advertising_set()
63
64    connection = await devices[1].connect(
65        devices[0].random_address, own_address_type=OwnAddressType.RANDOM
66    )
67
68    cid_ids = [2, 3]
69    cis_handles = await devices[1].setup_cig(
70        cig_id=1,
71        cis_id=cid_ids,
72        sdu_interval=(10000, 255),
73        framing=0,
74        max_sdu=(120, 0),
75        retransmission_number=13,
76        max_transport_latency=(100, 5),
77    )
78
79    def on_cis_request(
80        connection: Connection, cis_handle: int, _cig_id: int, _cis_id: int
81    ):
82        connection.abort_on('disconnection', devices[0].accept_cis_request(cis_handle))
83
84    devices[0].on('cis_request', on_cis_request)
85
86    cis_links = await devices[1].create_cis(
87        [(cis, connection.handle) for cis in cis_handles]
88    )
89
90    for cis_link in cis_links:
91        await cis_link.disconnect()
92
93    await asyncio.gather(
94        *[hci_transport.source.terminated for hci_transport in hci_transports]
95    )
96
97
98# -----------------------------------------------------------------------------
99logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
100asyncio.run(main())
101