Merge pull request #750 from google/gbg/rtk-driver-enhancement

gbg/rtk driver enhancement

bumble/drivers/rtk.py

@@ -489,6 +489,21 @@ class Driver(common.Driver):
 
         return True
 
+    @staticmethod
+    async def get_loaded_firmware_version(host):
+        response = await host.send_command(HCI_RTK_Read_ROM_Version_Command())
+
+        if response.return_parameters.status != hci.HCI_SUCCESS:
+            return None
+
+        response = await host.send_command(
+            hci.HCI_Read_Local_Version_Information_Command(), check_result=True
+        )
+        return (
+            response.return_parameters.hci_subversion << 16
+            | response.return_parameters.lmp_subversion
+        )
+
     @classmethod
     async def driver_info_for_host(cls, host):
         try:
@@ -592,7 +607,7 @@ class Driver(common.Driver):
             )
             if response.return_parameters.status != hci.HCI_SUCCESS:
                 logger.warning("can't get ROM version")
-                return
+                return None
             rom_version = response.return_parameters.version
             logger.debug(f"ROM version before download: {rom_version:04X}")
         else:
@@ -600,13 +615,14 @@ class Driver(common.Driver):
 
         firmware = Firmware(self.firmware)
         logger.debug(f"firmware: project_id=0x{firmware.project_id:04X}")
+        logger.debug(f"firmware: version=0x{firmware.version:04X}")
         for patch in firmware.patches:
             if patch[0] == rom_version + 1:
                 logger.debug(f"using patch {patch[0]}")
                 break
         else:
             logger.warning("no valid patch found for rom version {rom_version}")
-            return
+            return None
 
         # Append the config if there is one.
         if self.config:
@@ -642,7 +658,9 @@ class Driver(common.Driver):
             logger.warning("can't get ROM version")
         else:
             rom_version = response.return_parameters.version
-            logger.debug(f"ROM version after download: {rom_version:04X}")
+            logger.debug(f"ROM version after download: {rom_version:02X}")
+
+        return firmware.version
 
     async def download_firmware(self):
         if self.driver_info.rom == RTK_ROM_LMP_8723A:

tools/intel_fw_download.py

@@ -43,7 +43,8 @@ LINUX_KERNEL_GIT_SOURCE = "https://git.kernel.org/pub/scm/linux/kernel/git/firmw
 # -----------------------------------------------------------------------------
 def download_file(base_url, name):
     url = f"{base_url}/{name}"
-    with urllib.request.urlopen(url) as file:
+    request = urllib.request.Request(url, data=None, headers={"User-Agent": "Bumble"})
+    with urllib.request.urlopen(request) as file:
         data = file.read()
         print(f"Downloaded {name}: {len(data)} bytes")
         return data

tools/rtk_fw_download.py

@@ -25,6 +25,7 @@ import click
 from bumble.colors import color
 from bumble.drivers import rtk
 from bumble.tools import rtk_util
+import bumble.logging
 
 
 # -----------------------------------------------------------------------------
@@ -58,7 +59,9 @@ def download_file(base_url, name, remove_suffix):
         name = name.replace(".bin", "")
 
     url = f"{base_url}/{name}"
-    with urllib.request.urlopen(url) as file:
+    logger.debug(f"downloading {url}")
+    request = urllib.request.Request(url, data=None, headers={"User-Agent": "Bumble"})
+    with urllib.request.urlopen(request) as file:
         data = file.read()
         print(f"Downloaded {name}: {len(data)} bytes")
         return data
@@ -84,6 +87,7 @@ def download_file(base_url, name, remove_suffix):
 @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."""
+    bumble.logging.setup_basic_logging()
 
     # Check that the output dir exists
     if output_dir == '':

tools/rtk_util.py

@@ -20,7 +20,7 @@ import logging
 
 import click
 
-from bumble import transport
+from bumble import company_ids, hci, transport
 from bumble.host import Host
 from bumble.drivers import rtk
 import bumble.logging
@@ -62,10 +62,22 @@ async def do_load(usb_transport, force):
         # Get the driver.
         driver = await rtk.Driver.for_host(host, force)
         if driver is None:
-            print("Firmware already loaded or no supported driver for this device.")
+            # Try to see if there's already a FW image loaded
+            firmware_version = await rtk.Driver.get_loaded_firmware_version(host)
+            if firmware_version is None:
+                print("Device not supported")
+                return
+
+            print(f"Firmware already loaded: 0x{firmware_version:08X}")
             return
 
-        await driver.download_firmware()
+        firmware_version = await driver.download_firmware()
+
+        if firmware_version is None:
+            print("Failed to load firmware")
+            return
+
+        print(f"Loaded firmware version 0x{firmware_version:08X}")
 
 
 # -----------------------------------------------------------------------------
@@ -107,7 +119,13 @@ async def do_info(usb_transport, force):
                 f"  Config:   {driver_info.config_name}\n"
             )
         else:
-            print("Firmware already loaded or no supported driver for this device.")
+            # Try to see if there's already a FW image loaded
+            firmware_version = await rtk.Driver.get_loaded_firmware_version(host)
+            if firmware_version is None:
+                print("Device not supported")
+                return
+
+            print(f"Firmware loaded: 0x{firmware_version:08X}")
 
 
 # -----------------------------------------------------------------------------