You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

使用Python Socket连接自定义蓝牙设备失败求助

蓝牙Socket连接自定义BCI设备失败问题

我尝试用Python的Socket连接自定义蓝牙BCI设备,通过PyBluez发现设备并获取了地址,但连接失败。当前代码如下:

import bluetooth, subprocess
import socket

class BCI(object):
    """
    Bluetooth Connection to BCI
    """
    socket_ = None
    bluetooth_address = None
    connected = False
    port = 0x005
    dev_name = None

    cmds = {}        

    status = {
        0:  'ok',
        1:  'communication timeout',
        3:  'checksum error',
        4:  'unknown command',
        5:  'invalid access level',
        8:  'hardware error',
        10: 'device not ready',
    }

    def __init__(self, *args, **kwargs):
        self.bluetooth_address = kwargs.get("bluetooth_address", None)
        if self.bluetooth_address is None:
            self.find()

    def connect(self):
        try:         
            self.socket_ = socket.socket(socket.AF_BLUETOOTH, socket.SOCK_STREAM, socket.BTPROTO_RFCOMM)
            #self.socket_ = socket.socket(socket.AF_BLUETOOTH, socket.SOCK_SEQPACKET, socket.BTPROTO_L2CAP)
            self.socket_.connect((self.bluetooth_address, self.port))
            self.connected = True
        except:
            self.socket_.close()
            self.connected = False
            raise ConnectionError

    def find(self):
        dev = self.__class__.__name__
        print('Searching for ' + dev)
        nearby_devices = bluetooth.discover_devices(duration=8, lookup_names=True, flush_cache=True)
        for index, val in enumerate(nearby_devices):
            addr, name = val
            if dev in name.upper():
                self.dev_name = name
                self.bluetooth_address = addr
                print('Found BCI: ' + name + ' @', self.bluetooth_address)
                return        

    def find_bluetooth_services(self):
        services = bluetooth.find_service(address=self.bluetooth_address)
        if len(services) > 0:
            print("found %d services on %s" % (len(services), self.bluetooth_address))
            print(services)
        else:
            print("no services found")
            
    def close(self):
        self.socket_.close()
        
if __name__ == "__main__":
    try:
        device = BCI()
    except:
        print('No devices BCI found')
    
    try:
        print("Trying to connect with "+ device.__class__.__name__)
        device.connect()
    except ConnectionError:
        print ('Can\'t connect with ' + device.__class__.__name__)
    '''    
    try:
        device.find_bluetooth_services()
    except:
        print("Some problem occured")
    '''
    
    if device.connected:
        print('Connected {}'.format(device.dev_name)+ device.__class__.__name__ +'  @', device.bluetooth_address)

终端输出如下:

Searching for BCI
Found BCI: BCI-2016-SN005 @ 00:80:E1:BE:EE:F1
Trying to connect with BCI
(系统弹出配对确认请求,确认后)
Can't connect with BCI

改用connect_ex()替代connect()后得到错误码10065,不清楚该错误码在蓝牙场景下的含义,请问哪里操作有误?


问题排查与修正方案

1. 硬编码端口号错误

你当前使用的port = 0x005(十进制5)是硬编码值,但自定义BCI设备的RFCOMM服务不一定在这个端口。必须先通过find_bluetooth_services()获取设备实际开放的服务端口,而非盲目指定。

2. 错误处理逻辑漏洞

connect方法中的except捕获所有异常,且如果socket_未成功创建就调用close()会触发新的错误。应改为捕获具体异常类型,先判断socket_存在再执行关闭操作,同时打印错误详情。

3. 配对后未留等待时间

系统弹出配对确认后,设备需要短暂时间完成配对流程,直接发起连接会失败。可在配对确认后添加延迟(如2秒)再尝试连接。

4. 错误码10065解析

该错误码对应WSAEHOSTUNREACH,意为主机不可达,在蓝牙场景下通常是:

  • 目标设备未在指定端口提供服务
  • 配对完成后连接通道未建立
  • 设备处于未就绪状态

修正后的关键代码示例

def connect(self):
    import time
    # 先获取设备服务端口
    self.find_bluetooth_services()
    # 配对后延迟等待设备就绪
    time.sleep(2)
    try:
        self.socket_ = socket.socket(socket.AF_BLUETOOTH, socket.SOCK_STREAM, socket.BTPROTO_RFCOMM)
        # 使用实际获取的RFCOMM端口,若未获取到则尝试常用端口1
        target_port = self.service_port if hasattr(self, 'service_port') else 1
        self.socket_.connect((self.bluetooth_address, target_port))
        self.connected = True
    except socket.error as e:
        if self.socket_:
            self.socket_.close()
        self.connected = False
        print(f"连接错误: {e}, 错误码: {e.errno}")
        raise ConnectionError

def find_bluetooth_services(self):
    services = bluetooth.find_service(address=self.bluetooth_address)
    if len(services) > 0:
        print(f"找到 {len(services)} 个服务在 {self.bluetooth_address}")
        for svc in services:
            print(f"服务名称: {svc['name']}, 端口: {svc['port']}, 协议: {svc['protocol']}")
            # 筛选并记录RFCOMM服务端口
            if svc['protocol'] == 'RFCOMM':
                self.service_port = svc['port']
                print(f"使用RFCOMM端口: {self.service_port}")
                break
    else:
        print("未找到服务")

内容的提问来源于stack exchange,提问作者Avedik Ekizyan

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.27 19:44:54