蓝牙设备监控

from micropython import const
import bluetooth
import time

# BLE常量
_IRQ_SCAN_RESULT = const(5)
AD_TYPE_NAME_COMPLETE = const(0x09)
AD_TYPE_NAME_SHORT = const(0x08)

ble = bluetooth.BLE()
# 存储结构改为:{MAC地址字符串: {"name": 设备名, "last_tick": 最后时间戳}}
# 用MAC做唯一键,避免同名设备冲突
dev_cache = dict()
SCAN_RUNNING = False
# 设备超时时间:1500ms未收到广播判定离开
DEV_TIMEOUT = 1500

def _format_mac(addr_bytes):
    """辅助函数:将6字节MAC地址转换为 AA:BB:CC:DD:EE:FF 可读格式"""
    return ":".join("{:02X}".format(b) for b in addr_bytes)

def _parse_name(adv_data):
    """解析广播包中的蓝牙设备名称"""
    idx = 0
    data_len = len(adv_data)
    while idx < data_len:
        ad_len = adv_data[idx]
        if ad_len == 0:
            break
        ad_type = adv_data[idx + 1]
        payload = adv_data[idx + 2 : idx + 1 + ad_len]
        if ad_type in (AD_TYPE_NAME_COMPLETE, AD_TYPE_NAME_SHORT):
            try:
                return payload.decode("utf-8").strip()
            except Exception:
                return "Unreadable_Device"
        idx += ad_len + 1
    return "No_Name"

def ble_irq_callback(event, data):
    global dev_cache
    if event == _IRQ_SCAN_RESULT:
        # 完整解包扫描结果:地址类型、MAC地址、广播类型、信号强度、广播数据
        addr_type, addr, adv_type, rssi, adv_data = data
        dev_name = _parse_name(bytes(adv_data))
        mac_str = _format_mac(addr)  # 提取并格式化MAC地址
        now = time.ticks_ms()
        # 以MAC为唯一键更新缓存
        dev_cache[mac_str] = {
            "name": dev_name,
            "last_tick": now
        }

def start_background_full_scan():
    """启动后台永久高速扫描,匹配100ms间隔信标"""
    global SCAN_RUNNING
    if SCAN_RUNNING:
        return
    ble.active(True)
    ble.irq(ble_irq_callback)
    # gap_scan(永久扫描0, 扫描周期100000us=100ms, 监听窗口90000us=90ms, 主动扫描1)
    ble.gap_scan(0, 500000, 50000, 1)
    SCAN_RUNNING = True
    print("后台BLE扫描已启动 | 适配250ms间隔蓝牙信标")

def scan_ble_devices():
    """
    原有接口保留:返回当前在场蓝牙设备名称列表(兼容旧代码)
    return: list[str]
    """
    devs = get_ble_devices_with_mac()
    return [dev["name"] for dev in devs]

# ========== 新增:带MAC地址的设备查询方法 ==========
def get_ble_devices_with_mac():
    """
    返回当前在场蓝牙设备的完整信息(MAC+名称)
    自动清理超时消失的设备,无阻塞
    return: list[dict]  每个元素格式:{"mac": "AA:BB:CC:DD:EE:FF", "name": "设备名"}
    """
    global dev_cache
    if not SCAN_RUNNING:
        return []
    
    now = time.ticks_ms()
    valid_devs = []
    expired_macs = []
    
    # 遍历缓存:收集有效设备 + 标记超时设备
    for mac, info in dev_cache.items():
        if time.ticks_diff(now, info["last_tick"]) >= DEV_TIMEOUT:
            expired_macs.append(mac)
        else:
            valid_devs.append({
                "mac": mac,
                "name": info["name"]
            })
    
    # 从缓存中删除超时设备(避免缓存无限增长)
    for mac in expired_macs:
        del dev_cache[mac]
    
    return valid_devs

def stop_ble_scan():
    """停止扫描、关闭蓝牙,程序结束时调用"""
    global SCAN_RUNNING
    if not SCAN_RUNNING:
        return
    ble.gap_scan(None)
    ble.active(False)
    SCAN_RUNNING = False
    dev_cache.clear()
    print("BLE扫描已关闭")



class People:
    def __init__(self,pren=1):
        self.pren=pren
        self.sw=[]
    def dev(self,lis):
        plis=[]
        for i in lis:
            plis.append(i['mac'])
        return self.get_num(plis)
    def get_num(self,lis):
        self.push_list(lis)
        return self.people_num()
    def push_list(self,lis):
        if len(self.sw)<10:
            self.sw.append(lis)
        else:
            self.sw.pop(0)
            self.sw.append(lis)
    def people_num(self):
        new_list=[item for sub in self.sw for item in sub]
#         print(list(set(new_list)))
        return len(set(new_list))*self.pren
    
    
def mdatetime():
    t = time.localtime(time.time())
    return f"{t[0]}-{t[1]:02d}-{t[2]:02d} {t[3]:02d}:{t[4]:02d}:{t[5]:02d}"


#------------------- 测试示例 -------------------
start_background_full_scan()
# stop_ble_scan()
# get_ble_devices_with_mac()


people=People()
f=open("log.csv", "a")
i=0
while True:
    i=i+1
    # 调用新方法获取带MAC的设备列表
    dev_list = get_ble_devices_with_mac()
    print("==== 在场蓝牙设备 ====")
    if len(dev_list) == 0:
        print("无设备")
    else:
        num=people.dev(dev_list)
        print(mdatetime(),"PEOPLE",num)
        data={"time":mdatetime(),"PEOPLE":num}
        if i>10:
            f.write("{},{}\n".format(data["time"], data["PEOPLE"]))
            f.flush()
    time.sleep_ms(1000)  # 建议加延时,避免串口刷屏#

数据格式

==== 在场蓝牙设备 ====
['64:61:00:C0:21:6B', '72:4C:30:F4:32:E2', 'C5:65:56:F2:84:44', '52:7D:6D:92:08:5B', '0B:AA:FE:9C:1F:A1', 'A4:C1:38:3F:FA:23', 'E8:E6:09:13:3B:3E', '78:DF:72:DF:09:FF', 'B8:88:80:CE:85:54', '7B:C3:2E:84:AA:F0', 'D5:69:D5:BB:89:A2', 'CD:EC:A2:07:1C:EB', '6B:A2:91:34:5C:A2']
PEOPLE 13
- MAC: 72:4C:30:F4:32:E2  名称: No_Name
- MAC: 52:7D:6D:92:08:5B  名称: No_Name
- MAC: A4:C1:38:3F:FA:23  名称: No_Name
- MAC: 0B:AA:FE:9C:1F:A1  名称: No_Name
- MAC: E8:E6:09:13:3B:3E  名称: No_Name
- MAC: D5:69:D5:BB:89:A2  名称: No_Name
- MAC: 6B:A2:91:34:5C:A2  名称: No_Name

此条目发表在None分类目录。将固定链接加入收藏夹。

发表回复