企业官网建设流程全解析

做国内网站花费,大庆商城网站建设,网站建设的总体设计,做的网站有营销效果吗智能室内空气质量监控与净化系统一、实际应用场景描述场景背景在现代煤矿智能化开采的办公环境中#xff0c;由于煤矿设备维护区域、实验室、会议室等封闭空间可能存在甲醛#xff08;来自新设备、装修材料#xff09;和PM2.5#xff08;来自外部空气、设备运行#xff09…智能室内空气质量监控与净化系统一、实际应用场景描述场景背景在现代煤矿智能化开采的办公环境中由于煤矿设备维护区域、实验室、会议室等封闭空间可能存在甲醛来自新设备、装修材料和PM2.5来自外部空气、设备运行污染。长期暴露在这样的环境中会影响员工健康和工作效率。痛点分析1. 健康风险甲醛是致癌物PM2.5可引发呼吸道疾病2. 被动响应传统方式依赖人工发现污染并手动开启净化设备3. 能源浪费净化器常开造成电力浪费设备损耗4. 监管困难缺乏实时数据记录和分析难以优化空气质量二、核心逻辑讲解系统架构传感器数据采集 → 数据处理分析 → 阈值判断 → 设备控制 → 状态反馈控制逻辑1. 实时监测甲醛和PM2.5浓度2. 当任一指标超标时自动启动净化器3. 持续监测直至两项指标均达标4. 达标后延迟关闭防止频繁启停5. 记录所有数据用于分析和优化三、代码实现项目结构air_quality_system/├── main.py # 主程序├── sensors/│ ├── __init__.py│ ├── air_quality_sensor.py # 传感器模拟/接口│ └── sensor_interface.py # 传感器抽象接口├── controllers/│ ├── __init__.py│ └── purifier_controller.py # 净化器控制器├── config/│ ├── __init__.py│ └── settings.py # 配置文件├── utils/│ ├── __init__.py│ ├── logger.py # 日志模块│ └── data_handler.py # 数据处理├── requirements.txt # 依赖包└── README.md # 说明文档1. 配置文件 (config/settings.py)系统配置文件定义空气质量标准和系统参数class AirQualityStandard:空气质量标准类 (基于GB/T 18883-2022室内空气质量标准)# 甲醛浓度标准 (mg/m³)FORMALDEHYDE {excellent: 0.03, # 优级good: 0.05, # 良好limit: 0.08, # 国家标准限值warning: 0.10, # 警告阈值danger: 0.20 # 危险阈值}# PM2.5浓度标准 (μg/m³)PM25 {excellent: 15, # 优级good: 35, # 良好limit: 50, # 国家标准限值warning: 75, # 警告阈值danger: 150 # 危险阈值}class SystemConfig:系统运行参数配置# 监测间隔秒MONITOR_INTERVAL 5# 净化器控制参数PURIFIER_DELAY_OFF 60 # 达标后延迟关闭时间秒PURIFIER_MIN_RUN_TIME 300 # 净化器最小运行时间秒# 报警参数ENABLE_ALERT TrueALERT_INTERVAL 300 # 报警间隔秒# 数据记录LOG_ENABLED TrueDATA_SAVE_INTERVAL 60 # 数据保存间隔秒2. 传感器接口模块 (sensors/sensor_interface.py)传感器抽象接口模块定义统一的传感器接口规范from abc import ABC, abstractmethodimport timefrom datetime import datetimeclass AirQualitySensor(ABC):空气质量传感器抽象基类def __init__(self, sensor_id: str, location: str unknown):初始化传感器参数:sensor_id: 传感器唯一标识location: 传感器位置描述self.sensor_id sensor_idself.location locationself.status disconnectedself.last_update Noneabstractmethoddef connect(self) - bool:连接传感器设备passabstractmethoddef disconnect(self) - bool:断开传感器连接passabstractmethoddef read_data(self) - dict:读取传感器数据返回:包含空气质量数据的字典passdef get_status(self) - dict:获取传感器状态return {sensor_id: self.sensor_id,location: self.location,status: self.status,last_update: self.last_update}class SensorData:传感器数据封装类def __init__(self, formaldehyde: float, pm25: float, temperature: float None,humidity: float None, co2: float None):初始化传感器数据参数:formaldehyde: 甲醛浓度 (mg/m³)pm25: PM2.5浓度 (μg/m³)temperature: 温度 (°C)humidity: 湿度 (%)co2: 二氧化碳浓度 (ppm)self.formaldehyde formaldehydeself.pm25 pm25self.temperature temperatureself.humidity humidityself.co2 co2self.timestamp datetime.now()def to_dict(self) - dict:转换为字典格式return {timestamp: self.timestamp.isoformat(),formaldehyde: self.formaldehyde,pm25: self.pm25,temperature: self.temperature,humidity: self.humidity,co2: self.co2}def __str__(self) - str:字符串表示return (f时间: {self.timestamp.strftime(%Y-%m-%d %H:%M:%S)}, f甲醛: {self.formaldehyde:.3f} mg/m³, fPM2.5: {self.pm25:.1f} μg/m³)3. 传感器模拟实现 (sensors/air_quality_sensor.py)空气质量传感器实现模块包含真实传感器接口和模拟传感器import randomimport timefrom typing import Optionalfrom .sensor_interface import AirQualitySensor, SensorDataclass SimulatedAirQualitySensor(AirQualitySensor):模拟空气质量传感器用于开发和测试def __init__(self, sensor_id: str, location: str 模拟传感器,base_levels: Optional[dict] None):初始化模拟传感器参数:sensor_id: 传感器IDlocation: 传感器位置base_levels: 基础污染物水平super().__init__(sensor_id, location)# 基础污染物水平self.base_levels base_levels or {formaldehyde: 0.02, # 基础甲醛水平pm25: 20, # 基础PM2.5水平}# 污染事件参数self.pollution_event Falseself.event_start_time Noneself.event_duration 0def connect(self) - bool:连接模拟传感器print(f[INFO] 连接模拟传感器 {self.sensor_id} 在 {self.location})self.status connectedreturn Truedef disconnect(self) - bool:断开模拟传感器print(f[INFO] 断开模拟传感器 {self.sensor_id})self.status disconnectedreturn Truedef _simulate_pollution_event(self) - tuple:模拟污染事件current_time time.time()# 随机触发污染事件if not self.pollution_event and random.random() 0.05: # 5%概率触发self.pollution_event Trueself.event_start_time current_timeself.event_duration random.randint(300, 1800) # 5-30分钟# 处理污染事件if self.pollution_event:event_progress (current_time - self.event_start_time) / self.event_durationif event_progress 1: # 事件结束self.pollution_event Falsereturn 0, 0else:# 污染峰值在事件中期pollution_factor 4 * event_progress * (1 - event_progress)return pollution_factor * 0.3, pollution_factor * 200 # 甲醛和PM2.5增量return 0, 0def read_data(self) - dict:读取模拟传感器数据返回:传感器数据字典if self.status ! connected:raise ConnectionError(传感器未连接)# 获取基础水平formaldehyde_base self.base_levels[formaldehyde]pm25_base self.base_levels[pm25]# 模拟日常波动hour time.localtime().tm_hourtime_factor 0.5 0.5 * abs(12 - hour) / 12 # 中午最高# 模拟随机波动formaldehyde_variation random.uniform(-0.01, 0.01) * time_factorpm25_variation random.uniform(-5, 5) * time_factor# 模拟污染事件formaldehyde_event, pm25_event self._simulate_pollution_event()# 计算最终值formaldehyde max(0.01, formaldehyde_base formaldehyde_variation formaldehyde_event)pm25 max(5, pm25_base pm25_variation pm25_event)# 添加温湿度模拟temperature 20 random.uniform(-2, 2) (hour - 12) * 0.5humidity 50 random.uniform(-10, 10)# 创建数据对象sensor_data SensorData(formaldehydeformaldehyde,pm25pm25,temperatureround(temperature, 1),humidityround(humidity, 1))self.last_update time.time()return sensor_data.to_dict()class RealAirQualitySensor(AirQualitySensor):真实空气质量传感器接口示例需根据实际硬件实现def __init__(self, sensor_id: str, location: str, port: str, baudrate: int 9600):初始化真实传感器参数:sensor_id: 传感器IDlocation: 传感器位置port: 串口端口baudrate: 波特率super().__init__(sensor_id, location)self.port portself.baudrate baudrateself.serial_conn Nonedef connect(self) - bool:连接真实传感器try:import serialself.serial_conn serial.Serial(portself.port,baudrateself.baudrate,timeout2)self.status connectedprint(f[INFO] 成功连接传感器 {self.sensor_id} 在端口 {self.port})return Trueexcept Exception as e:print(f[ERROR] 连接传感器失败: {e})self.status disconnectedreturn Falsedef disconnect(self) - bool:断开传感器连接if self.serial_conn and self.serial_conn.is_open:self.serial_conn.close()self.status disconnectedreturn Truedef _parse_sensor_data(self, raw_data: bytes) - Optional[dict]:解析传感器原始数据注意此函数需要根据实际传感器协议实现# 示例解析逻辑需根据实际传感器调整try:# 假设传感器数据格式: HCHO:0.05,PM2.5:35,TEMP:25.0,HUM:50data_str raw_data.decode(utf-8).strip()data_dict {}for item in data_str.split(,):if : in item:key, value item.split(:)data_dict[key.strip()] float(value)# 映射到标准字段formaldehyde data_dict.get(HCHO, 0.0)pm25 data_dict.get(PM2.5, 0.0)temperature data_dict.get(TEMP, None)humidity data_dict.get(HUM, None)sensor_data SensorData(formaldehydeformaldehyde,pm25pm25,temperaturetemperature,humidityhumidity)return sensor_data.to_dict()except Exception as e:print(f[ERROR] 解析传感器数据失败: {e})return Nonedef read_data(self) - dict:从真实传感器读取数据if not self.serial_conn or not self.serial_conn.is_open:raise ConnectionError(传感器未连接)try:# 发送读取命令根据实际传感器协议self.serial_conn.write(bREAD\r\n)time.sleep(0.1)# 读取返回数据raw_data self.serial_conn.readline()# 解析数据data self._parse_sensor_data(raw_data)if data:self.last_update time.time()return dataelse:raise ValueError(无效的传感器数据)except Exception as e:print(f[ERROR] 读取传感器数据失败: {e})raise4. 净化器控制器 (controllers/purifier_controller.py)空气净化器控制器模块控制净化器的开启和关闭import timefrom abc import ABC, abstractmethodfrom datetime import datetime, timedeltaclass AirPurifierController(ABC):空气净化器控制器抽象基类def __init__(self, device_id: str, location: str unknown):初始化净化器控制器参数:device_id: 设备IDlocation: 设备位置self.device_id device_idself.location locationself.status off # off, on, errorself.start_time Noneself.total_runtime timedelta(0)self.energy_consumption 0 # 千瓦时abstractmethoddef turn_on(self) - bool:开启净化器passabstractmethoddef turn_off(self) - bool:关闭净化器passabstractmethoddef get_status(self) - dict:获取净化器状态passdef update_runtime(self):更新运行时间统计if self.status on and self.start_time:runtime datetime.now() - self.start_timeself.total_runtime runtime# 假设功率为50Wself.energy_consumption 0.05 * runtime.total_seconds() / 3600self.start_time datetime.now()class SimulatedPurifierController(AirPurifierController):模拟空气净化器控制器用于开发和测试def __init__(self, device_id: str, location: str 模拟净化器):初始化模拟净化器控制器super().__init__(device_id, location)self.power_consumption 0.05 # 千瓦self.filter_life 100 # 过滤器寿命百分比self.air_flow_level 2 # 1-3档def turn_on(self) - bool:开启模拟净化器if self.status ! on:print(f[INFO] 开启净化器 {self.device_id} 在 {self.location})self.status onself.start_time datetime.now()# 模拟过滤器损耗if self.filter_life 0:self.filter_life - 0.01return Truedef turn_off(self) - bool:关闭模拟净化器if self.status on:print(f[INFO] 关闭净化器 {self.device_id})self.update_runtime()self.status offself.start_time Nonereturn Truedef get_status(self) - dict:获取净化器状态self.update_runtime()return {device_id: self.device_id,location: self.location,status: self.status,total_runtime_hours: round(self.total_runtime.total_seconds() / 3600, 2),energy_consumption_kwh: round(self.energy_consumption, 2),filter_life_percent: round(self.filter_life, 1),air_flow_level: self.air_flow_level}class SmartPurifierController(AirPurifierController):智能净化器控制器支持多档位和模式def __init__(self, device_id: str, location: str, power: float 0.05):初始化智能净化器参数:device_id: 设备IDlocation: 设备位置power: 额定功率千瓦super().__init__(device_id, location)self.power powerself.mode auto # auto, manual, sleepself.fan_speed 0 # 0-3self.air_quality_history []self.max_history_size 100def turn_on(self, speed: int 2) - bool:开启净化器参数:speed: 风扇速度 (1-3)返回:操作是否成功if self.status ! on:print(f[INFO] 开启智能净化器 {self.device_id}风速{speed}档)self.status onself.fan_speed max(1, min(3, speed))self.start_time datetime.now()return Truedef turn_off(self) - bool:关闭净化器if self.status on:print(f[INFO] 关闭智能净化器 {self.device_id})self.update_runtime()self.status offself.fan_speed 0self.start_time Nonereturn Truedef set_mode(self, mode: str) - bool:设置净化器模式参数:mode: 模式 (auto, manual, sleep)返回:操作是否成功valid_modes [auto, manual, sleep]if mode in valid_modes:self.mode modeprint(f[INFO] 设置净化器 {self.device_id} 为 {mode} 模式)if mode sleep:self.fan_speed 1elif mode auto and self.status on:# 根据历史数据调整风速self._adjust_speed_auto()return Truereturn Falsedef _adjust_speed_auto(self):根据空气质量自动调整风速if not self.air_quality_history:return# 获取最近的平均空气质量recent_data self.air_quality_history[-10:] # 最近10个数据点avg_formaldehyde sum(d[formaldehyde] for d in recent_data) / len(recent_data)avg_pm25 sum(d[pm25] for d in recent_data) / len(recent_data)# 根据污染物水平调整风速if avg_formaldehyde 0.15 or avg_pm25 150:self.fan_speed 3elif avg_formaldehyde 0.08 or avg_pm25 75:self.fan_speed 2else:self.fan_speed 1def update_air_quality(self, formaldehyde: float, pm25: float):更新空气质量数据用于自动模式参数:formaldehyde: 甲醛浓度pm25: PM2.5浓度data_point {timestamp: datetime.now(),formaldehyde: formaldehyde,pm25: pm25}self.air_quality_history.append(data_point)# 保持历史数据大小if len(self.air_quality_history) self.max_history_size:self.air_quality_history.pop(0)# 自动模式下调整风速if self.mode auto and self.status on:self._adjust_speed_auto()def get_status(self) - dict:获取净化器详细状态self.update_runtime()status {device_id: self.device_id,location: self.location,status: self.status,mode: self.mode,fan_speed: self.fan_speed,total_runtime_hours: round(self.total_runtime.total_seconds() / 3600, 2),energy_consumption_kwh: round(self.energy_consumption, 2),power_kw: self.power}if self.air_quality_history:latest self.air_quality_history[-1]status[latest_formaldehyde] latest[formaldehyde]status[latest_pm25] latest[pm25]return status5. 数据处理器 (utils/data_handler.py)数据处理模块处理、分析和保存空气质量数据import jsonimport csvfrom datetime import datetime, timedeltafrom typing import List, Dict, Optionalimport statisticsclass AirQualityDataHandler:空气质量数据处理类def __init__(self, data_file: str air_quality_data.json):初始化数据处理器参数:data_file: 数据存储文件路径self.data_file data_fileself.history_data []self._load_data()def _load_data(self):从文件加载历史数据try:with open(self.data_file, r, encodingutf-8) as f:self.history_data json.load(f)print(f[INFO] 从 {self.data_file} 加载了 {len(self.history_data)} 条历史数据)except (FileNotFoundError, json.JSONDecodeError):self.history_data []print(f[INFO] 未找到历史数据文件创建新的数据存储)def save_data(self, data: dict):保存单条数据参数:data: 空气质量数据字典# 添加时间戳if timestamp not in data:data[timestamp] datetime.now().isoformat()self.history_data.append(data)# 定期保存到文件if len(self.history_data) % 10 0:self._save_to_file()def _save_to_file(self):保存数据到文件try:with open(self.data_file, w, encodingutf-8) as f:json.dump(self.history_data, f, ensure_asciiFalse, indent2)except Exception as e:print(f[ERROR] 保存数据失败: {e})def get_recent_data(self, hours: int 24) - List[dict]:获取最近指定小时的数据参数:hours: 小时数返回:最近的数据列表cutoff_time datetime.now() - timedelta(hourshours)recent_data []for data_point in self.history_data:try:data_time datetime.fromisoformat(data_point[timestamp])if data_time cutoff_time:recent_data.append(data_point)except (KeyError, ValueError):continuereturn recent_datadef analyze_air_quality(self, hours: int 24) - dict:分析空气质量数据参数:hours: 分析的时间范围小时返回:分析结果字典recent_data s如果你觉得这个工具好用欢迎关注我