json_encode($info[‘info_json’],JSON_UNESCAPED_UNICODE | JSON_NUMERIC_CHECK);

中文 数值 自动转换

sht40.py 文件 温湿度传感器 3.3V 【5v 也可以】

import machine
import time
class SHT40:
   def __init__(self, i2c, address=0x44):
       self.i2c = i2c
       self.address = address
   def read_temperature_humidity(self):
       # Send measurement command (High Precision)
       self.i2c.writeto(self.address, b'\xFD')
       time.sleep(0.01) # Wait for measurement to complete
       # Read 6 bytes of data
       data = self.i2c.readfrom(self.address, 6)
       # Convert raw data to temperature and humidity
       temp_raw = int.from_bytes(data[0:2], 'big')
       humidity_raw = int.from_bytes(data[3:5], 'big')
       temperature = -45 + (175 * (temp_raw / 65535.0))
       humidity = 100 * (humidity_raw / 65535.0)
       return temperature, humidity

调用

form sht40 import SHT40
import machine
i2c = machine.I2C(i2scl=machine.Pin(6), sda=machine.Pin(5), freq=100000)
# Initialize SHT40 sensor
sht40 = SHT40(i2c)
# Read and print temperature and humidity
while True:
   temp, hum = sht40.read_temperature_humidity()
   print("Temperature: {:.2f} °C".format(temp))
   print("Humidity: {:.2f} %".format(hum))
   time.sleep(1)

import machine
import time
class SHT40:
   def __init__(self, i2c, address=0x44):
       self.i2c = i2c
       self.address = address
   def read_temperature_humidity(self):
       # Send measurement command (High Precision)
       self.i2c.writeto(self.address, b'\xFD')
       time.sleep(0.01) # Wait for measurement to complete
       # Read 6 bytes of data
       data = self.i2c.readfrom(self.address, 6)
       # Convert raw data to temperature and humidity
       temp_raw = int.from_bytes(data[0:2], 'big')
       humidity_raw = int.from_bytes(data[3:5], 'big')
       temperature = -45 + (175 * (temp_raw / 65535.0))
       humidity = 100 * (humidity_raw / 65535.0)
       return temperature, humidity






# Initialize I2C
i2c = machine.I2C(i2scl=machine.Pin(6), sda=machine.Pin(5), freq=100000)
# Initialize SHT40 sensor
sht40 = SHT40(i2c)
# Read and print temperature and humidity
while True:
   temp, hum = sht40.read_temperature_humidity()
   print("Temperature: {:.2f} °C".format(temp))
   print("Humidity: {:.2f} %".format(hum))
   time.sleep(1)

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

https://github.com/cyberman54/ESP32-Paxcounter/releases

https://github.com/cyberman54/ESP32-Paxcounter