1# Copyright 2021-2022 Google LLC 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); 4# you may not use this file except in compliance with the License. 5# You may obtain a copy of the License at 6# 7# https://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12# See the License for the specific language governing permissions and 13# limitations under the License. 14 15# ----------------------------------------------------------------------------- 16# Imports 17# ----------------------------------------------------------------------------- 18import logging 19import asyncio 20import os 21import sys 22 23from bumble import hci, transport 24from bumble.bridge import HCI_Bridge 25 26# ----------------------------------------------------------------------------- 27# Logging 28# ----------------------------------------------------------------------------- 29logger = logging.getLogger(__name__) 30 31 32# ----------------------------------------------------------------------------- 33# Main 34# ----------------------------------------------------------------------------- 35async def async_main(): 36 if len(sys.argv) < 3: 37 print('Usage: hci_bridge.py <host-transport-spec> <controller-transport-spec> [command-short-circuit-list]') 38 print('example: python hci_bridge.py udp:0.0.0.0:9000,127.0.0.1:9001 serial:/dev/tty.usbmodem0006839912171,1000000 0x3f:0x0070,0x3f:0x0074,0x3f:0x0077,0x3f:0x0078') 39 return 40 41 print('>>> connecting to HCI...') 42 async with await transport.open_transport_or_link(sys.argv[1]) as (hci_host_source, hci_host_sink): 43 print('>>> connected') 44 45 print('>>> connecting to HCI...') 46 async with await transport.open_transport_or_link(sys.argv[2]) as (hci_controller_source, hci_controller_sink): 47 print('>>> connected') 48 49 command_short_circuits = [] 50 if len(sys.argv) >= 4: 51 for op_code_str in sys.argv[3].split(','): 52 if ':' in op_code_str: 53 ogf, ocf = op_code_str.split(':') 54 command_short_circuits.append(hci.hci_command_op_code(int(ogf, 16), int(ocf, 16))) 55 else: 56 command_short_circuits.append(int(op_code_str, 16)) 57 58 def host_to_controller_filter(hci_packet): 59 if hci_packet.hci_packet_type == hci.HCI_COMMAND_PACKET and hci_packet.op_code in command_short_circuits: 60 # Respond with a success response 61 logger.debug('short-circuiting packet') 62 response = hci.HCI_Command_Complete_Event( 63 num_hci_command_packets = 1, 64 command_opcode = hci_packet.op_code, 65 return_parameters = bytes([hci.HCI_SUCCESS]) 66 ) 67 # Return a packet with 'respond to sender' set to True 68 return (response.to_bytes(), True) 69 70 _ = HCI_Bridge( 71 hci_host_source, 72 hci_host_sink, 73 hci_controller_source, 74 hci_controller_sink, 75 host_to_controller_filter, 76 None 77 ) 78 await asyncio.get_running_loop().create_future() 79 80 81# ----------------------------------------------------------------------------- 82def main(): 83 logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'INFO').upper()) 84 asyncio.run(async_main()) 85 86 87# ----------------------------------------------------------------------------- 88if __name__ == '__main__': 89 main() 90