大棚AI全自动环境控制，输入CO2，温，湿，光照，处理，多因子联动控制，输出，通风/遮阳/喷淋指令。
大棚AI全自动环境控制系统
项目概述
实际应用场景描述
在江苏宿迁某现代农业产业园，种植大户李总经营着50个智能温室大棚，主要种植反季节番茄和黄瓜。每个大棚面积约800平方米，种植着2000株作物。传统的大棚管理完全依赖人工经验：早上6点工人凭感觉开天窗通风，上午10点看太阳强度决定是否拉遮阳网，下午3点根据温度计读数启动喷淋降温。这种粗放式管理导致环境波动极大：夏季中午温度常超过40℃，造成番茄日灼病；冬季夜间湿度高达95%，引发灰霉病爆发；CO₂浓度在上午9-11点严重不足，光合作用效率低下。
更糟糕的是，人工管理存在严重的滞后性。当工人发现叶片萎蔫时，作物已经遭受了2-3小时的高温胁迫。2023年，李总的基地因环境控制不当导致的病害损失达80万元，其中仅灰霉病一项就损失30万元。同时，人工成本不断攀升，每个大棚需要0.5个工时/天，50个大棚年人工成本达45万元。
本系统通过物联网传感器实时采集温、湿、光、CO₂等多维环境数据，运用机器学习模型预测环境变化趋势，基于作物生理模型和专家规则引擎，自动生成通风、遮阳、喷淋等精准控制指令，实现24小时无人值守的智能环境调控。
引入痛点
1. 环境波动剧烈：传统管理下温湿度日波动±8℃/±15%，远超作物适宜范围，导致生理紊乱
2. 响应严重滞后：人工巡检间隔2-4小时，关键环境胁迫发生时无法及时干预，损失不可逆
3. 经验依赖性强：管理效果取决于工人经验水平，新员工上岗需3个月培训期，人才流失风险大
4. 能源浪费严重：盲目开启设备导致能耗居高不下，夏季制冷能耗占运营成本40%
5. 病害频发难控：高湿环境持续6小时以上即诱发真菌病害，传统方法难以精确预警和防控
6. CO₂管理缺失：白天光合作用旺盛期CO₂浓度低于400ppm，严重影响产量和品质
7. 设备调度混乱：多台设备独立运行缺乏协调，风机与湿帘同时开启造成能源冲突
8. 数据孤岛严重：环境数据分散在各棚独立记录，无法进行大数据分析和优化决策
9. 应急处理迟缓：极端天气（暴雨、高温、寒潮）发生时响应延迟，损失扩大
10. 标准化困难：不同工人管理方式差异大，难以实现园区统一的环境控制标准
核心逻辑讲解
┌─────────────────────────────────────────────────────────────────────────────┐
│ 大棚AI全自动环境控制系统架构 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ 输入层： │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ 温度 │ │ 湿度 │ │ 光照 │ │ CO₂ │ │ 土壤 │ │
│ │ 传感器 │ │ 传感器 │ │ 传感器 │ │ 传感器 │ │ 传感器 │ │
│ └────┬────┘ └────┬────┘ └────┬────┘ └────┬────┘ └────┬────┘ │
│ │ │ │ │ │ │
│ └───────────┴─────┬─────┴───────────┴───────────┘ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 多源数据融合与预处理层 │ │
│ │ • 异常值检测与滤波 (卡尔曼滤波+滑动平均) │ │
│ │ • 传感器故障诊断与数据补偿 │ │
│ │ • 时空数据对齐 (多棚数据同步) │ │
│ └─────────────────────────┬───────────────────────────────────────────┘ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ AI环境预测与决策层 │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ LSTM预测 │ │ 作物生理 │ │ 专家规则 │ │ 强化学习 │ │ │
│ │ │ 模型 │ │ 模型 │ │ 引擎 │ │ 优化器 │ │ │
│ │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ │
│ │ └──────────────┼──────────────┼──────────────┘ │ │
│ │ ▼ ▼ │ │
│ │ ┌─────────────────────────────────────┐ │ │
│ │ │ 多因子联动决策中心 │ │ │
│ │ │ • 温湿度协同控制 │ │ │
│ │ │ • 光热耦合优化 │ │ │
│ │ │ • CO₂-光照-通风联动 │ │ │
│ │ │ • 能耗-效益平衡决策 │ │ │
│ │ └─────────────────┬───────────────────┘ │ │
│ └──────────────────────────────┼──────────────────────────────────────┘ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 执行控制层 │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ 通风控制 │ │ 遮阳控制 │ │ 喷淋控制 │ │ CO₂补充 │ │ │
│ │ │ 执行器 │ │ 执行器 │ │ 执行器 │ │ 执行器 │ │ │
│ │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ │
│ │ └──────────────┼──────────────┼──────────────┘ │ │
│ │ ▼ ▼ │ │
│ │ ┌─────────────────────────────────────┐ │ │
│ │ │ 设备协调调度器 │ │ │
│ │ │ • 设备冲突检测与解决 │ │ │
│ │ │ • 执行时序优化 │ │ │
│ │ │ • 故障切换与容错处理 │ │ │
│ │ └─────────────────┬───────────────────┘ │ │
│ └──────────────────────────────┼──────────────────────────────────────┘ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ 输出反馈层 │ │
│ │ • 控制指令JSON/Modbus/OPC-UA输出 │ │
│ │ • 设备运行状态实时监控 │ │
│ │ • 环境数据历史存储与分析 │ │
│ │ • 异常告警与事件日志 │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
核心技术流程：
1. 多源数据采集与融合：集成DHT22温湿度传感器、BH1750光照传感器、MH-Z19B CO₂传感器、TDR土壤水分传感器，通过卡尔曼滤波消除噪声，多传感器数据融合提高测量精度
2. 环境趋势预测：基于LSTM神经网络建立时间序列预测模型，预测未来2小时内温湿度、光照强度的变化趋势，提前识别环境胁迫风险
3. 作物生理建模：根据不同作物的生长阶段，建立温度-湿度-光照-CO₂四因子耦合的作物生理响应模型，计算当前环境下的作物胁迫指数
4. 多因子联动决策：运用模糊PID控制和专家规则引擎，综合考虑温度、湿度、光照、CO₂、能耗、作物生长阶段等多维因素，生成最优控制策略
5. 设备协调控制：基于设备特性矩阵和运行约束，协调风机、湿帘、遮阳网、喷淋系统、CO₂发生器的启停时序，避免设备冲突和能源浪费
6. 闭环反馈优化：实时监测控制效果，通过强化学习不断优化控制参数，形成自适应的智能控制体系
项目结构
greenhouse_ai_control_system/
├── README.md # 项目说明文档
├── requirements.txt # 依赖包列表
├── setup.py # 安装脚本
├── docker-compose.yml # Docker容器编排
├── .env.example # 环境变量模板
├── config/
│ ├── system_config.yaml # 系统主配置
│ ├── crop_profiles.yaml # 作物生理参数配置
│ ├── device_config.yaml # 设备参数配置
│ ├── control_rules.yaml # 控制规则配置
│ ├── alert_thresholds.yaml # 告警阈值配置
│ └── model_config.yaml # AI模型配置
├── src/
│ ├── __init__.py
│ ├── main.py # 程序入口
│ ├── greenhouse_app.py # 温室应用主类
│ │
│ ├── data_acquisition/ # 数据采集层
│ │ ├── __init__.py
│ │ ├── sensor_manager.py # 传感器管理器
│ │ ├── data_validator.py # 数据验证器
│ │ ├── kalman_filter.py # 卡尔曼滤波器
│ │ ├── fault_detector.py # 故障检测器
│ │ └── data_compensator.py # 数据补偿器
│ │
│ ├── preprocessing/ # 预处理层
│ │ ├── __init__.py
│ │ ├── noise_reducer.py # 噪声去除
│ │ ├── outlier_detector.py # 异常值检测
│ │ ├── data_normalizer.py # 数据标准化
│ │ ├── temporal_aligner.py # 时间对齐
│ │ └── feature_engineer.py # 特征工程
│ │
│ ├── prediction/ # 预测层
│ │ ├── __init__.py
│ │ ├── lstm_predictor.py # LSTM预测器
│ │ ├── arima_predictor.py # ARIMA预测器
│ │ ├── ensemble_predictor.py # 集成预测器
│ │ └── trend_analyzer.py # 趋势分析器
│ │
│ ├── decision/ # 决策层
│ │ ├── __init__.py
│ │ ├── crop_physiology_model.py # 作物生理模型
│ │ ├── rule_engine.py # 专家规则引擎
│ │ ├── fuzzy_controller.py # 模糊控制器
│ │ ├── optimization_engine.py # 优化引擎
│ │ └── decision_fusion.py # 决策融合器
│ │
│ ├── control/ # 控制层
│ │ ├── __init__.py
│ │ ├── ventilation_controller.py # 通风控制器
│ │ ├── shading_controller.py # 遮阳控制器
│ │ ├── misting_controller.py # 喷淋控制器
│ │ ├── co2_controller.py # CO₂控制器
│ │ ├── device_coordinator.py # 设备协调器
│ │ └── safety_monitor.py # 安全监控器
│ │
│ ├── execution/ # 执行层
│ │ ├── __init__.py
│ │ ├── command_generator.py # 指令生成器
│ │ ├── protocol_handler.py # 协议处理器
│ │ ├── modbus_client.py # Modbus客户端
│ │ ├── opcua_client.py # OPC-UA客户端
│ │ └── mqtt_publisher.py # MQTT发布器
│ │
│ ├── monitoring/ # 监控层
│ │ ├── __init__.py
│ │ ├── performance_monitor.py # 性能监控器
│ │ ├── energy_tracker.py # 能耗追踪器
│ │ ├── alert_manager.py # 告警管理器
│ │ └── dashboard_service.py # 仪表板服务
│ │
│ ├── storage/ # 存储层
│ │ ├── __init__.py
│ │ ├── timeseries_db.py # 时序数据库接口
│ │ ├── event_logger.py # 事件日志记录器
│ │ └── backup_manager.py # 备份管理器
│ │
│ └── utils/ # 工具层
│ ├── __init__.py
│ ├── constants.py # 常量定义
│ ├── helpers.py # 辅助函数
│ ├── logger.py # 日志工具
│ ├── validators.py # 验证工具
│ └── exceptions.py # 自定义异常
│
├── models/ # AI模型文件
│ ├── lstm_temperature/ # 温度预测模型
│ │ ├── model.h5
│ │ ├── scaler.pkl
│ │ └── metadata.json
│ ├── lstm_humidity/ # 湿度预测模型
│ ├── lstm_light/ # 光照预测模型
│ ├── physiology_models/ # 作物生理模型
│ │ ├── tomato_growth_stages.json
│ │ ├── cucumber_growth_stages.json
│ │ └── pepper_growth_stages.json
│ └── optimization_models/ # 优化模型
│ ├── energy_optimizer.pkl
│ └── resource_allocator.pkl
│
├── data/ # 数据目录
│ ├── raw/ # 原始数据
│ │ ├── sensor_logs/
│ │ └── device_logs/
│ ├── processed/ # 处理后数据
│ ├── training/ # 训练数据
│ └── backups/ # 数据备份
│
├── tests/ # 测试目录
│ ├── __init__.py
│ ├── unit/ # 单元测试
│ │ ├── test_sensor_manager.py
│ │ ├── test_prediction.py
│ │ ├── test_decision.py
│ │ └── test_control.py
│ ├── integration/ # 集成测试
│ └── performance/ # 性能测试
│
├── scripts/ # 脚本目录
│ ├── deploy.sh # 部署脚本
│ ├── backup.sh # 备份脚本
│ ├── calibrate_sensors.py # 传感器校准
│ └── train_models.py # 模型训练
│
├── docs/ # 文档目录
│ ├── api_documentation.md # API文档
│ ├── user_manual.md # 用户手册
│ ├── installation_guide.md # 安装指南
│ ├── troubleshooting.md # 故障排除
│ └── architecture_design.md # 架构设计
│
├── examples/ # 示例目录
│ ├── basic_usage.py # 基础使用示例
│ ├── custom_crop_profile.py # 自定义作物配置
│ └── integration_example.py # 集成示例
│
└── logs/ # 日志目录
├── application.log
├── error.log
└── audit.log
核心代码实现
1. 主程序入口 (src/main.py)
"""
大棚AI全自动环境控制系统
Author: Full Stack Developer with 5 years experience
Version: 3.0.0
Description: 基于物联网、机器学习和自动控制技术的智能温室环境管理系统，
实现温、湿、光、CO₂多因子联动的智能调控
"""
import os
import sys
import time
import signal
import logging
import threading
import asyncio
from pathlib import Path
from typing import Dict, List, Optional, Any, Callable, Tuple
from dataclasses import dataclass, field
from enum import Enum, auto
from datetime import datetime, timedelta
import json
import yaml
import numpy as np
import pandas as pd
from abc import ABC, abstractmethod
import queue
# 添加项目根目录到路径
sys.path.insert(0, str(Path(__file__).parent))
from src.utils.logger import setup_logger
from src.utils.exceptions import GreenhouseError, SensorError, ControlError
from src.utils.constants import (
SENSOR_UPDATE_INTERVAL,
CONTROL_CYCLE_DURATION,
PREDICTION_HORIZON,
DEFAULT_CROP_TYPE,
DEVICE_STATUS
)
from src.greenhouse_app import GreenhouseApplication
class SystemState(Enum):
"""系统状态枚举"""
INITIALIZING = auto()
RUNNING = auto()
PAUSED = auto()
STOPPING = auto()
STOPPED = auto()
ERROR = auto()
MAINTENANCE = auto()
class ControlMode(Enum):
"""控制模式枚举"""
AUTO = "automatic" # 全自动模式
SEMI_AUTO = "semi_automatic" # 半自动模式（人工确认）
MANUAL = "manual" # 手动模式
EMERGENCY = "emergency" # 紧急模式
SCHEDULED = "scheduled" # 计划模式
@dataclass
class SystemConfig:
"""系统配置数据类"""
# 基本配置
system_name: str = "Greenhouse AI Control System"
version: str = "3.0.0"
greenhouse_id: str = "GH001"
location: str = "Jiangsu, China"
# 数据采集配置
sensor_update_interval: int = 30 # 秒
data_buffer_size: int = 1000
enable_sensor_fault_detection: bool = True
enable_data_compensation: bool = True
# 预测配置
prediction_horizon: int = 120 # 分钟
enable_lstm_prediction: bool = True
enable_ensemble_prediction: bool = True
prediction_confidence_threshold: float = 0.7
# 控制配置
control_cycle_duration: int = 60 # 秒
control_mode: ControlMode = ControlMode.AUTO
enable_multi_factor_control: bool = True
enable_energy_optimization: bool = True
enable_safety_checks: bool = True
# 作物配置
crop_type: str = DEFAULT_CROP_TYPE
growth_stage: str = "vegetative"
planting_date: Optional[str] = None
# 设备配置
devices: Dict[str, Dict] = field(default_factory=lambda: {
'ventilation': {'enabled': True, 'priority': 1},
'shading': {'enabled': True, 'priority': 2},
'mistring': {'enabled': True, 'priority': 3},
'co2_injection': {'enabled': True, 'priority': 4}
})
# 告警配置
enable_alerts: bool = True
alert_channels: List[str] = field(default_factory=lambda: ['log', 'email'])
critical_alert_threshold: int = 3
# 存储配置
enable_data_storage: bool = True
storage_retention_days: int = 365
enable_backup: bool = True
backup_interval_hours: int = 24
# 网络配置
enable_remote_access: bool = True
api_port: int = 8080
mqtt_enabled: bool = True
mqtt_broker: str = "localhost"
mqtt_port: int = 1883
# 调试配置
debug_mode: bool = False
log_level: str = "INFO"
enable_performance_monitoring: bool = True
@classmethod
def from_yaml(cls, config_path: str) -> 'SystemConfig':
"""从YAML文件加载配置"""
with open(config_path, 'r', encoding='utf-8') as f:
config_dict = yaml.safe_load(f)
return cls(**config_dict.get('system', {}))
@classmethod
def default(cls) -> 'SystemConfig':
"""返回默认配置"""
return cls()
@dataclass
class EnvironmentalData:
"""环境数据数据类"""
timestamp: datetime
temperature: float # 温度 (°C)
humidity: float # 相对湿度 (%)
light_intensity: float # 光照强度 (lux)
co2_concentration: float # CO₂浓度 (ppm)
soil_moisture: float # 土壤湿度 (%)
soil_temperature: float # 土壤温度 (°C)
air_velocity: float # 空气流速 (m/s)
pressure: float # 大气压力 (hPa)
leaf_wetness: float # 叶面湿润度 (%)
vapor_pressure_deficit: float # 饱和水汽压差 (kPa)
# 数据质量指标
data_quality_score: float = 1.0
sensor_status: Dict[str, str] = field(default_factory=dict)
anomaly_flags: List[str] = field(default_factory=list)
def to_dict(self) -> Dict[str, Any]:
"""转换为字典格式"""
return {
'timestamp': self.timestamp.isoformat(),
'temperature': round(self.temperature, 2),
'humidity': round(self.humidity, 2),
'light_intensity': round(self.light_intensity, 2),
'co2_concentration': round(self.co2_concentration, 2),
'soil_moisture': round(self.soil_moisture, 2),
'soil_temperature': round(self.soil_temperature, 2),
'air_velocity': round(self.air_velocity, 2),
'pressure': round(self.pressure, 2),
'leaf_wetness': round(self.leaf_wetness, 2),
'vapor_pressure_deficit': round(self.vapor_pressure_deficit, 2),
'data_quality_score': round(self.data_quality_score, 3),
'sensor_status': self.sensor_status,
'anomaly_flags': self.anomaly_flags
}
def validate(self) -> Tuple[bool, List[str]]:
"""验证数据有效性"""
errors = []
if not -10 <= self.temperature <= 60:
errors.append(f"温度超出合理范围: {self.temperature}°C")
if not 0 <= self.humidity <= 100:
errors.append(f"湿度超出合理范围: {self.humidity}%")
if not 0 <= self.light_intensity <= 200000:
errors.append(f"光照强度超出合理范围: {self.light_intensity} lux")
if not 200 <= self.co2_concentration <= 2000:
errors.append(f"CO₂浓度超出合理范围: {self.co2_concentration} ppm")
if not 0 <= self.soil_moisture <= 100:
errors.append(f"土壤湿度超出合理范围: {self.soil_moisture}%")
is_valid = len(errors) == 0
return is_valid, errors
@dataclass
class ControlCommand:
"""控制指令数据类"""
command_id: str
timestamp: datetime
command_type: str # ventilation/shading/misting/co2_injection
action: str # on/off/set_value/adjust
target_value: Optional[float] = None
duration: Optional[int] = None # 秒
priority: int = 1
reason: str = ""
confidence: float = 1.0
energy_impact: float = 0.0 # 预估能耗变化 (kWh)
expected_effect: Dict[str, float] = field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
"""转换为字典格式"""
return {
'command_id': self.command_id,
'timestamp': self.timestamp.isoformat(),
'command_type': self.command_type,
'action': self.action,
'target_value': self.target_value,
'duration': self.duration,
'priority': self.priority,
'reason': self.reason,
'confidence': round(self.confidence, 3),
'energy_impact': round(self.energy_impact, 3),
'expected_effect': {k: round(v, 3) for k, v in self.expected_effect.items()}
}
def to_mqtt_payload(self) -> str:
"""转换为MQTT消息负载"""
payload = {
'cmd_id': self.command_id,
'type': self.command_type,
'action': self.action,
'value': self.target_value,
'ts': self.timestamp.isoformat()
}
return json.dumps(payload)
@dataclass
class ControlResult:
"""控制结果数据类"""
command: ControlCommand
success: bool
actual_execution_time: datetime
device_response: Dict[str, Any] = field(default_factory=dict)
error_message: Optional[str] = None
actual_effect: Dict[str, float] = field(default_factory=dict)
energy_consumed: float = 0.0
def to_dict(self) -> Dict[str, Any]:
"""转换为字典格式"""
return {
'command': self.command.to_dict(),
'success': self.success,
'actual_execution_time': self.actual_execution_time.isoformat(),
'device_response': self.device_response,
'error_message': self.error_message,
'actual_effect': self.actual_effect,
'energy_consumed': round(self.energy_consumed, 3)
}
class DataAcquisitionModule(ABC):
"""数据采集模块抽象基类"""
@abstractmethod
async def initialize(self) -> bool:
"""初始化模块"""
pass
@abstractmethod
async def read_sensors(self) -> EnvironmentalData:
"""读取传感器数据""
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！