Spaces:
Sleeping
Sleeping
File size: 14,268 Bytes
429a26d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 |
import os
import json
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import logging
try:
from PIL import Image
except ImportError:
Image = None
try:
import google.generativeai as genai
except ImportError:
genai = None
from models import DiseaseDetection, Farm, db
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class DiseaseDetectionService:
"""Service for AI-powered crop disease detection and treatment recommendations"""
def __init__(self, gemini_api_key: str):
self.gemini_api_key = gemini_api_key
genai.configure(api_key=gemini_api_key)
self.model = genai.GenerativeModel('gemini-2.0-flash')
# Common diseases database for quick reference
self.disease_database = {
'rice': {
'blast': {
'symptoms': 'Diamond-shaped lesions on leaves, brown borders with gray centers',
'treatment': 'Apply fungicides like Tricyclazole, improve drainage',
'prevention': 'Use resistant varieties, avoid excessive nitrogen'
},
'bacterial_blight': {
'symptoms': 'Water-soaked lesions that turn yellow then brown',
'treatment': 'Copper-based bactericides, remove infected plants',
'prevention': 'Use certified seeds, avoid overhead irrigation'
}
},
'wheat': {
'rust': {
'symptoms': 'Orange-red pustules on leaves and stems',
'treatment': 'Fungicides like Propiconazole, early application',
'prevention': 'Resistant varieties, proper spacing'
},
'powdery_mildew': {
'symptoms': 'White powdery growth on leaves',
'treatment': 'Sulfur-based fungicides, improve air circulation',
'prevention': 'Avoid dense planting, reduce humidity'
}
},
'tomato': {
'late_blight': {
'symptoms': 'Dark green water-soaked spots, white mold on leaf undersides',
'treatment': 'Copper fungicides, remove infected parts',
'prevention': 'Improve ventilation, avoid overhead watering'
},
'early_blight': {
'symptoms': 'Concentric rings on leaves, starts from bottom',
'treatment': 'Fungicides, remove lower leaves',
'prevention': 'Crop rotation, proper spacing'
}
}
}
def analyze_disease_from_text(self, farm_id: int, crop_name: str, symptoms: str) -> Dict:
"""Analyze disease based on text symptoms description"""
try:
farm = Farm.query.get(farm_id)
if not farm:
return {'error': 'Farm not found'}
# Create prompt for Gemini AI
prompt = self._create_disease_analysis_prompt(crop_name, symptoms)
# Get AI analysis
response = self.model.generate_content(prompt)
ai_analysis = self._parse_disease_response(response.text)
# Save to database
detection = DiseaseDetection(
farm_id=farm_id,
crop_name=crop_name,
disease_name=ai_analysis.get('disease_name', 'Unknown'),
confidence_score=ai_analysis.get('confidence', 0.0),
symptoms=symptoms,
treatment=ai_analysis.get('treatment', ''),
prevention=ai_analysis.get('prevention', ''),
ai_analysis=json.dumps(ai_analysis),
severity=ai_analysis.get('severity', 'unknown')
)
db.session.add(detection)
db.session.commit()
return {
'success': True,
'detection_id': detection.id,
'analysis': ai_analysis
}
except Exception as e:
logger.error(f"Error analyzing disease: {str(e)}")
return {'error': f'Analysis failed: {str(e)}'}
def analyze_disease_from_image(self, farm_id: int, crop_name: str, image_path: str, symptoms: str = "") -> Dict:
"""Analyze disease from uploaded image"""
try:
farm = Farm.query.get(farm_id)
if not farm:
return {'error': 'Farm not found'}
# Verify image exists
if not os.path.exists(image_path):
return {'error': 'Image file not found'}
# Create prompt for image analysis
prompt = self._create_image_analysis_prompt(crop_name, symptoms)
# Load and process image
image = Image.open(image_path)
# Get AI analysis with image
response = self.model.generate_content([prompt, image])
ai_analysis = self._parse_disease_response(response.text)
# Save to database
detection = DiseaseDetection(
farm_id=farm_id,
crop_name=crop_name,
disease_name=ai_analysis.get('disease_name', 'Unknown'),
confidence_score=ai_analysis.get('confidence', 0.0),
symptoms=symptoms,
treatment=ai_analysis.get('treatment', ''),
prevention=ai_analysis.get('prevention', ''),
image_path=image_path,
ai_analysis=json.dumps(ai_analysis),
severity=ai_analysis.get('severity', 'unknown')
)
db.session.add(detection)
db.session.commit()
return {
'success': True,
'detection_id': detection.id,
'analysis': ai_analysis
}
except Exception as e:
logger.error(f"Error analyzing disease from image: {str(e)}")
return {'error': f'Image analysis failed: {str(e)}'}
def _create_disease_analysis_prompt(self, crop_name: str, symptoms: str) -> str:
"""Create prompt for disease analysis"""
return f"""
You are an expert plant pathologist. Analyze the following crop disease symptoms and provide a detailed diagnosis.
CROP: {crop_name}
SYMPTOMS: {symptoms}
Please provide your analysis in the following JSON format:
{{
"disease_name": "Most likely disease name",
"confidence": 0.85,
"severity": "mild/moderate/severe",
"symptoms_analysis": "Detailed analysis of symptoms",
"treatment": "Immediate treatment recommendations",
"prevention": "Future prevention measures",
"additional_info": "Any additional relevant information",
"urgency": "low/medium/high"
}}
Consider:
1. Common diseases for this crop
2. Seasonal factors
3. Environmental conditions
4. Treatment urgency
5. Cost-effective solutions for farmers
Provide practical, actionable advice that farmers can easily implement.
"""
def _create_image_analysis_prompt(self, crop_name: str, symptoms: str = "") -> str:
"""Create prompt for image-based disease analysis"""
base_prompt = f"""
You are an expert plant pathologist. Analyze this image of {crop_name} crop for signs of disease or pest damage.
CROP: {crop_name}
"""
if symptoms:
base_prompt += f"REPORTED SYMPTOMS: {symptoms}\n"
base_prompt += """
Please examine the image carefully and provide your analysis in the following JSON format:
{
"disease_name": "Identified disease or pest issue",
"confidence": 0.85,
"severity": "mild/moderate/severe",
"visual_symptoms": "What you observe in the image",
"treatment": "Immediate treatment recommendations",
"prevention": "Future prevention measures",
"affected_area": "Percentage of plant affected",
"urgency": "low/medium/high",
"additional_observations": "Any other relevant findings"
}
Look for:
1. Leaf discoloration or spots
2. Wilting or deformation
3. Pest damage
4. Fungal growth
5. Nutrient deficiencies
6. Environmental stress signs
Provide practical, cost-effective treatment recommendations suitable for farmers.
"""
return base_prompt
def _parse_disease_response(self, response_text: str) -> Dict:
"""Parse AI response and extract disease information"""
try:
# Try to extract JSON from response
start_idx = response_text.find('{')
end_idx = response_text.rfind('}') + 1
if start_idx != -1 and end_idx != -1:
json_text = response_text[start_idx:end_idx]
analysis = json.loads(json_text)
# Ensure required fields
analysis.setdefault('disease_name', 'Unknown Disease')
analysis.setdefault('confidence', 0.5)
analysis.setdefault('severity', 'moderate')
analysis.setdefault('treatment', 'Consult agricultural expert')
analysis.setdefault('prevention', 'Maintain good crop hygiene')
analysis.setdefault('urgency', 'medium')
return analysis
else:
# Manual parsing if JSON not found
return self._manual_parse_response(response_text)
except json.JSONDecodeError:
return self._manual_parse_response(response_text)
except Exception as e:
logger.error(f"Error parsing disease response: {str(e)}")
return self._get_fallback_analysis()
def _manual_parse_response(self, response_text: str) -> Dict:
"""Manually parse response if JSON parsing fails"""
analysis = {
'disease_name': 'Unknown Disease',
'confidence': 0.5,
'severity': 'moderate',
'treatment': 'Consult agricultural expert',
'prevention': 'Maintain good crop hygiene',
'urgency': 'medium'
}
lines = response_text.lower().split('\n')
for line in lines:
if 'disease' in line and 'name' in line:
analysis['disease_name'] = line.split(':')[-1].strip()
elif 'treatment' in line:
analysis['treatment'] = line.split(':')[-1].strip()
elif 'prevention' in line:
analysis['prevention'] = line.split(':')[-1].strip()
elif 'severe' in line:
analysis['severity'] = 'severe'
elif 'mild' in line:
analysis['severity'] = 'mild'
return analysis
def _get_fallback_analysis(self) -> Dict:
"""Provide fallback analysis when AI fails"""
return {
'disease_name': 'Analysis Failed',
'confidence': 0.0,
'severity': 'unknown',
'treatment': 'Please consult with a local agricultural expert or extension officer',
'prevention': 'Maintain good crop hygiene and monitoring practices',
'urgency': 'medium',
'error': 'AI analysis unavailable'
}
def get_disease_history(self, farm_id: int, days: int = 30) -> List[DiseaseDetection]:
"""Get disease detection history for a farm"""
from_date = datetime.now() - timedelta(days=days)
return DiseaseDetection.query.filter(
DiseaseDetection.farm_id == farm_id,
DiseaseDetection.detected_at >= from_date
).order_by(DiseaseDetection.detected_at.desc()).all()
def update_treatment_status(self, detection_id: int, status: str, notes: str = "") -> bool:
"""Update treatment status for a disease detection"""
try:
detection = DiseaseDetection.query.get(detection_id)
if not detection:
return False
detection.status = status
if status == 'resolved':
detection.resolved_at = datetime.now()
db.session.commit()
return True
except Exception as e:
logger.error(f"Error updating treatment status: {str(e)}")
return False
def get_preventive_recommendations(self, crop_name: str, season: str = "") -> Dict:
"""Get preventive recommendations for common diseases"""
crop_lower = crop_name.lower()
if crop_lower in self.disease_database:
diseases = self.disease_database[crop_lower]
recommendations = []
for disease, info in diseases.items():
recommendations.append({
'disease': disease.replace('_', ' ').title(),
'prevention': info['prevention'],
'early_symptoms': info['symptoms']
})
return {
'crop': crop_name,
'recommendations': recommendations,
'general_tips': [
'Regular field monitoring',
'Proper crop rotation',
'Maintain field hygiene',
'Use disease-resistant varieties',
'Proper water management'
]
}
return {
'crop': crop_name,
'recommendations': [],
'general_tips': [
'Regular field monitoring is essential',
'Maintain proper spacing between plants',
'Ensure good drainage',
'Use certified seeds',
'Follow integrated pest management'
]
}
|