File size: 15,869 Bytes
5fd7795 970be68 5fd7795 970be68 5fd7795 970be68 5fd7795 | 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 | """
CropAI Database & Cloud Persistence Module
Supports Google Cloud Firestore for scalable Cloud Run deployment with
local thread-safe persistence fallback.
"""
import json
import sqlite3
import logging
import threading
import uuid
from typing import Dict, Any, List, Optional
from datetime import datetime, timezone
from pathlib import Path
from .config import settings
logger = logging.getLogger("cropai.database")
DB_FILE = Path(__file__).parent / "cropai_production.db"
class StorageManager:
"""Manages persistence for diagnoses and expert consultation tickets."""
def __init__(self):
self._lock = threading.Lock()
self._firestore_client = None
self._init_firestore()
self._init_local_db()
def _init_firestore(self):
"""Initializes Google Cloud Firestore if GCP environment is configured."""
if settings.gcp_project_id:
try:
from google.cloud import firestore
self._firestore_client = firestore.Client(project=settings.gcp_project_id)
logger.info(f"Initialized Google Cloud Firestore for project: {settings.gcp_project_id}")
except Exception as e:
logger.warning(f"Google Cloud Firestore not available: {e}. Using local storage.")
self._firestore_client = None
def _init_local_db(self):
"""Initializes local SQLite schema for resilient offline/container persistence."""
with self._lock:
conn = sqlite3.connect(DB_FILE)
try:
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS expert_tickets (
ticket_id TEXT PRIMARY KEY,
farmer_name TEXT NOT NULL,
phone_number TEXT NOT NULL,
location TEXT,
crop TEXT NOT NULL,
disease_suspected TEXT,
confidence_score REAL,
notes TEXT,
status TEXT DEFAULT 'PENDING_REVIEW',
assigned_expert TEXT,
created_at TEXT NOT NULL
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS diagnoses_log (
image_id TEXT PRIMARY KEY,
crop TEXT NOT NULL,
disease TEXT NOT NULL,
severity TEXT NOT NULL,
confidence REAL NOT NULL,
is_healthy INTEGER NOT NULL,
created_at TEXT NOT NULL,
payload_json TEXT NOT NULL
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS kisan_polls (
ballot_id TEXT PRIMARY KEY,
problem_id TEXT NOT NULL,
district TEXT NOT NULL,
state TEXT NOT NULL,
crop_affected TEXT,
farmer_comment TEXT,
device_id TEXT,
created_at TEXT NOT NULL
)
""")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_kisan_polls_problem ON kisan_polls(problem_id)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_kisan_polls_district ON kisan_polls(district)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_kisan_polls_device ON kisan_polls(device_id)")
conn.commit()
finally:
conn.close()
def save_ticket(self, ticket_data: Dict[str, Any]) -> str:
"""Saves an expert review ticket to Firestore or local SQLite."""
created_at_str = ticket_data.get("created_at") or datetime.now(timezone.utc).isoformat()
ticket_data["created_at"] = created_at_str
# 1. Try Firestore
if self._firestore_client:
try:
doc_ref = self._firestore_client.collection(settings.firestore_collection_tickets).document(ticket_data["ticket_id"])
doc_ref.set(ticket_data)
logger.info(f"Saved ticket {ticket_data['ticket_id']} to Firestore.")
except Exception as e:
logger.error(f"Firestore save error: {e}")
# 2. Persist to SQLite
with self._lock:
conn = sqlite3.connect(DB_FILE)
try:
cursor = conn.cursor()
cursor.execute("""
INSERT OR REPLACE INTO expert_tickets (
ticket_id, farmer_name, phone_number, location,
crop, disease_suspected, confidence_score, notes,
status, assigned_expert, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
ticket_data["ticket_id"],
ticket_data["farmer_name"],
ticket_data["phone_number"],
ticket_data.get("location"),
ticket_data["crop"],
ticket_data.get("disease_suspected"),
ticket_data.get("confidence_score"),
ticket_data.get("notes"),
ticket_data.get("status", "PENDING_REVIEW"),
ticket_data.get("assigned_expert", "Senior Agronomist"),
created_at_str
))
conn.commit()
finally:
conn.close()
return ticket_data["ticket_id"]
def log_diagnosis(self, diagnosis_data: Dict[str, Any]):
"""Logs an AI diagnosis event for audit, analytics, and disease surveillance."""
timestamp_str = diagnosis_data.get("timestamp") or datetime.now(timezone.utc).isoformat()
if self._firestore_client:
try:
doc_ref = self._firestore_client.collection(settings.firestore_collection_diagnoses).document(diagnosis_data["image_id"])
doc_ref.set(diagnosis_data)
except Exception as e:
logger.debug(f"Firestore diagnosis log error: {e}")
with self._lock:
conn = sqlite3.connect(DB_FILE)
try:
cursor = conn.cursor()
cursor.execute("""
INSERT OR REPLACE INTO diagnoses_log (
image_id, crop, disease, severity, confidence, is_healthy, created_at, payload_json
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", (
diagnosis_data["image_id"],
diagnosis_data["crop"],
diagnosis_data["disease"],
diagnosis_data["severity"],
diagnosis_data["confidence"],
1 if diagnosis_data.get("is_healthy", False) else 0,
timestamp_str,
json.dumps(diagnosis_data)
))
conn.commit()
finally:
conn.close()
def get_recent_tickets(self, limit: int = 20) -> List[Dict[str, Any]]:
"""Retrieves recent expert tickets."""
with self._lock:
conn = sqlite3.connect(DB_FILE)
try:
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("SELECT * FROM expert_tickets ORDER BY created_at DESC LIMIT ?", (limit,))
rows = cursor.fetchall()
return [dict(row) for row in rows]
finally:
conn.close()
def is_device_rate_limited(self, device_id: str, problem_id: str, hours: int = 24) -> bool:
"""Checks if a device has already cast a vote for the same issue in the last N hours."""
if not device_id:
return False
with self._lock:
conn = sqlite3.connect(DB_FILE)
try:
cursor = conn.cursor()
cursor.execute("""
SELECT created_at FROM kisan_polls
WHERE device_id = ? AND problem_id = ?
ORDER BY created_at DESC LIMIT 1
""", (device_id, problem_id))
row = cursor.fetchone()
if not row:
return False
try:
last_vote = datetime.fromisoformat(row[0].replace("Z", "+00:00"))
now = datetime.now(timezone.utc)
if (now - last_vote).total_seconds() < hours * 3600:
return True
except Exception:
pass
return False
finally:
conn.close()
def save_poll_ballot(self, ballot_data: Dict[str, Any]) -> str:
"""Persists a verified farmer poll ballot into SQLite."""
created_at = ballot_data.get("timestamp") or datetime.now(timezone.utc).isoformat()
ballot_id = ballot_data.get("ballot_id") or f"VOTE-{datetime.now().strftime('%Y%m%d')}-{uuid.uuid4().hex[:6].upper()}"
with self._lock:
conn = sqlite3.connect(DB_FILE)
try:
cursor = conn.cursor()
cursor.execute("""
INSERT OR REPLACE INTO kisan_polls (
ballot_id, problem_id, district, state,
crop_affected, farmer_comment, device_id, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", (
ballot_id,
ballot_data["problem_id"],
ballot_data.get("district", "Surguja"),
ballot_data.get("state", "Chhattisgarh"),
ballot_data.get("crop_affected", "Paddy (Rice)"),
ballot_data.get("farmer_comment", ""),
ballot_data.get("device_id", ""),
created_at
))
conn.commit()
finally:
conn.close()
return ballot_id
def get_poll_counts(self, district: Optional[str] = None) -> Dict[str, int]:
"""Returns live vote counts aggregated per problem ID."""
counts: Dict[str, int] = {}
with self._lock:
conn = sqlite3.connect(DB_FILE)
try:
cursor = conn.cursor()
if district:
cursor.execute("""
SELECT problem_id, COUNT(*) FROM kisan_polls
WHERE district = ? COLLATE NOCASE
GROUP BY problem_id
""", (district,))
else:
cursor.execute("SELECT problem_id, COUNT(*) FROM kisan_polls GROUP BY problem_id")
rows = cursor.fetchall()
for row in rows:
counts[row[0]] = row[1]
return counts
finally:
conn.close()
def get_recent_ballots(self, limit: int = 50, district: Optional[str] = None) -> List[Dict[str, Any]]:
"""Retrieves most recent cast ballots."""
with self._lock:
conn = sqlite3.connect(DB_FILE)
try:
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
if district:
cursor.execute("""
SELECT * FROM kisan_polls
WHERE district = ? COLLATE NOCASE
ORDER BY created_at DESC LIMIT ?
""", (district, limit))
else:
cursor.execute("SELECT * FROM kisan_polls ORDER BY created_at DESC LIMIT ?", (limit,))
rows = cursor.fetchall()
return [dict(row) for row in rows]
finally:
conn.close()
def seed_poll_baseline_if_empty(self, baseline_options: List[Dict[str, Any]]):
"""Seeds baseline grassroot ballots if the table is freshly initialized."""
with self._lock:
conn = sqlite3.connect(DB_FILE)
try:
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM kisan_polls")
total = cursor.fetchone()[0]
if total > 0:
return
logger.info("Initializing Kisan Jan-Mat SQLite database with baseline community sample votes...")
districts = ["Surguja", "Bilaspur", "Raipur", "Bastar", "Janjgir-Champa", "Rajnandgaon"]
sample_comments = {
"spurious_pesticides": "दुकानदार से ली गई दवा से इल्ली नहीं मरी, पूरा 2 एकड़ खेत खराब हुआ।",
"stray_cattle_wildlife": "रात में नीलगाय का झुंड आकर पूरी चना की फसल चट कर गया।",
"water_electricity_scarcity": "टेल-एंड तक नहर का पानी नहीं पहुंचा, बोरवेल में कम वोल्टेज से मोटर नहीं चली।",
"unseasonal_weather": "धान कटाई के समय ओले और तेज आंधी से बालियां झड़ गईं।",
"mandi_cold_storage": "मंडी में टमाटर ₹3 किलो बिका, किराया भी नहीं निकला।",
"uncontrolled_pest_epidemic": "अचानक भूरा माहू लग गया, 3 दिन में खेत जल गया।"
}
now_iso = datetime.now(timezone.utc).isoformat()
for opt in baseline_options:
p_id = opt["id"]
count = opt.get("default_votes", 50)
for i in range(count):
b_id = f"INIT-{p_id[:4].upper()}-{i+1:04d}"
dist = districts[i % len(districts)]
comment = sample_comments.get(p_id, "") if i % 15 == 0 else ""
cursor.execute("""
INSERT OR IGNORE INTO kisan_polls (
ballot_id, problem_id, district, state,
crop_affected, farmer_comment, device_id, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", (
b_id, p_id, dist, "Chhattisgarh",
"Paddy (Rice)", comment, f"DEV-SEED-{i%30}", now_iso
))
conn.commit()
logger.info("Kisan Jan-Mat baseline successfully seeded.")
finally:
conn.close()
def get_diagnosis_outbreak_clusters(self, days: int = 14) -> List[Dict[str, Any]]:
"""Aggregates real user diagnoses from diagnoses_log to detect actual field outbreak clusters."""
with self._lock:
conn = sqlite3.connect(DB_FILE)
try:
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("""
SELECT crop, disease, severity, COUNT(*) as report_count
FROM diagnoses_log
WHERE is_healthy = 0
GROUP BY crop, disease
ORDER BY report_count DESC
LIMIT 10
""")
rows = cursor.fetchall()
return [dict(row) for row in rows]
finally:
conn.close()
db_manager = StorageManager()
|