Precision Agriculture with AI Drones
Agriculture is being revolutionized by drone technology. AI-powered drones can monitor thousands of acres, detect crop stress before it's visible, and apply treatments precisely where needed—reducing costs and environmental impact.
The Agricultural Drone Workflow
Sponsored
Real-World Analogy
Think of agricultural drones as "doctors for crops." Just as a doctor uses different diagnostic tools (thermometer, X-ray, blood test), agricultural drones use different sensors (RGB cameras, multispectral, thermal) to diagnose plant health.
SVG: Agricultural Drone Workflow
NDVI Analysis
What is NDVI?
NDVI (Normalized Difference Vegetation Index) measures plant health by comparing near-infrared and red light reflectance. Healthy plants reflect more near-infrared light.
Formula: NDVI = (NIR - Red) / (NIR + Red)
NDVI Calculation Implementation
python
from PIL import Image
from typing import Tuple
class NDVIAnalyzer:
def __init__(self):
self.health_thresholds = {
'healthy': 0.6,
'moderate': 0.3,
'stressed': 0.1,
'bare_soil': -0.1
}
def calculate_ndvi(self, nir_band: np.ndarray,
red_band: np.ndarray) -> np.ndarray:
"""Calculate NDVI from near-infrared and red bands"""
nir = nir_band.astype(float)
red = red_band.astype(float)
# Avoid division by zero
denominator = nir + red
denominator[denominator == 0] = 1e-10
ndvi = (nir - red) / denominator
return np.clip(ndvi, -1, 1)
def classify_health(self, ndvi: np.ndarray) -> dict:
"""Classify vegetation health based on NDVI values"""
classification = {
'healthy': np.sum(ndvi > self.health_thresholds['healthy']),
'moderate': np.sum((ndvi > self.health_thresholds['moderate']) &
(ndvi <= self.health_thresholds['healthy'])),
'stressed': np.sum((ndvi > self.health_thresholds['stressed']) &
(ndvi <= self.health_thresholds['moderate'])),
'bare_soil': np.sum(ndvi <= self.health_thresholds['stressed'])
}
total_pixels = ndvi.size
return {k: v/total_pixels*100 for k, v in classification.items()}
def detect_anomalies(self, ndvi: np.ndarray,
kernel_size: int = 5) -> np.ndarray:
"""Detect anomalous regions using local statistics"""
from scipy.ndimage import uniform_filter
local_mean = uniform_filter(ndvi, size=kernel_size)
local_std = uniform_filter((ndvi - local_mean)**2, size=kernel_size)
local_std = np.sqrt(local_std)
# Anomalies are pixels deviating significantly from local mean
z_score = np.abs((ndvi - local_mean) / (local_std + 1e-10))
anomalies = z_score > 2.0
return anomalies
def generate_health_map(self, ndvi: np.ndarray) -> np.ndarray:
"""Generate RGB health visualization"""
h, w = ndvi.shape
health_map = np.zeros((h, w, 3), dtype=np.uint8)
# Red: stressed/bare
mask = ndvi <= self.health_thresholds['stressed']
health_map[mask] = [180, 50, 50]
# Yellow: moderate stress
mask = (ndvi > self.health_thresholds['stressed']) & \
(ndvi <= self.health_thresholds['moderate'])
health_map[mask] = [245, 158, 11]
# Light green: moderate health
mask = (ndvi > self.health_thresholds['moderate']) & \
(ndvi <= self.health_thresholds['healthy'])
health_map[mask] = [132, 204, 22]
# Dark green: healthy
mask = ndvi > self.health_thresholds['healthy']
health_map[mask] = [16, 185, 129]
return health_map
def generate_report(self, ndvi: np.ndarray) -> str:
"""Generate analysis report"""
stats = self.classify_health(ndvi)
report = f"""
NDVI Analysis Report
====================
Mean NDVI: {np.mean(ndvi):.3f}
Std NDVI: {np.std(ndvi):.3f}
Health Distribution:
- Healthy: {stats['healthy']:.1f}%
- Moderate: {stats['moderate']:.1f}%
- Stressed: {stats['stressed']:.1f}%
- Bare Soil: {stats['bare_soil']:.1f}%
Recommendations:
"""
if stats['stressed'] > 20:
report += "- CRITICAL: Significant stress detected. "
report += "Irrigation and fertilization recommended.\n"
if stats['healthy'] > 70:
report += "- GOOD: Majority of field is healthy.\n"
return report
Automated Spraying System
Sponsored
python
from dataclasses import dataclass
from typing import List, Tuple
@dataclass
class SprayZone:
center: np.ndarray
radius: float
ndvi_value: float
treatment: str
dosage_ml: float
class PrecisionSprayer:
def __init__(self, tank_capacity: float = 10.0):
self.tank_capacity = tank_capacity # liters
self.current_level = tank_capacity
self.spray_rate = 0.5 # liters per minute
self.flight_speed = 3.0 # m/s
self.swath_width = 3.0 # meters
def plan_treatment(self, ndvi_map: np.ndarray,
resolution: float) -> List[SprayZone]:
"""Plan treatment zones based on NDVI analysis"""
zones = []
h, w = ndvi_map.shape
for i in range(0, h, 5):
for j in range(0, w, 5):
patch = ndvi_map[i:min(i+5, h), j:min(j+5, w)]
mean_ndvi = np.mean(patch)
if mean_ndvi < 0.3: # Stressed area
center = np.array([i * resolution, j * resolution])
treatment = self._determine_treatment(mean_ndvi)
dosage = self._calculate_dosage(mean_ndvi)
zones.append(SprayZone(
center=center,
radius=5.0,
ndvi_value=mean_ndvi,
treatment=treatment,
dosage_ml=dosage
))
return self._optimize_path(zones)
def _determine_treatment(self, ndvi: float) -> str:
if ndvi < 0:
return "fungicide"
elif ndvi < 0.1:
return "fertilizer_heavy"
else:
return "fertilizer_light"
def _calculate_dosage(self, ndvi: float) -> float:
"""Calculate dosage based on stress level"""
base_dosage = 100 # ml per zone
stress_factor = max(0, (0.3 - ndvi) / 0.3)
return base_dosage * (1 + stress_factor)
def _optimize_path(self, zones: List[SprayZone]) -> List[SprayZone]:
"""Optimize spray path using nearest neighbor"""
if not zones:
return zones
optimized = [zones[0]]
remaining = zones[1:]
while remaining:
current = optimized[-1]
nearest = min(remaining,
key=lambda z: np.linalg.norm(z.center - current.center))
optimized.append(nearest)
remaining.remove(nearest)
return optimized
def estimate_consumption(self, zones: List[SprayZone]) -> dict:
"""Estimate chemical and time requirements"""
total_dosage = sum(z.dosage_ml for z in zones)
total_distance = sum(
np.linalg.norm(zones[i+1].center - zones[i].center)
for i in range(len(zones)-1)
)
flight_time = total_distance / self.flight_speed / 60 # minutes
spray_time = total_dosage / self.spray_rate # minutes
return {
'total_chemical_ml': total_dosage,
'total_distance_m': total_distance,
'flight_time_min': flight_time,
'spray_time_min': spray_time,
'total_time_min': flight_time + spray_time,
'tanks_needed': total_dosage / (self.tank_capacity * 1000)
}
Crop Disease Detection
python
from typing import Dict, List
class DiseaseDetector:
def __init__(self):
self.disease_signatures = {
'powdery_mildew': {
'ndvi_range': (0.2, 0.4),
'texture': 'white_patches',
'chlorophyll': 'reduced'
},
'rust': {
'ndvi_range': (0.1, 0.3),
'texture': 'orange_spots',
'chlorophyll': 'moderate'
},
'blight': {
'ndvi_range': (-0.1, 0.1),
'texture': 'brown_patches',
'chlorophyll': 'severely_reduced'
},
'nitrogen_deficiency': {
'ndvi_range': (0.3, 0.5),
'texture': 'uniform_yellowing',
'chlorophyll': 'reduced'
}
}
def analyze_region(self, rgb: np.ndarray, nir: np.ndarray,
thermal: np.ndarray = None) -> Dict:
"""Analyze a crop region for diseases"""
ndvi = (nir.astype(float) - rgb[:,:,0].astype(float)) / \
(nir.astype(float) + rgb[:,:,0].astype(float) + 1e-10)
mean_ndvi = np.mean(ndvi)
texture_score = self._compute_texture_score(rgb)
temperature = np.mean(thermal) if thermal is not None else 25.0
diagnosis = {
'mean_ndvi': mean_ndvi,
'health_score': self._health_score(mean_ndvi, texture_score),
'likely_issues': [],
'confidence': 0.0
}
for disease, sig in self.disease_signatures.items():
if sig['ndvi_range'][0] <= mean_ndvi <= sig['ndvi_range'][1]:
diagnosis['likely_issues'].append(disease)
diagnosis['confidence'] += 0.3
diagnosis['confidence'] = min(diagnosis['confidence'], 1.0)
return diagnosis
def _compute_texture_score(self, rgb: np.ndarray) -> float:
"""Simple texture analysis based on variance"""
gray = np.mean(rgb, axis=2)
return np.std(gray)
def _health_score(self, ndvi: float, texture: float) -> float:
"""Compute overall health score (0-1)"""
ndvi_score = max(0, min(1, (ndvi + 0.1) / 0.9))
texture_score = max(0, min(1, 1 - texture / 100))
return (ndvi_score * 0.7 + texture_score * 0.3)
Yield Prediction Model
python
from typing import List, Dict
class YieldPredictor:
def __init__(self, crop_type: str = "wheat"):
self.crop_type = crop_type
self.growth_stages = {
'vegetative': (0, 30),
'flowering': (30, 60),
'grain_fill': (60, 90),
'maturity': (90, 120)
}
def predict_yield(self, ndvi_history: List[float],
area_hectares: float,
day_of_season: int) -> Dict:
"""Predict yield based on NDVI history"""
current_ndvi = ndvi_history[-1] if ndvi_history else 0.5
# Simple yield model based on NDVI
base_yield = self._get_base_yield()
# Growth stage factor
stage = self._get_growth_stage(day_of_season)
stage_factor = self._get_stage_factor(stage)
# NDVI factor (higher NDVI = better yield)
ndvi_factor = max(0, min(2, current_ndvi / 0.6))
# Trend factor (improving or declining)
if len(ndvi_history) >= 3:
trend = np.polyfit(range(len(ndvi_history[-3:])),
ndvi_history[-3:], 1)[0]
trend_factor = 1 + trend * 0.1
else:
trend_factor = 1.0
predicted_yield = (base_yield * stage_factor *
ndvi_factor * trend_factor)
return {
'predicted_yield_kg_ha': predicted_yield,
'total_yield_kg': predicted_yield * area_hectares,
'confidence': self._prediction_confidence(ndvi_history),
'stage': stage,
'recommendations': self._get_recommendations(
current_ndvi, stage, trend_factor)
}
def _get_base_yield(self) -> float:
base_yields = {
'wheat': 3500,
'corn': 9000,
'soybean': 2800,
'rice': 4500
}
return base_yields.get(self.crop_type, 3500)
def _get_growth_stage(self, day: int) -> str:
for stage, (start, end) in self.growth_stages.items():
if start <= day < end:
return stage
return 'maturity'
def _get_stage_factor(self, stage: str) -> float:
factors = {
'vegetative': 0.3,
'flowering': 0.6,
'grain_fill': 0.85,
'maturity': 1.0
}
return factors.get(stage, 1.0)
def _prediction_confidence(self, history: List[float]) -> float:
if len(history) < 5:
return 0.3
elif len(history) < 10:
return 0.6
return 0.85
def _get_recommendations(self, ndvi: float, stage: str,
trend: float) -> List[str]:
recs = []
if ndvi < 0.4 and stage == 'flowering':
recs.append("Apply nitrogen fertilizer immediately")
if trend < -0.05:
recs.append("Investigate declining health trend")
if stage == 'grain_fill' and ndvi > 0.6:
recs.append("Optimal conditions - maintain current practices")
return recs
Soil Analysis
python
class SoilAnalyzer:
def __init__(self):
self.soil_types = {
'clay': {'water_holding': 0.8, 'drainage': 0.3},
'sandy': {'water_holding': 0.3, 'drainage': 0.8},
'loam': {'water_holding': 0.6, 'drainage': 0.6},
'silt': {'water_holding': 0.7, 'drainage': 0.5}
}
def analyze_moisture(self, thermal_data: np.ndarray,
depth_estimates: List[float]) -> Dict:
"""Analyze soil moisture from thermal imagery"""
surface_temp = np.mean(thermal_data[:10, :10])
deep_temp = np.mean(thermal_data[-10:, -10:])
temp_diff = surface_temp - deep_temp
# Inverse relationship between temp and moisture
moisture_index = max(0, min(1, 1 - (temp_diff / 20)))
return {
'surface_moisture': moisture_index,
'depth_gradient': temp_diff,
'irrigation_needed': moisture_index < 0.3,
'estimated_mm': moisture_index * 100
}
def classify_soil(self, rgb_mean: np.ndarray,
organic_matter: float) -> str:
"""Classify soil type from visual characteristics"""
r, g, b = rgb_mean
darkness = (r + g + b) / 3
reddish = r - b
if darkness < 80:
return 'clay'
elif reddish > 30:
return 'sandy'
elif darkness > 150:
return 'silt'
else:
return 'loam'
Hands-On Project: Complete Field Analysis System
Advertisement
python
# field_analysis_system.py
from typing import Dict, List, Tuple
class FieldAnalysisSystem:
def __init__(self, field_width: float, field_height: float):
self.width = field_width
self.height = field_height
self.ndvi_analyzer = NDVIAnalyzer()
self.sprayer = PrecisionSprayer()
self.predictor = YieldPredictor()
def full_analysis(self, flight_data: Dict) -> Dict:
"""Run complete field analysis pipeline"""
results = {}
# Step 1: Calculate NDVI
ndvi = self.ndvi_analyzer.calculate_ndvi(
flight_data['nir'],
flight_data['red']
)
results['ndvi'] = ndvi
results['health_map'] = self.ndvi_analyzer.generate_health_map(ndvi)
results['health_stats'] = self.ndvi_analyzer.classify_health(ndvi)
# Step 2: Detect anomalies
anomalies = self.ndvi_analyzer.detect_anomalies(ndvi)
results['anomaly_count'] = np.sum(anomalies)
results['anomaly_percentage'] = np.sum(anomalies) / ndvi.size * 100
# Step 3: Plan treatments
zones = self.sprayer.plan_treatment(ndvi, resolution=0.1)
results['treatment_zones'] = zones
results['spray_estimate'] = self.sprayer.estimate_consumption(zones)
# Step 4: Yield prediction
if 'ndvi_history' in flight_data:
results['yield_prediction'] = self.predictor.predict_yield(
flight_data['ndvi_history'],
area_hectares=self.width * self.height / 10000,
day_of_season=flight_data.get('day', 60)
)
# Step 5: Generate report
results['report'] = self._generate_report(results)
return results
def _generate_report(self, results: Dict) -> str:
report = "=" * 50 + "\n"
report += "FIELD ANALYSIS REPORT\n"
report += "=" * 50 + "\n\n"
stats = results['health_stats']
report += "Health Summary:\n"
for category, percentage in stats.items():
report += f" {category}: {percentage:.1f}%\n"
if results['anomaly_percentage'] > 5:
report += f"\nALERT: {results['anomaly_percentage']:.1f}% "
report += "anomalous regions detected\n"
if 'yield_prediction' in results:
yp = results['yield_prediction']
report += f"\nYield Prediction: "
report += f"{yp['predicted_yield_kg_ha']:.0f} kg/ha\n"
return report
# Generate sample flight data
def generate_sample_data():
np.random.seed(42)
size = 100
# Simulate NIR and Red bands
x, y = np.meshgrid(np.linspace(0, 2*np.pi, size),
np.linspace(0, 2*np.pi, size))
nir = (np.sin(x) * np.cos(y) + 1) * 0.5 * 1000 + 500
red = (np.cos(x) * np.sin(y) + 1) * 0.3 * 800 + 200
return {
'nir': nir,
'red': red,
'ndvi_history': [0.3, 0.35, 0.4, 0.45, 0.48],
'day': 60
}
# Run analysis
if __name__ == "__main__":
system = FieldAnalysisSystem(field_width=500, field_height=300)
data = generate_sample_data()
results = system.full_analysis(data)
print(results['report'])
Exercise
- Build a multi-field comparison tool that compares NDVI across multiple flights
- Implement irrigation scheduling based on soil moisture trends
- Create a pest outbreak predictor using weather and NDVI data
- Design a variable-rate application map for fertilizer
ℹ️
Precision agriculture drones can reduce pesticide use by up to 90% while increasing crop yields by 20-30%. Companies like DJI Agriculture and senseFly are leading this revolution.