无人机全自动巡田分析，输入航拍图，处理，拼接+长势分析，输出，农田健康地图。
无人机全自动巡田分析系统
项目概述
实际应用场景描述
随着智慧农业的发展，传统人工巡田方式面临诸多挑战。某大型农场拥有5000亩小麦种植区，传统巡田需要4名农技人员花费3天时间才能完成全面检查，不仅效率低下，还容易遗漏病虫害早期征兆。本系统通过无人机自动采集航拍图像，结合计算机视觉和深度学习技术，实现农田健康状况的智能分析与可视化。
引入痛点
1. 效率低下：人工巡田耗时耗力，无法及时发现作物异常
2. 主观性强：依赖个人经验判断，缺乏客观标准
3. 覆盖不全：地形复杂区域难以到达，存在监测盲区
4. 数据分散：影像资料难以整合分析，无法形成全局视图
5. 预警滞后：病虫害发现时已扩散，错过最佳防治期
核心逻辑讲解
┌─────────────────────────────────────────────────────────────────┐
│ 无人机全自动巡田分析系统 │
├─────────────────────────────────────────────────────────────────┤
│ 输入层：无人机航拍图像（RGB + 多光谱） │
│ ↓ │
│ 预处理层：图像去噪 → 几何校正 → 辐射定标 │
│ ↓ │
│ 拼接层：特征提取(SIFT/ORB) → 特征匹配 → 图像配准 → 无缝融合 │
│ ↓ │
│ 分析层：NDVI计算 → 长势评估 → 病虫害识别 → 杂草检测 │
│ ↓ │
│ 输出层：农田健康地图（热力图 + 统计报表 + 预警信息） │
└─────────────────────────────────────────────────────────────────┘
核心算法流程：
1. 图像预处理：使用OpenCV进行去畸变、白平衡、对比度增强
2. 图像拼接：基于特征点的全景图构建，采用RANSAC剔除误匹配
3. 植被指数计算：NDVI = (NIR - Red) / (NIR + Red)
4. 长势分析：结合颜色空间转换和纹理特征，使用K-means聚类
5. 健康评估：多维度评分模型，综合NDVI、叶面积指数、病害指数
项目结构
drone_field_analyzer/
├── README.md # 项目说明文档
├── requirements.txt # 依赖包列表
├── config/
│ └── settings.yaml # 配置文件
├── src/
│ ├── __init__.py
│ ├── main.py # 程序入口
│ ├── preprocessing/
│ │ ├── __init__.py
│ │ ├── image_denoiser.py # 图像去噪模块
│ │ ├── geometric_corrector.py # 几何校正模块
│ │ └── radiometric_calibrator.py # 辐射定标模块
│ ├── stitching/
│ │ ├── __init__.py
│ │ ├── feature_extractor.py # 特征提取模块
│ │ ├── image_matcher.py # 图像匹配模块
│ │ ├── image_registrator.py # 图像配准模块
│ │ └── panorama_builder.py # 全景图构建模块
│ ├── analysis/
│ │ ├── __init__.py
│ │ ├── ndvi_calculator.py # NDVI计算模块
│ │ ├── growth_assessor.py # 长势评估模块
│ │ ├── disease_detector.py # 病害检测模块
│ │ └── weed_detector.py # 杂草检测模块
│ ├── visualization/
│ │ ├── __init__.py
│ │ ├── health_map_generator.py # 健康地图生成器
│ │ └── report_generator.py # 报告生成器
│ └── utils/
│ ├── __init__.py
│ ├── logger.py # 日志工具
│ ├── file_handler.py # 文件处理工具
│ └── geo_utils.py # 地理坐标工具
├── tests/
│ ├── __init__.py
│ ├── test_preprocessing.py
│ ├── test_stitching.py
│ └── test_analysis.py
├── data/
│ ├── input/ # 输入图像目录
│ ├── output/ # 输出结果目录
│ └── models/ # 预训练模型目录
└── docs/
├── api_documentation.md # API文档
└── user_guide.md # 用户指南
核心代码实现
1. 主程序入口 (src/main.py)
"""
无人机全自动巡田分析系统 - 主程序入口
Author: Full Stack Developer with 5 years experience
Version: 1.0.0
Description: 集成图像预处理、拼接、长势分析的完整农田健康监测解决方案
"""
import os
import sys
import yaml
from pathlib import Path
from datetime import datetime
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, field
from enum import Enum
import logging
# 添加项目根目录到路径
sys.path.insert(0, str(Path(__file__).parent.parent))
from src.utils.logger import setup_logger
from src.utils.file_handler import FileHandler
from src.preprocessing.image_denoiser import ImageDenoiser
from src.preprocessing.geometric_corrector import GeometricCorrector
from src.preprocessing.radiometric_calibrator import RadiometricCalibrator
from src.stitching.panorama_builder import PanoramaBuilder
from src.analysis.ndvi_calculator import NDVICalculator
from src.analysis.growth_assessor import GrowthAssessor
from src.analysis.disease_detector import DiseaseDetector
from src.visualization.health_map_generator import HealthMapGenerator
class ProcessingStage(Enum):
"""处理阶段枚举"""
PREPROCESSING = "preprocessing"
STITCHING = "stitching"
ANALYSIS = "analysis"
VISUALIZATION = "visualization"
COMPLETED = "completed"
@dataclass
class ProcessingConfig:
"""处理配置数据类"""
input_dir: str
output_dir: str
enable_preprocessing: bool = True
enable_stitching: bool = True
enable_analysis: bool = True
enable_visualization: bool = True
stitch_method: str = "feature_based" # feature_based, direct
ndvi_threshold_low: float = 0.3
ndvi_threshold_high: float = 0.8
overlap_threshold: float = 0.3
tile_size: int = 512
log_level: str = "INFO"
@classmethod
def from_yaml(cls, config_path: str) -> 'ProcessingConfig':
"""从YAML文件加载配置"""
with open(config_path, 'r', encoding='utf-8') as f:
config_dict = yaml.safe_load(f)
return cls(**config_dict.get('processing', {}))
@classmethod
def default(cls) -> 'ProcessingConfig':
"""返回默认配置"""
return cls(
input_dir="./data/input",
output_dir="./data/output",
enable_preprocessing=True,
enable_stitching=True,
enable_analysis=True,
enable_visualization=True
)
@dataclass
class ProcessingResult:
"""处理结果数据类"""
stage: ProcessingStage
success: bool
message: str
data: Dict = field(default_factory=dict)
metrics: Dict = field(default_factory=dict)
timestamp: datetime = field(default_factory=datetime.now)
class DroneFieldAnalyzer:
"""
无人机全自动巡田分析系统核心类
该类实现了从原始航拍图像到农田健康地图的完整处理流水线，
集成了图像预处理、拼接、长势分析和可视化功能。
Attributes:
config: 处理配置对象
logger: 日志记录器
file_handler: 文件处理器
current_stage: 当前处理阶段
results: 各阶段处理结果记录
"""
def __init__(self, config: Optional[ProcessingConfig] = None):
"""
初始化分析系统
Args:
config: 处理配置，如果为None则使用默认配置
"""
self.config = config or ProcessingConfig.default()
self.logger = setup_logger(
name="DroneFieldAnalyzer",
level=self.config.log_level
)
self.file_handler = FileHandler(self.config.output_dir)
self.current_stage = ProcessingStage.PREPROCESSING
self.results: Dict[ProcessingStage, ProcessingResult] = {}
# 初始化各处理模块
self._initialize_modules()
self.logger.info("无人机全自动巡田分析系统初始化完成")
def _initialize_modules(self) -> None:
"""初始化所有处理模块"""
self.logger.info("正在初始化处理模块...")
# 预处理模块
self.denoiser = ImageDenoiser() if self.config.enable_preprocessing else None
self.geo_corrector = GeometricCorrector() if self.config.enable_preprocessing else None
self.radiometric_calibrator = RadiometricCalibrator() if self.config.enable_preprocessing else None
# 拼接模块
self.panorama_builder = PanoramaBuilder(
method=self.config.stitch_method
) if self.config.enable_stitching else None
# 分析模块
self.ndvi_calculator = NDVICalculator(
threshold_low=self.config.ndvi_threshold_low,
threshold_high=self.config.ndvi_threshold_high
) if self.config.enable_analysis else None
self.growth_assessor = GrowthAssessor() if self.config.enable_analysis else None
self.disease_detector = DiseaseDetector() if self.config.enable_analysis else None
# 可视化模块
self.health_map_generator = HealthMapGenerator() if self.config.enable_visualization else None
self.logger.info("所有处理模块初始化完成")
def process(self, progress_callback=None) -> ProcessingResult:
"""
执行完整的处理流水线
Args:
progress_callback: 进度回调函数，接收(stage, progress, message)
Returns:
ProcessingResult: 最终处理结果
"""
try:
self.logger.info("=" * 60)
self.logger.info("开始执行无人机全自动巡田分析流水线")
self.logger.info("=" * 60)
# 阶段1: 图像预处理
if self.config.enable_preprocessing:
result = self._run_preprocessing(progress_callback)
if not result.success:
return result
# 阶段2: 图像拼接
if self.config.enable_stitching:
result = self._run_stitching(progress_callback)
if not result.success:
return result
# 阶段3: 长势分析
if self.config.enable_analysis:
result = self._run_analysis(progress_callback)
if not result.success:
return result
# 阶段4: 可视化输出
if self.config.enable_visualization:
result = self._run_visualization(progress_callback)
if not result.success:
return result
# 生成最终报告
final_result = self._generate_final_report()
self.logger.info("=" * 60)
self.logger.info("无人机全自动巡田分析流水线执行完成!")
self.logger.info("=" * 60)
return final_result
except Exception as e:
self.logger.error(f"处理流水线执行失败: {str(e)}", exc_info=True)
return ProcessingResult(
stage=self.current_stage,
success=False,
message=f"处理失败: {str(e)}"
)
def _run_preprocessing(self, progress_callback=None) -> ProcessingResult:
"""执行图像预处理阶段"""
self.current_stage = ProcessingStage.PREPROCESSING
self.logger.info("[阶段1] 开始图像预处理...")
try:
input_images = self.file_handler.load_input_images(self.config.input_dir)
processed_images = []
total_images = len(input_images)
for idx, image_path in enumerate(input_images):
if progress_callback:
progress = (idx / total_images) * 100
progress_callback(self.current_stage, progress,
f"预处理图像 {idx+1}/{total_images}")
self.logger.debug(f"处理图像: {image_path}")
# 读取图像
image = self.file_handler.read_image(image_path)
# 图像去噪
if self.denoiser:
image = self.denoiser.denoise(image, method='bilateral')
# 几何校正
if self.geo_corrector:
image = self.geo_corrector.correct_distortion(image)
# 辐射定标
if self.radiometric_calibrator:
image = self.radiometric_calibrator.calibrate(image)
processed_images.append(image)
# 保存预处理后的图像
output_path = self.file_handler.save_processed_image(
image,
Path(image_path).stem + "_processed"
)
self.logger.debug(f"已保存预处理图像: {output_path}")
result = ProcessingResult(
stage=ProcessingStage.PREPROCESSING,
success=True,
message=f"成功预处理 {len(processed_images)} 张图像",
data={'processed_images': processed_images, 'count': len(processed_images)},
metrics={
'input_count': total_images,
'output_count': len(processed_images),
'avg_processing_time': 0.0 # 可添加实际计时
}
)
self.results[self.current_stage] = result
self.logger.info(f"[阶段1] 图像预处理完成: {result.message}")
return result
except Exception as e:
error_msg = f"图像预处理失败: {str(e)}"
self.logger.error(error_msg, exc_info=True)
return ProcessingResult(
stage=ProcessingStage.PREPROCESSING,
success=False,
message=error_msg
)
def _run_stitching(self, progress_callback=None) -> ProcessingResult:
"""执行图像拼接阶段"""
self.current_stage = ProcessingStage.STITCHING
self.logger.info("[阶段2] 开始图像拼接...")
try:
# 获取预处理后的图像
if self.results.get(ProcessingStage.PREPROCESSING):
images = self.results[ProcessingStage.PREPROCESSING].data['processed_images']
else:
images = self.file_handler.load_input_images(
self.config.input_dir,
pattern="*_processed.*"
)
if len(images) < 2:
return ProcessingResult(
stage=ProcessingStage.STITCHING,
success=False,
message="图像数量不足，至少需要2张图像进行拼接"
)
# 执行拼接
panorama, homographies = self.panorama_builder.build_panorama(
images,
progress_callback=lambda p, m: (
progress_callback(self.current_stage, p, m)
if progress_callback else None
)
)
# 保存拼接结果
output_path = self.file_handler.save_panorama(panorama, "field_panorama")
result = ProcessingResult(
stage=ProcessingStage.STITCHING,
success=True,
message=f"成功拼接 {len(images)} 张图像，生成全景图",
data={
'panorama': panorama,
'homographies': homographies,
'panorama_path': output_path
},
metrics={
'input_count': len(images),
'panorama_size': panorama.shape[:2],
'overlap_regions': len(homographies)
}
)
self.results[self.current_stage] = result
self.logger.info(f"[阶段2] 图像拼接完成: {result.message}")
return result
except Exception as e:
error_msg = f"图像拼接失败: {str(e)}"
self.logger.error(error_msg, exc_info=True)
return ProcessingResult(
stage=ProcessingStage.STITCHING,
success=False,
message=error_msg
)
def _run_analysis(self, progress_callback=None) -> ProcessingResult:
"""执行长势分析阶段"""
self.current_stage = ProcessingStage.ANALYSIS
self.logger.info("[阶段3] 开始长势分析...")
try:
# 获取拼接后的全景图
if self.results.get(ProcessingStage.STITCHING):
panorama = self.results[ProcessingStage.STITCHING].data['panorama']
else:
panorama = self.file_handler.load_panorama("field_panorama")
analysis_results = {}
# 计算NDVI
if self.ndvi_calculator:
if progress_callback:
progress_callback(self.current_stage, 20, "计算NDVI植被指数...")
ndvi_map, vegetation_mask = self.ndvi_calculator.calculate(panorama)
analysis_results['ndvi'] = {
'map': ndvi_map,
'mask': vegetation_mask,
'statistics': self.ndvi_calculator.get_statistics(ndvi_map)
}
# 保存NDVI图
self.file_handler.save_ndvi_map(ndvi_map, "ndvi_map")
self.logger.info("NDVI计算完成")
# 长势评估
if self.growth_assessor:
if progress_callback:
progress_callback(self.current_stage, 50, "评估作物长势...")
growth_map, growth_zones = self.growth_assessor.assess(
panorama,
ndvi_map if 'ndvi' in analysis_results else None
)
analysis_results['growth'] = {
'map': growth_map,
'zones': growth_zones,
'statistics': self.growth_assessor.get_statistics(growth_map)
}
# 保存长势图
self.file_handler.save_growth_map(growth_map, "growth_map")
self.logger.info("长势评估完成")
# 病害检测
if self.disease_detector:
if progress_callback:
progress_callback(self.current_stage, 80, "检测病虫害...")
disease_map, disease_instances = self.disease_detector.detect(
panorama,
ndvi_map if 'ndvi' in analysis_results else None
)
analysis_results['disease'] = {
'map': disease_map,
'instances': disease_instances,
'statistics': self.disease_detector.get_statistics(disease_instances)
}
# 保存病害检测结果
self.file_handler.save_disease_map(disease_map, "disease_map")
self.logger.info(f"病害检测完成: 发现 {len(disease_instances)} 处异常区域")
result = ProcessingResult(
stage=ProcessingStage.ANALYSIS,
success=True,
message="长势分析完成",
data=analysis_results,
metrics=self._calculate_analysis_metrics(analysis_results)
)
self.results[self.current_stage] = result
self.logger.info(f"[阶段3] 长势分析完成: {result.message}")
return result
except Exception as e:
error_msg = f"长势分析失败: {str(e)}"
self.logger.error(error_msg, exc_info=True)
return ProcessingResult(
stage=ProcessingStage.ANALYSIS,
success=False,
message=error_msg
)
def _run_visualization(self, progress_callback=None) -> ProcessingResult:
"""执行可视化输出阶段"""
self.current_stage = ProcessingStage.VISUALIZATION
self.logger.info("[阶段4] 开始生成健康地图...")
try:
# 获取分析结果
analysis_result = self.results.get(ProcessingStage.ANALYSIS)
if not analysis_result or not analysis_result.success:
return ProcessingResult(
stage=ProcessingStage.VISUALIZATION,
success=False,
message="无法进行可视化: 缺少有效的分析结果"
)
analysis_data = analysis_result.data
panorama = self.results[ProcessingStage.STITCHING].data['panorama']
if progress_callback:
progress_callback(self.current_stage, 30, "生成健康地图...")
# 生成健康地图
health_map = self.health_map_generator.generate(
panorama,
analysis_data.get('ndvi', {}).get('map'),
analysis_data.get('growth', {}).get('map'),
analysis_data.get('disease', {}).get('map')
)
# 保存健康地图
health_map_path = self.file_handler.save_health_map(health_map, "health_map")
if progress_callback:
progress_callback(self.current_stage, 70, "生成分析报告...")
# 生成统计报告
report = self.health_map_generator.generate_report(
analysis_data,
health_map
)
report_path = self.file_handler.save_report(report, "analysis_report")
# 生成预警信息
alerts = self.health_map_generator.generate_alerts(analysis_data)
alerts_path = self.file_handler.save_alerts(alerts, "alerts")
result = ProcessingResult(
stage=ProcessingStage.VISUALIZATION,
success=True,
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！