Python蓝牙家庭自动化：多进程检测iPhone存在时的设备忙问题

嘿，这个家庭自动化的挑战真的挺有意思的！我来帮你解决蓝牙查询冲突的问题——毕竟多进程抢蓝牙适配器导致的“Device busy”确实是这类场景里的常见坑。

核心问题根源 #

你的判断完全正确：树莓派的蓝牙适配器是单资源设备，多个进程同时调用bluetooth.lookup_name()或者hcitool name会互相抢占资源，导致查询无响应、卡住甚至直接失败。解决思路就是给蓝牙查询加“排队机制”，确保同一时间只有一个进程在操作蓝牙适配器。

方案1：用进程锁控制蓝牙访问 #

最简单的方式是用multiprocessing.Lock，让所有子进程在发起蓝牙查询前先获取锁，查询完成后释放锁。这样就能避免同时访问的冲突。

示例代码：

from multiprocessing import Process, Lock
import bluetooth
import subprocess
import time

def check_device(ip, mac, device_name, bt_lock):
    def ping_device():
        try:
            return subprocess.call(["ping", "-c", "1", "-W", "2", ip]) == 0
        except:
            return False
    
    def arp_check():
        try:
            output = subprocess.check_output(["arp", "-n", ip])
            return mac.lower() in output.decode().lower()
        except:
            return False
    
    def bt_check():
        with bt_lock:  # 关键：获取锁后再执行蓝牙查询
            try:
                return bluetooth.lookup_name(mac, timeout=2) is not None
            except Exception as e:
                print(f"{device_name}蓝牙查询出错: {str(e)}")
                return False

    # 优先用更轻量的ping和ARP检测，减少蓝牙调用
    present = ping_device() or arp_check()
    if not present:
        # 只有前两种检测失败时，才触发蓝牙查询
        present = bt_check()
    
    # 这里替换成你的智能家居开关控制逻辑
    print(f"{device_name}: {'在场' if present else '离开'}")
    # set_virtual_switch(device_name, present)

if __name__ == "__main__":
    # 创建全局蓝牙锁
    bluetooth_lock = Lock()
    # 你的四台iPhone信息
    devices = [
        ("192.168.1.10", "AA:BB:CC:DD:EE:01", "iPhone-爸爸"),
        ("192.168.1.11", "AA:BB:CC:DD:EE:02", "iPhone-妈妈"),
        ("192.168.1.12", "AA:BB:CC:DD:EE:03", "iPhone-孩子1"),
        ("192.168.1.13", "AA:BB:CC:DD:EE:04", "iPhone-孩子2"),
    ]

    # 启动检测进程
    processes = []
    for ip, mac, name in devices:
        p = Process(target=check_device, args=(ip, mac, name, bluetooth_lock))
        processes.append(p)
        p.start()
    
    # 保持主进程运行（实际场景可以加循环定时检测）
    try:
        for p in processes:
            p.join()
    except KeyboardInterrupt:
        print("脚本终止")

方案2：单独的蓝牙查询服务进程（更稳定） #

如果你的脚本长期运行，更推荐把蓝牙查询做成一个独立的服务进程，所有设备检测进程通过队列提交查询请求，由服务进程统一处理蓝牙操作。这种方式彻底避免了资源竞争，适合多设备场景。

示例代码：

from multiprocessing import Process, Queue
import bluetooth
import subprocess
import time

def bluetooth_service(request_queue, result_queue):
    """独立的蓝牙查询服务，处理所有请求"""
    while True:
        request = request_queue.get()
        if request is None:  # 退出信号
            break
        mac, request_id = request
        try:
            bt_present = bluetooth.lookup_name(mac, timeout=2) is not None
            result_queue.put((request_id, bt_present))
        except Exception as e:
            print(f"蓝牙服务出错: {str(e)}")
            result_queue.put((request_id, False))

def device_monitor(ip, mac, device_name, request_queue, result_queue, monitor_id):
    """单设备检测进程"""
    def ping_device():
        try:
            return subprocess.call(["ping", "-c", "1", "-W", "2", ip]) == 0
        except:
            return False
    
    def arp_check():
        try:
            output = subprocess.check_output(["arp", "-n", ip])
            return mac.lower() in output.decode().lower()
        except:
            return False

    while True:
        present = ping_device() or arp_check()
        if not present:
            # 提交蓝牙查询请求
            request_queue.put((mac, monitor_id))
            # 等待结果返回
            req_id, bt_result = result_queue.get()
            if req_id == monitor_id:
                present = bt_result
        
        print(f"{device_name}: {'在场' if present else '离开'}")
        # set_virtual_switch(device_name, present)
        time.sleep(60)  # 每分钟检测一次

if __name__ == "__main__":
    req_queue = Queue()
    res_queue = Queue()

    # 启动蓝牙服务进程
    bt_service_proc = Process(target=bluetooth_service, args=(req_queue, res_queue))
    bt_service_proc.start()

    # 启动设备监控进程
    devices = [
        ("192.168.1.10", "AA:BB:CC:DD:EE:01", "iPhone-爸爸", 0),
        ("192.168.1.11", "AA:BB:CC:DD:EE:02", "iPhone-妈妈", 1),
        ("192.168.1.12", "AA:BB:CC:DD:EE:03", "iPhone-孩子1", 2),
        ("192.168.1.13", "AA:BB:CC:DD:EE:04", "iPhone-孩子2", 3),
    ]
    monitor_procs = []
    for ip, mac, name, mid in devices:
        p = Process(target=device_monitor, args=(ip, mac, name, req_queue, res_queue, mid))
        monitor_procs.append(p)
        p.start()

    # 优雅退出逻辑
    try:
        for p in monitor_procs:
            p.join()
    except KeyboardInterrupt:
        req_queue.put(None)  # 通知蓝牙服务退出
        bt_service_proc.join()
        print("脚本终止")

额外优化建议 #

1. 减少蓝牙查询频率：只有连续2-3次ping和ARP都失败时，再触发蓝牙查询，避免不必要的资源消耗。
2. 利用iPhone的唤醒功能：在iPhone的设置-通用-后台App刷新里开启“唤醒局域网”，这样ping的时候可能唤醒设备的WiFi，提高检测准确率，减少蓝牙依赖。
3. 给subprocess加超时：如果你坚持用hcitool，一定要加timeout参数，比如subprocess.check_output(['hcitool', 'name', mac], timeout=2)，防止查询卡住。

这些方案都不需要配对iPhone，完全符合你的需求，而且能有效解决多进程蓝牙冲突的问题。毕竟这种资源调度的场景，也是学习Python多进程编程的好机会呀～

内容的提问来源于stack exchange，提问作者pittbull74