拍照秒出病虫害诊断系统
项目概述
实际应用场景描述
在农业生产一线，病虫害的快速准确诊断是保障粮食安全的关键环节。
某水稻种植合作社拥有3000亩稻田，每逢春夏之交，稻瘟病、纹枯病、二化螟等病虫害频发。
传统模式下，农户发现病叶后需要：拍照→咨询农技员→田间取样→实验室鉴定→等待诊断报告，整个过程往往需要2-3天时间。
等到确诊时，病害往往已经大面积扩散，造成不可逆的损失。
本系统通过手机拍照即时上传，结合深度学习和图像识别技术，实现30秒内输出病虫害诊断书和精准用药建议。
引入痛点
1.
诊断时效差：从发现病征到获得诊断需要1-3天，错过最佳防治窗口期
2.
专家资源稀缺：基层农技员数量有限，无法及时响应大量农户咨询
3.
误诊率高：农户凭经验判断，常将生理性病害误认为侵染性病害
4.
用药盲目：缺乏精准诊断导致农药滥用，既增加成本又污染环境
5.
知识断层：年轻农户缺乏传统农作经验，老农又不会使用智能设备
6.
疫情追踪难：无法建立区域病虫害发生发展动态数据库
核心逻辑讲解
┌─────────────────────────────────────────────────────────────────┐
│
拍照秒出病虫害诊断系统
│
├─────────────────────────────────────────────────────────────────┤
│
输入层：病叶照片（手机拍摄/无人机航拍/监控摄像头）
│
│
传统特征（颜色矩+纹理+形态学）
│
│
│
└─────────────────────────────────────────────────────────────────┘
核心技术流程：
1.
图像预处理：自适应直方图均衡化(CLAHE)增强病斑特征，超分辨率重建提升小目标检测能力
2.
+
病害库构建：建立包含200+种常见病虫害的标准化图像库，每种病害500+样本
4.
+
知识图谱推理：基于病虫害症状关联规则，输出综合诊断结论和治疗方案
项目结构
plant_disease_diagnoser/
├──
README.md
traditional_feature_extractor.py
传统特征提取器
│
test_feature_extraction.py
│
├──
test_disease_matching.py
│
└──
(src/main.py)
"""
拍照秒出病虫害诊断系统
主程序入口
Author:
基于深度学习的植物病虫害智能诊断系统，支持拍照即诊、秒出结果
"""
import
os
import
添加项目根目录到路径
sys.path.insert(0,
str(Path(__file__).parent.parent))
from
src.utils.logger
src.preprocessing.image_enhancer
import
src.preprocessing.roi_extractor
import
src.feature_extraction.deep_feature_extractor
import
DeepFeatureExtractor
from
src.feature_extraction.traditional_feature_extractor
import
TraditionalFeatureExtractor
from
src.feature_extraction.feature_fusion
import
src.disease_matching.disease_database
import
src.disease_matching.similarity_matcher
import
src.disease_matching.decision_fuser
import
src.diagnosis.disease_classifier
import
src.diagnosis.confidence_evaluator
import
src.recommendation.pesticide_recommender
import
PesticideRecommender
from
src.report_generator.diagnosis_report
import
DiagnosisReportGenerator
from
src.report_generator.treatment_plan
import
TreatmentPlanGenerator
class
ProcessingStage(Enum):
"""处理阶段枚举"""
PREPROCESSING
=
"preprocessing"
FEATURE_EXTRACTION
=
"feature_extraction"
DISEASE_MATCHING
=
"disease_matching"
DIAGNOSIS
=
"diagnosis"
RECOMMENDATION
=
"recommendation"
REPORT_GENERATION
=
"report_generation"
COMPLETED
=
"completed"
@dataclass
class
SystemConfig:
"""系统配置数据类"""
input_dir:
str
"./data/sample_images"
output_dir:
str
"./data/output"
model_dir:
str
"./data/models"
disease_db_path:
str
"./database/disease_knowledge.db"
pesticide_db_path:
str
"./database/pesticide_knowledge.db"
enable_preprocessing:
bool
True
enable_deep_features:
bool
True
enable_traditional_features:
bool
"INFO"
@classmethod
def
from_yaml(cls,
'SystemConfig':
"""从YAML文件加载配置"""
with
open(config_path,
cls(**config_dict.get('system',
{}))
@classmethod
def
default(cls)
'SystemConfig':
"""返回默认配置"""
return
cls()
@dataclass
class
DiagnosisResult:
"""诊断结果数据类"""
stage:
bool
message:
field(default_factory=dict)
metrics:
Dict[str,
field(default_factory=dict)
timestamp:
datetime
field(default_factory=datetime.now)
def
to_dict(self)
Dict:
"""转换为字典格式"""
return
self.stage.value,
'success':
self.success,
'message':
self.message,
'data':
self.data,
'metrics':
self.metrics,
'timestamp':
self.timestamp.isoformat()
}
class
PlantDiseaseDiagnoser:
"""
植物病虫害智能诊断系统核心类
该系统实现了从病叶图像到诊断报告的全自动处理流水线，
集成了图像预处理、特征提取、病害匹配、智能诊断和处方推荐功能。
Attributes:
config:
系统配置对象
logger:
模型管理器
"""
def
__init__(self,
None):
"""
初始化诊断系统
Args:
config:
系统配置，如果为None则使用默认配置
"""
self.config
=
SystemConfig.default()
self.logger
=
setup_logger(
name="PlantDiseaseDiagnoser",
level=self.config.log_level
)
self.file_handler
=
FileHandler(self.config.output_dir)
self.current_stage
=
ProcessingStage.PREPROCESSING
self.results:
DiagnosisResult]
ModelManager(self.config.model_dir)
#
初始化各处理模块
self._initialize_modules()
self.logger.info("="
60)
self.logger.info("植物病虫害智能诊断系统
v2.0.0
初始化完成")
self.logger.info("="
60)
def
None:
"""初始化所有处理模块"""
self.logger.info("正在初始化处理模块...")
#
预处理模块
if
self.config.enable_preprocessing:
self.image_enhancer
=
ImageEnhancer()
self.roi_extractor
=
ROIExtractor()
self.logger.info("✓
特征提取模块
self.deep_feature_extractor
=
None
self.traditional_feature_extractor
=
self.config.enable_deep_features:
self.deep_feature_extractor
=
DeepFeatureExtractor(
model_manager=self.model_manager,
use_gpu=self.config.use_gpu
)
self.logger.info("✓
self.config.enable_traditional_features:
self.traditional_feature_extractor
=
TraditionalFeatureExtractor()
self.logger.info("✓
self.traditional_feature_extractor:
self.feature_fusion
=
FeatureFusion(
use_deep_features=self.config.enable_deep_features,
use_traditional_features=self.config.enable_traditional_features
)
self.logger.info("✓
病害匹配模块
self.disease_database
=
DiseaseDatabase(self.config.disease_db_path)
self.similarity_matcher
=
SimilarityMatcher(
disease_database=self.disease_database,
top_k=self.config.top_k_matches
)
self.decision_fuser
=
DecisionFuser(
confidence_threshold=self.config.confidence_threshold
)
self.logger.info("✓
诊断模块
self.disease_classifier
=
DiseaseClassifier(
disease_database=self.disease_database
)
self.confidence_evaluator
=
ConfidenceEvaluator()
self.logger.info("✓
推荐模块
self.pesticide_recommender
=
PesticideRecommender(
pesticide_db_path=self.config.pesticide_db_path
)
self.logger.info("✓
报告生成模块
self.diagnosis_report_generator
=
DiagnosisReportGenerator()
self.treatment_plan_generator
=
TreatmentPlanGenerator(
pesticide_recommender=self.pesticide_recommender
)
self.logger.info("✓
报告生成模块初始化完成")
self.logger.info("所有处理模块初始化完成")
def
diagnose(
self,
image_path:
str
"rice",
progress_callback=None
)
->
DiagnosisResult:
"""
执行完整的诊断流水线
Args:
image_path:
作物类型
(rice/wheat/corn/cotton/etc.)
progress_callback:
进度回调函数
Returns:
DiagnosisResult:
最终诊断结果
"""
try:
start_time
=
time.time()
self.logger.info("="
60)
self.logger.info(f"开始诊断流程，图像:
{image_path},
{crop_type}")
self.logger.info("="
60)
#
os.path.exists(image_path):
raise
FileNotFoundError(f"图像文件不存在:
阶段1:
self._run_preprocessing(image_path,
preprocessed_image
DiagnosisResult(
stage=ProcessingStage.PREPROCESSING,
success=False,
message="图像预处理失败"
)
#
阶段2:
self._run_feature_extraction(preprocessed_image,
feature_vector
DiagnosisResult(
stage=ProcessingStage.FEATURE_EXTRACTION,
success=False,
message="特征提取失败"
)
#
阶段3:
self._run_disease_matching(feature_vector,
crop_type,
DiagnosisResult(
stage=ProcessingStage.DISEASE_MATCHING,
success=False,
message="病害匹配失败"
)
#
阶段4:
self._run_diagnosis(matching_results,
diagnosis
DiagnosisResult(
stage=ProcessingStage.DIAGNOSIS,
success=False,
message="智能诊断失败"
)
#
阶段5:
self._run_recommendation(diagnosis,
crop_type,
self._run_report_generation(
diagnosis,
recommendations,
preprocessed_image,
progress_callback
)
#
=
start_time
final_result.metrics['total_processing_time']
=
total_time
final_result.message
=
{total_time:.2f}秒"
self.logger.info("="
60)
self.logger.info(f"诊断流程完成!
结果:
{final_result.data.get('disease_name',
'未知病害')}")
self.logger.info(f"总耗时:
{total_time:.2f}秒")
self.logger.info("="
60)
return
e:
self.logger.error(f"诊断流程执行失败:
{str(e)}",
exc_info=True)
return
DiagnosisResult(
stage=self.current_stage,
success=False,
message=f"诊断失败:
{str(e)}"
)
def
_run_preprocessing(
self,
image_path:
str,
progress_callback=None
)
->
Optional[np.ndarray]:
"""执行图像预处理阶段"""
self.current_stage
=
ProcessingStage.PREPROCESSING
self.logger.info("[阶段1]
开始图像预处理...")
try:
if
progress_callback:
progress_callback(self.current_stage,
10,
self.file_handler.read_image(image_path)
self.logger.debug(f"图像尺寸:
progress_callback:
progress_callback(self.current_stage,
30,
self.image_enhancer.enhance(image)
if
progress_callback:
progress_callback(self.current_stage,
50,
提取ROI（病斑区域）
roi_image,
roi_mask
self.roi_extractor.extract(enhanced_image)
if
progress_callback:
progress_callback(self.current_stage,
80,
self.roi_extractor.calibrate_color(roi_image)
if
progress_callback:
progress_callback(self.current_stage,
100,
DiagnosisResult(
stage=ProcessingStage.PREPROCESSING,
success=True,
message="图像预处理完成",
data={
'original_image':
image,
'enhanced_image':
enhanced_image,
'roi_image':
roi_image,
'calibrated_image':
calibrated_image,
'roi_mask':
roi_mask
},
metrics={
'original_size':
image.shape[:2],
'roi_size':
roi_image.shape[:2],
'processing_time':
0.0
}
)
self.results[self.current_stage]
=
result
self.logger.info(f"[阶段1]
{roi_image.shape[:2]}")
return
Exception
{str(e)}"
self.logger.error(error_msg,
exc_info=True)
return
None
def
_run_feature_extraction(
self,
image:
np.ndarray,
progress_callback=None
)
->
Optional[np.ndarray]:
"""执行特征提取阶段"""
self.current_stage
=
ProcessingStage.FEATURE_EXTRACTION
self.logger.info("[阶段2]
开始特征提取...")
try:
feature_parts
=
progress_callback:
progress_callback(self.current_stage,
10,
self.deep_feature_extractor:
if
progress_callback:
progress_callback(self.current_stage,
30,
"提取深度特征...")
deep_features
=
self.deep_feature_extractor.extract(image)
feature_parts.append(deep_features)
self.logger.debug(f"深度特征维度:
{deep_features.shape}")
#
传统特征提取
if
self.traditional_feature_extractor:
if
progress_callback:
progress_callback(self.current_stage,
60,
"提取传统特征...")
traditional_features
=
self.traditional_feature_extractor.extract(image)
feature_parts.append(traditional_features)
self.logger.debug(f"传统特征维度:
{traditional_features.shape}")
if
not
RuntimeError("没有可用的特征提取器")
if
progress_callback:
progress_callback(self.current_stage,
85,
self.feature_fusion.fuse(feature_parts)
if
progress_callback:
progress_callback(self.current_stage,
100,
DiagnosisResult(
stage=ProcessingStage.FEATURE_EXTRACTION,
success=True,
message="特征提取完成",
data={
'deep_features':
feature_parts[0]
None,
'traditional_features':
feature_parts[1]
None,
'fused_features':
fused_features
},
metrics={
'feature_dimension':
fused_features.shape[0],
'deep_feature_dim':
len(feature_parts)
0,
'traditional_feature_dim':
len(feature_parts)
0
}
)
self.results[self.current_stage]
=
result
self.logger.info(f"[阶段2]
{fused_features.shape[0]}")
return
Exception
{str(e)}"
self.logger.error(error_msg,
exc_info=True)
return
None
def
_run_disease_matching(
self,
features:
str,
progress_callback=None
)
->
Optional[List[Dict]]:
"""执行病害匹配阶段"""
self.current_stage
=
ProcessingStage.DISEASE_MATCHING
self.logger.info("[阶段3]
开始病害匹配...")
try:
if
progress_callback:
progress_callback(self.current_stage,
20,
"检索病害数据库...")
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！
