RootCanal: Write helper script for dumping the controller configuration

#!/usr/bin/env python3

# Copyright 2015 The Android Open Source Project

#

# 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

#

# http://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.

"""Dump the configuration of a Bluetooth controller.

The script expects to find the generated module hci_packets

in PYTHONPATH.

The controller is expected to be available through HCI over TCP

at the port passed as parameter."""

import argparse

import asyncio

import collections

import sys

import hci_packets as hci

H4_IDC_CMD = 0x01

H4_IDC_ACL = 0x02

H4_IDC_SCO = 0x03

H4_IDC_EVT = 0x04

H4_IDC_ISO = 0x05

HCI_HEADER_SIZES = dict([(H4_IDC_CMD, 3), (H4_IDC_ACL, 4), (H4_IDC_SCO, 3), (H4_IDC_EVT, 2), (H4_IDC_ISO, 4)])

class Host:

    def __init__(self):

        self.evt_queue = collections.deque()

        self.evt_queue_event = asyncio.Event()

    async def connect(self, ip: str, port: int):

        reader, writer = await asyncio.open_connection(ip, port)

        self.reader = asyncio.create_task(self._read(reader))

        self.writer = writer

    async def _read(self, reader):

        try:

            while True:

                idc = await reader.readexactly(1)

                assert idc[0] in HCI_HEADER_SIZES

                header = await reader.readexactly(HCI_HEADER_SIZES[idc[0]])

                if idc[0] == H4_IDC_EVT:

                    evt = hci.Event.parse_all(header + (await reader.readexactly(header[1])))

                    #print(f"<< {evt.__class__.__name__}")

                    self.evt_queue.append(evt)

                    self.evt_queue_event.set()

                else:

                    assert False

        except Exception as exn:

            print(f"Reader interrupted: {exn}")

            return

    async def send_cmd(self, cmd: hci.Command):

        #print(f">> {cmd.__class__.__name__}")

        packet = bytes([H4_IDC_CMD]) + cmd.serialize()

        self.writer.write(packet)

    async def recv_evt(self) -> hci.Event:

        while not self.evt_queue:

            await self.evt_queue_event.wait()

            self.evt_queue_event.clear()

        return self.evt_queue.popleft()

    async def expect_evt(self, expected_evt: type) -> hci.Event:

        evt = await self.recv_evt()

        assert isinstance(evt, expected_evt)

        if not isinstance(evt, expected_evt):

            print("Received unexpected event:")

            evt.show()

            print(f"Expected event of type {expected_evt.__name__}")

            print(f"{list(evt.payload)}")

        return evt

async def br_edr_properties(host: Host):

    await host.send_cmd(hci.ReadLocalSupportedFeatures())

    page0 = await host.expect_evt(hci.ReadLocalSupportedFeaturesComplete)

    await host.send_cmd(hci.ReadLocalExtendedFeatures(page_number=1))

    page1 = await host.expect_evt(hci.ReadLocalExtendedFeaturesComplete)

    await host.send_cmd(hci.ReadLocalExtendedFeatures(page_number=2))

    page2 = await host.expect_evt(hci.ReadLocalExtendedFeaturesComplete)

    print(

        f"lmp_features: {{ 0x{page0.lmp_features:x}, 0x{page1.extended_lmp_features:x}, 0x{page2.extended_lmp_features:x} }}"

    )

    await host.send_cmd(hci.ReadBufferSize())

    evt = await host.expect_evt(hci.ReadBufferSizeComplete)

    print(f"acl_data_packet_length: {evt.acl_data_packet_length}")

    print(f"total_num_acl_data_packets: {evt.total_num_acl_data_packets}")

    print(f"sco_data_packet_length: {evt.synchronous_data_packet_length}")

    print(f"total_num_sco_data_packets: {evt.total_num_synchronous_data_packets}")

    await host.send_cmd(hci.ReadNumberOfSupportedIac())

    evt = await host.expect_evt(hci.ReadNumberOfSupportedIacComplete)

    print(f"num_supported_iac: {evt.num_support_iac}")

async def le_properties(host: Host):

    await host.send_cmd(hci.LeReadLocalSupportedFeatures())

    evt = await host.expect_evt(hci.LeReadLocalSupportedFeaturesComplete)

    print(f"le_features: 0x{evt.le_features:x}")

    await host.send_cmd(hci.LeReadBufferSizeV2())

    evt = await host.expect_evt(hci.LeReadBufferSizeV2Complete)

    print(f"le_acl_data_packet_length: {evt.le_buffer_size.le_data_packet_length}")

    print(f"total_num_le_acl_data_packets: {evt.le_buffer_size.total_num_le_packets}")

    print(f"iso_data_packet_length: {evt.iso_buffer_size.le_data_packet_length}")

    print(f"total_num_iso_data_packets: {evt.iso_buffer_size.total_num_le_packets}")

    await host.send_cmd(hci.LeReadFilterAcceptListSize())

    evt = await host.expect_evt(hci.LeReadFilterAcceptListSizeComplete)

    print(f"le_filter_accept_list_size: {evt.filter_accept_list_size}")

    await host.send_cmd(hci.LeReadResolvingListSize())

    evt = await host.expect_evt(hci.LeReadResolvingListSizeComplete)

    print(f"le_resolving_list_size: {evt.resolving_list_size}")

    await host.send_cmd(hci.LeReadSupportedStates())

    evt = await host.expect_evt(hci.LeReadSupportedStatesComplete)

    print(f"le_supported_states: 0x{evt.le_states:x}")

    await host.send_cmd(hci.LeReadMaximumAdvertisingDataLength())

    evt = await host.expect_evt(hci.LeReadMaximumAdvertisingDataLengthComplete)

    print(f"le_max_advertising_data_length: {evt.maximum_advertising_data_length}")

    await host.send_cmd(hci.LeReadNumberOfSupportedAdvertisingSets())

    evt = await host.expect_evt(hci.LeReadNumberOfSupportedAdvertisingSetsComplete)

    print(f"le_num_supported_advertising_sets: {evt.number_supported_advertising_sets}")

    await host.send_cmd(hci.LeReadPeriodicAdvertiserListSize())

    evt = await host.expect_evt(hci.LeReadPeriodicAdvertiserListSizeComplete)

    print(f"le_periodic_advertiser_list_size: {evt.periodic_advertiser_list_size}")

async def run(tcp_port: int):

    host = Host()

    await host.connect('127.0.0.1', tcp_port)

    await host.send_cmd(hci.Reset())

    await host.expect_evt(hci.ResetComplete)

    await host.send_cmd(hci.ReadLocalVersionInformation())

    evt = await host.expect_evt(hci.ReadLocalVersionInformationComplete)

    print(f"hci_version: {evt.local_version_information.hci_version}")

    print(f"hci_subversion: 0x{evt.local_version_information.hci_revision:x}")

    print(f"lmp_version: {evt.local_version_information.lmp_version}")

    print(f"lmp_subversion: 0x{evt.local_version_information.lmp_subversion:x}")

    print(f"company_identifier: 0x{evt.local_version_information.manufacturer_name:x}")

    await host.send_cmd(hci.ReadLocalSupportedCommands())

    evt = await host.expect_evt(hci.ReadLocalSupportedCommandsComplete)

    print(f"supported_commands: {{ {', '.join([f'0x{b:x}' for b in evt.supported_commands])} }}")

    try:

        await br_edr_properties(host)

    except Exception:

        pass

    try:

        await le_properties(host)

    except Exception:

        pass

def main() -> int:

    """Generate cxx PDL backend."""

    parser = argparse.ArgumentParser(description=__doc__)

    parser.add_argument('tcp_port', type=int, help='HCI port')

    return asyncio.run(run(**vars(parser.parse_args())))

if __name__ == '__main__':

    sys.exit(main())