import streamlit as st import google.generativeai as genai from geopy.geocoders import Nominatim # from geopy.exc import GeocoderTimedOut, GeocoderUnavailable # Not explicitly caught, requests.timeout handles import folium from streamlit_folium import st_folium import pandas as pd import requests import re import os from datetime import datetime, timedelta # --- Page Configuration --- st.set_page_config( layout="wide", page_title="Landslide Factor Explorer | India", # More professional title page_icon="๐๏ธ", # Favicon initial_sidebar_state="collapsed" ) # Custom CSS for enhanced UI st.markdown(""" """, unsafe_allow_html=True) # --- Gemini API Key Handling --- API_KEY = os.getenv("GOOGLE_API_KEY", "AIzaSyDkiYr-eSkqIXpZ1fHlik_YFsFtfQoFi0w") # Use yours, or allow env var if not API_KEY or API_KEY == "YOUR_API_KEY_HERE": # Default check st.sidebar.error("๐ด GOOGLE_API_KEY not set. Please set it as an environment variable or enter below.") API_KEY = st.sidebar.text_input("Enter your Gemini API Key:", type="password", key="api_key_input_explorer_v4") if API_KEY and API_KEY != "YOUR_API_KEY_HERE": try: genai.configure(api_key=API_KEY) except Exception as e: st.error(f"Error configuring Gemini API: {e}") st.stop() else: st.error("๐ด Gemini API Key is required to run this application.") st.stop() # --- Services & Constants --- geolocator = Nominatim(user_agent="india_landslide_explorer_v4") FORECAST_DAYS = 14 SEISMIC_RADIUS_KM = 150 SEISMIC_MIN_MAGNITUDE = 4.0 SEISMIC_DAYS_AGO = 30 # --- Session State Initialization --- if 'map_center_india' not in st.session_state: st.session_state.map_center_india = [20.5937, 78.9629] if 'map_zoom_india' not in st.session_state: st.session_state.map_zoom_india = 4 if 'selected_lat_lon' not in st.session_state: st.session_state.selected_lat_lon = None if 'location_name' not in st.session_state: st.session_state.location_name = "" if 'exploration_output' not in st.session_state: st.session_state.exploration_output = {"kpi_data": {}, "detailed_text": {}} if 'api_data_fetched' not in st.session_state: st.session_state.api_data_fetched = {} if 'is_fetching_data' not in st.session_state: st.session_state.is_fetching_data = False # --- Helper Functions (get_elevation, fetch_rainfall_data, fetch_seismic_data, reverse_geocode) --- def get_elevation(lat, lon): try: url = f"https://api.open-meteo.com/v1/elevation?latitude={lat}&longitude={lon}" response = requests.get(url, timeout=10) response.raise_for_status() data = response.json() return data['elevation'][0] except Exception: return "N/A" def fetch_rainfall_data(lat, lon, forecast_days_count=FORECAST_DAYS): url = "https://api.open-meteo.com/v1/forecast" params = { "latitude": lat, "longitude": lon, "daily": "precipitation_sum,precipitation_hours", "current": "precipitation,rain,showers,snowfall", "forecast_days": forecast_days_count, "timezone": "auto" } try: response = requests.get(url, params=params, timeout=15) response.raise_for_status() data = response.json() current_data = data.get("current", {}) daily_data = data.get("daily", {}) df_daily_forecast = pd.DataFrame() if daily_data.get("time") and daily_data.get("precipitation_sum"): df_daily_forecast = pd.DataFrame({ "Date": pd.to_datetime(daily_data["time"]), "Rainfall_Sum (mm)": daily_data["precipitation_sum"], "Precipitation_Hours (hrs)": daily_data.get("precipitation_hours", [0]*len(daily_data["time"])) }).set_index("Date") return { "current_precipitation_mm": current_data.get("precipitation", "N/A"), "current_rain_mm": current_data.get("rain", "N/A"), "current_showers_mm": current_data.get("showers", "N/A"), "current_snowfall_cm": current_data.get("snowfall", "N/A"), "daily_forecast_df": df_daily_forecast } except Exception as e: st.toast(f"Weather fetch error: {e}", icon="๐ฆ๏ธ") return {"current_precipitation_mm": "Error", "daily_forecast_df": pd.DataFrame()} def fetch_seismic_data(lat, lon, radius_km=SEISMIC_RADIUS_KM, min_mag=SEISMIC_MIN_MAGNITUDE, days_ago=SEISMIC_DAYS_AGO): try: end_time = datetime.utcnow() start_time = end_time - timedelta(days=days_ago) url = "https://earthquake.usgs.gov/fdsnws/event/1/query" params = { "format": "geojson", "latitude": lat, "longitude": lon, "maxradiuskm": radius_km, "minmagnitude": min_mag, "starttime": start_time.strftime("%Y-%m-%dT%H:%M:%S"), "endtime": end_time.strftime("%Y-%m-%dT%H:%M:%S"), "orderby": "time" } response = requests.get(url, params=params, timeout=15) response.raise_for_status() data = response.json() earthquakes = [] for feature in data.get("features", []): props = feature.get("properties", {}); geom = feature.get("geometry", {}) if props and geom and props.get("mag") is not None and geom.get("coordinates"): earthquakes.append({ "place": props.get("place", "Unknown"), "magnitude": props.get("mag"), "time": datetime.utcfromtimestamp(props.get("time") / 1000).strftime('%Y-%m-%d %H:%M UTC'), "depth_km": geom.get("coordinates")[2] if len(geom.get("coordinates", [])) > 2 else "N/A", "url": props.get("url")}) return earthquakes except Exception as e: st.toast(f"Seismic fetch error: {e}", icon="๐"); return [] def reverse_geocode(lat, lon): try: location = geolocator.reverse((lat, lon), exactly_one=True, timeout=10) return location.address if location else "Unknown location" except Exception: return "Could not determine address" # --- Gemini Prompt and Parsing V4 --- def get_gemini_exploration_v4(location_name, lat_lon, api_data): model = genai.GenerativeModel('gemini-1.5-flash-latest') elevation_str = f"{api_data.get('elevation_m', 'N/A')}" weather_data = api_data.get('weather', {}) current_precip_str = f"{weather_data.get('current_precipitation_mm', 'N/A')}" forecast_df = weather_data.get('daily_forecast_df') forecast_summary_str = "N/A" if forecast_df is not None and not forecast_df.empty: summary_days = min(7, len(forecast_df)) forecast_days_summary = [f"Day {i+1} ({forecast_df.index[i].strftime('%Y-%m-%d')}): {forecast_df['Rainfall_Sum (mm)'].iloc[i] if pd.notna(forecast_df['Rainfall_Sum (mm)'].iloc[i]) else 'N/A'} mm" for i in range(summary_days)] forecast_summary_str = "; ".join(forecast_days_summary) if forecast_days_summary else "No forecast data." elif isinstance(forecast_df, pd.DataFrame) and forecast_df.empty: forecast_summary_str = "Forecast data empty/unavailable." seismic_events = api_data.get('seismic', []) seismic_summary_str = "No significant recent seismic activity reported by USGS in the vicinity." if seismic_events: event_strs = [f"Mag {event['magnitude']} event near {event['place'].split('of')[-1].strip() if 'of' in event['place'] else event['place']}, on {event['time'].split(' ')[0]}" for event in seismic_events[:2]] seismic_summary_str = "Recent Seismic Activity: " + "; ".join(event_strs) if len(seismic_events) > 2: seismic_summary_str += f"; and {len(seismic_events)-2} more similar events." prompt = f""" You are an AI assistant for an advanced educational landslide factor exploration tool focused on INDIA (Version 4 - Visual Focus). This tool DOES NOT use specific user observations of local conditions. Your discussion will be based on the provided general location, fetched API data (elevation, weather, recent seismic activity), and your broad knowledge of Indian geography, geology, land cover, and climate. This is strictly for educational purposes to explore POTENTIAL factors for a TYPE of area, NOT a real-time prediction or specific site assessment. Location & Fetched Data: - Approximate Location Name: "{location_name}" (Lat/Lon: {lat_lon[0]:.4f}, {lat_lon[1]:.4f}) - Elevation: {elevation_str} meters - Current Precipitation Summary: {current_precip_str} mm - Rainfall Forecast Summary (e.g., next 7 days): {forecast_summary_str} - Recent Seismic Activity Summary (within ~{SEISMIC_RADIUS_KM}km, M{SEISMIC_MIN_MAGNITUDE}+, last {SEISMIC_DAYS_AGO} days): {seismic_summary_str} Task: Based on the above information and your general knowledge, please provide the following structured exploration. First, provide specific KPI data, then provide the detailed textual explanations. KPI_DATA_START GENERAL_SUSCEPTIBILITY_LEVEL: [Provide one single category: Low / Moderate / High / Very High - based on typical regional characteristics for this type of area] RAINFALL_IMPACT_ASSESSMENT: [Provide one single category: Low Concern / Moderate Concern / Significant Concern / High Concern - regarding its potential to trigger landslides in this type of area given the forecast and typical seasonal patterns] SEISMIC_IMPACT_ASSESSMENT: [Provide one single category: Negligible / Low Potential / Moderate Potential / Significant Potential - as a landslide trigger in this type of area, considering reported activity and general regional seismicity] TOP_HYPOTHETICAL_LANDSLIDE_TYPES: [List up to 3 most common/likely landslide types for similar regions in India, separated by commas, e.g., Debris Flow, Rockfall, Rotational Slump] KEY_CONTRIBUTING_FACTORS_POINTS: - [Brief point (max 10 words) on a key natural factor, e.g., Steep topography typical of the region] - [Brief point (max 10 words) on a key human-induced factor, e.g., Unplanned construction if prevalent in similar areas] - [Brief point (max 10 words) on another critical factor, e.g., Intense monsoon rainfall patterns] TYPICAL_LAND_COVER_INFERRED: [Describe in 1-3 words the most typical general land cover you infer for this type of region, e.g., Forested Slopes, Agricultural Terraces, Urbanizing Hillsides, Barren Rocky Terrain] KPI_DATA_END Now, provide the detailed textual explanations, structured with the following headers: HEADER_KEY_INSIGHTS_SUMMARY Provide 2-3 bullet points elaborating on the most critical potential landslide-related insights or considerations for this TYPE of area in India, building upon the KPI data. HEADER_SUSCEPTIBILITY_DISCUSSION A. General Discussion of Landslide Susceptibility for this TYPE of Area: (Elaborate on the GENERAL_SUSCEPTIBILITY_LEVEL. Discuss typical geological features, soil types, or topographical characteristics for this type of area. Ensure the output for GENERAL_SUSCEPTIBILITY_LEVEL provided in KPI_DATA_START is consistent with this discussion and includes "General Susceptibility for this type of area: " before the level, e.g., "General Susceptibility for this type of area: Moderate"). HEADER_DATA_ANALYSIS B. Analysis of Fetched Data in Context of Potential Landslides: (Elaborate on RAINFALL_IMPACT_ASSESSMENT and SEISMIC_IMPACT_ASSESSMENT. Discuss how fetched rainfall, elevation, and seismic data influence landslide potential, considering seasonal patterns and regional context). HEADER_HYPOTHETICAL_FACTORS C. Hypothetical Contributing Factors (Beyond Fetched Data): (Elaborate on KEY_CONTRIBUTING_FACTORS_POINTS and TYPICAL_LAND_COVER_INFERRED. Discuss typical land cover and other natural/human-induced factors common to such regions in India). HEADER_COMMON_LANDSLIDE_TYPES D. Common Landslide Types in Similar Indian Regions: (Elaborate on TOP_HYPOTHETICAL_LANDSLIDE_TYPES. Describe their characteristics and triggers relevant to the scenario). HEADER_CRITICAL_LOCAL_DATA_NEED E. Critical Importance of Local Site-Specific Data (Emphasize very strongly!): (Explain why absence of local observations makes specific risk assessment impossible. Detail necessary local data). HEADER_AWARENESS_PREPAREDNESS F. General Awareness & Preparedness Ideas (India Context): (Suggest general, non-site-specific educational points on landslide awareness/preparedness). HEADER_OFFICIAL_RESOURCES G. Official Indian Resources & Further Learning: (List key Indian government agencies and information sources). Structure your response exactly with the specified KPI_DATA_START/END and HEADER_ SECTION NAMES. Maintain an educational tone. Explicitly and repeatedly state the limitations. """ try: response = model.generate_content(prompt) return response.text except Exception as e: st.error(f"Error communicating with Gemini API: {e}") return None def parse_gemini_output_v4(text): if not text: return {"kpi_data": {}, "detailed_text": {}} kpi_data = {} default_kpi_values = { "GENERAL_SUSCEPTIBILITY_LEVEL": "N/A", "RAINFALL_IMPACT_ASSESSMENT": "N/A", "SEISMIC_IMPACT_ASSESSMENT": "N/A", "TOP_HYPOTHETICAL_LANDSLIDE_TYPES": "Not specified", "KEY_CONTRIBUTING_FACTORS_POINTS": [], "TYPICAL_LAND_COVER_INFERRED": "N/A" } kpi_data.update(default_kpi_values) detailed_text_sections_map = { "HEADER_KEY_INSIGHTS_SUMMARY": "๐ Key Insights Summary", "HEADER_SUSCEPTIBILITY_DISCUSSION": "๐ง General Susceptibility Discussion", "HEADER_DATA_ANALYSIS": "๐ Analysis of Fetched Data", "HEADER_HYPOTHETICAL_FACTORS": "๐ค Contributing Factors", "HEADER_COMMON_LANDSLIDE_TYPES": "๐๏ธ Common Landslide Types", "HEADER_CRITICAL_LOCAL_DATA_NEED": "โCRUCIAL: Need for Local Site-Specific Dataโ", "HEADER_AWARENESS_PREPAREDNESS": "๐ก General Awareness & Preparedness", "HEADER_OFFICIAL_RESOURCES": "๐ฎ๐ณ Official Resources & Further Learning" } parsed_detailed_text = {display_name: [] for _, display_name in detailed_text_sections_map.items()} in_kpi_section = False current_detailed_section_key = None key_factors_collecting = False kpi_regex_map = { "GENERAL_SUSCEPTIBILITY_LEVEL": re.compile(r"GENERAL_SUSCEPTIBILITY_LEVEL:\s*(.+)", re.IGNORECASE), "RAINFALL_IMPACT_ASSESSMENT": re.compile(r"RAINFALL_IMPACT_ASSESSMENT:\s*(.+)", re.IGNORECASE), "SEISMIC_IMPACT_ASSESSMENT": re.compile(r"SEISMIC_IMPACT_ASSESSMENT:\s*(.+)", re.IGNORECASE), "TOP_HYPOTHETICAL_LANDSLIDE_TYPES": re.compile(r"TOP_HYPOTHETICAL_LANDSLIDE_TYPES:\s*(.+)", re.IGNORECASE), "TYPICAL_LAND_COVER_INFERRED": re.compile(r"TYPICAL_LAND_COVER_INFERRED:\s*(.+)", re.IGNORECASE), } for line in text.splitlines(): line_strip = line.strip() if not line_strip: continue if line_strip == "KPI_DATA_START": in_kpi_section = True; continue if line_strip == "KPI_DATA_END": in_kpi_section = False; key_factors_collecting = False; continue if in_kpi_section: matched_specific_kpi = False for key, pattern in kpi_regex_map.items(): match = pattern.match(line_strip) if match: kpi_data[key] = match.group(1).strip() matched_specific_kpi = True; break if matched_specific_kpi: continue if line_strip.startswith("KEY_CONTRIBUTING_FACTORS_POINTS:"): key_factors_collecting = True; kpi_data["KEY_CONTRIBUTING_FACTORS_POINTS"] = [] # Reset for new parse continue if key_factors_collecting and line_strip.startswith("-"): kpi_data["KEY_CONTRIBUTING_FACTORS_POINTS"].append(line_strip.lstrip("- ").strip()) continue found_new_header = False for header_key_from_prompt, display_name in detailed_text_sections_map.items(): if line_strip == header_key_from_prompt: current_detailed_section_key = display_name found_new_header = True; break if not found_new_header and current_detailed_section_key: parsed_detailed_text[current_detailed_section_key].append(line) final_detailed_text = {k: "\n".join(v).strip() for k, v in parsed_detailed_text.items()} return {"kpi_data": kpi_data, "detailed_text": final_detailed_text} # --- UI Rendering with Enhanced Styling --- st.markdown('
Welcome! Begin by selecting a location on the map or using the search bar to find a specific place in India. The tool will then fetch publicly available data (elevation, weather forecast, recent seismic activity) for the chosen area. After data retrieval, you can initiate an AI-powered exploration. The AI will provide a generalized discussion on potential landslide susceptibility and contributing factors relevant to that type of area in India, based on the fetched data and its broad geographical knowledge.