编写程序，让智能香烟检测仪检测烟雾时，室内新风系统自动启动提示，

智能香烟烟雾检测与新风联动系统

一、实际应用场景描述

在现代办公场所、餐厅包间、家庭客厅等封闭或半封闭环境中，二手烟已成为严重的室内空气污染问题。传统解决方案存在诸多局限：

- 被动式排风：依赖人工发现烟雾后手动开启新风系统，响应滞后

- 定时通风：无法根据实际污染情况动态调整，要么过度通风浪费能源，要么通风不足

- 单一检测：仅能检测烟雾存在与否，无法量化污染程度

- 无联动机制：检测系统与通风设备分离，形成信息孤岛

- 误报漏报：对烹饪油烟、香薰等类似气溶胶缺乏区分能力

- 缺乏记录：无法追溯污染事件，不利于环境管理和健康评估

本系统通过Python程序与MQ-2烟雾传感器、SGP30空气质量传感器、继电器模块、OLED显示屏、蜂鸣器集成，实现：

- 实时监测室内烟雾浓度（PPM）和空气质量指数（TVOC/eCO2）

- 智能判断污染等级，自动触发新风系统

- 多传感器数据融合，提高检测准确性

- 分级预警和自动联动控制

- 事件日志记录和空气质量趋势分析

- 可配置的灵敏度阈值和联动策略

二、引入痛点

1. 健康隐患：二手烟中的PM2.5、尼古丁、一氧化碳等有害物质持续危害人体

2. 响应延迟：人工发现到开启新风存在时间差，期间已造成污染扩散

3. 能源浪费：固定时长的通风模式在轻度污染时过度运行

4. 误判频繁：烹饪、焚香等活动触发误报，影响系统可信度

5. 管理盲区：无法记录污染事件，难以追溯和评估环境改善效果

6. 设备割裂：检测、报警、通风各自独立，操作繁琐

7. 法律合规：部分场所对禁烟有严格规定，缺乏有效监管手段

三、核心逻辑讲解

烟雾检测科学原理

污染等级 PM2.5等效(μg/m³) 烟雾浓度(PPM) 系统响应 新风档位

优 0-35 0-200 绿灯常亮 关闭

良 35-75 200-400 绿灯闪烁 低速

轻度污染 75-115 400-800 黄灯常亮 中速

中度污染 115-150 800-1500 橙灯闪烁 高速

重度污染 >150 >1500 红灯常亮+蜂鸣 最大档

MQ-2传感器特性曲线

Rs/Ro比值 vs 气体浓度(LPG/烟雾)

Rs/Ro = 1 : 浓度≈200ppm (洁净空气基准)

Rs/Ro = 2 : 浓度≈400ppm (轻度污染)

Rs/Ro = 5 : 浓度≈1000ppm (中度污染)

Rs/Ro = 10 : 浓度≈2000ppm (重度污染)

核心算法流程

1. 多传感器数据采集：

- MQ-2模拟量读取（烟雾浓度）

- SGP30数字量读取（TVOC/eCO2）

- DHT22温湿度读取（补偿因子）

2. 数据预处理：

- 滑动平均滤波消除ADC噪声

- 温度湿度补偿MQ-2读数

- SGP30基线校准和恢复

3. 污染等级判定：

- 烟雾浓度阈值判断

- TVOC超标检测（烹饪油烟识别）

- 综合污染指数计算

4. 联动决策引擎：

- 分级响应策略

- 延时启动防误触

- 渐进式风量调节

5. 执行控制：

- 继电器控制新风电机

- PWM调速实现无级变速

- 状态指示灯和OLED显示

6. 事件记录与分析：

- CSV日志存储

- 污染趋势图表生成

- 统计分析报告

关键技术难点及解决方案

难点 解决方案

MQ-2预热时间长 启动时预热提示，等待传感器稳定

交叉敏感性 SGP30 TVOC检测区分烟雾vs油烟

误报问题 多条件联合判断+延时确认机制

继电器抖动 软件消抖+硬件RC滤波

功耗优化 休眠唤醒机制+动态采样频率

四、代码模块化（含安全设计）

项目结构

smart_smoke_detector/

├── main.py # 主程序入口

├── sensors/ # 传感器模块

│ ├── __init__.py

│ ├── mq2_sensor.py # MQ-2烟雾传感器驱动

│ ├── sgp30_sensor.py # SGP30空气质量传感器驱动

│ ├── dht22_sensor.py # DHT22温湿度传感器驱动

│ └── sensor_fusion.py # 多传感器数据融合

├── actuators/ # 执行器模块

│ ├── __init__.py

│ ├── relay_controller.py # 继电器控制模块

│ ├── pwm_fan.py # PWM风扇调速模块

│ ├── oled_display.py # OLED显示屏控制

│ ├── buzzer.py # 蜂鸣器控制

│ └── led_indicator.py # RGB LED指示灯控制

├── core/ # 核心逻辑模块

│ ├── __init__.py

│ ├── pollution_analyzer.py # 污染分析引擎

│ ├── control_strategy.py # 联动控制策略

│ └── event_manager.py # 事件管理器

├── storage/ # 数据存储模块

│ ├── __init__.py

│ ├── data_logger.py # 数据记录器

│ └── report_generator.py # 报表生成器

├── config.py # 系统配置

├── requirements.txt # 依赖清单

├── README.md # 项目文档

└── utils/

├── __init__.py

├── logger.py # 日志工具

└── decorators.py # 装饰器工具

核心代码实现

1. config.py（系统配置）

"""

智能香烟烟雾检测与新风联动系统配置文件

包含传感器、执行器、控制策略等所有配置参数

"""

from dataclasses import dataclass, field

from typing import List, Optional, Dict, Any, Tuple

from enum import Enum, auto

from pathlib import Path

import os

import math

class FanSpeed(Enum):

"""风扇速度档位"""

OFF = 0

LOW = 1

MEDIUM = 2

HIGH = 3

MAX = 4

class PollutionLevel(Enum):

"""污染等级"""

EXCELLENT = auto() # 优

GOOD = auto() # 良

LIGHT_POLLUTION = auto() # 轻度污染

MODERATE_POLLUTION = auto() # 中度污染

HEAVY_POLLUTION = auto() # 重度污染

HAZARDOUS = auto() # 危险

@dataclass

class SensorConfig:

"""传感器配置"""

# MQ-2烟雾传感器

mq2_analog_pin: int = 0 # ADC通道

mq2_heating_time: int = 20 # 预热时间（秒）

mq2_ro_clean_air: float = 9.83 # 洁净空气中Ro值

mq2_attenuation: float = 0.5 # 信号衰减系数

# SGP30空气质量传感器

sgp30_i2c_address: int = 0x58

sgp30_i2c_bus: int = 1

sgp30_baseline_file: str = "sgp30_baseline.json"

# DHT22温湿度传感器

dht22_gpio_pin: int = 4

dht22_retry_count: int = 3

# 采样配置

sample_interval: float = 1.0 # 采样间隔（秒）

filter_window: int = 10 # 移动平均窗口

temperature_compensation: bool = True

# 模拟模式

simulation_mode: bool = True # 开发阶段使用

@dataclass

class ThresholdConfig:

"""污染阈值配置"""

# 烟雾浓度阈值 (PPM)

smoke_excellent_max: float = 200.0

smoke_good_max: float = 400.0

smoke_light_max: float = 800.0

smoke_moderate_max: float = 1500.0

smoke_heavy_max: float = 3000.0

# TVOC阈值 (ppb)

tvoc_excellent_max: float = 50.0

tvoc_good_max: float = 200.0

tvoc_light_max: float = 500.0

tvoc_moderate_max: float = 1000.0

# eCO2阈值 (ppm)

eco2_excellent_max: float = 400.0

eco2_good_max: float = 1000.0

eco2_light_max: float = 1500.0

eco2_moderate_max: float = 2000.0

# 综合污染指数权重

smoke_weight: float = 0.5

tvoc_weight: float = 0.3

eco2_weight: float = 0.2

# 确认延时（防误报）

detection_delay: float = 3.0 # 连续检测到污染才确认

recovery_delay: float = 10.0 # 连续清洁才恢复

@dataclass

class ControlConfig:

"""控制策略配置"""

# 继电器配置

relay_pin: int = 17

relay_active_low: bool = True # 低电平触发

# 风扇PWM配置

pwm_pin: int = 18

pwm_frequency: int = 25000 # 25kHz

# 各档位PWM占空比

fan_speed_duty_cycles: Dict[FanSpeed, float] = field(default_factory=lambda: {

FanSpeed.OFF: 0.0,

FanSpeed.LOW: 30.0,

FanSpeed.MEDIUM: 55.0,

FanSpeed.HIGH: 75.0,

FanSpeed.MAX: 100.0

})

# 联动策略

enable_auto_control: bool = True

enable_manual_override: bool = True

gradual_speed_change: bool = True # 渐进式调速

speed_transition_time: float = 2.0 # 档位切换过渡时间（秒）

# 延时启动（给用户反应时间）

activation_delay: float = 5.0 # 检测到污染后延时启动

@dataclass

class DisplayConfig:

"""显示配置"""

# OLED配置

oled_width: int = 128

oled_height: int = 64

oled_i2c_address: int = 0x3C

oled_i2c_bus: int = 1

# 显示内容

show_smoke_concentration: bool = True

show_air_quality_index: bool = True

show_fan_status: bool = True

show_pollution_level: bool = True

show_temperature_humidity: bool = True

show_timestamp: bool = True

# 刷新率

refresh_rate: float = 1.0 # 刷新间隔（秒）

@dataclass

class AlertConfig:

"""提醒配置"""

# LED配置

led_red_pin: int = 22

led_green_pin: int = 27

led_blue_pin: int = 24

# 蜂鸣器配置

buzzer_pin: int = 23

buzzer_frequency_normal: int = 1000

buzzer_frequency_warning: int = 1500

buzzer_frequency_alert: int = 2000

# 提醒模式

visual_alert_enabled: bool = True

audio_alert_enabled: bool = True

alert_cooldown: float = 5.0 # 提醒冷却时间

@dataclass

class StorageConfig:

"""存储配置"""

# 数据记录

log_enabled: bool = True

log_directory: str = "logs"

log_filename: str = "smoke_detection_log.csv"

log_interval: float = 60.0 # 记录间隔（秒）

# 事件记录

event_log_filename: str = "events_log.csv"

# 历史数据

max_history_points: int = 1000

export_enabled: bool = True

export_format: str = "csv"

@dataclass

class SystemConfig:

"""系统总配置"""

# 子配置

sensor: SensorConfig = field(default_factory=SensorConfig)

thresholds: ThresholdConfig = field(default_factory=ThresholdConfig)

control: ControlConfig = field(default_factory=ControlConfig)

display: DisplayConfig = field(default_factory=DisplayConfig)

alert: AlertConfig = field(default_factory=AlertConfig)

storage: StorageConfig = field(default_factory=StorageConfig)

# 系统设置

debug_mode: bool = False

log_level: str = "INFO"

log_file: str = "smoke_detector.log"

# 安全限制

max_adc_value: int = 1023

min_adc_value: int = 0

max_smoke_ppm: float = 10000.0

min_smoke_ppm: float = 0.0

max_tvoc_ppb: float = 60000.0

max_eco2_ppm: float = 40000.0

# 数据路径

data_directory: str = "data"

def __post_init__(self):

# 确保目录存在

for directory in [self.storage.log_directory, self.data_directory]:

Path(directory).mkdir(parents=True, exist_ok=True)

# 全局配置实例

config = SystemConfig()

# 污染等级颜色映射

POLLUTION_LEVEL_COLORS = {

PollutionLevel.EXCELLENT: (0, 255, 0), # 绿色

PollutionLevel.GOOD: (0, 200, 100), # 浅绿

PollutionLevel.LIGHT_POLLUTION: (255, 255, 0), # 黄色

PollutionLevel.MODERATE_POLLUTION: (255, 165, 0), # 橙色

PollutionLevel.HEAVY_POLLUTION: (255, 0, 0), # 红色

PollutionLevel.HAZARDOUS: (128, 0, 128), # 紫色

}

# 风扇档位对应的风速描述

FAN_SPEED_DESCRIPTIONS = {

FanSpeed.OFF: "关闭",

FanSpeed.LOW: "低速",

FanSpeed.MEDIUM: "中速",

FanSpeed.HIGH: "高速",

FanSpeed.MAX: "最大档"

}

# 污染等级描述

POLLUTION_LEVEL_DESCRIPTIONS = {

PollutionLevel.EXCELLENT: "空气质量优",

PollutionLevel.GOOD: "空气质量良",

PollutionLevel.LIGHT_POLLUTION: "轻度污染",

PollutionLevel.MODERATE_POLLUTION: "中度污染",

PollutionLevel.HEAVY_POLLUTION: "重度污染",

PollutionLevel.HAZARDOUS: "危险级别"

}

2. sensors/mq2_sensor.py（MQ-2烟雾传感器模块）

"""

MQ-2烟雾传感器模块

支持模拟量读取、温度补偿、数据滤波

用于检测LPG、烟雾、酒精、甲烷等可燃气体

"""

import time

import logging

import random

import math

from typing import Optional, List, Tuple, Deque

from dataclasses import dataclass, field

from enum import Enum, auto

from collections import deque

from config import config, SafetyLimits, PollutionLevel

logger = logging.getLogger(__name__)

class SensorError(Exception):

"""传感器异常类"""

pass

class CalibrationError(SensorError):

"""校准异常类"""

pass

@dataclass

class SmokeReading:

"""烟雾读数数据类"""

timestamp: float

raw_adc_value: int

voltage: float

rs_value: float

ratio_rs_ro: float

smoke_ppm: float

temperature: Optional[float] = None

humidity: Optional[float] = None

is_valid: bool = True

quality: float = 1.0 # 数据质量分数 0-1

class MQ2SmokeSensor:

"""

MQ-2烟雾传感器驱动

特性：

- 模拟量ADC读取

- 温度湿度补偿

- 滑动平均滤波

- 自动校准功能

- 模拟模式支持

- 预热状态管理

- 多气体检测（主要检测烟雾）

工作原理：

1. 加热元件使敏感层达到工作温度

2. 目标气体吸附改变敏感层电阻

3. 通过测量负载电阻分压计算气体浓度

"""

# MQ-2特性参数

# 对数坐标下，Rs/Ro与气体浓度的关系

# 烟雾/LPG: log(y) = a*log(x) + b

SMOKE_LPG_PARAMS = {

'a': -0.42, # 斜率

'b': 1.98, # 截距

'ro_ref': 9.83 # 洁净空气中Ro值

}

def __init__(self, analog_pin: int = 0, ro_clean_air: float = 9.83):

"""

初始化MQ-2传感器

Args:

analog_pin: ADC通道号

ro_clean_air: 洁净空气中的Ro值

"""

self.analog_pin = analog_pin

self.ro_clean_air = ro_clean_air

# 当前Ro值（会随使用老化）

self._ro_current: float = ro_clean_air

# 滤波缓冲区

self._filter_buffer: Deque[float] = deque(

maxlen=config.sensor.filter_window

)

# 状态管理

self._is_initialized = False

self._is_preheated = False

self._preheat_start_time: Optional[float] = None

self._error_count = 0

self._max_errors = 10

# 模拟模式

self._simulation_mode = config.simulation_mode

self._simulation_base_ppm = 50.0 # 模拟基础烟雾浓度

# 统计信息

self._readings_count = 0

self._calibration_count = 0

logger.info(

f"MQ-2烟雾传感器初始化: 模拟引脚={analog_pin}, "

f"Ro洁净空气={ro_clean_air}"

)

def initialize(self) -> bool:

"""

初始化传感器硬件

Returns:

bool: 初始化是否成功

"""

try:

if self._simulation_mode:

logger.info("MQ-2传感器运行在模拟模式")

self._is_initialized = True

self._start_preheat()

return True

# 实际硬件初始化

# import spidev

# self.spi = spidev.SpiDev()

# self.spi.open(0, 0)

# self.spi.max_speed_hz = 1350000

self._is_initialized = True

self._start_preheat()

logger.info("MQ-2烟雾传感器初始化成功")

return True

except Exception as e:

logger.error(f"MQ-2传感器初始化失败: {e}")

return False

def _start_preheat(self):

"""启动预热过程"""

self._preheat_start_time = time.time()

self._is_preheated = False

logger.info(f"MQ-2传感器开始预热，预计需要{config.sensor.mq2_heating_time}秒")

def _check_preheat_status(self) -> Tuple[bool, float]:

"""

检查预热状态

Returns:

Tuple[bool, float]: (是否预热完成, 剩余时间)

"""

if self._preheat_start_time is None:

return True, 0.0

elapsed = time.time() - self._preheat_start_time

remaining = config.sensor.mq2_heating_time - elapsed

if remaining <= 0:

self._is_preheated = True

logger.info("MQ-2传感器预热完成")

return True, 0.0

return False, remaining

def _read_raw_adc(self) -> Optional[int]:

"""

读取原始ADC值

Returns:

Optional[int]: 10位ADC值（0-1023）

"""

if not self._is_initialized:

logger.error("传感器未初始化")

return None

try:

if self._simulation_mode:

# 模拟数据：模拟吸烟过程和自然衰减

elapsed = time.time() % 300 # 5分钟周期

# 模拟吸烟事件

smoking_factor = 0.0

if 60 < elapsed < 120: # 第1-2分钟模拟吸烟

smoking_factor = math.sin((elapsed - 60) / 60 * math.pi) * 0.8

elif 180 < elapsed < 210: # 第3-3.5分钟再次吸烟

smoking_factor = math.sin((elapsed - 180) / 30 * math.pi) * 0.6

# 基础浓度 + 吸烟影响 + 随机噪声

simulated_value = (

self._simulation_base_ppm +

smoking_factor * 1500 +

random.uniform(-20, 20)

)

# 转换为ADC值（假设3.3V供电，MQ-2输出电压与浓度正相关）

adc_value = int(min(1023, simulated_value / 3000 * 1023))

return adc_value

# 实际硬件读取

# command = 0x01 << 7 | self.analog_pin << 4

# response = self.spi.xfer2([command, 0x00, 0x00])

# adc_value = ((response[1] & 0x03) << 8) | response[2]

# return adc_value

return int(random.uniform(100, 600))

except Exception as e:

logger.error(f"读取ADC失败: {e}")

self._error_count += 1

if self._error_count >= self._max_errors:

logger.critical("传感器错误过多，进入保护模式")

self._is_initialized = False

return None

def read_smoke(self, temperature: Optional[float] = None,

humidity: Optional[float] = None) -> Optional[SmokeReading]:

"""

读取烟雾浓度

Args:

temperature: 环境温度（用于补偿）

humidity: 环境湿度（用于补偿）

Returns:

Optional[SmokeReading]: 烟雾读数对象

"""

# 检查预热状态

preheated, remaining = self._check_preheat_status()

if not preheated:

logger.debug(f"传感器预热中，剩余{remaining:.1f}秒")

# 预热期间返回最后一次有效读数或默认值

return self._create_preheat_reading(remaining)

raw_value = self._read_raw_adc()

if raw_value is None:

return None

# 重置错误计数

self._error_count = 0

# 转换为电压

voltage = self._adc_to_voltage(raw_value)

# 计算Rs值（传感器电阻）

rs_value = self._calculate_rs(voltage)

# 计算Rs/Ro比值

ratio_rs_ro = rs_value / self._ro_current

# 温度湿度补偿

compensated_ratio = self._apply_environmental_compensation(

ratio_rs_ro, temperature, humidity

)

# 转换为烟雾浓度（PPM）

smoke_ppm = self._ratio_to_smoke_ppm(compensated_ratio)

# 滤波处理

filtered_ppm = self._apply_filter(smoke_ppm)

# 创建读数对象

reading = SmokeReading(

timestamp=time.time(),

raw_adc_value=raw_value,

voltage=round(voltage, 3),

rs_value=round(rs_value, 2),

ratio_rs_ro=round(ratio_rs_ro, 3),

smoke_ppm=round(filtered_ppm, 1),

temperature=temperature,

humidity=humidity,

is_valid=True,

quality=self._calculate_data_quality(raw_value)

)

self._last_reading = reading

self._readings_count += 1

logger.debug(

f"烟雾读数: ADC={raw_value}, 电压={voltage:.3f}V, "

f"Rs={rs_value:.2f}Ω, Rs/Ro={ratio_rs_ro:.3f}, "

f"烟雾={filtered_ppm:.1f}PPM"

)

return reading

def _adc_to_voltage(self, adc_value: int) -> float:

"""ADC值转电压"""

return (adc_value / config.sensor.max_adc_value) * 3.3

def _calculate_rs(self, voltage: float) -> float:

"""

计算传感器电阻Rs

MQ-2电路：Vcc -> Ro -> Rs -> GND

中间抽头接ADC

Vout = Vcc * Rs / (Ro + Rs)

Args:

voltage: ADC测量的电压值

Returns:

float: Rs电阻值（欧姆）

"""

# 假设负载电阻Ro为10K（典型值）

load_resistance = 10000.0 # 10KΩ

if voltage <= 0 or voltage >= 3.3:

return load_resistance

# Rs = Ro * (Vcc - Vout) / Vout

rs = load_resistance * (3.3 - voltage) / voltage

return rs

def _apply_environmental_compensation(self, ratio: float,

temperature: Optional[float],

humidity: Optional[float]) -> float:

"""

应用温度湿度补偿

MQ-2对温度和湿度敏感，需要进行补偿

Args:

ratio: Rs/Ro比值

temperature: 温度（摄氏度）

humidity: 湿度（%）

Returns:

float: 补偿后的比值

"""

if not config.sensor.temperature_compensation:

return ratio

if temperature is None or humidity is None:

return ratio

# 温度补偿系数（典型值，需根据实际传感器校准）

temp_coefficient = 0.005 # 每度温度变化的影响

humidity_coefficient = 0.002 # 每1%湿度变化的影响

# 补偿公式

# 温度升高，Rs减小；湿度升高，Rs增大

temp_compensation = 1.0 + temp_coefficient * (temperature - 20.0)

humidity_compensation = 1.0 - humidity_coefficient * (humidity - 50.0)

compensated_ratio = ratio * temp_compensation * humidity_compensation

logger.debug(

f"环境补偿: 温度={temperature}°C, 湿度={humidity}%, "

f"补偿系数={temp_compensation*humidity_compensation:.4f}"

)

return compensated_ratio

def _ratio_to_smoke_ppm(self, ratio: float) -> float:

"""

将Rs/Ro比值转换为烟雾浓度（PPM）

使用对数线性关系：

log(PPM) = a * log(Rs/Ro) + b

Args:

ratio: Rs/Ro比值

Returns:

float: 烟雾浓度（PPM）

"""

params = self.SMOKE_LPG_PARAMS

if ratio <= 0:

return 0.0

# 对数转换

log_ratio = math.log10(ratio)

log_ppm = params['a'] * log_ratio + params['b']

# 反对数

ppm = math.pow(10, log_ppm)

# 限制范围

ppm = max(config.min_smoke_ppm, min(config.max_smoke_ppm, ppm))

return ppm

def _apply_filter(self, ppm: float) -> float:

"""应用滑动平均滤波"""

self._filter_buffer.append(ppm)

if len(self._filter_buffer) == 0:

return ppm

# 加权平均，最近的值权重更高

weights = [i + 1 for i in range(len(self._filter_buffer))]

weighted_sum = sum(p * w for p, w in zip(self._filter_buffer, weights))

total_weight = sum(weights)

return weighted_sum / total_weight

def _calculate_data_quality(self, raw_value: int) -> float:

"""计算数据质量分数"""

# 基于ADC值的合理性评估

if raw_value < 50 or raw_value > 900:

return 0.7 # 边缘值，质量较低

elif raw_value < 100 or raw_value > 800:

return 0.85 # 接近边缘，质量中等

else:

return 1.0 # 正常范围，质量高

def _create_preheat_reading(self, remaining_time: float) -> SmokeReading:

"""创建预热期间的读数""

利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！