智能作物产量预测与优化系统
一、实际应用场景与痛点分析
应用场景
农业生产面临着气候变化、资源限制和市场波动等多重挑战。本系统面向现代农业管理者、农业合作社、大型农场和农业科技公司，通过智能预测作物产量和优化管理方案，提高农业生产效率和可持续性。
主要痛点
1. 预测不准确 - 传统经验法无法精确预测作物产量
2. 资源浪费 - 水、肥、农药等资源使用缺乏科学依据
3. 气候风险 - 无法有效应对极端天气和气候变化
4. 数据孤岛 - 土壤、气象、作物等多源数据难以整合
5. 决策困难 - 缺乏科学的种植决策支持
6. 成本控制 - 农业生产成本难以精确控制
7. 品质不一 - 农产品品质和产量稳定性差
8. 环境影响 - 农业生产对环境影响评估不足
二、核心逻辑与智能控制原理
系统架构
数据层 → 分析层 → 决策层 → 执行层
↓ ↓ ↓ ↓
多源数据 → 特征工程 → 智能预测 → 优化方案
监测网络 → 模型训练 → 模糊推理 → 精准管理
历史数据 → 深度学习 → 专家系统 → 自动控制
核心智能控制原理
1. 模糊控制 - 处理"土壤肥沃"、"气候适宜"等模糊概念
2. 神经网络 - 深度学习预测作物产量
3. 专家系统 - 基于农业知识的种植决策
4. 遗传算法 - 优化施肥和灌溉方案
5. 时间序列分析 - 作物生长过程建模
6. 多目标优化 - 平衡产量、成本、环境多个目标
三、代码实现
主程序：smart_crop_yield.py
#!/usr/bin/env python3
"""
智能作物产量预测与优化系统
基于智能控制原理的农业生产决策支持系统
"""
import json
import datetime
import time
import math
import numpy as np
from typing import Dict, List, Tuple, Optional, Any, Set, Deque
from dataclasses import dataclass, asdict, field
from enum import Enum
import matplotlib.pyplot as plt
from matplotlib.patches import Rectangle, Circle, Patch
import matplotlib.colors as mcolors
from collections import defaultdict, deque
import uuid
import random
import logging
from dataclasses_json import dataclass_json
import os
import pickle
from scipy import stats
from scipy.optimize import differential_evolution
import pandas as pd
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from sklearn.neural_network import MLPRegressor
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split
import warnings
warnings.filterwarnings('ignore')
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('crop_yield.log', encoding='utf-8'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
class CropType(Enum):
"""作物类型枚举"""
WHEAT = "小麦"
CORN = "玉米"
RICE = "水稻"
SOYBEAN = "大豆"
COTTON = "棉花"
POTATO = "马铃薯"
TOMATO = "番茄"
APPLE = "苹果"
GRAPE = "葡萄"
TEA = "茶叶"
class SoilType(Enum):
"""土壤类型枚举"""
SANDY = "沙土"
LOAMY = "壤土"
CLAY = "黏土"
SILTY = "粉土"
PEAT = "泥炭土"
SALINE = "盐碱土"
class FertilizerType(Enum):
"""肥料类型枚举"""
NITROGEN = "氮肥"
PHOSPHORUS = "磷肥"
POTASSIUM = "钾肥"
COMPOUND = "复合肥"
ORGANIC = "有机肥"
MICRO = "微肥"
class IrrigationType(Enum):
"""灌溉类型枚举"""
FLOOD = "漫灌"
SPRINKLER = "喷灌"
DRIP = "滴灌"
SUBSURFACE = "地下灌溉"
FURROW = "沟灌"
class GrowthStage(Enum):
"""生长阶段枚举"""
SEEDLING = "苗期"
VEGETATIVE = "营养生长期"
REPRODUCTIVE = "生殖生长期"
MATURITY = "成熟期"
HARVEST = "收获期"
@dataclass_json
@dataclass
class Location:
"""地理位置"""
id: str
name: str
latitude: float
longitude: float
altitude: float = 0.0 # 海拔，米
region: str = "" # 所属区域
climate_zone: str = "" # 气候带
def __hash__(self):
return hash(self.id)
@dataclass_json
@dataclass
class SoilData:
"""土壤数据"""
location_id: str
sampling_time: datetime.datetime
soil_type: SoilType
ph: float # pH值
organic_matter: float # 有机质含量 %
nitrogen: float # 氮含量 mg/kg
phosphorus: float # 磷含量 mg/kg
potassium: float # 钾含量 mg/kg
moisture: float # 土壤湿度 %
salinity: float # 盐分 %
bulk_density: float # 容重 g/cm³
porosity: float # 孔隙度 %
def get_soil_fertility_index(self) -> float:
"""计算土壤肥力指数"""
# 各项指标的权重
weights = {
'organic_matter': 0.25,
'nitrogen': 0.20,
'phosphorus': 0.20,
'potassium': 0.20,
'ph': 0.10,
'moisture': 0.05
}
# 标准化各项指标
normalized = {
'organic_matter': min(1.0, self.organic_matter / 5.0), # 有机质5%为优
'nitrogen': min(1.0, self.nitrogen / 200.0), # 氮200mg/kg为优
'phosphorus': min(1.0, self.phosphorus / 100.0), # 磷100mg/kg为优
'potassium': min(1.0, self.potassium / 300.0), # 钾300mg/kg为优
'ph': 1.0 - abs(self.ph - 6.8) / 3.0, # pH 6.5-7.0为最佳
'moisture': min(1.0, self.moisture / 30.0) # 湿度30%为优
}
# 计算加权和
fertility = sum(weights[key] * normalized[key] for key in weights)
return min(1.0, fertility)
def get_soil_quality_level(self) -> str:
"""获取土壤质量等级"""
fertility = self.get_soil_fertility_index()
if fertility >= 0.8:
return "优"
elif fertility >= 0.6:
return "良"
elif fertility >= 0.4:
return "中"
else:
return "差"
@dataclass_json
@dataclass
class WeatherData:
"""气象数据"""
location_id: str
time: datetime.datetime
temperature: float # 温度 ℃
humidity: float # 湿度 %
precipitation: float # 降水量 mm
wind_speed: float # 风速 m/s
solar_radiation: float # 太阳辐射 MJ/m²
sunshine_hours: float # 日照时数
pressure: float # 气压 hPa
evapotranspiration: float # 蒸散量 mm
def get_growing_degree_days(self, base_temp: float = 10.0) -> float:
"""计算生长度日"""
if self.temperature <= base_temp:
return 0.0
return self.temperature - base_temp
def get_weather_suitability(self, crop_type: CropType,
growth_stage: GrowthStage) -> float:
"""计算天气适宜度"""
# 不同作物不同生长阶段的适宜条件
conditions = {
CropType.WHEAT: {
GrowthStage.SEEDLING: {'temp': (10, 20), 'rain': (10, 30)},
GrowthStage.VEGETATIVE: {'temp': (15, 25), 'rain': (20, 40)},
GrowthStage.REPRODUCTIVE: {'temp': (18, 28), 'rain': (15, 35)},
GrowthStage.MATURITY: {'temp': (20, 30), 'rain': (5, 20)}
},
CropType.CORN: {
GrowthStage.SEEDLING: {'temp': (15, 25), 'rain': (15, 35)},
GrowthStage.VEGETATIVE: {'temp': (20, 30), 'rain': (25, 45)},
GrowthStage.REPRODUCTIVE: {'temp': (22, 32), 'rain': (20, 40)},
GrowthStage.MATURITY: {'temp': (20, 30), 'rain': (10, 25)}
},
CropType.RICE: {
GrowthStage.SEEDLING: {'temp': (20, 30), 'rain': (20, 40)},
GrowthStage.VEGETATIVE: {'temp': (25, 35), 'rain': (30, 50)},
GrowthStage.REPRODUCTIVE: {'temp': (25, 35), 'rain': (25, 45)},
GrowthStage.MATURITY: {'temp': (20, 30), 'rain': (10, 25)}
}
}
if crop_type not in conditions or growth_stage not in conditions[crop_type]:
return 0.5 # 默认值
cond = conditions[crop_type][growth_stage]
temp_range = cond['temp']
rain_range = cond['rain']
# 温度适宜度
if self.temperature < temp_range[0]:
temp_score = 1.0 - (temp_range[0] - self.temperature) / 10.0
elif self.temperature > temp_range[1]:
temp_score = 1.0 - (self.temperature - temp_range[1]) / 10.0
else:
temp_score = 1.0
# 降水适宜度
if self.precipitation < rain_range[0]:
rain_score = self.precipitation / rain_range[0]
elif self.precipitation > rain_range[1]:
rain_score = rain_range[1] / self.precipitation
else:
rain_score = 1.0
# 日照适宜度
sunshine_score = min(1.0, self.sunshine_hours / 8.0)
# 综合适宜度
suitability = 0.4 * temp_score + 0.3 * rain_score + 0.3 * sunshine_score
return max(0.0, min(1.0, suitability))
@dataclass_json
@dataclass
class FertilizationRecord:
"""施肥记录"""
record_id: str
field_id: str
application_time: datetime.datetime
fertilizer_type: FertilizerType
amount: float # 用量 kg/ha
method: str # 施用方法
depth: float = 0.0 # 施肥深度 cm
cost: float = 0.0 # 成本 元/ha
def get_nutrient_content(self) -> Dict[str, float]:
"""获取养分含量"""
# 不同肥料的养分含量（%）
nutrient_content = {
FertilizerType.NITROGEN: {'N': 46.0, 'P': 0.0, 'K': 0.0}, # 尿素
FertilizerType.PHOSPHORUS: {'N': 0.0, 'P': 46.0, 'K': 0.0}, # 过磷酸钙
FertilizerType.POTASSIUM: {'N': 0.0, 'P': 0.0, 'K': 60.0}, # 氯化钾
FertilizerType.COMPOUND: {'N': 15.0, 'P': 15.0, 'K': 15.0}, # 复合肥
FertilizerType.ORGANIC: {'N': 2.0, 'P': 1.5, 'K': 2.0}, # 有机肥
FertilizerType.MICRO: {'N': 0.0, 'P': 0.0, 'K': 0.0} # 微肥
}
return nutrient_content.get(self.fertilizer_type, {'N': 0, 'P': 0, 'K': 0})
@dataclass_json
@dataclass
class IrrigationRecord:
"""灌溉记录"""
record_id: str
field_id: str
irrigation_time: datetime.datetime
irrigation_type: IrrigationType
amount: float # 灌溉量 mm
duration: float # 灌溉时长 小时
efficiency: float = 0.8 # 灌溉效率
cost: float = 0.0 # 成本 元/ha
def get_effective_water(self) -> float:
"""获取有效水量"""
return self.amount * self.efficiency
@dataclass_json
@dataclass
class CropField:
"""农田地块"""
field_id: str
location_id: str
area: float # 面积 公顷
crop_type: CropType
variety: str # 品种
planting_date: datetime.datetime
expected_harvest_date: datetime.datetime
soil_data: Optional[SoilData] = None
historical_yield: float = 0.0 # 历史产量 kg/ha
target_yield: float = 0.0 # 目标产量 kg/ha
def get_growth_stage(self, current_date: datetime.datetime = None) -> GrowthStage:
"""获取当前生长阶段"""
if current_date is None:
current_date = datetime.datetime.now()
days_after_planting = (current_date - self.planting_date).days
# 不同作物的生长阶段划分
growth_periods = {
CropType.WHEAT: {
GrowthStage.SEEDLING: (0, 30),
GrowthStage.VEGETATIVE: (31, 100),
GrowthStage.REPRODUCTIVE: (101, 140),
GrowthStage.MATURITY: (141, 180)
},
CropType.CORN: {
GrowthStage.SEEDLING: (0, 20),
GrowthStage.VEGETATIVE: (21, 70),
GrowthStage.REPRODUCTIVE: (71, 110),
GrowthStage.MATURITY: (111, 140)
},
CropType.RICE: {
GrowthStage.SEEDLING: (0, 30),
GrowthStage.VEGETATIVE: (31, 80),
GrowthStage.REPRODUCTIVE: (81, 120),
GrowthStage.MATURITY: (121, 150)
}
}
if self.crop_type not in growth_periods:
return GrowthStage.VEGETATIVE
periods = growth_periods[self.crop_type]
for stage, (start, end) in periods.items():
if start <= days_after_planting <= end:
return stage
return GrowthStage.MATURITY
@dataclass_json
@dataclass
class ManagementPlan:
"""管理计划"""
plan_id: str
field_id: str
created_date: datetime.datetime
fertilization_plans: List[Dict] = field(default_factory=list)
irrigation_plans: List[Dict] = field(default_factory=list)
pest_control_plans: List[Dict] = field(default_factory=list)
expected_yield: float = 0.0
expected_cost: float = 0.0
expected_profit: float = 0.0
risk_assessment: Dict = field(default_factory=dict)
class FuzzyYieldPredictor:
"""
模糊产量预测器
处理模糊农业概念
"""
def __init__(self):
# 土壤肥力模糊集合
self.soil_fertility_sets = {
'poor': {'center': 0.2, 'range': 0.2}, # 贫瘠
'fair': {'center': 0.4, 'range': 0.2}, # 一般
'good': {'center': 0.6, 'range': 0.2}, # 良好
'excellent': {'center': 0.8, 'range': 0.2} # 优秀
}
# 天气适宜度模糊集合
self.weather_suitability_sets = {
'poor': {'center': 0.3, 'range': 0.2},
'fair': {'center': 0.5, 'range': 0.2},
'good': {'center': 0.7, 'range': 0.2},
'excellent': {'center': 0.9, 'range': 0.1}
}
# 管理水平模糊集合
self.management_level_sets = {
'low': {'center': 0.3, 'range': 0.2},
'medium': {'center': 0.6, 'range': 0.2},
'high': {'center': 0.8, 'range': 0.2}
}
# 产量潜力模糊集合
self.yield_potential_sets = {
'low': {'center': 0.3, 'range': 0.2},
'medium': {'center': 0.5, 'range': 0.2},
'high': {'center': 0.7, 'range': 0.2},
'very_high': {'center': 0.9, 'range': 0.1}
}
# 模糊规则库
self.rules = self._initialize_fuzzy_rules()
def _initialize_fuzzy_rules(self) -> List[Dict]:
"""初始化模糊规则"""
return [
# 规则1: 如果土壤肥力高且天气适宜度高，则产量潜力高
{
'antecedent': lambda soil, weather, mgmt: (
self._fuzzify(soil, self.soil_fertility_sets)['excellent'] > 0.7 and
self._fuzzify(weather, self.weather_suitability_sets)['excellent'] > 0.7
),
'consequent': 'very_high',
'weight': 0.9
},
# 规则2: 如果土壤肥力高且管理水平高，则产量潜力高
{
'antecedent': lambda soil, weather, mgmt: (
self._fuzzify(soil, self.soil_fertility_sets)['excellent'] > 0.7 and
self._fuzzify(mgmt, self.management_level_sets)['high'] > 0.7
),
'consequent': 'high',
'weight': 0.8
},
# 规则3: 如果天气适宜度低，则产量潜力低
{
'antecedent': lambda soil, weather, mgmt: (
self._fuzzify(weather, self.weather_suitability_sets)['poor'] > 0.7
),
'consequent': 'low',
'weight': 0.7
},
# 规则4: 如果土壤肥力低且管理水平低，则产量潜力低
{
'antecedent': lambda soil, weather, mgmt: (
self._fuzzify(soil, self.soil_fertility_sets)['poor'] > 0.7 and
self._fuzzify(mgmt, self.management_level_sets)['low'] > 0.7
),
'consequent': 'low',
'weight': 0.8
},
# 规则5: 如果天气适宜度高且管理水平高，则产量潜力高
{
'antecedent': lambda soil, weather, mgmt: (
self._fuzzify(weather, self.weather_suitability_sets)['good'] > 0.6 and
self._fuzzify(mgmt, self.management_level_sets)['high'] > 0.6
),
'consequent': 'high',
'weight': 0.7
}
]
def _fuzzify(self, value: float, fuzzy_sets: Dict) -> Dict[str, float]:
"""模糊化"""
memberships = {}
for level, params in fuzzy_sets.items():
center = params['center']
width = params['range']
if value <= center - width/2 or value >= center + width/2:
membership = 0
elif value <= center:
membership = (value - (center - width/2)) / (width/2)
else:
membership = ((center + width/2) - value) / (width/2)
memberships[level] = max(0, min(1, membership))
return memberships
def predict_yield_potential(self, soil_fertility: float,
weather_suitability: float,
management_level: float) -> Dict:
"""预测产量潜力"""
# 模糊化输入
soil_fuzzy = self._fuzzify(soil_fertility, self.soil_fertility_sets)
weather_fuzzy = self._fuzzify(weather_suitability, self.weather_suitability_sets)
mgmt_fuzzy = self._fuzzify(management_level, self.management_level_sets)
# 初始化输出模糊集合
output_memberships = defaultdict(float)
# 应用模糊规则
for rule in self.rules:
try:
if rule['antecedent'](soil_fertility, weather_suitability, management_level):
consequent = rule['consequent']
activation = min(
soil_fuzzy.get('excellent', 0) if 'excellent' in str(rule['antecedent']) else 1.0,
weather_fuzzy.get('excellent', 0) if 'excellent' in str(rule['antecedent']) else 1.0,
mgmt_fuzzy.get('high', 0) if 'high' in str(rule['antecedent']) else 1.0
)
output_memberships[consequent] = max(
output_memberships[consequent],
activation * rule['weight']
)
except Exception as e:
logger.warning(f"规则应用失败: {e}")
continue
# 如果没有规则激活，使用默认
if not any(output_memberships.values()):
# 计算平均模糊值
avg_fuzzy = (soil_fertility + weather_suitability + management_level) / 3
output_fuzzy = self._fuzzify(avg_fuzzy, self.yield_potential_sets)
for level, membership in output_fuzzy.items():
output_memberships[level] = membership
# 去模糊化
yield_potential = self._defuzzify(output_memberships, self.yield_potential_sets)
return {
'yield_potential': yield_potential,
'potential_level': self._get_potential_level(yield_potential),
'soil_contribution': soil_fertility,
'weather_contribution': weather_suitability,
'management_contribution': management_level,
'fuzzy_memberships': dict(output_memberships)
}
def _defuzzify(self, memberships: Dict[str, float], fuzzy_sets: Dict) -> float:
"""去模糊化"""
if not memberships:
return 0.5
# 使用重心法
numerator = 0
denominator = 0
for level, membership in memberships.items():
if membership > 0 and level in fuzzy_sets:
center = fuzzy_sets[level]['center']
numerator += center * membership
denominator += membership
if denominator == 0:
return 0.5
return numerator / denominator
def _get_potential_level(self, potential: float) -> str:
"""获取潜力等级"""
if potential >= 0.8:
return "很高"
elif potential >= 0.6:
return "高"
elif potential >= 0.4:
return "中等"
else:
return "低"
def estimate_yield(self, field: CropField, historical_weather: List[WeatherData],
management_level: float = 0.7) -> Dict:
"""估计产量"""
if not field.soil_data:
return {'error': '缺少土壤数据'}
# 计算土壤肥力
soil_fertility = field.soil_data.get_soil_fertility_index()
# 计算生长季天气适宜度
weather_suitability = self._calculate_season_suitability(
field, historical_weather
)
# 预测产量潜力
prediction = self.predict_yield_potential(
soil_fertility, weather_suitability, management_level
)
# 转换为实际产量（kg/ha）
# 不同作物的基准产量
base_yields = {
CropType.WHEAT: 6000, # kg/ha
CropType.CORN: 8000,
CropType.RICE: 7000,
CropType.SOYBEAN: 3000,
CropType.COTTON: 1500
}
base_yield = base_yields.get(field.crop_type, 5000)
# 计算预测产量
potential = prediction['yield_potential']
estimated_yield = base_yield * (0.5 + 0.8 * potential) # 调整系数
# 考虑历史产量
if field.historical_yield > 0:
# 历史产量占30%权重
estimated_yield = 0.7 * estimated_yield + 0.3 * field.historical_yield
return {
'estimated_yield_kg_per_ha': estimated_yield,
'estimated_total_yield_kg': estimated_yield * field.area,
'yield_potential': potential,
如果你觉得这个工具好用，欢迎关注我！
)
