1 人员手机和平板一半是没有名称
2 可以添加一个信号强度在字典里面用于过滤距离
d = 10^((abs(RSSI) – A) / (10 * n)) d – 计算所得距离 RSSI – 接收信号强度（负值） A – 发射端和接收端相隔1米时的信号强度 n – 环境衰减因子
最新代码
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()
# 缓存新增rssi字段存储实时信号强度
# {MAC地址字符串: {"name": 设备名, "last_tick": 最后时间戳, "rssi": 信号强度dBm}}
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()
# 存入最新rssi信号值
dev_cache[mac_str] = {
"name": dev_name,
"last_tick": now,
"rssi": rssi
}
def start_background_full_scan():
"""启动后台永久高速扫描，匹配100ms间隔信标"""
global SCAN_RUNNING
if SCAN_RUNNING:
return
ble.active(True)
ble.irq(ble_irq_callback)
# gap_scan(永久扫描0, 扫描周期250000us=250ms, 监听窗口60000us=60ms, 主动扫描1)
ble.gap_scan(0, 250000, 60000, 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地址+RSSI信号强度的设备查询方法 ==========
def get_ble_devices_with_mac():
"""
返回当前在场蓝牙设备的完整信息（MAC+名称+信号强度rssi）
自动清理超时消失的设备，无阻塞
return: list[dict] 每个元素格式：{"mac": "AA:BB:CC:DD:EE:FF", "name": "设备名", "rssi": -XX}
"""
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"],
"rssi": info["rssi"]
})
# 从缓存中删除超时设备（避免缓存无限增长）
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 BDrive:
def __init__(self,multiplier=1,rssi_set=-90,distance=10):
self.multiplier=multiplier
self.distance=distance
self.mac=[]
self.names=[]
self.all_names=[]
self.all_drives=[]
self.all_devs=[]
self.rssi_set=rssi_set
self.devs=[]
self.come={}
self.go={}
self.stay={}
def service_come(self):
#扫描到新蓝牙触发
for one in self.all_devs:
if one['mac'] in self.stay:
self.come.pop(one['mac'],None)
else:
self.stay[one['mac']]=one['name']
self.come[one['mac']]=one['name']
def service_go(self):
#蓝牙离开触发
for one_key in list(self.stay):
if one_key not in self.all_mac:
self.go[one_key]=self.stay[one_key]
del self.stay[one_key]
def dev(self,ones):
mac=[]
names=[]
devs=[]
for i in ones:
if i['rssi']>=self.rssi_set:
mac.append(i['mac'])
names.append(i['name'])
devs.append(i)
return self.get_num(mac,names,devs)
def get_num(self,mac,names,devs):
self.push_list(mac,names,devs)
return self.people_num()
def push_list(self,lis,names,devs):
if len(self.mac)<10:
self.mac.append(lis)
self.names.append(names)
self.devs.append(devs)
else:
self.mac.pop(0)
self.names.pop(0)
self.devs.pop(0)
self.mac.append(lis)
self.devs.append(devs)
self.names.append(names)
def people_num(self):
new_list=[item for sub in self.mac for item in sub]
new_list_2=[item for sub in self.names for item in sub]
self.all_drives=list(set(new_list))
self.all_names=list(set(new_list_2))
self.get_all_devs()
self.noname()
self.service_come()
self.service_go()
return len(self.all_drives)*self.multiplier
def noname(self):
self.nn=0
for i in self.all_devs:
if i['name']=='No_Name':
self.nn=self.nn+1
def get_all_devs(self):
self.all_mac={}
self.all_distance={}
self.all_devs=[]
for i in self.devs:
for j in i:
if j['mac'] in self.all_drives:
if j['mac'] not in self.all_mac:
self.all_devs.append(j)
self.all_mac[j['mac']]=j['rssi']
else:
if abs(self.all_mac[j['mac']]-j['rssi'])>self.distance:
if self.all_mac[j['mac']]-j['rssi']>0:
self.all_distance[j['mac']]=-1
#远离
else:
self.all_distance[j['mac']]=1
#靠近
else:
self.all_distance[j['mac']]=0
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()
bdrive=BDrive(1,-80,10)
#dev_list = get_ble_devices_with_mac()
#num=bdrive.dev(dev_list)
#bdrive.all_devs
#全部设备
#bdrive.nn
#无名称设备
# f=open("log.csv", "a")
i=0
while True:
i=i+1
# 调用新方法获取带MAC、RSSI、名称的设备列表
dev_list = get_ble_devices_with_mac()
print("==== 在场蓝牙设备 ====")
if len(dev_list) == 0:
print("无设备")
else:
num=bdrive.dev(dev_list)
print(mdatetime(),"蓝牙在场人数估算:",num)
# 打印每台设备完整信息：MAC/名称/信号强度
for dev in dev_list:
print(f"MAC:{dev['mac']} | 名称:{dev['name']} | RSSI:{dev['rssi']} dBm")
print('全部设备名集合:',bdrive.all_names)
print('全部MAC集合:',bdrive.all_drives)
print('全部设备情况:',bdrive.all_devs)
print('全部MAC信号集合:',bdrive.all_mac)
print('全部MAC情况集合:',bdrive.all_distance)
print('全部no name情况集合:',bdrive.nn)
data={"time":mdatetime(),"PEOPLE":num}
# if i>10:
# f.write("{},{}\n".format(data["time"], data["PEOPLE"]))
# f.flush()
time.sleep_ms(1000) # 建议加延时，避免串口刷屏
旧代码
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, 250000, 60000, 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 BDrive:
def __init__(self,multiplier=1):
self.multiplier=multiplier
self.mac=[]
self.names=[]
self.all_names=[]
self.all_drives=[]
def dev(self,ones):
mac=[]
names=[]
for i in ones:
mac.append(i['mac'])
names.append(i['name'])
return self.get_num(mac,names)
def get_num(self,mac,names):
self.push_list(mac,names)
return self.people_num()
def push_list(self,lis,names):
if len(self.mac)<10:
self.mac.append(lis)
self.names.append(names)
else:
self.mac.pop(0)
self.names.pop(0)
self.mac.append(lis)
self.names.append(names)
def people_num(self):
new_list=[item for sub in self.mac for item in sub]
new_list_2=[item for sub in self.names for item in sub]
self.all_drives=list(set(new_list))
self.all_names=list(set(new_list_2))
return len(self.all_drives)*self.multiplier
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()
bdrive=BDrive()
# 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=bdrive.dev(dev_list)
print(mdatetime(),"Bluetooth Drive",num,'\n',bdrive.all_names,'\n',bdrive.all_drives)
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
