# Copyright 2021-2025 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 # ----------------------------------------------------------------------------- from __future__ import annotations import asyncio import collections import dataclasses import logging from collections.abc import Awaitable, Callable from typing import TYPE_CHECKING, Any, TypeVar, cast, overload from bumble import drivers, hci, utils from bumble.colors import color from bumble.core import ( ConnectionPHY, InvalidStateError, PhysicalTransport, ) from bumble.l2cap import L2CAP_PDU from bumble.snoop import Snooper from bumble.transport.common import TransportLostError if TYPE_CHECKING: from bumble.transport.common import TransportSink, TransportSource # ----------------------------------------------------------------------------- # Logging # ----------------------------------------------------------------------------- logger = logging.getLogger(__name__) # ----------------------------------------------------------------------------- class DataPacketQueue(utils.EventEmitter): """ Flow-control queue for host->controller data packets (ACL, ISO). The queue holds packets associated with a connection handle. The packets are sent to the controller, up to a maximum total number of packets in flight. A packet is considered to be "in flight" when it has been sent to the controller but not completed yet. Packets are no longer "in flight" when the controller declares them as completed. The queue emits a 'flow' event whenever one or more packets are completed. """ max_packet_size: int class PerConnectionState: def __init__(self) -> None: self.in_flight = 0 self.drained = asyncio.Event() def __init__( self, max_packet_size: int, max_in_flight: int, send: Callable[[hci.HCI_Packet], None], ) -> None: super().__init__() self.max_packet_size = max_packet_size self.max_in_flight = max_in_flight self._in_flight = 0 # Total number of packets in flight across all connections self._connection_state: dict[int, DataPacketQueue.PerConnectionState] = ( collections.defaultdict(DataPacketQueue.PerConnectionState) ) self._drained_per_connection: dict[int, asyncio.Event] = ( collections.defaultdict(asyncio.Event) ) self._send = send self._packets: collections.deque[tuple[hci.HCI_Packet, int]] = ( collections.deque() ) self._queued = 0 self._completed = 0 @property def queued(self) -> int: """Total number of packets queued since creation.""" return self._queued @property def completed(self) -> int: """Total number of packets completed since creation.""" return self._completed @property def pending(self) -> int: """Number of packets that have been queued but not completed.""" return self._queued - self._completed def enqueue(self, packet: hci.HCI_Packet, connection_handle: int) -> None: """Enqueue a packet associated with a connection""" self._packets.appendleft((packet, connection_handle)) self._queued += 1 self._check_queue() if self._packets: logger.debug( f'{self._in_flight} packets in flight, {len(self._packets)} in queue' ) def flush(self, connection_handle: int) -> None: """ Remove all packets associated with a connection. All packets associated with the connection that are in flight are implicitly marked as completed, but no 'flow' event is emitted. """ packets_to_keep = [ (packet, handle) for (packet, handle) in self._packets if handle != connection_handle ] if flushed_count := len(self._packets) - len(packets_to_keep): self._completed += flushed_count self._packets = collections.deque(packets_to_keep) if connection_state := self._connection_state.pop(connection_handle, None): in_flight = connection_state.in_flight self._completed += in_flight self._in_flight -= in_flight connection_state.drained.set() def _check_queue(self) -> None: while self._packets and self._in_flight < self.max_in_flight: packet, connection_handle = self._packets.pop() self._send(packet) self._in_flight += 1 connection_state = self._connection_state[connection_handle] connection_state.in_flight += 1 connection_state.drained.clear() def on_packets_completed(self, packet_count: int, connection_handle: int) -> None: """Mark one or more packets associated with a connection as completed.""" if connection_handle not in self._connection_state: logger.warning( f'received completion for unknown connection {connection_handle}' ) return connection_state = self._connection_state[connection_handle] if packet_count <= connection_state.in_flight: connection_state.in_flight -= packet_count else: logger.warning( f'{packet_count} completed for {connection_handle} ' f'but only {connection_state.in_flight} in flight' ) connection_state.in_flight = 0 if connection_state.in_flight == 0: connection_state.drained.set() if packet_count <= self._in_flight: self._in_flight -= packet_count self._completed += packet_count else: logger.warning( f'{packet_count} completed but only {self._in_flight} in flight' ) self._in_flight = 0 self._completed = self._queued self._check_queue() self.emit('flow') async def drain(self, connection_handle: int) -> None: """Wait until there are no pending packets for a connection.""" if not (connection_state := self._connection_state.get(connection_handle)): raise ValueError('no such connection') await connection_state.drained.wait() # ----------------------------------------------------------------------------- class Connection: def __init__( self, host: Host, handle: int, peer_address: hci.Address, transport: PhysicalTransport, ): self.host = host self.handle = handle self.peer_address = peer_address self.assembler = hci.HCI_AclDataPacketAssembler(self.on_acl_pdu) self.transport = transport acl_packet_queue: DataPacketQueue | None = ( host.le_acl_packet_queue if transport == PhysicalTransport.LE else host.acl_packet_queue ) assert acl_packet_queue self.acl_packet_queue = acl_packet_queue def on_hci_acl_data_packet(self, packet: hci.HCI_AclDataPacket) -> None: self.assembler.feed_packet(packet) def on_acl_pdu(self, pdu: bytes) -> None: l2cap_pdu = L2CAP_PDU.from_bytes(pdu) self.host.on_l2cap_pdu(self, l2cap_pdu.cid, l2cap_pdu.payload) def __str__(self) -> str: return ( f'Connection(transport={self.transport}, peer_address={self.peer_address})' ) # ----------------------------------------------------------------------------- @dataclasses.dataclass class ScoLink: peer_address: hci.Address connection_handle: int # ----------------------------------------------------------------------------- @dataclasses.dataclass class IsoLink: handle: int packet_queue: DataPacketQueue = dataclasses.field(repr=False) packet_sequence_number: int = 0 # ----------------------------------------------------------------------------- _RP = TypeVar('_RP', bound=hci.HCI_ReturnParameters) class Host(utils.EventEmitter): connections: dict[int, Connection] cis_links: dict[int, IsoLink] bis_links: dict[int, IsoLink] sco_links: dict[int, ScoLink] bigs: dict[int, set[int]] acl_packet_queue: DataPacketQueue | None = None le_acl_packet_queue: DataPacketQueue | None = None iso_packet_queue: DataPacketQueue | None = None hci_sink: TransportSink | None = None hci_metadata: dict[str, Any] long_term_key_provider: Callable[[int, bytes, int], Awaitable[bytes | None]] | None link_key_provider: Callable[[hci.Address], Awaitable[bytes | None]] | None def __init__( self, controller_source: TransportSource | None = None, controller_sink: TransportSink | None = None, ) -> None: super().__init__() self.hci_metadata = {} self.ready = False # True when we can accept incoming packets self.connections = {} # Connections, by connection handle self.cis_links = {} # CIS links, by connection handle self.bis_links = {} # BIS links, by connection handle self.sco_links = {} # SCO links, by connection handle self.bigs = {} # BIG Handle to BIS Handles self.pending_command: hci.HCI_SyncCommand | hci.HCI_AsyncCommand | None = None self.pending_response: ( asyncio.Future[ hci.HCI_Command_Complete_Event | hci.HCI_Command_Status_Event ] | None ) = None self.number_of_supported_advertising_sets = 0 self.maximum_advertising_data_length = 31 self.local_version: ( hci.HCI_Read_Local_Version_Information_ReturnParameters | None ) = None self.local_supported_commands = 0 self.local_le_features = hci.LeFeatureMask(0) # LE features self.local_lmp_features = hci.LmpFeatureMask(0) # Classic LMP features self.suggested_max_tx_octets = 251 # Max allowed self.suggested_max_tx_time = 2120 # Max allowed self.command_semaphore = asyncio.Semaphore(1) self.long_term_key_provider = None self.link_key_provider = None self.pairing_io_capability_provider = None # Classic only self.snooper: Snooper | None = None # Connect to the source and sink if specified if controller_source: self.set_packet_source(controller_source) if controller_sink: self.set_packet_sink(controller_sink) def find_connection_by_bd_addr( self, bd_addr: hci.Address, transport: int | None = None, check_address_type: bool = False, ) -> Connection | None: for connection in self.connections.values(): if bytes(connection.peer_address) == bytes(bd_addr): if ( check_address_type and connection.peer_address.address_type != bd_addr.address_type ): continue if transport is None or connection.transport == transport: return connection return None async def flush(self) -> None: # Make sure no command is pending await self.command_semaphore.acquire() # Flush current host state, then release command semaphore self.emit('flush') self.command_semaphore.release() async def reset(self, driver_factory=drivers.get_driver_for_host) -> None: if self.ready: self.ready = False await self.flush() # Instantiate and init a driver for the host if needed. # NOTE: we don't keep a reference to the driver here, because we don't # currently have a need for the driver later on. But if the driver interface # evolves, it may be required, then, to store a reference to the driver in # an object property. reset_needed = True if driver_factory is not None: if driver := await driver_factory(self): await driver.init_controller() reset_needed = False # Send a reset command unless a driver has already done so. if reset_needed: await self.send_sync_command(hci.HCI_Reset_Command()) self.ready = True response1 = await self.send_sync_command( hci.HCI_Read_Local_Supported_Commands_Command() ) self.local_supported_commands = int.from_bytes( response1.supported_commands, 'little' ) if self.supports_command(hci.HCI_READ_LOCAL_VERSION_INFORMATION_COMMAND): self.local_version = await self.send_sync_command( hci.HCI_Read_Local_Version_Information_Command() ) if self.supports_command(hci.HCI_LE_READ_ALL_LOCAL_SUPPORTED_FEATURES_COMMAND): response2 = await self.send_sync_command( hci.HCI_LE_Read_All_Local_Supported_Features_Command() ) self.local_le_features = hci.LeFeatureMask( int.from_bytes(response2.le_features, 'little') ) elif self.supports_command(hci.HCI_LE_READ_LOCAL_SUPPORTED_FEATURES_COMMAND): response3 = await self.send_sync_command( hci.HCI_LE_Read_Local_Supported_Features_Command() ) self.local_le_features = hci.LeFeatureMask( int.from_bytes(response3.le_features, 'little') ) if self.supports_command(hci.HCI_READ_LOCAL_EXTENDED_FEATURES_COMMAND): max_page_number = 0 page_number = 0 lmp_features = 0 while page_number <= max_page_number: response4 = await self.send_sync_command( hci.HCI_Read_Local_Extended_Features_Command( page_number=page_number ) ) lmp_features |= int.from_bytes( response4.extended_lmp_features, 'little' ) << (64 * page_number) max_page_number = response4.maximum_page_number page_number += 1 self.local_lmp_features = hci.LmpFeatureMask(lmp_features) elif self.supports_command(hci.HCI_READ_LOCAL_SUPPORTED_FEATURES_COMMAND): response5 = await self.send_sync_command( hci.HCI_Read_Local_Supported_Features_Command() ) self.local_lmp_features = hci.LmpFeatureMask( int.from_bytes(response5.lmp_features, 'little') ) await self.send_sync_command( hci.HCI_Set_Event_Mask_Command( event_mask=hci.HCI_Set_Event_Mask_Command.mask( [ hci.HCI_INQUIRY_COMPLETE_EVENT, hci.HCI_INQUIRY_RESULT_EVENT, hci.HCI_CONNECTION_COMPLETE_EVENT, hci.HCI_CONNECTION_REQUEST_EVENT, hci.HCI_DISCONNECTION_COMPLETE_EVENT, hci.HCI_AUTHENTICATION_COMPLETE_EVENT, hci.HCI_REMOTE_NAME_REQUEST_COMPLETE_EVENT, hci.HCI_ENCRYPTION_CHANGE_EVENT, hci.HCI_CHANGE_CONNECTION_LINK_KEY_COMPLETE_EVENT, hci.HCI_LINK_KEY_TYPE_CHANGED_EVENT, hci.HCI_READ_REMOTE_SUPPORTED_FEATURES_COMPLETE_EVENT, hci.HCI_READ_REMOTE_VERSION_INFORMATION_COMPLETE_EVENT, hci.HCI_QOS_SETUP_COMPLETE_EVENT, hci.HCI_HARDWARE_ERROR_EVENT, hci.HCI_FLUSH_OCCURRED_EVENT, hci.HCI_ROLE_CHANGE_EVENT, hci.HCI_MODE_CHANGE_EVENT, hci.HCI_RETURN_LINK_KEYS_EVENT, hci.HCI_PIN_CODE_REQUEST_EVENT, hci.HCI_LINK_KEY_REQUEST_EVENT, hci.HCI_LINK_KEY_NOTIFICATION_EVENT, hci.HCI_LOOPBACK_COMMAND_EVENT, hci.HCI_DATA_BUFFER_OVERFLOW_EVENT, hci.HCI_MAX_SLOTS_CHANGE_EVENT, hci.HCI_READ_CLOCK_OFFSET_COMPLETE_EVENT, hci.HCI_CONNECTION_PACKET_TYPE_CHANGED_EVENT, hci.HCI_QOS_VIOLATION_EVENT, hci.HCI_PAGE_SCAN_REPETITION_MODE_CHANGE_EVENT, hci.HCI_FLOW_SPECIFICATION_COMPLETE_EVENT, hci.HCI_INQUIRY_RESULT_WITH_RSSI_EVENT, hci.HCI_READ_REMOTE_EXTENDED_FEATURES_COMPLETE_EVENT, hci.HCI_SYNCHRONOUS_CONNECTION_COMPLETE_EVENT, hci.HCI_SYNCHRONOUS_CONNECTION_CHANGED_EVENT, hci.HCI_SNIFF_SUBRATING_EVENT, hci.HCI_EXTENDED_INQUIRY_RESULT_EVENT, hci.HCI_ENCRYPTION_KEY_REFRESH_COMPLETE_EVENT, hci.HCI_IO_CAPABILITY_REQUEST_EVENT, hci.HCI_IO_CAPABILITY_RESPONSE_EVENT, hci.HCI_USER_CONFIRMATION_REQUEST_EVENT, hci.HCI_USER_PASSKEY_REQUEST_EVENT, hci.HCI_REMOTE_OOB_DATA_REQUEST_EVENT, hci.HCI_SIMPLE_PAIRING_COMPLETE_EVENT, hci.HCI_LINK_SUPERVISION_TIMEOUT_CHANGED_EVENT, hci.HCI_ENHANCED_FLUSH_COMPLETE_EVENT, hci.HCI_USER_PASSKEY_NOTIFICATION_EVENT, hci.HCI_KEYPRESS_NOTIFICATION_EVENT, hci.HCI_REMOTE_HOST_SUPPORTED_FEATURES_NOTIFICATION_EVENT, hci.HCI_LE_META_EVENT, ] ) ) ) if self.supports_command(hci.HCI_SET_EVENT_MASK_PAGE_2_COMMAND): await self.send_sync_command( hci.HCI_Set_Event_Mask_Page_2_Command( event_mask_page_2=hci.HCI_Set_Event_Mask_Page_2_Command.mask( [hci.HCI_ENCRYPTION_CHANGE_V2_EVENT] ) ) ) if ( self.local_version is not None and self.local_version.hci_version <= hci.HCI_VERSION_BLUETOOTH_CORE_4_0 ): # Some older controllers don't like event masks with bits they don't # understand le_event_mask = bytes.fromhex('1F00000000000000') else: le_event_mask = hci.HCI_LE_Set_Event_Mask_Command.mask( [ hci.HCI_LE_CONNECTION_COMPLETE_EVENT, hci.HCI_LE_ADVERTISING_REPORT_EVENT, hci.HCI_LE_CONNECTION_UPDATE_COMPLETE_EVENT, hci.HCI_LE_READ_REMOTE_FEATURES_COMPLETE_EVENT, hci.HCI_LE_LONG_TERM_KEY_REQUEST_EVENT, hci.HCI_LE_REMOTE_CONNECTION_PARAMETER_REQUEST_EVENT, hci.HCI_LE_DATA_LENGTH_CHANGE_EVENT, hci.HCI_LE_READ_LOCAL_P_256_PUBLIC_KEY_COMPLETE_EVENT, hci.HCI_LE_GENERATE_DHKEY_COMPLETE_EVENT, hci.HCI_LE_ENHANCED_CONNECTION_COMPLETE_EVENT, hci.HCI_LE_ENHANCED_CONNECTION_COMPLETE_V2_EVENT, hci.HCI_LE_DIRECTED_ADVERTISING_REPORT_EVENT, hci.HCI_LE_PHY_UPDATE_COMPLETE_EVENT, hci.HCI_LE_EXTENDED_ADVERTISING_REPORT_EVENT, hci.HCI_LE_PERIODIC_ADVERTISING_SYNC_ESTABLISHED_EVENT, hci.HCI_LE_PERIODIC_ADVERTISING_REPORT_EVENT, hci.HCI_LE_PERIODIC_ADVERTISING_SYNC_LOST_EVENT, hci.HCI_LE_SCAN_TIMEOUT_EVENT, hci.HCI_LE_ADVERTISING_SET_TERMINATED_EVENT, hci.HCI_LE_SCAN_REQUEST_RECEIVED_EVENT, hci.HCI_LE_CONNECTIONLESS_IQ_REPORT_EVENT, hci.HCI_LE_CONNECTION_IQ_REPORT_EVENT, hci.HCI_LE_CTE_REQUEST_FAILED_EVENT, hci.HCI_LE_PERIODIC_ADVERTISING_SYNC_TRANSFER_RECEIVED_EVENT, hci.HCI_LE_CIS_ESTABLISHED_EVENT, hci.HCI_LE_CIS_REQUEST_EVENT, hci.HCI_LE_CREATE_BIG_COMPLETE_EVENT, hci.HCI_LE_TERMINATE_BIG_COMPLETE_EVENT, hci.HCI_LE_BIG_SYNC_ESTABLISHED_EVENT, hci.HCI_LE_BIG_SYNC_LOST_EVENT, hci.HCI_LE_REQUEST_PEER_SCA_COMPLETE_EVENT, hci.HCI_LE_PATH_LOSS_THRESHOLD_EVENT, hci.HCI_LE_TRANSMIT_POWER_REPORTING_EVENT, hci.HCI_LE_BIGINFO_ADVERTISING_REPORT_EVENT, hci.HCI_LE_SUBRATE_CHANGE_EVENT, hci.HCI_LE_READ_ALL_REMOTE_FEATURES_COMPLETE_EVENT, hci.HCI_LE_CS_READ_REMOTE_SUPPORTED_CAPABILITIES_COMPLETE_EVENT, hci.HCI_LE_CS_PROCEDURE_ENABLE_COMPLETE_EVENT, hci.HCI_LE_CS_SECURITY_ENABLE_COMPLETE_EVENT, hci.HCI_LE_CS_CONFIG_COMPLETE_EVENT, hci.HCI_LE_CS_SUBEVENT_RESULT_EVENT, hci.HCI_LE_CS_SUBEVENT_RESULT_CONTINUE_EVENT, hci.HCI_LE_MONITORED_ADVERTISERS_REPORT_EVENT, hci.HCI_LE_FRAME_SPACE_UPDATE_COMPLETE_EVENT, hci.HCI_LE_UTP_RECEIVE_EVENT, hci.HCI_LE_CONNECTION_RATE_CHANGE_EVENT, ] ) await self.send_sync_command( hci.HCI_LE_Set_Event_Mask_Command(le_event_mask=le_event_mask) ) if self.supports_command(hci.HCI_READ_BUFFER_SIZE_COMMAND): response6 = await self.send_sync_command(hci.HCI_Read_Buffer_Size_Command()) hc_acl_data_packet_length = response6.hc_acl_data_packet_length hc_total_num_acl_data_packets = response6.hc_total_num_acl_data_packets logger.debug( 'HCI ACL flow control: ' f'hc_acl_data_packet_length={hc_acl_data_packet_length},' f'hc_total_num_acl_data_packets={hc_total_num_acl_data_packets}' ) self.acl_packet_queue = DataPacketQueue( max_packet_size=hc_acl_data_packet_length, max_in_flight=hc_total_num_acl_data_packets, send=self.send_hci_packet, ) le_acl_data_packet_length = 0 total_num_le_acl_data_packets = 0 iso_data_packet_length = 0 total_num_iso_data_packets = 0 if self.supports_command(hci.HCI_LE_READ_BUFFER_SIZE_V2_COMMAND): response7 = await self.send_sync_command( hci.HCI_LE_Read_Buffer_Size_V2_Command() ) le_acl_data_packet_length = response7.le_acl_data_packet_length total_num_le_acl_data_packets = response7.total_num_le_acl_data_packets iso_data_packet_length = response7.iso_data_packet_length total_num_iso_data_packets = response7.total_num_iso_data_packets logger.debug( 'HCI LE flow control: ' f'le_acl_data_packet_length={le_acl_data_packet_length},' f'total_num_le_acl_data_packets={total_num_le_acl_data_packets},' f'iso_data_packet_length={iso_data_packet_length},' f'total_num_iso_data_packets={total_num_iso_data_packets}' ) elif self.supports_command(hci.HCI_LE_READ_BUFFER_SIZE_COMMAND): response8 = await self.send_sync_command( hci.HCI_LE_Read_Buffer_Size_Command() ) le_acl_data_packet_length = response8.le_acl_data_packet_length total_num_le_acl_data_packets = response8.total_num_le_acl_data_packets logger.debug( 'HCI LE ACL flow control: ' f'le_acl_data_packet_length={le_acl_data_packet_length},' f'total_num_le_acl_data_packets={total_num_le_acl_data_packets}' ) if le_acl_data_packet_length == 0 or total_num_le_acl_data_packets == 0: # LE and Classic share the same queue self.le_acl_packet_queue = self.acl_packet_queue else: # Create a separate queue for LE self.le_acl_packet_queue = DataPacketQueue( max_packet_size=le_acl_data_packet_length, max_in_flight=total_num_le_acl_data_packets, send=self.send_hci_packet, ) if iso_data_packet_length and total_num_iso_data_packets: self.iso_packet_queue = DataPacketQueue( max_packet_size=iso_data_packet_length, max_in_flight=total_num_iso_data_packets, send=self.send_hci_packet, ) if self.supports_command( hci.HCI_LE_READ_SUGGESTED_DEFAULT_DATA_LENGTH_COMMAND ) and self.supports_command( hci.HCI_LE_WRITE_SUGGESTED_DEFAULT_DATA_LENGTH_COMMAND ): response9 = await self.send_sync_command( hci.HCI_LE_Read_Suggested_Default_Data_Length_Command() ) suggested_max_tx_octets = response9.suggested_max_tx_octets suggested_max_tx_time = response9.suggested_max_tx_time if ( suggested_max_tx_octets != self.suggested_max_tx_octets or suggested_max_tx_time != self.suggested_max_tx_time ): await self.send_sync_command( hci.HCI_LE_Write_Suggested_Default_Data_Length_Command( suggested_max_tx_octets=self.suggested_max_tx_octets, suggested_max_tx_time=self.suggested_max_tx_time, ) ) if self.supports_command( hci.HCI_LE_READ_NUMBER_OF_SUPPORTED_ADVERTISING_SETS_COMMAND ): try: response10 = await self.send_sync_command( hci.HCI_LE_Read_Number_Of_Supported_Advertising_Sets_Command() ) self.number_of_supported_advertising_sets = ( response10.num_supported_advertising_sets ) except hci.HCI_Error: logger.warning('Failed to read number of supported advertising sets') if self.supports_command( hci.HCI_LE_READ_MAXIMUM_ADVERTISING_DATA_LENGTH_COMMAND ): try: response11 = await self.send_sync_command( hci.HCI_LE_Read_Maximum_Advertising_Data_Length_Command() ) self.maximum_advertising_data_length = ( response11.max_advertising_data_length ) except hci.HCI_Error: logger.warning('Failed to read maximum advertising data length') @property def controller(self) -> TransportSink | None: return self.hci_sink @controller.setter def controller(self, controller) -> None: self.set_packet_sink(controller) if controller: self.set_packet_source(controller) def set_packet_sink(self, sink: TransportSink | None) -> None: self.hci_sink = sink def set_packet_source(self, source: TransportSource) -> None: source.set_packet_sink(self) self.hci_metadata = getattr(source, 'metadata', self.hci_metadata) def send_hci_packet(self, packet: hci.HCI_Packet) -> None: logger.debug(f'{color("### HOST -> CONTROLLER", "blue")}: {packet}') if self.snooper: self.snooper.snoop(bytes(packet), Snooper.Direction.HOST_TO_CONTROLLER) if self.hci_sink: self.hci_sink.on_packet(bytes(packet)) async def _send_command( self, command: hci.HCI_SyncCommand | hci.HCI_AsyncCommand, response_timeout: float | None = None, ) -> hci.HCI_Command_Complete_Event | hci.HCI_Command_Status_Event: # Wait until we can send (only one pending command at a time) await self.command_semaphore.acquire() # Create a future value to hold the eventual response assert self.pending_command is None assert self.pending_response is None self.pending_response = asyncio.get_running_loop().create_future() self.pending_command = command response: ( hci.HCI_Command_Complete_Event | hci.HCI_Command_Status_Event | None ) = None try: self.send_hci_packet(command) response = await asyncio.wait_for( self.pending_response, timeout=response_timeout ) return response except Exception: logger.exception(color("!!! Exception while sending command:", "red")) raise finally: self.pending_command = None self.pending_response = None if ( response is not None and response.num_hci_command_packets and self.command_semaphore.locked() ): self.command_semaphore.release() @overload async def send_command( self, command: hci.HCI_SyncCommand[_RP], check_result: bool = False, response_timeout: float | None = None, ) -> hci.HCI_Command_Complete_Event[_RP]: ... @overload async def send_command( self, command: hci.HCI_AsyncCommand, check_result: bool = False, response_timeout: float | None = None, ) -> hci.HCI_Command_Status_Event: ... async def send_command( self, command: hci.HCI_SyncCommand[_RP] | hci.HCI_AsyncCommand, check_result: bool = False, response_timeout: float | None = None, ) -> hci.HCI_Command_Complete_Event[_RP] | hci.HCI_Command_Status_Event: response = await self._send_command(command, response_timeout) # Check the return parameters if required if check_result: if isinstance(response, hci.HCI_Command_Status_Event): status = response.status # type: ignore[attr-defined] elif isinstance(response.return_parameters, int): status = response.return_parameters elif isinstance(response.return_parameters, bytes): # return parameters first field is a one byte status code status = response.return_parameters[0] elif isinstance( response.return_parameters, hci.HCI_GenericReturnParameters ): # FIXME: temporary workaround # NO STATUS status = hci.HCI_SUCCESS else: status = response.return_parameters.status if status != hci.HCI_SUCCESS: logger.warning( f'{command.name} failed ' f'({hci.HCI_Constant.error_name(status)})' ) raise hci.HCI_Error(status) return response async def send_sync_command( self, command: hci.HCI_SyncCommand[_RP], response_timeout: float | None = None ) -> _RP: response = await self.send_sync_command_raw(command, response_timeout) return_parameters = response.return_parameters # Check the return parameters's status if isinstance(return_parameters, hci.HCI_StatusReturnParameters): status = return_parameters.status elif isinstance(return_parameters, hci.HCI_GenericReturnParameters): # if the payload has at least one byte, assume the first byte is the status if not return_parameters.data: raise RuntimeError('no status byte in return parameters') status = hci.HCI_ErrorCode(return_parameters.data[0]) else: raise RuntimeError( f'unexpected return parameters type ({type(return_parameters)})' ) if status != hci.HCI_ErrorCode.SUCCESS: logger.warning( f'{command.name} failed ' f'({hci.HCI_Constant.error_name(status)})' ) raise hci.HCI_Error(status) return return_parameters async def send_sync_command_raw( self, command: hci.HCI_SyncCommand[_RP], response_timeout: float | None = None, ) -> hci.HCI_Command_Complete_Event[_RP]: response = await self._send_command(command, response_timeout) # For unknown HCI commands, some controllers return Command Status instead of # Command Complete. if ( isinstance(response, hci.HCI_Command_Status_Event) and response.status == hci.HCI_ErrorCode.UNKNOWN_HCI_COMMAND_ERROR ): return hci.HCI_Command_Complete_Event( num_hci_command_packets=response.num_hci_command_packets, command_opcode=command.op_code, return_parameters=hci.HCI_StatusReturnParameters( status=hci.HCI_ErrorCode(response.status) ), # type: ignore ) # Check that the response is of the expected type assert isinstance(response, hci.HCI_Command_Complete_Event) return response async def send_async_command( self, command: hci.HCI_AsyncCommand, check_status: bool = True, response_timeout: float | None = None, ) -> hci.HCI_ErrorCode: response = await self._send_command(command, response_timeout) # For unknown HCI commands, some controllers return Command Complete instead of # Command Status. if isinstance(response, hci.HCI_Command_Complete_Event): # Assume the first byte of the return parameters is the status if ( status := hci.HCI_ErrorCode(response.parameters[3]) ) != hci.HCI_ErrorCode.UNKNOWN_HCI_COMMAND_ERROR: logger.warning(f'unexpected return paramerers status {status}') else: assert isinstance(response, hci.HCI_Command_Status_Event) status = hci.HCI_ErrorCode(response.status) # Check the status if required if check_status: if status != hci.HCI_CommandStatus.PENDING: logger.warning(f'{command.name} failed ' f'({status.name})') raise hci.HCI_Error(status) return status @utils.deprecated("Use utils.AsyncRunner.spawn() instead.") def send_command_sync(self, command: hci.HCI_AsyncCommand) -> None: utils.AsyncRunner.spawn(self.send_async_command(command)) def send_acl_sdu(self, connection_handle: int, sdu: bytes) -> None: if not (connection := self.connections.get(connection_handle)): logger.warning(f'connection 0x{connection_handle:04X} not found') return packet_queue = connection.acl_packet_queue if packet_queue is None: logger.warning( f'no ACL packet queue for connection 0x{connection_handle:04X}' ) return # Send the data to the controller via ACL packets max_packet_size = packet_queue.max_packet_size for offset in range(0, len(sdu), max_packet_size): pdu = sdu[offset : offset + max_packet_size] acl_packet = hci.HCI_AclDataPacket( connection_handle=connection_handle, pb_flag=1 if offset > 0 else 0, bc_flag=0, data_total_length=len(pdu), data=pdu, ) logger.debug( '>>> ACL packet enqueue: (handle=0x%04X) %s', connection_handle, pdu.hex(), ) packet_queue.enqueue(acl_packet, connection_handle) def send_sco_sdu(self, connection_handle: int, sdu: bytes) -> None: self.send_hci_packet( hci.HCI_SynchronousDataPacket( connection_handle=connection_handle, packet_status=0, data_total_length=len(sdu), data=sdu, ) ) def send_l2cap_pdu(self, connection_handle: int, cid: int, pdu: bytes) -> None: self.send_acl_sdu(connection_handle, bytes(L2CAP_PDU(cid, pdu))) def get_data_packet_queue(self, connection_handle: int) -> DataPacketQueue | None: if connection := self.connections.get(connection_handle): return connection.acl_packet_queue if iso_link := self.cis_links.get(connection_handle) or self.bis_links.get( connection_handle ): return iso_link.packet_queue return None def send_iso_sdu(self, connection_handle: int, sdu: bytes) -> None: if not ( iso_link := self.cis_links.get(connection_handle) or self.bis_links.get(connection_handle) ): logger.warning(f"no ISO link for connection handle {connection_handle}") return if iso_link.packet_queue is None: logger.warning("ISO link has no data packet queue") return bytes_remaining = len(sdu) offset = 0 while bytes_remaining: is_first_fragment = offset == 0 header_length = 4 if is_first_fragment else 0 assert iso_link.packet_queue.max_packet_size > header_length fragment_length = min( bytes_remaining, iso_link.packet_queue.max_packet_size - header_length ) is_last_fragment = bytes_remaining == fragment_length iso_sdu_fragment = sdu[offset : offset + fragment_length] iso_link.packet_queue.enqueue( ( hci.HCI_IsoDataPacket( connection_handle=connection_handle, data_total_length=header_length + fragment_length, packet_sequence_number=iso_link.packet_sequence_number, pb_flag=0b10 if is_last_fragment else 0b00, packet_status_flag=0, iso_sdu_length=len(sdu), iso_sdu_fragment=iso_sdu_fragment, ) if is_first_fragment else hci.HCI_IsoDataPacket( connection_handle=connection_handle, data_total_length=fragment_length, pb_flag=0b11 if is_last_fragment else 0b01, iso_sdu_fragment=iso_sdu_fragment, ) ), connection_handle, ) offset += fragment_length bytes_remaining -= fragment_length iso_link.packet_sequence_number = (iso_link.packet_sequence_number + 1) & 0xFFFF def remove_big(self, big_handle: int) -> None: if big := self.bigs.pop(big_handle, None): for connection_handle in big: if bis_link := self.bis_links.pop(connection_handle, None): bis_link.packet_queue.flush(bis_link.handle) def supports_command(self, op_code: int) -> bool: return ( self.local_supported_commands & hci.HCI_SUPPORTED_COMMANDS_MASKS.get(op_code, 0) ) != 0 @property def supported_commands(self) -> set[int]: return set( op_code for op_code, mask in hci.HCI_SUPPORTED_COMMANDS_MASKS.items() if self.local_supported_commands & mask ) def supports_le_features(self, features: hci.LeFeatureMask) -> bool: return (self.local_le_features & features) == features def supports_lmp_features(self, features: hci.LmpFeatureMask) -> bool: return self.local_lmp_features & (features) == features @property def supported_le_features(self) -> list[hci.LeFeature]: return [ feature for feature in hci.LeFeature if self.local_le_features & (1 << feature) ] # Packet Sink protocol (packets coming from the controller via HCI) def on_packet(self, packet: bytes) -> None: try: hci_packet = hci.HCI_Packet.from_bytes(packet) except Exception: logger.exception('!!! error parsing packet from bytes') return if self.ready or ( isinstance(hci_packet, hci.HCI_Command_Complete_Event) and hci_packet.command_opcode == hci.HCI_RESET_COMMAND ): self.on_hci_packet(hci_packet) else: logger.debug( f'reset not done, ignoring packet from controller: {hci_packet}' ) def on_transport_lost(self): # Called by the source when the transport has been lost. if self.pending_response: self.pending_response.set_exception(TransportLostError('transport lost')) self.emit('flush') def on_hci_packet(self, packet: hci.HCI_Packet) -> None: logger.debug(f'{color("### CONTROLLER -> HOST", "green")}: {packet}') if self.snooper: self.snooper.snoop(bytes(packet), Snooper.Direction.CONTROLLER_TO_HOST) # If the packet is a command, invoke the handler for this packet if packet.hci_packet_type == hci.HCI_COMMAND_PACKET: self.on_hci_command_packet(cast(hci.HCI_Command, packet)) elif packet.hci_packet_type == hci.HCI_EVENT_PACKET: self.on_hci_event_packet(cast(hci.HCI_Event, packet)) elif packet.hci_packet_type == hci.HCI_ACL_DATA_PACKET: self.on_hci_acl_data_packet(cast(hci.HCI_AclDataPacket, packet)) elif packet.hci_packet_type == hci.HCI_SYNCHRONOUS_DATA_PACKET: self.on_hci_sco_data_packet(cast(hci.HCI_SynchronousDataPacket, packet)) elif packet.hci_packet_type == hci.HCI_ISO_DATA_PACKET: self.on_hci_iso_data_packet(cast(hci.HCI_IsoDataPacket, packet)) else: logger.warning(f'!!! unknown packet type {packet.hci_packet_type}') def on_hci_command_packet(self, command: hci.HCI_Command) -> None: logger.warning(f'!!! unexpected command packet: {command}') def on_hci_event_packet(self, event: hci.HCI_Event) -> None: handler_name = f'on_{event.name.lower()}' handler = getattr(self, handler_name, self.on_hci_event) handler(event) def on_hci_acl_data_packet(self, packet: hci.HCI_AclDataPacket) -> None: # Look for the connection to which this data belongs if connection := self.connections.get(packet.connection_handle): connection.on_hci_acl_data_packet(packet) def on_hci_sco_data_packet(self, packet: hci.HCI_SynchronousDataPacket) -> None: # Experimental self.emit('sco_packet', packet.connection_handle, packet) def on_hci_iso_data_packet(self, packet: hci.HCI_IsoDataPacket) -> None: # Experimental self.emit('iso_packet', packet.connection_handle, packet) def on_l2cap_pdu(self, connection: Connection, cid: int, pdu: bytes) -> None: self.emit('l2cap_pdu', connection.handle, cid, pdu) def on_command_processed( self, event: hci.HCI_Command_Complete_Event | hci.HCI_Command_Status_Event ): if self.pending_response: # Check that it is what we were expecting if self.pending_command is None: logger.warning('!!! pending_command is None ') elif self.pending_command.op_code != event.command_opcode: logger.warning( '!!! command result mismatch, expected ' f'0x{self.pending_command.op_code:X} but got ' f'0x{event.command_opcode:X}' ) self.pending_response.set_result(event) else: logger.warning('!!! no pending response future to set') if event.num_hci_command_packets and self.command_semaphore.locked(): self.command_semaphore.release() ############################################################ # HCI handlers ############################################################ def on_hci_event(self, event: hci.HCI_Event): logger.warning(f'{color(f"--- Ignoring event {event}", "red")}') def on_hci_command_complete_event(self, event: hci.HCI_Command_Complete_Event): if event.command_opcode == 0: # This is used just for the Num_HCI_Command_Packets field, not related to # an actual command logger.debug('no-command event for flow control') # Release the command semaphore if needed if event.num_hci_command_packets and self.command_semaphore.locked(): logger.debug('command complete event releasing semaphore') self.command_semaphore.release() return return self.on_command_processed(event) def on_hci_command_status_event(self, event: hci.HCI_Command_Status_Event): return self.on_command_processed(event) def on_hci_number_of_completed_packets_event( self, event: hci.HCI_Number_Of_Completed_Packets_Event ) -> None: for connection_handle, num_completed_packets in zip( event.connection_handles, event.num_completed_packets ): if queue := self.get_data_packet_queue(connection_handle): queue.on_packets_completed(num_completed_packets, connection_handle) continue if connection_handle not in self.sco_links: logger.warning( 'received packet completion event for unknown handle ' f'0x{connection_handle:04X}' ) # Classic only def on_hci_connection_request_event(self, event: hci.HCI_Connection_Request_Event): # Notify the listeners self.emit( 'connection_request', event.bd_addr, event.class_of_device, event.link_type, ) def on_hci_le_connection_complete_event( self, event: ( hci.HCI_LE_Connection_Complete_Event | hci.HCI_LE_Enhanced_Connection_Complete_Event | hci.HCI_LE_Enhanced_Connection_Complete_V2_Event ), ): # Check if this is a cancellation if event.status == hci.HCI_SUCCESS: # Create/update the connection logger.debug( f'### LE CONNECTION: [0x{event.connection_handle:04X}] ' f'{event.peer_address} as {hci.HCI_Constant.role_name(event.role)}' ) connection = self.connections.get(event.connection_handle) if connection is None: connection = Connection( self, event.connection_handle, event.peer_address, PhysicalTransport.LE, ) self.connections[event.connection_handle] = connection # Notify the client self.emit( 'le_connection', event.connection_handle, event.peer_address, getattr(event, 'local_resolvable_private_address', None), getattr(event, 'peer_resolvable_private_address', None), hci.Role(event.role), event.connection_interval, event.peripheral_latency, event.supervision_timeout, ) else: logger.debug(f'### CONNECTION FAILED: {event.status}') # Notify the listeners self.emit( 'connection_failure', PhysicalTransport.LE, event.peer_address, event.status, ) def on_hci_le_enhanced_connection_complete_event( self, event: ( hci.HCI_LE_Enhanced_Connection_Complete_Event | hci.HCI_LE_Enhanced_Connection_Complete_V2_Event ), ): # Just use the same implementation as for the non-enhanced event for now self.on_hci_le_connection_complete_event(event) def on_hci_le_enhanced_connection_complete_v2_event( self, event: hci.HCI_LE_Enhanced_Connection_Complete_V2_Event ): # Just use the same implementation as for the v1 event for now self.on_hci_le_enhanced_connection_complete_event(event) def on_hci_connection_complete_event( self, event: hci.HCI_Connection_Complete_Event ): if event.status == hci.HCI_SUCCESS: # Create/update the connection logger.debug( f'### BR/EDR CONNECTION: [0x{event.connection_handle:04X}] ' f'{event.bd_addr}' ) connection = self.connections.get(event.connection_handle) if connection is None: connection = Connection( self, event.connection_handle, event.bd_addr, PhysicalTransport.BR_EDR, ) self.connections[event.connection_handle] = connection # Notify the client self.emit( 'classic_connection', event.connection_handle, event.bd_addr, ) else: logger.debug(f'### BR/EDR CONNECTION FAILED: {event.status}') # Notify the client self.emit( 'connection_failure', PhysicalTransport.BR_EDR, event.bd_addr, event.status, ) def on_hci_disconnection_complete_event( self, event: hci.HCI_Disconnection_Complete_Event ): # Find the connection handle = event.connection_handle if ( connection := ( self.connections.get(handle) or self.cis_links.get(handle) or self.sco_links.get(handle) ) ) is None: logger.warning('!!! DISCONNECTION COMPLETE: unknown handle') return if event.status == hci.HCI_SUCCESS: logger.debug(f'### DISCONNECTION: {connection}, reason={event.reason}') # Notify the listeners self.emit('disconnection', handle, event.reason) # Remove the handle reference _ = ( self.connections.pop(handle, 0) or self.cis_links.pop(handle, 0) or self.sco_links.pop(handle, 0) ) # Flush the data queues if self.acl_packet_queue: self.acl_packet_queue.flush(handle) if self.le_acl_packet_queue: self.le_acl_packet_queue.flush(handle) if self.iso_packet_queue: self.iso_packet_queue.flush(handle) else: logger.debug(f'### DISCONNECTION FAILED: {event.status}') # Notify the listeners self.emit('disconnection_failure', handle, event.status) def on_hci_le_connection_update_complete_event( self, event: hci.HCI_LE_Connection_Update_Complete_Event ): if (connection := self.connections.get(event.connection_handle)) is None: logger.warning('!!! CONNECTION UPDATE COMPLETE: unknown handle') return # Notify the client if event.status == hci.HCI_SUCCESS: self.emit( 'connection_parameters_update', connection.handle, event.connection_interval, event.peripheral_latency, event.supervision_timeout, ) else: self.emit( 'connection_parameters_update_failure', connection.handle, event.status ) def on_hci_le_connection_rate_change_event( self, event: hci.HCI_LE_Connection_Rate_Change_Event ): if (connection := self.connections.get(event.connection_handle)) is None: logger.warning('!!! CONNECTION RATE CHANGE: unknown handle') return # Notify the client if event.status == hci.HCI_SUCCESS: self.emit( 'le_connection_rate_change', connection.handle, event.connection_interval, event.subrate_factor, event.peripheral_latency, event.continuation_number, event.supervision_timeout, ) else: self.emit( 'le_connection_rate_change_failure', connection.handle, event.status ) def on_hci_le_phy_update_complete_event( self, event: hci.HCI_LE_PHY_Update_Complete_Event ): if (connection := self.connections.get(event.connection_handle)) is None: logger.warning('!!! CONNECTION PHY UPDATE COMPLETE: unknown handle') return # Notify the client if event.status == hci.HCI_SUCCESS: self.emit( 'connection_phy_update', connection.handle, ConnectionPHY(event.tx_phy, event.rx_phy), ) else: self.emit('connection_phy_update_failure', connection.handle, event.status) def on_hci_le_advertising_report_event( self, event: ( hci.HCI_LE_Advertising_Report_Event | hci.HCI_LE_Extended_Advertising_Report_Event ), ): for report in event.reports: self.emit('advertising_report', report) def on_hci_le_extended_advertising_report_event( self, event: hci.HCI_LE_Extended_Advertising_Report_Event ): self.on_hci_le_advertising_report_event(event) def on_hci_le_advertising_set_terminated_event( self, event: hci.HCI_LE_Advertising_Set_Terminated_Event ): self.emit( 'advertising_set_termination', event.status, event.advertising_handle, event.connection_handle, event.num_completed_extended_advertising_events, ) def on_hci_le_periodic_advertising_sync_established_event( self, event: hci.HCI_LE_Periodic_Advertising_Sync_Established_Event ): self.emit( 'periodic_advertising_sync_establishment', event.status, event.sync_handle, event.advertising_sid, event.advertiser_address, event.advertiser_phy, event.periodic_advertising_interval, event.advertiser_clock_accuracy, ) def on_hci_le_periodic_advertising_sync_lost_event( self, event: hci.HCI_LE_Periodic_Advertising_Sync_Lost_Event ): self.emit('periodic_advertising_sync_loss', event.sync_handle) def on_hci_le_periodic_advertising_report_event( self, event: hci.HCI_LE_Periodic_Advertising_Report_Event ): self.emit('periodic_advertising_report', event.sync_handle, event) def on_hci_le_biginfo_advertising_report_event( self, event: hci.HCI_LE_BIGInfo_Advertising_Report_Event ): self.emit('biginfo_advertising_report', event.sync_handle, event) def on_hci_le_cis_request_event(self, event: hci.HCI_LE_CIS_Request_Event): self.emit( 'cis_request', event.acl_connection_handle, event.cis_connection_handle, event.cig_id, event.cis_id, ) def on_hci_le_create_big_complete_event( self, event: hci.HCI_LE_Create_BIG_Complete_Event ): self.bigs[event.big_handle] = set(event.connection_handle) if self.iso_packet_queue is None: raise InvalidStateError("BIS established but ISO packets not supported") for connection_handle in event.connection_handle: self.bis_links[connection_handle] = IsoLink( connection_handle, self.iso_packet_queue ) self.emit( 'big_establishment', event.status, event.big_handle, event.connection_handle, event.big_sync_delay, event.transport_latency_big, event.phy, event.nse, event.bn, event.pto, event.irc, event.max_pdu, event.iso_interval, ) def on_hci_le_big_sync_established_event( self, event: hci.HCI_LE_BIG_Sync_Established_Event ): self.bigs[event.big_handle] = set(event.connection_handle) if self.iso_packet_queue is None: raise InvalidStateError("BIS established but ISO packets not supported") for connection_handle in event.connection_handle: self.bis_links[connection_handle] = IsoLink( connection_handle, self.iso_packet_queue ) self.emit( 'big_sync_establishment', event.status, event.big_handle, event.transport_latency_big, event.nse, event.bn, event.pto, event.irc, event.max_pdu, event.iso_interval, event.connection_handle, ) def on_hci_le_big_sync_lost_event(self, event: hci.HCI_LE_BIG_Sync_Lost_Event): self.remove_big(event.big_handle) self.emit('big_sync_lost', event.big_handle, event.reason) def on_hci_le_terminate_big_complete_event( self, event: hci.HCI_LE_Terminate_BIG_Complete_Event ): self.remove_big(event.big_handle) self.emit('big_termination', event.reason, event.big_handle) def on_hci_le_periodic_advertising_sync_transfer_received_event( self, event: hci.HCI_LE_Periodic_Advertising_Sync_Transfer_Received_Event ): self.emit( 'periodic_advertising_sync_transfer', event.status, event.connection_handle, event.sync_handle, event.advertising_sid, event.advertiser_address, event.advertiser_phy, event.periodic_advertising_interval, event.advertiser_clock_accuracy, ) def on_hci_le_periodic_advertising_sync_transfer_received_v2_event( self, event: hci.HCI_LE_Periodic_Advertising_Sync_Transfer_Received_V2_Event ): self.emit( 'periodic_advertising_sync_transfer', event.status, event.connection_handle, event.sync_handle, event.advertising_sid, event.advertiser_address, event.advertiser_phy, event.periodic_advertising_interval, event.advertiser_clock_accuracy, ) def on_hci_le_cis_established_event(self, event: hci.HCI_LE_CIS_Established_Event): # The remaining parameters are unused for now. if event.status == hci.HCI_SUCCESS: if self.iso_packet_queue is None: raise InvalidStateError("CIS established but ISO packets not supported") self.cis_links[event.connection_handle] = IsoLink( handle=event.connection_handle, packet_queue=self.iso_packet_queue ) self.emit( 'cis_establishment', event.connection_handle, event.cig_sync_delay, event.cis_sync_delay, event.transport_latency_c_to_p, event.transport_latency_p_to_c, event.phy_c_to_p, event.phy_p_to_c, event.nse, event.bn_c_to_p, event.bn_p_to_c, event.ft_c_to_p, event.ft_p_to_c, event.max_pdu_c_to_p, event.max_pdu_p_to_c, event.iso_interval, ) else: self.emit( 'cis_establishment_failure', event.connection_handle, event.status ) def on_hci_le_remote_connection_parameter_request_event( self, event: hci.HCI_LE_Remote_Connection_Parameter_Request_Event ): if event.connection_handle not in self.connections: logger.warning('!!! REMOTE CONNECTION PARAMETER REQUEST: unknown handle') return # For now, just accept everything # TODO: delegate the decision utils.AsyncRunner.spawn( self.send_sync_command( hci.HCI_LE_Remote_Connection_Parameter_Request_Reply_Command( connection_handle=event.connection_handle, interval_min=event.interval_min, interval_max=event.interval_max, max_latency=event.max_latency, timeout=event.timeout, min_ce_length=0, max_ce_length=0, ) ) ) def on_hci_le_long_term_key_request_event( self, event: hci.HCI_LE_Long_Term_Key_Request_Event ): if (connection := self.connections.get(event.connection_handle)) is None: logger.warning('!!! LE LONG TERM KEY REQUEST: unknown handle') return async def send_long_term_key(): if self.long_term_key_provider is None: logger.debug('no long term key provider') long_term_key = None else: long_term_key = await utils.cancel_on_event( self, 'flush', # pylint: disable-next=not-callable self.long_term_key_provider( connection.handle, event.random_number, event.encryption_diversifier, ), ) if long_term_key: response = hci.HCI_LE_Long_Term_Key_Request_Reply_Command( connection_handle=event.connection_handle, long_term_key=long_term_key, ) else: response = hci.HCI_LE_Long_Term_Key_Request_Negative_Reply_Command( connection_handle=event.connection_handle ) await self.send_sync_command(response) utils.AsyncRunner.spawn(send_long_term_key()) def on_hci_synchronous_connection_complete_event( self, event: hci.HCI_Synchronous_Connection_Complete_Event ): if event.status == hci.HCI_SUCCESS: # Create/update the connection logger.debug( f'### SCO CONNECTION: [0x{event.connection_handle:04X}] {event.bd_addr}' ) self.sco_links[event.connection_handle] = ScoLink( peer_address=event.bd_addr, connection_handle=event.connection_handle, ) # Notify the client self.emit( 'sco_connection', event.bd_addr, event.connection_handle, event.link_type, ) else: logger.debug(f'### SCO CONNECTION FAILED: {event.status}') # Notify the client self.emit('sco_connection_failure', event.bd_addr, event.status) def on_hci_synchronous_connection_changed_event( self, event: hci.HCI_Synchronous_Connection_Changed_Event ): pass def on_hci_mode_change_event(self, event: hci.HCI_Mode_Change_Event): self.emit( 'mode_change', event.connection_handle, event.status, event.current_mode, event.interval, ) def on_hci_role_change_event(self, event: hci.HCI_Role_Change_Event): if event.status == hci.HCI_SUCCESS: logger.debug( f'role change for {event.bd_addr}: ' f'{hci.HCI_Constant.role_name(event.new_role)}' ) self.emit('role_change', event.bd_addr, hci.Role(event.new_role)) else: logger.debug( f'role change for {event.bd_addr} failed: ' f'{hci.HCI_Constant.error_name(event.status)}' ) self.emit('role_change_failure', event.bd_addr, event.status) def on_hci_le_data_length_change_event( self, event: hci.HCI_LE_Data_Length_Change_Event ): if event.connection_handle not in self.connections: logger.warning('!!! DATA LENGTH CHANGE: unknown handle') return self.emit( 'connection_data_length_change', event.connection_handle, event.max_tx_octets, event.max_tx_time, event.max_rx_octets, event.max_rx_time, ) def on_hci_authentication_complete_event( self, event: hci.HCI_Authentication_Complete_Event ): # Notify the client if event.status == hci.HCI_SUCCESS: self.emit('connection_authentication', event.connection_handle) else: self.emit( 'connection_authentication_failure', event.connection_handle, event.status, ) def on_hci_encryption_change_event(self, event: hci.HCI_Encryption_Change_Event): # Notify the client if event.status == hci.HCI_SUCCESS: self.emit( 'connection_encryption_change', event.connection_handle, event.encryption_enabled, 0, ) else: self.emit( 'connection_encryption_failure', event.connection_handle, event.status ) def on_hci_encryption_change_v2_event( self, event: hci.HCI_Encryption_Change_V2_Event ): # Notify the client if event.status == hci.HCI_SUCCESS: self.emit( 'connection_encryption_change', event.connection_handle, event.encryption_enabled, event.encryption_key_size, ) else: self.emit( 'connection_encryption_failure', event.connection_handle, event.status ) def on_hci_encryption_key_refresh_complete_event( self, event: hci.HCI_Encryption_Key_Refresh_Complete_Event ): # Notify the client if event.status == hci.HCI_SUCCESS: self.emit('connection_encryption_key_refresh', event.connection_handle) else: self.emit( 'connection_encryption_key_refresh_failure', event.connection_handle, event.status, ) def on_hci_qos_setup_complete_event(self, event: hci.HCI_QOS_Setup_Complete_Event): if event.status == hci.HCI_SUCCESS: self.emit( 'connection_qos_setup', event.connection_handle, event.service_type ) else: self.emit( 'connection_qos_setup_failure', event.connection_handle, event.status, ) def on_hci_link_supervision_timeout_changed_event( self, event: hci.HCI_Link_Supervision_Timeout_Changed_Event ): pass def on_hci_max_slots_change_event(self, event: hci.HCI_Max_Slots_Change_Event): pass def on_hci_page_scan_repetition_mode_change_event( self, event: hci.HCI_Page_Scan_Repetition_Mode_Change_Event ): pass def on_hci_link_key_notification_event( self, event: hci.HCI_Link_Key_Notification_Event ): logger.debug( f'link key for {event.bd_addr}: {event.link_key.hex()}, ' f'type={hci.HCI_Constant.link_key_type_name(event.key_type)}' ) self.emit('link_key', event.bd_addr, event.link_key, event.key_type) def on_hci_simple_pairing_complete_event( self, event: hci.HCI_Simple_Pairing_Complete_Event ): logger.debug( f'simple pairing complete for {event.bd_addr}: ' f'status={hci.HCI_Constant.status_name(event.status)}' ) if event.status == hci.HCI_SUCCESS: self.emit('classic_pairing', event.bd_addr) else: self.emit('classic_pairing_failure', event.bd_addr, event.status) def on_hci_pin_code_request_event(self, event: hci.HCI_PIN_Code_Request_Event): self.emit('pin_code_request', event.bd_addr) def on_hci_link_key_request_event(self, event: hci.HCI_Link_Key_Request_Event): async def send_link_key(): if self.link_key_provider is None: logger.debug('no link key provider') link_key = None else: link_key = await utils.cancel_on_event( self, 'flush', # pylint: disable-next=not-callable self.link_key_provider(event.bd_addr), ) if link_key: response = hci.HCI_Link_Key_Request_Reply_Command( bd_addr=event.bd_addr, link_key=link_key ) else: response = hci.HCI_Link_Key_Request_Negative_Reply_Command( bd_addr=event.bd_addr ) await self.send_sync_command(response) utils.AsyncRunner.spawn(send_link_key()) def on_hci_io_capability_request_event( self, event: hci.HCI_IO_Capability_Request_Event ): self.emit('authentication_io_capability_request', event.bd_addr) def on_hci_io_capability_response_event( self, event: hci.HCI_IO_Capability_Response_Event ): self.emit( 'authentication_io_capability_response', event.bd_addr, event.io_capability, event.authentication_requirements, ) def on_hci_user_confirmation_request_event( self, event: hci.HCI_User_Confirmation_Request_Event ): self.emit( 'authentication_user_confirmation_request', event.bd_addr, event.numeric_value, ) def on_hci_user_passkey_request_event( self, event: hci.HCI_User_Passkey_Request_Event ): self.emit('authentication_user_passkey_request', event.bd_addr) def on_hci_user_passkey_notification_event( self, event: hci.HCI_User_Passkey_Notification_Event ): self.emit( 'authentication_user_passkey_notification', event.bd_addr, event.passkey ) def on_hci_inquiry_complete_event(self, _event: hci.HCI_Inquiry_Complete_Event): self.emit('inquiry_complete') def on_hci_inquiry_result_with_rssi_event( self, event: hci.HCI_Inquiry_Result_With_RSSI_Event ): for bd_addr, class_of_device, rssi in zip( event.bd_addr, event.class_of_device, event.rssi ): self.emit( 'inquiry_result', bd_addr, class_of_device, b'', rssi, ) def on_hci_extended_inquiry_result_event( self, event: hci.HCI_Extended_Inquiry_Result_Event ): self.emit( 'inquiry_result', event.bd_addr, event.class_of_device, event.extended_inquiry_response, event.rssi, ) def on_hci_remote_name_request_complete_event( self, event: hci.HCI_Remote_Name_Request_Complete_Event ): if event.status != hci.HCI_SUCCESS: self.emit('remote_name_failure', event.bd_addr, event.status) else: utf8_name = event.remote_name terminator = utf8_name.find(0) if terminator >= 0: utf8_name = utf8_name[0:terminator] self.emit('remote_name', event.bd_addr, utf8_name) def on_hci_remote_host_supported_features_notification_event( self, event: hci.HCI_Remote_Host_Supported_Features_Notification_Event ): self.emit( 'remote_host_supported_features', event.bd_addr, event.host_supported_features, ) def on_hci_le_read_remote_features_complete_event( self, event: hci.HCI_LE_Read_Remote_Features_Complete_Event ): if event.status != hci.HCI_SUCCESS: self.emit( 'le_remote_features_failure', event.connection_handle, event.status ) return self.emit( 'le_remote_features', event.connection_handle, hci.LeFeatureMask(int.from_bytes(event.le_features, 'little')), ) def on_hci_le_cs_read_remote_supported_capabilities_complete_event( self, event: hci.HCI_LE_CS_Read_Remote_Supported_Capabilities_Complete_Event ): self.emit('cs_remote_supported_capabilities', event) def on_hci_le_cs_security_enable_complete_event( self, event: hci.HCI_LE_CS_Security_Enable_Complete_Event ): self.emit('cs_security', event) def on_hci_le_cs_config_complete_event( self, event: hci.HCI_LE_CS_Config_Complete_Event ): self.emit('cs_config', event) def on_hci_le_cs_procedure_enable_complete_event( self, event: hci.HCI_LE_CS_Procedure_Enable_Complete_Event ): self.emit('cs_procedure', event) def on_hci_le_cs_subevent_result_event( self, event: hci.HCI_LE_CS_Subevent_Result_Event ): self.emit('cs_subevent_result', event) def on_hci_le_cs_subevent_result_continue_event( self, event: hci.HCI_LE_CS_Subevent_Result_Continue_Event ): self.emit('cs_subevent_result_continue', event) def on_hci_le_subrate_change_event(self, event: hci.HCI_LE_Subrate_Change_Event): if event.status != hci.HCI_SUCCESS: self.emit( 'le_subrate_change_failure', event.connection_handle, event.status ) return self.emit( 'le_subrate_change', event.connection_handle, event.subrate_factor, event.peripheral_latency, event.continuation_number, event.supervision_timeout, ) def on_hci_vendor_event(self, event: hci.HCI_Vendor_Event): self.emit('vendor_event', event)