From e55285830517e8522fab1214064ce8e51f9bffc4 Mon Sep 17 00:00:00 2001 From: Charlie Boutier Date: Wed, 13 Dec 2023 00:16:44 +0000 Subject: [PATCH] intel driver: adapt to new way to load driver --- bumble/drivers/__init__.py | 3 +-- bumble/drivers/intel.py | 43 +++++++++++++++++++++++++------------- bumble/host.py | 15 +++++++------ tools/intel_util.py | 11 +++++----- 4 files changed, 42 insertions(+), 30 deletions(-) diff --git a/bumble/drivers/__init__.py b/bumble/drivers/__init__.py index d6d5f84c..a6862bac 100644 --- a/bumble/drivers/__init__.py +++ b/bumble/drivers/__init__.py @@ -24,9 +24,8 @@ import pathlib import platform from typing import Dict, Iterable, Optional, Type, TYPE_CHECKING -from . import rtk, intel -from . import rtk +from . import rtk, intel from .common import Driver if TYPE_CHECKING: diff --git a/bumble/drivers/intel.py b/bumble/drivers/intel.py index 25c76d9c..e23f2b94 100644 --- a/bumble/drivers/intel.py +++ b/bumble/drivers/intel.py @@ -20,20 +20,23 @@ # ----------------------------------------------------------------------------- # Imports # ----------------------------------------------------------------------------- -import logging +import asyncio from dataclasses import dataclass +from enum import IntEnum +import logging +import os +import pathlib +import platform +from typing import Optional, Tuple + from bumble.hci import ( hci_vendor_command_op_code, # type: ignore HCI_Command, + HCI_Reset_Command, STATUS_SPEC, # type: ignore HCI_SUCCESS, ) -from enum import IntEnum -from typing import Optional -import asyncio -import os -import pathlib -import platform +from bumble.drivers import common # ----------------------------------------------------------------------------- # Logging @@ -217,18 +220,22 @@ async def download_fw_payload(host, fw: bytes, header_offset: int): frag_len = 0 -class Driver: - def __init__(self, host, version: IntelVersionTLV, firmware: bytes): +class Driver(common.Driver): + def __init__(self, host, version: IntelVersionTLV, firmware: bytes, fw_name: str): self.host = host self.version = version self.firmware = firmware + self.fw_name = fw_name @classmethod - async def for_host(cls, host): # type: ignore + async def for_host(cls, host, force=False): # type: ignore try: + if not force and not cls.check(host): + return None + version = await fetch_intel_version(host) # type: ignore - fw = prepare_firmware(version) - return cls(host, version, fw) + fw, fw_name = prepare_firmware(version) + return cls(host, version, fw, fw_name) except Exception: logging.exception("Error preparing the firmware") return None @@ -236,10 +243,18 @@ async def for_host(cls, host): # type: ignore async def init_controller(self): try: await download_firmware(self.host, self.version, self.firmware) + await self.host.send_command(HCI_Reset_Command(), check_result=True) + logger.info(f"loaded FW image {self.fw_name}") except Exception: logging.exception("Failed to download the firmware") return None + @staticmethod + def check(host): + if host.hci_metadata.get('driver') == 'intel': + # Forced driver + return True + @staticmethod def find_binary_path(file_name: str) -> Optional[pathlib.Path]: # First check if an environment variable is set @@ -334,7 +349,7 @@ def fetch_firmware_name(version: IntelVersionTLV) -> str: return f"ibt-{upper_name:04x}-{lower_name:04x}.sfi" -def prepare_firmware(version: IntelVersionTLV) -> bytes: +def prepare_firmware(version: IntelVersionTLV) -> Tuple[bytes, str]: fw_name = fetch_firmware_name(version) logging.debug(f"Firmware: {fw_name}") fw_path = Driver.find_binary_path(fw_name) @@ -346,7 +361,7 @@ def prepare_firmware(version: IntelVersionTLV) -> bytes: raise ValueError( "Firmware size is less then the minimum required size of 644 bytes" ) - return fw + return (fw, fw_name) async def download_firmware(host, version: IntelVersionTLV, fw: bytes): diff --git a/bumble/host.py b/bumble/host.py index 45862b2a..190ab89e 100644 --- a/bumble/host.py +++ b/bumble/host.py @@ -469,14 +469,13 @@ def supported_le_features(self): # Packet Sink protocol (packets coming from the controller via HCI) def on_packet(self, packet: bytes) -> None: hci_packet = HCI_Packet.from_bytes(packet) - self.on_hci_packet(hci_packet) - # if self.ready or ( - # isinstance(hci_packet, HCI_Command_Complete_Event) - # and hci_packet.command_opcode == HCI_RESET_COMMAND - # ): - # self.on_hci_packet(hci_packet) - # else: - # logger.debug('`reset not done`, ignoring packet from controller') + if self.ready or ( + isinstance(hci_packet, HCI_Command_Complete_Event) + and hci_packet.command_opcode == HCI_RESET_COMMAND + ): + self.on_hci_packet(hci_packet) + else: + logger.debug('reset not done, ignoring packet from controller') def on_transport_lost(self): # Called by the source when the transport has been lost. diff --git a/tools/intel_util.py b/tools/intel_util.py index b714ed15..f140a72e 100644 --- a/tools/intel_util.py +++ b/tools/intel_util.py @@ -38,10 +38,8 @@ async def do_load(transport: str, force: bool): ): # Create a host to communicate with the device host = Host(hci_source, hci_sink) - if not force: - await host.reset(driver_factory=None) - driver = await intel.Driver.for_host(host) + driver = await intel.Driver.for_host(host, force) if not driver: print("Firmware already loaded or no supported driver for this device.") return @@ -61,12 +59,13 @@ async def do_info(transport: str, force: bool): ): # Create a host to communicate with the device host = Host(hci_source, hci_sink) - if not force: - await host.reset(driver_factory=None) + if not force and not intel.Driver.check(host): + print("Device not supported by this Intel driver") + return version = await intel.fetch_intel_version(host) if not version: - print("Device not supported by this RTK driver") + print("Device not supported by this Intel driver") return try: fw_name = intel.fetch_firmware_name(version)