Spaces:

bapi143
/

CropAI

Running on Zero

CropAI / backend /farm_memory.py

Deploy CropAI Deep Learning backend, YOLO11 pest model and Audio-CNN

b3555b4 4 days ago

13.9 kB

# -*- 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()

	# -- 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()