ShrishtiAI-backend / server /services /gee_chat_fallbacks.py
tanmaydesai07's picture
fixed chatbot
207961e
Raw
History Blame Contribute Delete
4.2 kB
"""
Known-good GEE snippets when the AI emits invalid index method names.
"""
import re
from typing import Dict, Optional, Tuple
# Common typos / hallucinated ee.Image methods → normalizedDifference bands
_METHOD_FIXES: Tuple[Tuple[re.Pattern, str], ...] = (
# AI often invents this when user types "NBI" (built-up) — not NBR burn
(re.compile(r"\.normalizedBurnIndex\s*\(\s*\)", re.I), ".normalizedDifference(['B11', 'B8'])"),
(re.compile(r"\.normalizedBuiltUpIndex\s*\(\s*\)", re.I), ".normalizedDifference(['B11', 'B8'])"),
(re.compile(r"\.normalizedDifferenceBuiltUpIndex\s*\(\s*\)", re.I), ".normalizedDifference(['B11', 'B8'])"),
(re.compile(r"\.normalizedBuiltUp\s*\(\s*\)", re.I), ".normalizedDifference(['B11', 'B8'])"),
(re.compile(r"\.nbi\s*\(\s*\)", re.I), ".normalizedDifference(['B11', 'B8'])"),
(re.compile(r"\.ndbi\s*\(\s*\)", re.I), ".normalizedDifference(['B11', 'B8'])"),
)
_REGIONS: Tuple[Tuple[str, str], ...] = (
("andhra pradesh", "Andhra Pradesh"),
("tamil nadu", "Tamil Nadu"),
("uttar pradesh", "Uttar Pradesh"),
("madhya pradesh", "Madhya Pradesh"),
("west bengal", "West Bengal"),
("karnataka", "Karnataka"),
("maharashtra", "Maharashtra"),
("kerala", "Kerala"),
("goa", "Goa"),
("india", "India"),
("rajasthan", "Rajasthan"),
("gujarat", "Gujarat"),
("punjab", "Punjab"),
("bihar", "Bihar"),
("assam", "Assam"),
)
def sanitize_ai_gee_code(code: str) -> str:
"""Rewrite hallucinated index helpers to valid normalizedDifference calls."""
sanitized = code
for pattern, replacement in _METHOD_FIXES:
sanitized = pattern.sub(replacement, sanitized)
return sanitized
def _detect_region(message: str) -> str:
msg = message.lower()
for needle, label in _REGIONS:
if needle in msg:
return label
return "India"
def _region_fc_snippet(region_name: str) -> str:
if region_name == "India":
return (
'ee.FeatureCollection("FAO/GAUL/2015/level0")'
'.filter(ee.Filter.eq("ADM0_NAME", "India"))'
)
return (
'ee.FeatureCollection("FAO/GAUL/2015/level1")'
f'.filter(ee.Filter.eq("ADM1_NAME", "{region_name}"))'
)
def build_chat_gee_fallback(user_message: str) -> Optional[Tuple[str, Dict]]:
"""
Return (gee_code, vis_params) for common chat requests when AI code fails.
"""
msg = user_message.lower()
region = _detect_region(user_message)
region_fc = _region_fc_snippet(region)
s2 = (
"ee.ImageCollection('COPERNICUS/S2_SR_HARMONIZED')"
".filterBounds(region)"
".filterDate('2024-03-01','2024-04-01')"
".filter(ee.Filter.lt('CLOUDY_PIXEL_PERCENTAGE', 30))"
".mean()"
)
# NBI / NDBI — built-up index (user often types "NBI" without the D)
if re.search(r"\b(ndbi|nbi|built[\s-]?up)\b", msg) and not re.search(
r"\b(nbr|burn|wildfire)\b", msg
):
code = f"""region = {region_fc}
image = {s2}.normalizedDifference(['B11','B8']).clipToCollection(region)"""
vis = {"min": -0.2, "max": 0.6, "palette": ["#0000ff", "#ffff00", "#ff0000"]}
return code, vis
if re.search(r"\bndvi\b", msg):
code = f"""region = {region_fc}
image = {s2}.normalizedDifference(['B8','B4']).clipToCollection(region)"""
vis = {"min": -0.1, "max": 0.9, "palette": ["#d73027", "#fee08b", "#1a9850"]}
return code, vis
if re.search(r"\bndwi\b", msg):
code = f"""region = {region_fc}
image = {s2}.normalizedDifference(['B3','B8']).clipToCollection(region)"""
vis = {"min": -0.3, "max": 0.5, "palette": ["#ffffcc", "#a1dab4", "#41b6c4", "#2c7fb8", "#253494"]}
return code, vis
if re.search(r"\bsentinel[\s-]?1\b|\bsar\b", msg):
code = f"""region = {region_fc}
image = ee.ImageCollection('COPERNICUS/S1_GRD').filterBounds(region).filter(ee.Filter.eq('instrumentMode', 'IW')).filter(ee.Filter.listContains('transmitterReceiverPolarisation', 'VV')).filterDate('2024-03-01','2024-04-01').select('VV').mean().clipToCollection(region)"""
vis = {"min": -25, "max": 0}
return code, vis
return None