mirror of
https://github.com/google/bumble.git
synced 2026-04-16 00:25:31 +00:00
add downloader tool
This commit is contained in:
Gilles Boccon-Gibod
38
tools/generate_company_id_list.py
Normal file
38
tools/generate_company_id_list.py
Normal file
@@ -0,0 +1,38 @@
|
||||
# Copyright 2021-2022 Google LLC
|
||||
#
|
||||
# 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
|
||||
#
|
||||
# https://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.
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# This script generates a python-syntax list of dictionary entries for the
|
||||
# company IDs listed at: https://www.bluetooth.com/specifications/assigned-numbers/company-identifiers/
|
||||
# The input to this script is the CSV file that can be obtained at that URL
|
||||
# -----------------------------------------------------------------------------
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Imports
|
||||
# -----------------------------------------------------------------------------
|
||||
import sys
|
||||
import csv
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
with open(sys.argv[1], newline='') as csvfile:
|
||||
reader = csv.reader(csvfile, delimiter=',', quotechar='"')
|
||||
lines = []
|
||||
for row in reader:
|
||||
if len(row) == 3 and row[1].startswith('0x'):
|
||||
company_id = row[1]
|
||||
company_name = row[2]
|
||||
escaped_company_name = company_name.replace('"', '\\"')
|
||||
lines.append(f' {company_id}: "{escaped_company_name}"')
|
||||
|
||||
print(',\n'.join(reversed(lines)))
|
||||
153
tools/rtk_fw_download.py
Normal file
153
tools/rtk_fw_download.py
Normal file
@@ -0,0 +1,153 @@
|
||||
# Copyright 2021-2023 Google LLC
|
||||
#
|
||||
# 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
|
||||
#
|
||||
# https://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.
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Imports
|
||||
# -----------------------------------------------------------------------------
|
||||
import logging
|
||||
import pathlib
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
|
||||
import click
|
||||
|
||||
from bumble.colors import color
|
||||
from bumble.drivers import rtk
|
||||
from bumble.tools import rtk_util
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Logging
|
||||
# -----------------------------------------------------------------------------
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Constants
|
||||
# -----------------------------------------------------------------------------
|
||||
LINUX_KERNEL_GIT_SOURCE = (
|
||||
"https://git.kernel.org/pub/scm/linux/kernel/git/firmware/linux-firmware.git/plain/rtl_bt",
|
||||
False,
|
||||
)
|
||||
REALTEK_OPENSOURCE_SOURCE = (
|
||||
"https://github.com/Realtek-OpenSource/android_hardware_realtek/raw/rtk1395/bt/rtkbt/Firmware/BT",
|
||||
True,
|
||||
)
|
||||
LINUX_FROM_SCRATCH_SOURCE = (
|
||||
"https://anduin.linuxfromscratch.org/sources/linux-firmware/rtl_bt",
|
||||
False
|
||||
)
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Functions
|
||||
# -----------------------------------------------------------------------------
|
||||
def download_file(base_url, name, remove_suffix):
|
||||
if remove_suffix:
|
||||
name = name.replace(".bin", "")
|
||||
|
||||
url = f"{base_url}/{name}"
|
||||
with urllib.request.urlopen(url) as file:
|
||||
data = file.read()
|
||||
print(f"Downloaded {name}: {len(data)} bytes")
|
||||
return data
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
@click.command
|
||||
@click.option(
|
||||
"--output-dir",
|
||||
default=".",
|
||||
help="Output directory where the files will be saved",
|
||||
show_default=True,
|
||||
)
|
||||
@click.option(
|
||||
"--source",
|
||||
type=click.Choice(["linux-kernel", "realtek-opensource", "linux-from-scratch"]),
|
||||
default="linux-kernel",
|
||||
show_default=True,
|
||||
)
|
||||
@click.option("--single", help="Only download a single image set, by its base name")
|
||||
@click.option("--force", is_flag=True, help="Overwrite files if they already exist")
|
||||
@click.option("--parse", is_flag=True, help="Parse the FW image after saving")
|
||||
def main(output_dir, source, single, force, parse):
|
||||
"""Download RTK firmware images and configs."""
|
||||
|
||||
# Check that the output dir exists
|
||||
output_dir = pathlib.Path(output_dir)
|
||||
if not output_dir.is_dir():
|
||||
print("Output dir does not exist or is not a directory")
|
||||
return
|
||||
|
||||
base_url, remove_suffix = {
|
||||
"linux-kernel": LINUX_KERNEL_GIT_SOURCE,
|
||||
"realtek-opensource": REALTEK_OPENSOURCE_SOURCE,
|
||||
"linux-from-scratch": LINUX_FROM_SCRATCH_SOURCE
|
||||
}[source]
|
||||
|
||||
print("Downloading")
|
||||
print(color("FROM:", "green"), base_url)
|
||||
print(color("TO:", "green"), output_dir)
|
||||
|
||||
if single:
|
||||
images = [(f"{single}_fw.bin", f"{single}_config.bin", True)]
|
||||
else:
|
||||
images = [
|
||||
(
|
||||
driver_info.fw_name,
|
||||
driver_info.config_name,
|
||||
driver_info.config_needed
|
||||
)
|
||||
for driver_info in rtk.Driver.DRIVER_INFOS
|
||||
]
|
||||
|
||||
for (fw_name, config_name, config_needed) in images:
|
||||
print(color("---", "yellow"))
|
||||
fw_image_out = output_dir / fw_name
|
||||
if not force and fw_image_out.exists():
|
||||
print(color(f"{fw_image_out} already exists, skipping", "red"))
|
||||
continue
|
||||
if config_name:
|
||||
config_image_out = output_dir / config_name
|
||||
if not force and config_image_out.exists():
|
||||
print(color("f{config_out} already exists, skipping", "red"))
|
||||
continue
|
||||
|
||||
try:
|
||||
fw_image = download_file(base_url, fw_name, remove_suffix)
|
||||
except urllib.error.HTTPError as error:
|
||||
print(f"Failed to download {fw_name}: {error}")
|
||||
continue
|
||||
|
||||
config_image = None
|
||||
if config_name:
|
||||
try:
|
||||
config_image = download_file(base_url, config_name, remove_suffix)
|
||||
except urllib.error.HTTPError as error:
|
||||
if config_needed:
|
||||
print(f"Failed to download {config_name}: {error}")
|
||||
continue
|
||||
else:
|
||||
print(f"No config available as {config_name}")
|
||||
|
||||
fw_image_out.write_bytes(fw_image)
|
||||
if parse and config_name:
|
||||
print(color("Parsing:", "cyan"), fw_name)
|
||||
rtk_util.do_parse(fw_image_out)
|
||||
if config_image:
|
||||
config_image_out.write_bytes(config_image)
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
161
tools/rtk_util.py
Normal file
161
tools/rtk_util.py
Normal file
@@ -0,0 +1,161 @@
|
||||
# Copyright 2021-2023 Google LLC
|
||||
#
|
||||
# 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
|
||||
#
|
||||
# https://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.
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Imports
|
||||
# -----------------------------------------------------------------------------
|
||||
import logging
|
||||
import asyncio
|
||||
import os
|
||||
|
||||
import click
|
||||
|
||||
from bumble import transport
|
||||
from bumble.host import Host
|
||||
from bumble.drivers import rtk
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Logging
|
||||
# -----------------------------------------------------------------------------
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
def do_parse(firmware_path):
|
||||
with open(firmware_path, 'rb') as firmware_file:
|
||||
firmware_data = firmware_file.read()
|
||||
firmware = rtk.Firmware(firmware_data)
|
||||
print(
|
||||
f"Firmware: version=0x{firmware.version:08X} "
|
||||
f"project_id=0x{firmware.project_id:04X}"
|
||||
)
|
||||
for patch in firmware.patches:
|
||||
print(
|
||||
f" Patch: chip_id=0x{patch[0]:04X}, "
|
||||
f"{len(patch[1])} bytes, "
|
||||
f"SVN Version={patch[2]:08X}"
|
||||
)
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
async def do_load(usb_transport, force):
|
||||
async with await transport.open_transport_or_link(usb_transport) as (
|
||||
hci_source,
|
||||
hci_sink,
|
||||
):
|
||||
# Create a host to communicate with the device
|
||||
host = Host(hci_source, hci_sink)
|
||||
await host.reset(raw=True)
|
||||
|
||||
# Get the driver.
|
||||
driver = await rtk.Driver.for_host(host, force)
|
||||
if driver is None:
|
||||
if not force:
|
||||
print("Firmware already loaded or no supported driver for this device.")
|
||||
return
|
||||
|
||||
await driver.download_firmware()
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
async def do_drop(usb_transport):
|
||||
async with await transport.open_transport_or_link(usb_transport) as (
|
||||
hci_source,
|
||||
hci_sink,
|
||||
):
|
||||
# Create a host to communicate with the device
|
||||
host = Host(hci_source, hci_sink)
|
||||
await host.reset(raw=True)
|
||||
|
||||
# Tell the device to reset/drop any loaded patch
|
||||
await rtk.Driver.drop_firmware(host)
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
async def do_info(usb_transport, force):
|
||||
async with await transport.open_transport(usb_transport) as (
|
||||
hci_source,
|
||||
hci_sink,
|
||||
):
|
||||
# Create a host to communicate with the device
|
||||
host = Host(hci_source, hci_sink)
|
||||
await host.reset(raw=True)
|
||||
|
||||
# Check if this is a supported device.
|
||||
if not force and not rtk.Driver.check(host):
|
||||
print("USB device not supported by this RTK driver")
|
||||
return
|
||||
|
||||
# Get the driver info.
|
||||
driver_info = await rtk.Driver.driver_info_for_host(host)
|
||||
if driver_info:
|
||||
print(
|
||||
"Driver:\n"
|
||||
f" ROM: {driver_info.rom:04X}\n"
|
||||
f" Firmware: {driver_info.fw_name}\n"
|
||||
f" Config: {driver_info.config_name}\n"
|
||||
)
|
||||
else:
|
||||
print("Firmware already loaded or no supported driver for this device.")
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
@click.group()
|
||||
def main():
|
||||
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'INFO').upper())
|
||||
|
||||
|
||||
@main.command
|
||||
@click.argument("firmware_path")
|
||||
def parse(firmware_path):
|
||||
"""Parse a firmware image."""
|
||||
do_parse(firmware_path)
|
||||
|
||||
|
||||
@main.command
|
||||
@click.argument("usb_transport")
|
||||
@click.option(
|
||||
"--force",
|
||||
is_flag=True,
|
||||
default=False,
|
||||
help="Load even if the USB info doesn't match",
|
||||
)
|
||||
def load(usb_transport, force):
|
||||
"""Load a firmware image into the USB dongle."""
|
||||
asyncio.run(do_load(usb_transport, force))
|
||||
|
||||
|
||||
@main.command
|
||||
@click.argument("usb_transport")
|
||||
def drop(usb_transport):
|
||||
"""Drop a firmware image from the USB dongle."""
|
||||
asyncio.run(do_drop(usb_transport))
|
||||
|
||||
|
||||
@main.command
|
||||
@click.argument("usb_transport")
|
||||
@click.option(
|
||||
"--force",
|
||||
is_flag=True,
|
||||
default=False,
|
||||
help="Try to get the device info even if the USB info doesn't match",
|
||||
)
|
||||
def info(usb_transport, force):
|
||||
"""Get the firmware info from a USB dongle."""
|
||||
asyncio.run(do_info(usb_transport, force))
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user