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

sgp41.py 驱动文件

import time
from machine import I2C

class SGP41:
    I2C_ADDR = 0x59  # Fixed SGP41 I2C Address

    def __init__(self, i2c: I2C):
        self.i2c = i2c
        # Verify connection
        if self.I2C_ADDR not in self.i2c.scan():
            raise RuntimeError("SGP41 sensor not found on I2C bus.")
        self.reset()

    def _crc8(self, data: bytes) -> int:
        """Calculate Sensirion 8-bit CRC checksum."""
        crc = 0xFF
        for byte in data:
            crc ^= byte
            for _ in range(8):
                if crc & 0x80:
                    crc = (crc << 1) ^ 0x31
                else:
                    crc <<= 1
                crc &= 0xFF
        return crc

    def _send_command(self, cmd: int, arguments: list = None) -> bytes:
        """Send command with optional data arguments and return raw response bytes."""
        buf = bytearray([(cmd >> 8) & 0xFF, cmd & 0xFF])
        
        if arguments:
            for arg in arguments:
                arg_bytes = bytearray([(arg >> 8) & 0xFF, arg & 0xFF])
                buf.extend(arg_bytes)
                buf.append(self._crc8(arg_bytes))
                
        self.i2c.writeto(self.I2C_ADDR, buf)
        
        # Give sensor precise processing window time based on command
        if cmd == 0x2619:  # Execute conditioning
            time.sleep_ms(50)
            return self.i2c.readfrom(self.I2C_ADDR, 3)
        elif cmd == 0x2612:  # Measure raw signals
            time.sleep_ms(50)
            return self.i2c.readfrom(self.I2C_ADDR, 6)
        elif cmd == 0x0006:  # Reset
            time.sleep_ms(10)
        return b''

    def reset(self):
        """Soft reset the sensor."""
        self._send_command(0x0006)

    def execute_conditioning(self) -> int:
        """
        Executes pixel conditioning. Must be run for 10 seconds on startup 
        to ensure proper NOx readings. Returns raw VOC ticks.
        """
        # Default environment data: 50% RH and 25C mapped into sensor values
        default_rh = 0x8000 
        default_t = 0x6666
        
        reply = self._send_command(0x2619, [default_rh, default_t])
        if self._crc8(reply[0:2]) != reply[2]:
            raise RuntimeError("CRC Check Match Failure")
        return (reply[0] << 8) | reply[1]

    def measure_raw(self, humidity: float = None, temperature: float = None) -> tuple:
        """
        Measures raw VOC and NOx ticks.
        Accepts optional float context parameters to feed compensation loops.
        """
        if humidity is not None and temperature is not None:
            # Convert raw inputs to compensated ticks per Sensirion spec
            rh_ticks = int(round(humidity * 65535 / 100)) & 0xFFFF
            t_ticks = int(round((temperature + 45) * 65535 / 175)) & 0xFFFF
        else:
            rh_ticks = 0x8000 # 50%
            t_ticks = 0x6666  # 25C

        reply = self._send_command(0x2612, [rh_ticks, t_ticks])
        
        # Parse data blocks & validate checksums
        if (self._crc8(reply[0:2]) != reply[2]) or (self._crc8(reply[3:5]) != reply[5]):
            raise RuntimeError("CRC Check Match Failure")
            
        sraw_voc = (reply[0] << 8) | reply[1]
        sraw_nox = (reply[3] << 8) | reply[4]
        return sraw_voc, sraw_nox

使用代码 【必须要10秒启动否则会使用 期间只能VOC】

添加温湿度值可以提升精度

from machine import I2C, Pin
import time
from sgp41 import SGP41


i2c=I2C(-1, scl=Pin(14), sda=Pin(13))
sensor = SGP41(i2c)

#初始化10秒 制度VOC 否则NOX无法使用
for i in range(10):
    raw_voc = sensor.execute_conditioning()
    print(f"Warm-up Second {i+1}/10 - Raw VOC: {raw_voc}")
    time.sleep(1.0)
    
#读取VOC NOX    
voc_raw, nox_raw = sensor.measure_raw()
#请按照1HZ 的速度读取数据

#sensor.measure_raw(humidity=45.2, temperature=23.5)
#使用温湿度提升精度

Gas index 模拟算法

刚刚到只能读取raw值
sensirion_gas_index.py 算法

import math

class SensirionGasIndexAlgorithm:
    def __init__(self, is_nox=False):
        self.is_nox = is_nox
        
        # 官方标准预设参数
        if not is_nox:
            self.algorithm_type = 1.0   # VOC
            self.index_gain = 250.0
            self.index_offset = 100.0   # VOC 基准分
            self.gamma = 0.015          # VOC 学习率
        else:
            self.algorithm_type = 2.0   # NOx
            self.index_gain = 1.0       # NOx 增益系数
            self.index_offset = 1.0     # NOx 基准分
            self.gamma = 0.003          # NOx 学习率

        self.sraw_minimum = 0.0
        self.sraw_maximum = 65535.0
        
        # 算法内部状态机变量
        self.sraw_mean = None
        self.mean_variance = 0.0
        self.sraw_std_dev = 0.0

    def process(self, sraw: int) -> int:
        """输入 SGP41 的原始 Ticks，输出官方标准的 1-500 指数"""
        # 1. 边界裁剪
        sraw_f = float(sraw)
        if sraw_f < self.sraw_minimum: sraw_f = self.sraw_minimum
        if sraw_f > self.sraw_maximum: sraw_f = self.sraw_maximum

        # 2. 官方核心：自适应均值与方差迭代 (学习空气环境基线)
        if self.sraw_mean is None:
            self.sraw_mean = sraw_f
            self.mean_variance = 0.0
        else:
            delta = sraw_f - self.sraw_mean
            # 滚动更新均值
            self.sraw_mean += self.gamma * delta
            # 滚动更新方差
            self.mean_variance += self.gamma * ((delta * delta) - self.mean_variance)
        
        # 计算标准差，防止分母为 0 加入 0.001
        self.sraw_std_dev = math.sqrt(abs(self.mean_variance)) + 0.001

        # 3. 计算相对偏差归一化值 (Z-score)
        x = (sraw_f - self.sraw_mean) / self.sraw_std_dev

        # 4. 官方核心：Sigmoid 非线性感知映射
        try:
            if self.algorithm_type == 1.0:
                # VOC 公式：Ticks 变小代表浓度变高，Index 上升
                index_raw = self.index_offset + (self.index_gain / (1.0 + math.exp(x)))
            else:
                # NOx 公式：Ticks 变大代表浓度变高，Index 上升
                # 官方 NOx 指数较为特殊，常态通常极低(靠近1)，有污染时暴增
                index_raw = self.index_offset + (self.index_gain / (1.0 + math.exp(-x)))
        except OverflowError:
            # 防止 math.exp(x) 产生溢出崩溃
            index_raw = 500.0 if x < 0 else 1.0

        # 5. 边界剪裁限制在 1 - 500 之间
        final_index = int(round(index_raw))
        if final_index < 1: final_index = 1
        if final_index > 500: final_index = 500

        return final_index

使用

import time
from machine import Pin, I2C
from sgp41 import SGP41
from sensirion_gas_index import SensirionGasIndexAlgorithm

# 初始化硬件
i2c = I2C(0, scl=Pin(22), sda=Pin(21), freq=100000)
sensor = SGP41(i2c)

# 2. 实例化官方算法引擎
voc_engine = SensirionGasIndexAlgorithm(is_nox=False)
nox_engine = SensirionGasIndexAlgorithm(is_nox=True)
#初始化10秒 制度VOC 否则NOX无法使用
for i in range(10):
    raw_voc = sensor.execute_conditioning()
    print(f"Warm-up Second {i+1}/10 - Raw VOC: {raw_voc}")
    time.sleep(1.0)


print("正在以 1Hz 频率读取标准官方空气指数...")


while True:
    start_time = time.ticks_ms()
    
    try:
        # 读取原始数据
        voc_raw, nox_raw = sensor.measure_raw()
        
        # 使用官方翻译过来的算法处理
        voc_index = voc_engine.process(voc_raw)
        nox_index = nox_engine.process(nox_raw)
        
        print(f"标准 VOC 指数: {voc_index} | 标准 NOx 指数: {nox_index}")
        
    except Exception as e:
        print("错误:", e)
        
    # 严格保持 1Hz 频率
    elapsed = time.ticks_diff(time.ticks_ms(), start_time)
    time.sleep_ms(max(0, 1000 - elapsed))
   

FLASH 保存 sensirion_gas_index.py

import math
import json
import time

class SensirionGasIndexAlgorithm:
    def __init__(self, is_nox=False, storage_filename=None):
        self.is_nox = is_nox
        self.storage_filename = storage_filename 
        
        if not is_nox:
            self.algorithm_type = 1.0
            self.index_gain = 250.0
            self.index_offset = 100.0
            self.gamma = 0.015          
        else:
            self.algorithm_type = 2.0
            self.index_gain = 1.0
            self.index_offset = 1.0
            self.gamma = 0.003          

        self.sraw_minimum = 0.0
        self.sraw_maximum = 65535.0
        
        self.sraw_mean = None
        self.mean_variance = 0.0
        
        self.last_saved_mean = 0.0
        self.last_save_time = time.ticks_ms()
        self.total_samples = 0
        
        self.has_restored = False
        self._load_from_flash()

    def _load_from_flash(self):
        """开机时尝试从本地 Flash 读取基线"""
        if not self.storage_filename:
            return
        try:
            with open(self.storage_filename, "r") as f:
                data = json.load(f)
                self.sraw_mean = float(data["mean"])
                self.mean_variance = float(data["variance"])
                self.last_saved_mean = self.sraw_mean
                self.total_samples = int(data.get("samples", 1000))
                self.has_restored = True
                print(f"[{self.storage_filename}] 成功从 Flash 恢复历史基线: Mean={self.sraw_mean:.1f}")
        except Exception:
            print(f"[{self.storage_filename}] 未找到历史基线或文件损坏，将重新开始自学习。")

    def _save_to_flash(self, force=False):
        """控制写入次数：第一秒强行创建，之后每5分钟无条件记录一次"""
        if not self.storage_filename or self.sraw_mean is None:
            return
            
        current_time = time.ticks_ms()
        # 计算距离上一次保存过去了多少秒
        elapsed_seconds = time.ticks_diff(current_time, self.last_save_time) / 1000.0
        
        # 【修改后的逻辑】：
        if not force:
            # 1. 如果是出厂第一次运行且在开机第 1 秒（第 1 个样本），立刻无条件放行创建文件
            if not self.has_restored and self.total_samples <= 1:
                pass  
            else:
                # 2. 之后的正常运行期，不管空气有没有变，只要距离上次保存没到 300 秒（5分钟），就拦截
                if elapsed_seconds < 300.0:
                    return

        # 满足 5 分钟时间到了，执行物理写入
        try:
            data = {
                "mean": self.sraw_mean,
                "variance": self.mean_variance,
                "samples": self.total_samples
            }
            with open(self.storage_filename, "w") as f:
                json.dump(data, f)
            self.last_saved_mean = self.sraw_mean
            self.last_save_time = current_time  # 重置时间锚点
            self.has_restored = True
            print(f"--> [{self.storage_filename}] 已满5分钟，数据成功定时同步至 Flash (总样本数: {self.total_samples})")
        except Exception as e:
            print("Flash 写入失败:", e)

    def process(self, sraw: int) -> int:
        sraw_f = float(sraw)
        if sraw_f < self.sraw_minimum: sraw_f = self.sraw_minimum
        if sraw_f > self.sraw_maximum: sraw_f = self.sraw_maximum

        self.total_samples += 1

        if self.sraw_mean is None:
            self.sraw_mean = sraw_f
            self.mean_variance = 0.0
            self._save_to_flash()  # 触发第1秒瞬间创建文件
            return int(self.index_offset)
        else:
            delta = sraw_f - self.sraw_mean
            self.sraw_mean += self.gamma * delta
            self.mean_variance += self.gamma * ((delta * delta) - self.mean_variance)
        
        self._save_to_flash()  # 每次循环都评估是否到了 5 分钟

        sraw_std_dev = math.sqrt(abs(self.mean_variance))
        if sraw_std_dev < 0.01:
            sraw_std_dev = 0.01

        x = (sraw_f - self.sraw_mean) / sraw_std_dev

        if x > 70.0:   x = 70.0
        elif x < -70.0: x = -70.0

        if self.algorithm_type == 1.0:
            index_raw = self.index_offset + (self.index_gain / (1.0 + math.exp(x)))
        else:
            index_raw = self.index_offset + (self.index_gain / (1.0 + math.exp(-x)))

        final_index = int(round(index_raw))
        if final_index < 1: final_index = 1
        if final_index > 500: final_index = 500

        return final_index

使用

import time
from machine import Pin, I2C
from sgp41 import SGP41
from sensirion_gas_index import SensirionGasIndexAlgorithm

i2c = I2C(0, scl=Pin(22), sda=Pin(21), freq=100000)
sensor = SGP41(i2c)

# 【核心改变】：传入不同的文件名，分别存储 VOC 和 NOx 的基线
voc_engine = SensirionGasIndexAlgorithm(is_nox=False, storage_filename="voc_base.json")
nox_engine = SensirionGasIndexAlgorithm(is_nox=True, storage_filename="nox_base.json")

print("\n开始 1Hz 空气质量监测...")

while True:
    start_time = time.ticks_ms()
    
    try:
        voc_raw, nox_raw = sensor.measure_raw()
        
        # 运行算法（内部会自动处理掉电读取与动态保存）
        voc_index = voc_engine.process(voc_raw)
        nox_index = nox_engine.process(nox_raw)
        
        print(f"VOC 指数: {voc_index} | NOx 指数: {nox_index}")
        
    except Exception as e:
        print("循环报错:", e)
        
    # 严格保持 1Hz  pacing 步长
    elapsed = time.ticks_diff(time.ticks_ms(), start_time)
    time.sleep_ms(max(0, 1000 - elapsed))

最终成品

from machine import I2C, Pin
import time
from sgp41 import SGP41
from sensirion_gas_index import SensirionGasIndexAlgorithm


i2c=I2C(-1, scl=Pin(14), sda=Pin(13))
sensor = SGP41(i2c)

# 2. 实例化官方算法引擎
#voc_engine = SensirionGasIndexAlgorithm(is_nox=False)
#nox_engine = SensirionGasIndexAlgorithm(is_nox=True)

voc_engine = SensirionGasIndexAlgorithm(is_nox=False, storage_filename="voc_base.json")
nox_engine = SensirionGasIndexAlgorithm(is_nox=True, storage_filename="nox_base.json")


#初始化10秒 制度VOC 否则NOX无法使用
for i in range(10):
    raw_voc = sensor.execute_conditioning()
    print(f"Warm-up Second {i+1}/10 - Raw VOC: {raw_voc}")
    time.sleep(1.0)
    
print("正在以 1Hz 频率读取标准官方空气指数...")


while True:
    start_time = time.ticks_ms()
    
    try:
        # 读取原始数据
        voc_raw, nox_raw = sensor.measure_raw()
        
        # 使用官方翻译过来的算法处理
        voc_index = voc_engine.process(voc_raw)
        nox_index = nox_engine.process(nox_raw)
        
        print(f"标准 VOC 指数: {voc_index} | 标准 NOx 指数: {nox_index}")
        
    except Exception as e:
        print("错误:", e)
        
    # 严格保持 1Hz 频率
    elapsed = time.ticks_diff(time.ticks_ms(), start_time)
    time.sleep_ms(max(0, 1000 - elapsed))