app.py

# ============================================================
# GeoMate: Hugging Face Gradio App
# AOI Soil / Flood / Seismic Report Generator
# ============================================================

import os, json
import ee
import gradio as gr
from datetime import datetime
import pandas as pd
from reportlab.lib.pagesizes import A4
from reportlab.platypus import Paragraph, SimpleDocTemplate, Spacer, Table, TableStyle
from reportlab.lib.styles import getSampleStyleSheet
from reportlab.lib import colors

# ---------------------------
# Authenticate Earth Engine
# ---------------------------
# IMPORTANT: In your HF Space Settings → Secrets:
#   EE_SERVICE_ACCOUNT = "your-service-account@project.iam.gserviceaccount.com"
#   EE_PRIVATE_KEY     = contents of your service account JSON
# ---------------------------

# ---------------------------
# Authenticate Earth Engine (HF Secrets)
# ---------------------------
# ---------------------------
# Authenticate Earth Engine (with HF Secrets)
# ---------------------------
import ee, os, tempfile

service_account = os.environ.get("EE_SERVICE_ACCOUNT")   # service account email
key_json = os.environ.get("EE_PRIVATE_KEY")              # the JSON text as stored in HF Secrets

if service_account and key_json:
    # Write the JSON string to a temporary file
    with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
        f.write(key_json)
        temp_key_path = f.name

    # Authenticate using the file path
    creds = ee.ServiceAccountCredentials(service_account, temp_key_path)
    ee.Initialize(creds)
else:
    raise RuntimeError("Earth Engine authentication failed. Did you set HF secrets?")
# Helper Functions
# ---------------------------
def _fmt_pct(x):
    try:
        return f"{float(x):.1f}"
    except Exception:
        return "NA"

def compute_soil_stats(aoi: ee.Geometry) -> dict:
    stats = {"source": "ISRIC SoilGrids 250m (0–5 cm)", "sand_pct": None, "silt_pct": None, "clay_pct": None}
    try:
        soil = ee.ImageCollection("projects/soilgrids-isric/soilgrids250m").first()
        bands = soil.select(["sand_0-5cm_mean","silt_0-5cm_mean","clay_0-5cm_mean"])
        bands_pct = bands.divide(10).rename(["sand_pct","silt_pct","clay_pct"])
        result = bands_pct.reduceRegion(ee.Reducer.mean(), geometry=aoi, scale=250, maxPixels=1e9)
        stats["sand_pct"] = result.get("sand_pct").getInfo()
        stats["silt_pct"] = result.get("silt_pct").getInfo()
        stats["clay_pct"] = result.get("clay_pct").getInfo()
    except Exception as e:
        stats["error"] = str(e)
    return stats

def compute_flood_stats(aoi: ee.Geometry) -> dict:
    stats = {"source": "JRC Global Surface Water (1984–2020)", "mean_occurrence_pct": None, "p75_occurrence_pct": None}
    try:
        water = ee.Image("JRC/GSW1_4/GlobalSurfaceWater").select("occurrence")
        reducer = ee.Reducer.mean().combine(reducer2=ee.Reducer.percentile([75]), sharedInputs=True)
        res = water.reduceRegion(reducer, geometry=aoi, scale=30, maxPixels=1e9)
        stats["mean_occurrence_pct"] = res.get("occurrence_mean").getInfo()
        stats["p75_occurrence_pct"] = res.get("occurrence_p75").getInfo()
    except Exception as e:
        stats["error"] = str(e)
    return stats

def compute_seismic_stats(aoi: ee.Geometry) -> dict:
    stats = {"source": None, "pga_g": None, "eq_count_10yr_M4p": None}
    try:
        gem_pga = ee.Image("GEM/2018/PGA_10in50")
        pga = gem_pga.select(0)
        res = pga.reduceRegion(ee.Reducer.mean(), geometry=aoi, scale=1000, maxPixels=1e9)
        stats["pga_g"] = res.get(pga.bandNames().get(0)).getInfo()
        stats["source"] = "GEM GSHM 2018 – PGA (10% in 50y)"
        return stats
    except Exception:
        pass
    try:
        end = ee.Date(datetime.now().strftime("%Y-%m-%d"))
        start = end.advance(-10, "year")
        fc = ee.FeatureCollection("USGS/earthquakes").filter(ee.Filter.date(start, end)).filter(ee.Filter.gte("mag", 4))
        count = fc.filterBounds(aoi).size().getInfo()
        stats["eq_count_10yr_M4p"] = int(count)
        stats["source"] = "USGS earthquake catalog – counts last 10y, M≥4"
    except Exception as e:
        stats["source"] = "Seismic hazard unavailable"
        stats["error"] = str(e)
    return stats

def classify_texture(sand, silt, clay):
    if any(v is None for v in [sand, silt, clay]):
        return "Unknown"
    sand, silt, clay = float(sand), float(silt), float(clay)
    if clay >= 40: return "Clay/Clayey"
    if sand >= 70 and clay < 15: return "Sandy"
    if silt >= 70 and clay < 12: return "Silty"
    if 20 <= clay <= 35 and silt < 28: return "Clay Loam"
    if 7 <= clay <= 27 and 28 <= silt <= 50 and sand <= 52: return "Loam"
    return "Mixed"

def summarize(lat, lon, buffer_m):
    point = ee.Geometry.Point([lon, lat])
    geom = point.buffer(buffer_m).bounds()
    soil = compute_soil_stats(geom)
    flood = compute_flood_stats(geom)
    seis = compute_seismic_stats(geom)
    texture = classify_texture(soil.get("sand_pct"), soil.get("silt_pct"), soil.get("clay_pct"))
    return {"soil": soil, "flood": flood, "seismic": seis, "texture_class": texture}

# ---------------------------
# PDF + CSV Generation
# ---------------------------
def generate_pdf(aoi_summary: dict, out_path: str, aoi_name: str = "Selected Area"):
    doc = SimpleDocTemplate(out_path, pagesize=A4)
    styles = getSampleStyleSheet()
    story = []
    story.append(Paragraph(f"GeoMate Site Report – {aoi_name}", styles['Title']))
    story.append(Spacer(1, 12))
    story.append(Paragraph(f"Generated: {datetime.utcnow().strftime('%Y-%m-%d %H:%M UTC')}", styles['Normal']))
    story.append(Spacer(1, 12))

    # Soil
    soil = aoi_summary['soil']
    soil_rows = [["Metric","Value","Units"],
                 ["Sand (0–5 cm)", _fmt_pct(soil.get('sand_pct')),"%"],
                 ["Silt (0–5 cm)", _fmt_pct(soil.get('silt_pct')),"%"],
                 ["Clay (0–5 cm)", _fmt_pct(soil.get('clay_pct')),"%"]]
    soil_table = Table(soil_rows, hAlign='LEFT')
    soil_table.setStyle(TableStyle([('BACKGROUND',(0,0),(-1,0),colors.lightgrey),('GRID',(0,0),(-1,-1),0.5,colors.grey)]))
    story.append(Paragraph("Soil", styles['Heading2'])); story.append(soil_table)

    # Flood
    flood = aoi_summary['flood']
    flood_rows = [["Metric","Value","Units"],
                  ["Mean Occurrence", _fmt_pct(flood.get('mean_occurrence_pct')),"%"],
                  ["75th Percentile", _fmt_pct(flood.get('p75_occurrence_pct')),"%"]]
    flood_table = Table(flood_rows, hAlign='LEFT')
    flood_table.setStyle(TableStyle([('BACKGROUND',(0,0),(-1,0),colors.lightgrey),('GRID',(0,0),(-1,-1),0.5,colors.grey)]))
    story.append(Paragraph("Flood Proxy", styles['Heading2'])); story.append(flood_table)

    # Seismic
    seis = aoi_summary['seismic']
    seis_rows = [["Metric","Value","Units/Notes"],
                 ["PGA (10% in 50y)", _fmt_pct(seis.get('pga_g')),"g"],
                 ["Earthquakes (10y, M≥4)", str(seis.get('eq_count_10yr_M4p') or "NA"),"count"]]
    seis_table = Table(seis_rows, hAlign='LEFT')
    seis_table.setStyle(TableStyle([('BACKGROUND',(0,0),(-1,0),colors.lightgrey),('GRID',(0,0),(-1,-1),0.5,colors.grey)]))
    story.append(Paragraph("Seismic", styles['Heading2'])); story.append(seis_table)

    story.append(Spacer(1, 12))
    story.append(Paragraph(f"Texture Class: <b>{aoi_summary.get('texture_class')}</b>", styles['Heading3']))
    doc.build(story)

def export_csv(aoi_summary: dict, out_csv: str):
    soil, flood, seis = aoi_summary['soil'], aoi_summary['flood'], aoi_summary['seismic']
    row = {'sand_pct':soil.get('sand_pct'),'silt_pct':soil.get('silt_pct'),'clay_pct':soil.get('clay_pct'),
           'mean_water_occurrence_pct':flood.get('mean_occurrence_pct'),'p75_water_occurrence_pct':flood.get('p75_occurrence_pct'),
           'pga_g':seis.get('pga_g'),'eq_count_10yr_M4p':seis.get('eq_count_10yr_M4p')}
    df = pd.DataFrame([row]); df.to_csv(out_csv,index=False); return df

# ---------------------------
# Gradio App
# ---------------------------
def run(lat, lon, buffer_m, name):
    try:
        summary = summarize(lat, lon, buffer_m)
        pdf_path, csv_path = "GeoMate_Report.pdf", "GeoMate_Data.csv"
        generate_pdf(summary, pdf_path, aoi_name=name)
        export_csv(summary, csv_path)
        return f"✅ Report generated for {name}", pdf_path, csv_path
    except Exception as e:
        return f"❌ Error: {str(e)}", None, None

with gr.Blocks(theme=gr.themes.Soft()) as demo:
    gr.Markdown("# 🌍 GeoMate – Site Soil/Flood/Seismic Report")
    with gr.Row():
        lat = gr.Number(label="Latitude", value=30.3753)
        lon = gr.Number(label="Longitude", value=69.3451)
    buffer_m = gr.Slider(0, 20000, value=1000, step=100, label="Buffer (m)")
    name = gr.Textbox(label="Site Name", value="My Site")
    run_btn = gr.Button("Generate Report")
    status = gr.Label()
    pdf_out = gr.File(label="Download PDF")
    csv_out = gr.File(label="Download CSV")

    run_btn.click(run, [lat, lon, buffer_m, name], [status, pdf_out, csv_out])

if __name__ == "__main__":
    demo.launch()