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))