4. 连接远程设备

任务说明

本案例实现与指定蓝牙设备的异步连接（ACL）建立过程。给定目标设备地址（BD_ADDR），通过HCI命令完成从查询到连接建立的完整流程，并处理连接完成事件。核心交付物为一个可独立运行的Python脚本，模拟主机控制器接口（HCI）层的关键命令交互，验证连接建立的状态机转换。

HCI连接建立流程

ACL连接建立涉及以下HCI命令与事件的时序配合：

  1. HCI_Inquiry：发现周围设备（本例直接指定BD_ADDR，跳过查询）
  2. HCI_Create_Connection：向链路管理器发起ACL连接请求
  3. HCI_Connection_Complete：异步事件，报告连接建立成功或失败

核心难点在于异步性：Create_Connection命令立即返回状态，而连接结果通过后续事件通知。实现需维护命令与事件的对应关系，并处理超时与重试。

代码实现

import struct
import time
import threading
from enum import Enum

class HCI_Event_Code(Enum):
    CONNECTION_COMPLETE = 0x03
    COMMAND_COMPLETE = 0x0E
    COMMAND_STATUS = 0x0F

class HCI_Command_Opcode(Enum):
    CREATE_CONNECTION = 0x0405
    INQUIRY = 0x0401

class ACL_Connection_State(Enum):
    IDLE = 0
    PENDING = 1
    CONNECTED = 2
    FAILED = 3

class HCI_Connection_Manager:
    def __init__(self, timeout_sec=5.0):
        self.state = ACL_Connection_State.IDLE
        self.timeout_sec = timeout_sec
        self.connection_handle = None
        self._event_queue = []
        self._lock = threading.Lock()
        self._pending_opcode = None

    def _send_command(self, opcode, params):
        """构造并发送HCI命令包（模拟发送到控制器）"""
        # 参数格式: opcode(2字节) + 参数长度(1字节) + 参数
        pkt = struct.pack('<HB', opcode.value, len(params)) + params
        print(f"[TX] HCI Command: opcode=0x{opcode.value:04X}, params={params.hex()}")
        # 实际实现中此处通过UART/USB发送给蓝牙控制器
        return pkt

    def _receive_event(self, event_code, params):
        """接收并处理来自控制器的事件（模拟异步回调）"""
        with self._lock:
            self._event_queue.append((event_code, params))
        print(f"[RX] HCI Event: code=0x{event_code.value:02X}, params={params.hex()}")

    def _process_events(self):
        """处理所有待处理事件（由事件分发线程调用）"""
        while True:
            with self._lock:
                if not self._event_queue:
                    break
                event_code, params = self._event_queue.pop(0)
            
            if event_code == HCI_Event_Code.COMMAND_STATUS:
                # 命令已被接受，等待后续事件
                status = params[0]
                if status != 0x00:
                    self.state = ACL_Connection_State.FAILED
                    print(f"[ERR] Command rejected, status=0x{status:02X}")
            
            elif event_code == HCI_Event_Code.CONNECTION_COMPLETE:
                status = params[0]
                handle = struct.unpack('<H', params[1:3])[0]
                if status == 0x00:
                    self.state = ACL_Connection_State.CONNECTED
                    self.connection_handle = handle
                    print(f"[OK] Connection established, handle=0x{handle:04X}")
                else:
                    self.state = ACL_Connection_State.FAILED
                    print(f"[ERR] Connection failed, status=0x{status:02X}")

    def connect(self, bd_addr):
        """发起ACL连接（异步）"""
        if self.state not in (ACL_Connection_State.IDLE, ACL_Connection_State.FAILED):
            raise RuntimeError(f"Invalid state for connect: {self.state}")
        
        self.state = ACL_Connection_State.PENDING
        # 参数: BD_ADDR(6字节) + packet_type(2字节) + page_scan_mode(1字节)
        bd_addr_bytes = bytes.fromhex(bd_addr.replace(':', ''))
        packet_type = struct.pack('<H', 0xCC18)  # DM1/DH1/DM3/DH3/DM5/DH5
        page_scan_mode = bytes([0x00])
        params = bd_addr_bytes + packet_type + page_scan_mode
        
        self._send_command(HCI_Command_Opcode.CREATE_CONNECTION, params)
        
        # 模拟控制器异步响应（实际由硬件中断触发）
        def _simulate_controller():
            time.sleep(0.1)  # 模拟链路建立耗时
            # 模拟Command Status事件
            self._receive_event(HCI_Event_Code.COMMAND_STATUS, bytes([0x00, 0x01, 0x00]))
            # 模拟Connection Complete事件（成功）
            self._receive_event(HCI_Event_Code.CONNECTION_COMPLETE, 
                              bytes([0x00]) + struct.pack('<H', 0x0042) + bd_addr_bytes + bytes([0x00]))
        
        threading.Thread(target=_simulate_controller, daemon=True).start()
        
        # 等待连接完成或超时
        start = time.time()
        while self.state == ACL_Connection_State.PENDING:
            self._process_events()
            if time.time() - start > self.timeout_sec:
                self.state = ACL_Connection_State.FAILED
                print("[TIMEOUT] Connection attempt timed out")
                break
            time.sleep(0.01)
        
        return self.state

    def disconnect(self):
        """断开当前连接"""
        if self.state != ACL_Connection_State.CONNECTED:
            return
        # 实际发送HCI_Disconnect命令
        print(f"[TX] HCI Disconnect: handle=0x{self.connection_handle:04X}")
        self.state = ACL_Connection_State.IDLE
        self.connection_handle = None

# 使用示例
if __name__ == "__main__":
    mgr = HCI_Connection_Manager()
    target_addr = "AA:BB:CC:DD:EE:FF"
    
    print(f"=== 连接远程设备 {target_addr} ===")
    result = mgr.connect(target_addr)
    
    if result == ACL_Connection_State.CONNECTED:
        print(f"连接成功，句柄: 0x{mgr.connection_handle:04X}")
        mgr.disconnect()
    else:
        print("连接失败")

关键机制说明

代码实现了以下核心逻辑：

  • 状态机管理：IDLE→PENDING→CONNECTED/FAILED，确保连接流程状态可追踪
  • 异步事件处理：通过事件队列和独立线程模拟控制器异步响应，符合真实蓝牙协议栈行为
  • 超时保护：5秒超时防止无限等待
  • 命令构造：严格按蓝牙Core 5.3规范构造HCI_Create_Connection命令参数

验证方法

运行脚本后观察输出：

  1. 应依次看到[TX] HCI Command（Create_Connection）和两条[RX] HCI Event（Command Status、Connection Complete）
  2. 最终状态为CONNECTED，连接句柄为0x0042
  3. 若修改模拟事件中的状态字节为0x01（失败），则状态应变为FAILED

判定标准：脚本输出包含完整的命令/事件交互序列，且最终状态与模拟事件内容一致，即验证通过。

HCI连接时序

主机 (Host) 控制器 (Controller) HCI传输层 HCI_Create_Connection Command Status (0x0F) 链路建立 Connection Complete (0x03) 状态: PENDING → CONNECTED

上图展示了ACL连接建立过程中主机与控制器之间的HCI命令/事件交互时序。主机发送Create_Connection后进入PENDING状态，控制器先回复Command Status表示已接受命令，待链路建立完成后异步上报Connection Complete事件，主机据此更新连接状态。

← 上一章 📖 返回目录 下一章 →