# -*- coding: utf-8 -*- """ CropAI AgriMemory™ — Farm Disease Database & Seasonal Predictive Alarm Engine Stores farm diagnosis history, tracks multi-year outbreak recurrence patterns across Chhattisgarh (Surguja, Bilaspur, Raipur, Durg, Bastar, Bemetara, Kabirdham, Rajnandgaon, Dhamtari) and India, correlates real-time micro-climate with historical outbreaks, and generates proactive 4-phase prophylactic care alerts with exact knapsack tank chemical and organic dosages. """ import json import os import math from datetime import datetime from typing import Dict, Any, List, Optional from .historical_outbreaks import PRELOADED_HISTORICAL_OUTBREAKS from .knowledge_base import get_disease_info class AgriMemoryEngine: """Personal Farm Disease History & Seasonal Predictive Alarm Engine.""" def __init__(self, storage_path: str = "farm_memory_store.json"): self.storage_path = storage_path self._ensure_storage() def _ensure_storage(self): """Ensures farm memory storage exists and is pre-seeded with multi-year historical outbreaks.""" should_seed = False if not os.path.exists(self.storage_path): should_seed = True else: try: with open(self.storage_path, "r", encoding="utf-8") as f: data = json.load(f) if not data.get("records"): should_seed = True except Exception: should_seed = True if should_seed: try: with open(self.storage_path, "w", encoding="utf-8") as f: json.dump({"records": PRELOADED_HISTORICAL_OUTBREAKS}, f, indent=2, ensure_ascii=False) except Exception as e: pass def _load_records(self) -> List[Dict[str, Any]]: """Loads all historical outbreak records.""" try: with open(self.storage_path, "r", encoding="utf-8") as f: data = json.load(f) records = data.get("records", []) if not records: return PRELOADED_HISTORICAL_OUTBREAKS return records except Exception: return PRELOADED_HISTORICAL_OUTBREAKS def _save_records(self, records: List[Dict[str, Any]]): """Saves records to the JSON store.""" with open(self.storage_path, "w", encoding="utf-8") as f: json.dump({"records": records}, f, indent=2, ensure_ascii=False) def save_diagnosis_record( self, crop: str, disease: str, disease_id: str, severity: str, confidence: float, temperature_c: float, humidity_pct: float, weather_condition: str, location_name: str = "Ambikapur, Surguja", gps_lat: float = 23.1188, gps_lng: float = 83.1979, farmer_notes: str = "", diagnosis_date: Optional[str] = None ) -> Dict[str, Any]: """Saves a verified crop diagnosis with full environmental and geographical context.""" records = self._load_records() now = datetime.now() dt = datetime.fromisoformat(diagnosis_date) if diagnosis_date else now record_id = f"MEM-{dt.strftime('%Y%m%d')}-{len(records) + 1:04d}" prophylactic_plan = self.generate_prophylactic_plan(crop, disease_id) record = { "record_id": record_id, "crop": crop, "disease": disease, "disease_id": disease_id, "severity": severity, "confidence": round(confidence, 2), "diagnosis_date": dt.isoformat(), "calendar_month": dt.month, "calendar_day": dt.day, "location_name": location_name, "gps_latitude": gps_lat, "gps_longitude": gps_lng, "weather_snapshot": { "temperature_c": round(temperature_c, 1), "humidity_pct": round(humidity_pct, 1), "weather_condition": weather_condition, "dew_point_c": round(self._calculate_dew_point(temperature_c, humidity_pct), 1) }, "prophylactic_plan": prophylactic_plan, "farmer_notes": farmer_notes, "is_healthy": ( severity.lower() == "healthy" or "healthy" in disease.lower() or "healthy" in disease_id.lower() or "स्वस्थ" in disease ), "resolved": False } records.append(record) self._save_records(records) return {"status": "success", "record": record} def get_all_records(self, crop: Optional[str] = None) -> List[Dict[str, Any]]: """Retrieves historical farm diagnosis records, optionally filtered by crop name.""" records = self._load_records() if crop: records = [r for r in records if crop.lower() in r.get("crop", "").lower()] return sorted(records, key=lambda x: x.get("diagnosis_date", ""), reverse=True) def check_seasonal_disease_alerts( self, current_date: Optional[str] = None, current_temp_c: float = 27.0, current_humidity_pct: float = 85.0 ) -> List[Dict[str, Any]]: """ Evaluates historical farm memory against current calendar day and weather conditions. Triggers proactive seasonal warnings within +/- 30 days of historical outbreak windows. """ records = self._load_records() if not records: records = PRELOADED_HISTORICAL_OUTBREAKS now = datetime.fromisoformat(current_date) if current_date else datetime.now() current_day_of_year = now.timetuple().tm_yday alerts = [] seen_diseases = set() for rec in records: # 1. Strictly filter out healthy crops - healthy condition is NEVER an outbreak! dis_lower = rec.get("disease", "").lower() dis_id_lower = rec.get("disease_id", "").lower() sev_lower = rec.get("severity", "").lower() if ( rec.get("is_healthy", False) or sev_lower == "healthy" or "healthy" in dis_lower or "स्वस्थ" in rec.get("disease", "") or "healthy" in dis_id_lower or "no crop" in dis_lower or "अमान्य" in rec.get("disease", "") ): continue rec_date_str = rec.get("diagnosis_date", "2024-09-01T10:00:00") try: rec_dt = datetime.fromisoformat(rec_date_str) except Exception: continue # 2. Strict Date Filter: Must be from a prior year / season (at least 120 days ago and prior year) days_ago = (now - rec_dt).days if days_ago < 120 or rec_dt.year >= now.year: continue past_day_of_year = rec_dt.timetuple().tm_yday # Day distance accounting for annual wrap-around day_diff = abs(current_day_of_year - past_day_of_year) if day_diff > 182: day_diff = 365 - day_diff # Proximity factor: within +/- 30 days if day_diff <= 30: disease_id = rec.get("disease_id", "") if disease_id in seen_diseases: continue seen_diseases.add(disease_id) # Weather similarity correlation hist_weather = rec.get("weather_snapshot", {}) hist_temp = hist_weather.get("temperature_c", 26.0) hist_hum = hist_weather.get("humidity_pct", 85.0) temp_match = max(0.0, 1.0 - abs(current_temp_c - hist_temp) / 15.0) hum_match = max(0.0, 1.0 - abs(current_humidity_pct - hist_hum) / 40.0) recurrence_risk_score = 0.5 * (1.0 - day_diff / 30.0) + 0.3 * temp_match + 0.2 * hum_match threat_level = "CRITICAL" if recurrence_risk_score >= 0.80 else ("HIGH" if recurrence_risk_score >= 0.65 else "MODERATE") days_until_window = past_day_of_year - current_day_of_year if days_until_window < 0: time_desc = f"{abs(days_until_window)} दिन पहले इसी सीजन में" elif days_until_window == 0: time_desc = "आज ही के दिन पिछले साल" else: time_desc = f"अगले {days_until_window} दिनों के अंदर" prophylactic_plan = self.generate_prophylactic_plan(rec["crop"], disease_id) alerts.append({ "alert_id": f"ALT-{rec.get('record_id', 'HIST')}", "crop": rec.get("crop", "Fasal"), "disease": rec.get("disease", "Bimari"), "disease_id": disease_id, "location_recorded": rec.get("location_name", "Chhattisgarh"), "historical_outbreak_date": rec_dt.strftime("%d %B %Y"), "threat_level": threat_level, "recurrence_risk_score": round(recurrence_risk_score * 100.0, 1), "window_description": time_desc, "alert_title": f"⚠️ मौसमी अलर्ट: {rec['crop']} में '{rec['disease']}' का प्रकोप समय!", "alert_message": ( f"ऐतिहासिक डेटा ({rec_dt.strftime('%d %B')} - {rec.get('location_name', 'छत्तीसगढ़')}) के अनुसार " f"इस समय '{rec['disease']}' का गंभीर संक्रमण दर्ज हुआ था। " f"वर्तमान तापमान ({current_temp_c}°C) और आर्द्रता ({current_humidity_pct}%) भी रोगजनक अनुकूल हैं। " f"फसल की तुरंत जांच करें और नीचे दिए गए 4-चरणीय सुरक्षा प्रोटोकॉल का पालन करें।" ), "prophylactic_plan": prophylactic_plan }) return alerts @classmethod def generate_prophylactic_plan(cls, crop: str, disease_id: str) -> Dict[str, Any]: """Generates a 4-phase proactive prophylactic pre-season care plan using the agronomic knowledge base.""" disease_info = get_disease_info(disease_id) disease_name = disease_info.get("name", f"{crop} Disease Prevention") treatment = disease_info.get("treatment", {}) immediate = treatment.get("immediate_action", []) organic = treatment.get("organic_control", []) chemical = treatment.get("chemical_control", []) prevention = disease_info.get("prevention", []) phases = [ { "phase_name": "Phase 1: बुवाई/रोपाई पूर्व तैयारी (15-20 Days Pre)", "action": "बीज शोधन एवं मृदा उपचार (Seed Inoculation & Sanitation)", "details": prevention[0] if prevention else "ट्राइकोडर्मा विरिडी (Trichoderma viride) 5-10g/kg बीज से बीजोपचार करें।" }, { "phase_name": "Phase 2: वानस्पतिक वृद्धि एवं पोषण (Vegetative Stage)", "action": "संतुलित NPK एवं कैनोपी वेंटिलेशन", "details": immediate[0] if immediate else "अत्यधिक यूरिया का छिड़काव रोकें। पोटाश (MOP 20-25 kg/एकड़) का संतुलित प्रयोग करें।" }, { "phase_name": "Phase 3: मौसमी जोखिम विंडो (Prophylactic Bio-Shield)", "action": "प्रिवेंटिव जैविक छिड़काव (Organic Shield)", "details": organic[0] if organic else "स्यूडोमोनास फ्लोरेसेंस @ 5g/L या 5% नीम तेल (10,000 PPM @ 3ml/L) का 16L नैपसैक टैंक से छिड़काव करें।" }, { "phase_name": "Phase 4: आपातकालीन कवकनाशी/कीटनाशी सुरक्षा (Targeted Intervention)", "action": "लक्षित रासायनिक छिड़काव (Chemical Rescue Dosage)", "details": chemical[0] if chemical else "संक्रमण के शुरुआती लक्षण दिखने पर अनुशंसित सिस्टेमिक रसायन का 16L नैपसैक टैंक में घोल बनाकर छिड़कें।" } ] return { "disease_name": disease_name, "crop": crop, "scientific_name": disease_info.get("scientific_name", "N/A"), "phases": phases, "recommended_spray_tank_volume": "16 Liters Knapsack Tank", "organic_shield": organic[:2], "chemical_rescue": chemical[:2] } @staticmethod def _calculate_dew_point(temp: float, humidity: float) -> float: a, b = 17.27, 237.7 alpha = ((a * temp) / (b + temp)) + math.log(max(1.0, humidity) / 100.0) return (b * alpha) / (a - alpha) farm_memory_engine = AgriMemoryEngine()