# app.py โ€” GeoMate V2 (single-file) # Save this as app.py in your HuggingFace Space (or local folder) # 0) Page config (must be first Streamlit command) import streamlit as st st.set_page_config(page_title="GeoMate V2", page_icon="๐ŸŒ", layout="wide", initial_sidebar_state="expanded") # 1) Standard imports import os import io import json import math import base64 import tempfile from datetime import datetime from typing import Dict, Any, Tuple, List, Optional # Visualization & PDF import matplotlib.pyplot as plt from reportlab.lib.pagesizes import A4 from reportlab.lib import colors from reportlab.lib.units import mm from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle, Image as RLImage, PageBreak from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle # Optional imports handled gracefully try: import geemap import ee EE_AVAILABLE = True except Exception: EE_AVAILABLE = False try: from fpdf import FPDF FPDF_AVAILABLE = True except Exception: FPDF_AVAILABLE = False try: import faiss FAISS_AVAILABLE = True except Exception: FAISS_AVAILABLE = False try: import pytesseract from PIL import Image OCR_AVAILABLE = True except Exception: OCR_AVAILABLE = False # Groq client import โ€” we will require key try: from groq import Groq GROQ_AVAILABLE = True except Exception: GROQ_AVAILABLE = False # 2) Secrets check (strict) REQUIRED_SECRETS = ["GROQ_API_KEY", "SERVICE_ACCOUNT", "EARTH_ENGINE_KEY"] missing = [s for s in REQUIRED_SECRETS if not os.environ.get(s)] if missing: st.sidebar.error(f"Missing required secrets: {', '.join(missing)}. Please add these to your HF Space secrets.") st.error("Required secrets missing. Please set GROQ_API_KEY, SERVICE_ACCOUNT, and EARTH_ENGINE_KEY in Secrets and reload the app.") st.stop() # If Groq lib missing, still stop because user requested Groq usage if not GROQ_AVAILABLE: st.sidebar.error("Python package 'groq' not installed. Add it to requirements.txt and redeploy.") st.error("Missing required library 'groq'. Please add to requirements and redeploy.") st.stop() # 3) Global constants & helper functions MAX_SITES = 4 # ---------------------------- # Soil Recognizer Page (Integrated 6-Class ResNet18) # ---------------------------- import torch import torch.nn as nn import torchvision.models as models import torchvision.transforms as T from PIL import Image import streamlit as st # ---------------------------- # Load Soil Model (6 Classes) # ---------------------------- @st.cache_resource def load_soil_model(path="soil_best_model.pth"): device = torch.device("cuda" if torch.cuda.is_available() else "cpu") try: model = models.resnet18(pretrained=False) num_ftrs = model.fc.in_features model.fc = nn.Linear(num_ftrs, 6) # 6 soil classes # Load checkpoint state_dict = torch.load(path, map_location=device) model.load_state_dict(state_dict) model = model.to(device) model.eval() return model, device except Exception as e: st.error(f"โš ๏ธ Could not load soil model: {e}") return None, device soil_model, device = load_soil_model() # ---------------------------- # Soil Classes & Transform # ---------------------------- SOIL_CLASSES = ["Clay", "Gravel", "Loam", "Peat", "Sand", "Silt"] transform = T.Compose([ T.Resize((224, 224)), T.ToTensor(), T.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]) ]) # ---------------------------- # Prediction Function # ---------------------------- def predict_soil(img: Image.Image): if soil_model is None: return "Model not loaded", {} img = img.convert("RGB") inp = transform(img).unsqueeze(0).to(device) with torch.no_grad(): logits = soil_model(inp) probs = torch.softmax(logits[0], dim=0) top_idx = torch.argmax(probs).item() predicted_class = SOIL_CLASSES[top_idx] result = {SOIL_CLASSES[i]: float(probs[i]) for i in range(len(SOIL_CLASSES))} return predicted_class, result # ---------------------------- # Soil Recognizer Page # ---------------------------- def soil_recognizer_page(): st.header("๐Ÿ–ผ๏ธ Soil Recognizer (ResNet18)") site = st.session_state["sites"] # your existing site getter if site is None: st.warning("โš ๏ธ No active site selected. Please add or select a site from the sidebar.") return uploaded = st.file_uploader("Upload soil image", type=["jpg", "jpeg", "png"]) if uploaded is not None: img = Image.open(uploaded) st.image(img, caption="Uploaded soil image", use_column_width=True) predicted_class, confidence_scores = predict_soil(img) st.success(f"โœ… Predicted: **{predicted_class}**") st.subheader("Confidence Scores") for cls, score in confidence_scores.items(): st.write(f"{cls}: {score:.2%}") if st.button("Save to site"): # Save predicted soil class into Soil Profile field st.session_state["sites"][st.session_state["active_site"]]["Soil Class"] = predicted_class st.session_state["sites"][st.session_state["active_site"]]["Soil Recognizer Confidence"] = confidence_scores[predicted_class] save_active_site(site) st.success("Saved prediction to active site memory.") # Pre-defined dropdown text mappings (as you requested) โ€” exact text with mapping numbers for logic backend DILATANCY_OPTIONS = [ "1. Quick to slow", "2. None to very slow", "3. Slow", "4. Slow to none", "5. None", "6. Null?" ] TOUGHNESS_OPTIONS = [ "1. None", "2. Medium", "3. Slight?", "4. Slight to Medium?", "5. High", "6. Null?" ] DRY_STRENGTH_OPTIONS = [ "1. None to slight", "2. Medium to high", "3. Slight to Medium", "4. High to very high", "5. Null?" ] # Map option text to numeric codes used in your USCS logic DILATANCY_MAP = {DILATANCY_OPTIONS[i]: i+1 for i in range(len(DILATANCY_OPTIONS))} TOUGHNESS_MAP = {TOUGHNESS_OPTIONS[i]: i+1 for i in range(len(TOUGHNESS_OPTIONS))} DRY_STRENGTH_MAP = {DRY_STRENGTH_OPTIONS[i]: i+1 for i in range(len(DRY_STRENGTH_OPTIONS))} # Engineering characteristics dictionary (expanded earlier; trimmed to representative entries but detailed) ENGINEERING_CHARACTERISTICS = { "Gravel": { "Settlement": "None", "Quicksand": "Impossible", "Frost-heaving": "None", "Groundwater_lowering": "Possible", "Cement_grouting": "Possible", "Silicate_bitumen_injections": "Unsuitable", "Compressed_air": "Possible (see notes)" }, "Coarse sand": { "Settlement": "None", "Quicksand": "Impossible", "Frost-heaving": "None", "Groundwater_lowering": "Possible", "Cement_grouting": "Possible only if very coarse", "Silicate_bitumen_injections": "Suitable", "Compressed_air": "Suitable" }, "Medium sand": { "Settlement": "None", "Quicksand": "Unlikely", "Frost-heaving": "None", "Groundwater_lowering": "Suitable", "Cement_grouting": "Impossible", "Silicate_bitumen_injections": "Suitable", "Compressed_air": "Suitable" }, "Fine sand": { "Settlement": "None", "Quicksand": "Liable", "Frost-heaving": "None", "Groundwater_lowering": "Suitable", "Cement_grouting": "Impossible", "Silicate_bitumen_injections": "Not possible in very fine sands", "Compressed_air": "Suitable" }, "Silt": { "Settlement": "Occurs", "Quicksand": "Liable (very coarse silts may behave differently)", "Frost-heaving": "Occurs", "Groundwater_lowering": "Generally not suitable (electro-osmosis possible)", "Cement_grouting": "Impossible", "Silicate_bitumen_injections": "Impossible", "Compressed_air": "Suitable" }, "Clay": { "Settlement": "Occurs", "Quicksand": "Impossible", "Frost-heaving": "None", "Groundwater_lowering": "Impossible (generally)", "Cement_grouting": "Only in stiff fissured clay", "Silicate_bitumen_injections": "Impossible", "Compressed_air": "Used for support only in special cases" } } # USCS & AASHTO verbatim logic (function) from math import floor def classify_uscs_aashto(inputs: Dict[str, Any]) -> Tuple[str, str, int, Dict[str, str], str]: """ Verbatim USCS + AASHTO classifier based on the logic you supplied. inputs: dictionary expected keys: opt: 'y' or 'n' P2 (float): % passing #200 (0.075 mm) P4 (float): % passing #4 (4.75 mm) D60, D30, D10 (float mm) - can be 0 if unknown LL, PL (float) nDS, nDIL, nTG (int) mapped from dropdowns Returns: result_text (markdown), aashto_str, GI, engineering_characteristics (dict), uscs_str """ opt = str(inputs.get("opt","n")).lower() if opt == 'y': uscs = "Pt" uscs_expl = "Peat / organic soil โ€” compressible, high organic content; poor engineering properties for load-bearing without special treatment." aashto = "Organic (special handling)" GI = 0 chars = {"summary":"Highly organic peat โ€” large settlement, low strength, not suitable for foundations without improvement."} res_text = f"According to USCS, the soil is **{uscs}** โ€” {uscs_expl}\n\nAccording to AASHTO, the soil is **{aashto}**." return res_text, aashto, GI, chars, uscs # parse numeric inputs with defaults P2 = float(inputs.get("P2", 0.0)) P4 = float(inputs.get("P4", 0.0)) D60 = float(inputs.get("D60", 0.0)) D30 = float(inputs.get("D30", 0.0)) D10 = float(inputs.get("D10", 0.0)) LL = float(inputs.get("LL", 0.0)) PL = float(inputs.get("PL", 0.0)) PI = LL - PL if (LL is not None and PL is not None) else 0.0 Cu = (D60 / D10) if (D10 > 0 and D60 > 0) else 0.0 Cc = ((D30 ** 2) / (D10 * D60)) if (D10 > 0 and D30 > 0 and D60 > 0) else 0.0 uscs = "Unknown" uscs_expl = "" if P2 <= 50: # Coarse-Grained Soils if P4 <= 50: # Gravels if Cu != 0 and Cc != 0: if Cu >= 4 and 1 <= Cc <= 3: uscs = "GW"; uscs_expl = "Well-graded gravel (good engineering properties, high strength, good drainage)." else: uscs = "GP"; uscs_expl = "Poorly-graded gravel (less favorable gradation)." else: if PI < 4 or PI < 0.73 * (LL - 20): uscs = "GM"; uscs_expl = "Silty gravel (fines may reduce permeability and strength)." elif PI > 7 and PI > 0.73 * (LL - 20): uscs = "GC"; uscs_expl = "Clayey gravel (clayey fines increase plasticity)." else: uscs = "GM-GC"; uscs_expl = "Gravel with mixed silt/clay fines." else: # Sands if Cu != 0 and Cc != 0: if Cu >= 6 and 1 <= Cc <= 3: uscs = "SW"; uscs_expl = "Well-graded sand (good compaction and drainage)." else: uscs = "SP"; uscs_expl = "Poorly-graded sand (uniform or gap-graded)." else: if PI < 4 or PI <= 0.73 * (LL - 20): uscs = "SM"; uscs_expl = "Silty sand (fines are low-plasticity silt)." elif PI > 7 and PI > 0.73 * (LL - 20): uscs = "SC"; uscs_expl = "Clayey sand (clayey fines present; higher plasticity)." else: uscs = "SM-SC"; uscs_expl = "Transition between silty sand and clayey sand." else: # Fine-Grained Soils nDS = int(inputs.get("nDS", 5)) nDIL = int(inputs.get("nDIL", 6)) nTG = int(inputs.get("nTG", 6)) if LL < 50: if 20 <= LL < 50 and PI <= 0.73 * (LL - 20): if nDS == 1 or nDIL == 3 or nTG == 3: uscs = "ML"; uscs_expl = "Silt (low plasticity)." elif nDS == 3 or nDIL == 3 or nTG == 3: uscs = "OL"; uscs_expl = "Organic silt (low plasticity)." else: uscs = "ML-OL"; uscs_expl = "Mixed silt/organic silt." elif 10 <= LL <= 30 and 4 <= PI <= 7 and PI > 0.72 * (LL - 20): if nDS == 1 or nDIL == 1 or nTG == 1: uscs = "ML"; uscs_expl = "Silt" elif nDS == 2 or nDIL == 2 or nTG == 2: uscs = "CL"; uscs_expl = "Clay (low plasticity)." else: uscs = "ML-CL"; uscs_expl = "Mixed silt/clay" else: uscs = "CL"; uscs_expl = "Clay (low plasticity)." else: if PI < 0.73 * (LL - 20): if nDS == 3 or nDIL == 4 or nTG == 4: uscs = "MH"; uscs_expl = "Silt (high plasticity)" elif nDS == 2 or nDIL == 2 or nTG == 4: uscs = "OH"; uscs_expl = "Organic silt/clay (high plasticity)" else: uscs = "MH-OH"; uscs_expl = "Mixed high-plasticity silt/organic" else: uscs = "CH"; uscs_expl = "Clay (high plasticity)" # === AASHTO (verbatim) === if P2 <= 35: if P2 <= 15 and P4 <= 30 and PI <= 6: aashto = "A-1-a" elif P2 <= 25 and P4 <= 50 and PI <= 6: aashto = "A-1-b" elif P2 <= 35 and P4 > 0: if LL <= 40 and PI <= 10: aashto = "A-2-4" elif LL >= 41 and PI <= 10: aashto = "A-2-5" elif LL <= 40 and PI >= 11: aashto = "A-2-6" elif LL >= 41 and PI >= 11: aashto = "A-2-7" else: aashto = "A-2" else: aashto = "A-3" else: if LL <= 40 and PI <= 10: aashto = "A-4" elif LL >= 41 and PI <= 10: aashto = "A-5" elif LL <= 40 and PI >= 11: aashto = "A-6" else: aashto = "A-7-5" if PI <= (LL - 30) else "A-7-6" # Group Index a = P2 - 35 a = 0 if a < 0 else (40 if a > 40 else a) b = P2 - 15 b = 0 if b < 0 else (40 if b > 40 else b) c = LL - 40 c = 0 if c < 0 else (20 if c > 20 else c) d = PI - 10 d = 0 if d < 0 else (20 if d > 20 else d) GI = floor(0.2 * a + 0.005 * a * c + 0.01 * b * d) aashto_expl = f"{aashto} (Group Index = {GI})" # engineering characteristics pick char_summary = {} found_key = None for key in ENGINEERING_CHARACTERISTICS: if key.lower() in uscs.lower() or key.lower() in uscs_expl.lower(): found_key = key break if found_key: char_summary = ENGINEERING_CHARACTERISTICS[found_key] else: # fallback selection by starting letter if uscs.startswith("G") or uscs.startswith("S"): char_summary = ENGINEERING_CHARACTERISTICS.get("Coarse sand", {}) else: char_summary = ENGINEERING_CHARACTERISTICS.get("Silt", {}) res_text_lines = [ f"According to USCS, the soil is **{uscs}** โ€” {uscs_expl}", f"According to AASHTO, the soil is **{aashto_expl}**", "", "Engineering characteristics (summary):" ] for k,v in char_summary.items(): res_text_lines.append(f"- **{k}**: {v}") result_text = "\n".join(res_text_lines) return result_text, aashto_expl, GI, char_summary, uscs # Helper: GSD interpolation to find diameters D10,D30,D60 def compute_gsd_metrics(diams: List[float], passing: List[float]) -> Dict[str, float]: """ diams: list of diameters in mm (descending) passing: corresponding % passing (0-100) returns D10, D30, D60, Cu, Cc """ # ensure descending diam, convert to float arrays if len(diams) < 2 or len(diams) != len(passing): raise ValueError("Diameters and passing arrays must match and have at least 2 items.") # linear interpolation on log(d) import numpy as np d = np.array(diams) p = np.array(passing) # make sure p is decreasing or increasing? passing decreases as diameter decreases, but we will handle general interpolation by sorting by diameter descending order = np.argsort(-d) d = d[order] p = p[order] # drop duplicates etc # function to find Dx = diameter at which passing = x (percent) def find_D(x): if x <= p.min(): return float(d[p.argmin()]) if x >= p.max(): return float(d[p.argmax()]) # linear interpolation on p vs log(d) from math import log, exp ld = np.log(d) # interpolate ld as function of p ld_interp = np.interp(x, p[::-1], ld[::-1]) # reverse because interp expects ascending x return float(math.exp(ld_interp)) D10 = find_D(10.0) D30 = find_D(30.0) D60 = find_D(60.0) Cu = D60 / D10 if D10 > 0 else 0.0 Cc = (D30 ** 2) / (D10 * D60) if (D10 > 0 and D60 > 0) else 0.0 return {"D10":D10, "D30":D30, "D60":D60, "Cu":Cu, "Cc":Cc} # PDF builder (reportlab) โ€” create professional document similar to uploaded sample def build_full_geotech_pdf(site: Dict[str, Any], filename: str, include_map_image: Optional[bytes]=None, ext_refs: Optional[List[str]]=None): """ site: dictionary of site data filename: output file path include_map_image: bytes of image to embed (optional) ext_refs: list of external refs (strings) """ styles = getSampleStyleSheet() title_style = ParagraphStyle("title", parent=styles["Title"], fontSize=20, alignment=1, textColor=colors.HexColor("#FF7A00")) h1 = ParagraphStyle("h1", parent=styles["Heading1"], fontSize=14, textColor=colors.HexColor("#1F4E79"), spaceAfter=6) body = ParagraphStyle("body", parent=styles["BodyText"], fontSize=10.5, leading=13) bullet = ParagraphStyle("bullet", parent=body, leftIndent=12, bulletIndent=6) doc = SimpleDocTemplate(filename, pagesize=A4, leftMargin=18*mm, rightMargin=18*mm, topMargin=18*mm, bottomMargin=18*mm) elems = [] # Cover elems.append(Paragraph("GEOTECHNICAL INVESTIGATION REPORT", title_style)) elems.append(Spacer(1,6)) elems.append(Paragraph(f"Project: {site.get('Project Name','-')}", body)) elems.append(Paragraph(f"Site: {site.get('Site Name','-')}", body)) elems.append(Paragraph(f"Date: {datetime.today().strftime('%Y-%m-%d')}", body)) elems.append(Spacer(1,8)) elems.append(Paragraph("Prepared by: GeoMate AI", body)) elems.append(PageBreak()) # Summary elems.append(Paragraph("SUMMARY", h1)) summary_bullets = [ f"Site: {site.get('Site Name','-')}.", f"General geology: {site.get('Soil Profile','Not provided')}.", f"Key lab tests: {', '.join([r.get('sampleId','') for r in site.get('Laboratory Results',[])]) if site.get('Laboratory Results') else 'No lab results provided.'}", f"Classification: USCS = {site.get('USCS','Not provided')}; AASHTO = {site.get('AASHTO','Not provided')}.", "Primary recommendation: See Recommendations section." ] for s in summary_bullets: elems.append(Paragraph(f"โ€ข {s}", bullet)) elems.append(PageBreak()) # 1.0 Introduction elems.append(Paragraph("1.0 INTRODUCTION", h1)) intro_text = site.get("Project Description", "Project description not provided.") elems.append(Paragraph(intro_text, body)) # 2.0 Site description and geology elems.append(Paragraph("2.0 SITE DESCRIPTION AND GEOLOGY", h1)) site_geo = [] site_geo.append(f"Topography: {site.get('Topography','Not provided')}") site_geo.append(f"Drainage: {site.get('Drainage','Not provided')}") site_geo.append(f"Current land use: {site.get('Current Land Use','Not provided')}") site_geo.append(f"Regional geology: {site.get('Regional Geology','Not provided')}") for t in site_geo: elems.append(Paragraph(t, body)) elems.append(PageBreak()) # 3.0 Field investigation and laboratory testing elems.append(Paragraph("3.0 FIELD INVESTIGATION & LABORATORY TESTING", h1)) if site.get("Field Investigation"): for item in site["Field Investigation"]: elems.append(Paragraph(f"{item.get('id','Test')} โ€” depth {item.get('depth','-')}", body)) for layer in item.get("layers",[]): elems.append(Paragraph(f"- {layer.get('depth','')} : {layer.get('description','')}", body)) else: elems.append(Paragraph("No field investigation data supplied.", body)) # Lab table lab_rows = site.get("Laboratory Results", []) if lab_rows: elems.append(Spacer(1,6)) elems.append(Paragraph("Laboratory Results", h1)) data = [["Sample ID","Material","LL","PI","Linear Shrinkage","%Clay","%Silt","%Sand","%Gravel","Expansiveness"]] for r in lab_rows: data.append([ r.get("sampleId","-"), r.get("material","-"), str(r.get("liquidLimit","-")), str(r.get("plasticityIndex","-")), str(r.get("linearShrinkage","-")), str(r.get("percentClay","-")), str(r.get("percentSilt","-")), str(r.get("percentSand","-")), str(r.get("percentGravel","-")), r.get("potentialExpansiveness","-") ]) t = Table(data, repeatRows=1, colWidths=[40*mm,40*mm,18*mm,18*mm,22*mm,20*mm,20*mm,20*mm,20*mm,30*mm]) t.setStyle(TableStyle([ ('BACKGROUND',(0,0),(-1,0),colors.HexColor("#1F4E79")), ('TEXTCOLOR',(0,0),(-1,0),colors.white), ('GRID',(0,0),(-1,-1),0.4,colors.grey), ('BOX',(0,0),(-1,-1),1,colors.HexColor("#FF7A00")) ])) elems.append(t) elems.append(PageBreak()) # 4.0 Evaluation & 5.0 Classification & 6.0 Recommendations elems.append(Paragraph("4.0 EVALUATION OF GEOTECHNICAL PROPERTIES", h1)) elems.append(Paragraph(site.get("Evaluation","Evaluation not provided."), body)) elems.append(Paragraph("5.0 PROVISIONAL SITE CLASSIFICATION", h1)) elems.append(Paragraph(site.get("Provisional Classification","Not provided."), body)) elems.append(Paragraph("6.0 RECOMMENDATIONS", h1)) elems.append(Paragraph(site.get("Recommendations","Not provided."), body)) # Map if include_map_image: try: tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".png") tmp.write(include_map_image) tmp.flush() elems.append(PageBreak()) elems.append(Paragraph("Map Snapshot", h1)) elems.append(RLImage(tmp.name, width=160*mm, height=90*mm)) except Exception: pass # External refs if ext_refs: elems.append(PageBreak()) elems.append(Paragraph("References", h1)) for r in ext_refs: elems.append(Paragraph(f"- {r}", body)) doc.build(elems) return filename # 4) Session state initialization if "sites" not in st.session_state: # initialize with a default site st.session_state["sites"] = [{ "Site Name": None, "Project Name": "Project", "Site ID": None, "Soil Class": None, "Soil Recognizer Confidence": None, "Coordinates": "", "lat": None, "lon": None, "Project Description": "", # --------------------------- # Site Characterization # --------------------------- "Topography": None, # manual topo entry "Drainage": None, # manual drainage notes "Current Land Use": None, # can be linked to Environmental Data "Regional Geology": None, # manual geology notes # --------------------------- # Investigations & Lab # --------------------------- "Field Investigation": [], "Laboratory Results": [], "GSD": None, "USCS": None, "AASHTO": None, "GI": None, # --------------------------- # Geotechnical Parameters # --------------------------- "Load Bearing Capacity": None, "Skin Shear Strength": None, "Relative Compaction": None, "Rate of Consolidation": None, "Nature of Construction": None, # --------------------------- # Earth Engine Data # --------------------------- "Soil Profile": { # SoilGrids (multi-parameter) "Clay": None, # e.g. % clay at 200 cm "Sand": None, # % sand "Silt": None, # % silt "OrganicCarbon": None, # % organic carbon "pH": None # soil pH if available }, "Topo Data": None, # Avg elevation (SRTM DEM) "Seismic Data": None, # PGA/g (GEM hazard) "Flood Data": None, # JRC Surface Water occurrence "Environmental Data": { # Landcover, vegetation, urban, etc. "Landcover Stats": None, # histogram by class "Forest Loss": None, # future add: Hansen dataset "Urban Fraction": None # optional calc from landcover }, "Weather Data": { # daily/monthly climate summaries "Rainfall": None, "Temperature": None, "Humidity": None }, "Atmospheric Data": { # optional: pollution, aerosols "AerosolOpticalDepth": None, "NO2": None, "CO": None }, # --------------------------- # Map & Visualization # --------------------------- "map_snapshot": None, # --------------------------- # AI / Reporting # --------------------------- "chat_history": [], "classifier_inputs": {}, "classifier_decision": None, "report_convo_state": 0, "report_missing_fields": [], "report_answers": {} }) if "active_site" not in st.session_state: st.session_state["active_site"] = 0 if "llm_model" not in st.session_state: st.session_state["llm_model"] = "groq/compound" # Groq client (simple wrapper) GROQ_API_KEY = os.environ.get("GROQ_API_KEY") def groq_generate(prompt: str, model: str = None, max_tokens: int = 512) -> str: """Call Groq. If call fails, return an explanatory text.""" try: client = Groq(api_key=GROQ_API_KEY) model_name = model or st.session_state["llm_model"] completion = client.chat.completions.create( model=model_name, messages=[{"role":"user","content":prompt}], temperature=0.2, max_tokens=max_tokens ) text = completion.choices[0].message.content return text except Exception as e: return f"[LLM error or offline: {e}]" # 5) UI helper: nice CSS for chat bubbles & page styling st.markdown(""" """, unsafe_allow_html=True) # 6) Sidebar: navigation, site selector, model selector from streamlit_option_menu import option_menu with st.sidebar: st.markdown("

GeoMate V2

", unsafe_allow_html=True) # LLM model selector st.session_state["llm_model"] = st.selectbox("Select LLM model", options=[ "meta-llama/llama-4-maverick-17b-128e-instruct", "llama-3.1-8b-instant", "meta-llama/llama-guard-4-12b", "llama-3.3-70b-versatile", "groq/compound" ], index=0) st.markdown("---") # Site management controls st.markdown("### Project Sites") site_names = [s.get("Site Name", f"Site {i}") for i,s in enumerate(st.session_state["sites"])] # Add new site input new_site_name = st.text_input("New site name", value="", key="new_site_name_input") if st.button("โž• Add / Create Site"): if new_site_name.strip() == "": st.warning("Enter a name for the new site.") elif len(st.session_state["sites"]) >= MAX_SITES: st.error(f"Maximum of {MAX_SITES} sites allowed.") else: idx = len(st.session_state["sites"]) #idx = len(st.session_state["sites"]) + 1 st.session_state["sites"].append({ "Site Name": new_site_name.strip(), "Project Name": "Project - " + new_site_name.strip(), "Site ID": idx, "Soil Class": None, "Soil Recognizer Confidence": None, "Coordinates": "", "lat": None, "lon": None, "Project Description": "", # --------------------------- # Site Characterization # --------------------------- "Topography": None, # manual topo entry "Drainage": None, # manual drainage notes "Current Land Use": None, # can be linked to Environmental Data "Regional Geology": None, # manual geology notes # --------------------------- # Investigations & Lab # --------------------------- "Field Investigation": [], "Laboratory Results": [], "GSD": None, "USCS": None, "AASHTO": None, "GI": None, # --------------------------- # Geotechnical Parameters # --------------------------- "Load Bearing Capacity": None, "Skin Shear Strength": None, "Relative Compaction": None, "Rate of Consolidation": None, "Nature of Construction": None, # --------------------------- # Earth Engine Data # --------------------------- "Soil Profile": { # SoilGrids (multi-parameter) "Clay": None, # e.g. % clay at 200 cm "Sand": None, # % sand "Silt": None, # % silt "OrganicCarbon": None, # % organic carbon "pH": None # soil pH if available }, "Topo Data": None, # Avg elevation (SRTM DEM) "Seismic Data": None, # PGA/g (GEM hazard) "Flood Data": None, # JRC Surface Water occurrence "Environmental Data": { # Landcover, vegetation, urban, etc. "Landcover Stats": None, # histogram by class "Forest Loss": None, # future add: Hansen dataset "Urban Fraction": None # optional calc from landcover }, "Weather Data": { # daily/monthly climate summaries "Rainfall": None, "Temperature": None, "Humidity": None }, "Atmospheric Data": { # optional: pollution, aerosols "AerosolOpticalDepth": None, "NO2": None, "CO": None }, # --------------------------- # Map & Visualization # --------------------------- "map_snapshot": None, # --------------------------- # AI / Reporting # --------------------------- "chat_history": [], "classifier_inputs": {}, "classifier_decision": None, "report_convo_state": 0, "report_missing_fields": [], "report_answers": {} }) st.success(f"Site '{new_site_name.strip()}' created.") st.session_state["active_site"] = idx st.rerun() # Active site selector if site_names: active_index = st.selectbox("Active Site", options=list(range(len(site_names))), format_func=lambda x: site_names[x], index=st.session_state["active_site"]) st.session_state["active_site"] = active_index st.markdown("---") st.write("Active Site JSON (live)") st.json(st.session_state["sites"][st.session_state["active_site"]]) st.markdown("---") st.markdown("ยฉ GeoMate โ€ข Advanced geotechnical copilot", unsafe_allow_html=True) # 7) Pages implementation def landing_page(): st.markdown("
" "
" "๐Ÿ›ฐ๏ธ
" "

GeoMate V2

" "
AI geotechnical copilot โ€” soil recognition, classification, locator, RAG, and reports
", unsafe_allow_html=True) st.markdown("---") col1, col2 = st.columns([2,1]) with col1: st.markdown("
", unsafe_allow_html=True) st.write("GeoMate is built to help geotechnical engineers: classify soils (USCS/AASHTO), plot GSD, fetch Earth Engine data, chat with a RAG-backed LLM, and generate professional geotechnical reports.") st.markdown("
", unsafe_allow_html=True) st.markdown("### Quick Actions") c1, c2, c3 = st.columns(3) if c1.button("๐Ÿงช Classifier"): st.session_state["page"] = "Classifier"; st.rerun() if c2.button("๐Ÿ“ˆ GSD Curve"): st.session_state["page"] = "GSD"; st.rerun() if c3.button("๐ŸŒ Locator"): st.session_state["page"] = "Locator"; st.rerun() c4, c5, c6 = st.columns(3) if c4.button("๐Ÿค– GeoMate Ask"): st.session_state["page"] = "RAG"; st.rerun() if c5.button("๐Ÿ“ท OCR"): st.session_state["page"] = "OCR"; st.rerun() if c6.button("๐Ÿ“‘ Reports"): st.session_state["page"] = "Reports"; st.rerun() with col2: st.markdown("
", unsafe_allow_html=True) st.markdown("

Live Site Summary

", unsafe_allow_html=True) site = st.session_state["sites"][st.session_state["active_site"]] st.write(f"Site: **{site.get('Site Name')}**") st.write(f"USCS: {site.get('USCS')}, AASHTO: {site.get('AASHTO')}") st.write(f"GSD saved: {'Yes' if site.get('GSD') else 'No'}") st.markdown("
", unsafe_allow_html=True) # Soil Classifier page (conversational, step-by-step) def soil_classifier_page(): st.header("๐Ÿงช Soil Classifier โ€” Conversational (USCS & AASHTO)") site = st.session_state["sites"][st.session_state["active_site"]] # conversation state machine: steps list steps = [ {"id":"intro", "bot":"Hello โ€” I am the GeoMate Soil Classifier. Ready to start?"}, {"id":"organic", "bot":"Is the soil at this site organic (contains high organic matter, feels spongy or has odour)?", "type":"choice", "choices":["No","Yes"]}, {"id":"P2", "bot":"Please enter the percentage passing the #200 sieve (0.075 mm). Example: 12", "type":"number"}, {"id":"P4", "bot":"What is the percentage passing the sieve no. 4 (4.75 mm)? (enter 0 if unknown)", "type":"number"}, {"id":"hasD", "bot":"Do you know the D10, D30 and D60 diameters (in mm)?", "type":"choice","choices":["No","Yes"]}, {"id":"D60", "bot":"Enter D60 (diameter in mm corresponding to 60% passing).", "type":"number"}, {"id":"D30", "bot":"Enter D30 (diameter in mm corresponding to 30% passing).", "type":"number"}, {"id":"D10", "bot":"Enter D10 (diameter in mm corresponding to 10% passing).", "type":"number"}, {"id":"LL", "bot":"What is the liquid limit (LL)?", "type":"number"}, {"id":"PL", "bot":"What is the plastic limit (PL)?", "type":"number"}, {"id":"dry", "bot":"Select the observed dry strength of the fine soil (if applicable).", "type":"select", "options":DRY_STRENGTH_OPTIONS}, {"id":"dilat", "bot":"Select the observed dilatancy behaviour.", "type":"select", "options":DILATANCY_OPTIONS}, {"id":"tough", "bot":"Select the observed toughness.", "type":"select", "options":TOUGHNESS_OPTIONS}, {"id":"confirm", "bot":"Would you like me to classify now?", "type":"choice", "choices":["No","Yes"]} ] if "classifier_step" not in st.session_state: st.session_state["classifier_step"] = 0 if "classifier_inputs" not in st.session_state: st.session_state["classifier_inputs"] = dict(site.get("classifier_inputs", {})) step_idx = st.session_state["classifier_step"] # chat history display st.markdown("
", unsafe_allow_html=True) st.markdown("
{}
".format("GeoMate: Hello โ€” soil classifier ready. Use the controls below to answer step-by-step."), unsafe_allow_html=True) # Show stored user answers sequentially for context # render question up to current step for i in range(step_idx+1): s = steps[i] # show bot prompt st.markdown(f"
{s['bot']}
", unsafe_allow_html=True) # show user answer if exists in classifier_inputs key = s["id"] val = st.session_state["classifier_inputs"].get(key) if val is not None: st.markdown(f"
{val}
", unsafe_allow_html=True) st.markdown("
", unsafe_allow_html=True) # Render input widget for current step current = steps[step_idx] step_id = current["id"] proceed = False user_answer = None cols = st.columns([1,1,1]) with cols[0]: if current.get("type") == "choice": choice = st.radio(current["bot"], options=current["choices"], index=0, key=f"cls_{step_id}") user_answer = choice elif current.get("type") == "number": # numeric input without +/- spinner (we use text_input and validate) raw = st.text_input(current["bot"], value=str(st.session_state["classifier_inputs"].get(step_id,"")), key=f"cls_{step_id}_num") # validate numeric try: if raw.strip() == "": user_answer = None else: user_answer = float(raw) except: st.warning("Please enter a valid number (e.g., 12 or 0).") user_answer = None elif current.get("type") == "select": opts = current.get("options", []) sel = st.selectbox(current["bot"], options=opts, index=0, key=f"cls_{step_id}_sel") user_answer = sel else: # just a message step โ€” proceed user_answer = None # controls: Next / Back coln, colb, colsave = st.columns([1,1,1]) with coln: if st.button("โžก๏ธ Next", key=f"next_{step_id}"): # store answer if provided if current.get("type") == "number": if user_answer is None: st.warning("Please enter a numeric value or enter 0 if unknown.") else: st.session_state["classifier_inputs"][step_id] = user_answer st.session_state["classifier_step"] = min(step_idx+1, len(steps)-1) st.rerun() elif current.get("type") in ("choice","select"): st.session_state["classifier_inputs"][step_id] = user_answer st.session_state["classifier_step"] = min(step_idx+1, len(steps)-1) st.rerun() else: # message-only step st.session_state["classifier_step"] = min(step_idx+1, len(steps)-1) st.rerun() with colb: if st.button("โฌ…๏ธ Back", key=f"back_{step_id}"): st.session_state["classifier_step"] = max(0, step_idx-1) st.rerun() with colsave: if st.button("๐Ÿ’พ Save & Classify now", key="save_and_classify"): # prepare inputs in required format for classify_uscs_aashto ci = st.session_state["classifier_inputs"].copy() # Normalize choices into expected codes if isinstance(ci.get("dry"), str): ci["nDS"] = DRY_STRENGTH_MAP.get(ci.get("dry"), 5) if isinstance(ci.get("dilat"), str): ci["nDIL"] = DILATANCY_MAP.get(ci.get("dilat"), 6) if isinstance(ci.get("tough"), str): ci["nTG"] = TOUGHNESS_MAP.get(ci.get("tough"), 6) # map 'Yes'/'No' for organic and hasD ci["opt"] = "y" if ci.get("organic","No")=="Yes" or ci.get("organic",ci.get("organic"))=="Yes" else ci.get("organic","n") # our field names in CI may differ: convert organic stored under 'organic' step to 'opt' if "organic" in ci: ci["opt"] = "y" if ci["organic"]=="Yes" else "n" # map D entries: D60 etc may be present # call classification try: res_text, aashto, GI, chars, uscs = classify_uscs_aashto(ci) except Exception as e: st.error(f"Classification error: {e}") res_text = f"Error during classification: {e}" aashto = "N/A"; GI = 0; chars = {}; uscs = "N/A" # save into active site site["USCS"] = uscs site["AASHTO"] = aashto site["GI"] = GI site["classifier_inputs"] = ci site["classifier_decision"] = res_text st.success("Classification complete. Results saved to site.") st.write("### Classification Results") st.markdown(res_text) # Keep classifier_step at end so user can review st.session_state["classifier_step"] = len(steps)-1 # GSD Curve Page def gsd_page(): st.header("๐Ÿ“ˆ Grain Size Distribution (GSD) Curve") site = st.session_state["sites"][st.session_state["active_site"]] st.markdown("Enter diameters (mm) and % passing (comma-separated). Use descending diameters (largest to smallest).") diam_input = st.text_area("Diameters (mm) comma-separated", value=site.get("GSD",{}).get("diameters","75,50,37.5,25,19,12.5,9.5,4.75,2,0.85,0.425,0.25,0.18,0.15,0.075") if site.get("GSD") else "75,50,37.5,25,19,12.5,9.5,4.75,2,0.85,0.425,0.25,0.18,0.15,0.075") pass_input = st.text_area("% Passing comma-separated", value=site.get("GSD",{}).get("passing","100,98,96,90,85,78,72,65,55,45,35,25,18,14,8") if site.get("GSD") else "100,98,96,90,85,78,72,65,55,45,35,25,18,14,8") if st.button("Compute GSD & Save"): try: diams = [float(x.strip()) for x in diam_input.split(",") if x.strip()] passing = [float(x.strip()) for x in pass_input.split(",") if x.strip()] metrics = compute_gsd_metrics(diams, passing) # plot fig, ax = plt.subplots(figsize=(7,4)) ax.semilogx(diams, passing, marker='o') ax.set_xlabel("Particle size (mm)") ax.set_ylabel("% Passing") ax.invert_xaxis() ax.grid(True, which='both', linestyle='--', linewidth=0.5) ax.set_title("Grain Size Distribution") st.pyplot(fig) # save into site site["GSD"] = {"diameters":diams, "passing":passing, **metrics} st.success(f"Saved GSD for site. D10={metrics['D10']:.4g} mm, D30={metrics['D30']:.4g} mm, D60={metrics['D60']:.4g} mm") except Exception as e: st.error(f"GSD error: {e}") # OCR Page def ocr_page(): st.header("๐Ÿ“ท OCR โ€” extract values from an image") site = st.session_state["sites"][st.session_state["active_site"]] if not OCR_AVAILABLE: st.warning("OCR dependencies not available (pytesseract/PIL). Add pytesseract and pillow to requirements to enable OCR.") uploaded = st.file_uploader("Upload an image (photo of textbook question or sieve data)", type=["png","jpg","jpeg"]) if uploaded: if OCR_AVAILABLE: try: img = Image.open(uploaded) st.image(img, caption="Uploaded", use_column_width=True) text = pytesseract.image_to_string(img) st.text_area("Extracted text", value=text, height=180) # Basic parsing: try to find LL, PL, D10 etc via regex import re found = {} for key in ["LL","PL","D10","D30","D60","P2","P4","CBR"]: pattern = re.compile(rf"{key}[:=]?\s*([0-9]+\.?[0-9]*)", re.I) m = pattern.search(text) if m: found[key] = float(m.group(1)) site.setdefault("classifier_inputs",{})[key] = float(m.group(1)) if found: st.success(f"Parsed values: {found}") st.write("Values saved into classifier inputs.") else: st.info("No clear numeric matches found automatically.") except Exception as e: st.error(f"OCR failed: {e}") else: st.warning("OCR not available in this deployment.") # Locator Page (with Earth Engine auth at top) import os import json import streamlit as st import geemap.foliumap as geemap import ee import matplotlib.pyplot as plt from datetime import datetime from io import BytesIO import base64 def locator_page(): """ Robust locator page: - Uses your initialize_ee() auth routine (expects EARTHENGINE_TOKEN / SERVICE_ACCOUNT in env) - Shows interactive map with many basemaps and overlays - Safe reducers with fallbacks and caching - Stores results in st.session_state['soil_json'] AND in the active site entry under Earth Engine fields """ st.title("๐ŸŒ GeoMate Interactive Earth Explorer") st.markdown( "Draw a polygon (or rectangle) on the map using the drawing tool. " "The app will compute regional summaries (soil clay, elevation, seismic, flood occurrence, landcover, NDVI) " "and save results for reports." ) # ---------------------------- # Use your existing EE init function if present EARTHENGINE_TOKEN = os.getenv("EARTHENGINE_TOKEN") SERVICE_ACCOUNT = os.getenv("SERVICE_ACCOUNT") # optional: service account email def initialize_ee(): """Initialize Earth Engine with multiple fallbacks.""" if "ee_initialized" in st.session_state and st.session_state["ee_initialized"]: return True if EARTHENGINE_TOKEN and SERVICE_ACCOUNT: try: creds = ee.ServiceAccountCredentials( email=SERVICE_ACCOUNT, key_data=EARTHENGINE_TOKEN ) ee.Initialize(creds) st.session_state["ee_initialized"] = True return True except Exception as e: st.warning(f"Service account init failed: {e} โ€” trying default/interactive auth...") try: ee.Initialize() st.session_state["ee_initialized"] = True return True except Exception: try: ee.Authenticate() ee.Initialize() st.session_state["ee_initialized"] = True return True except Exception as e: st.error(f"Earth Engine authentication failed: {e}") return False if not initialize_ee(): st.stop() # ---------------------------- # I assume your file defines initialize_ee() exactly like your earlier message. try: init_ok = initialize_ee() # call your auth initializer except NameError: st.error("Auth initializer `initialize_ee()` not found. Ensure your auth code exists above this function.") return if not init_ok: return # ---------------------------- # Helper: safe reducers + caching # ---------------------------- def safe_get_reduce(region, image, band, scale=1000, default=None, max_pixels=int(1e7)): """Return float or None. Uses reduceRegion mean safely.""" cache_key = f"reduce::{region.toGeoJSONString()[:200]}::{str(image)}::{band}::{scale}" # check cache cache = st.session_state.setdefault("_ee_cache", {}) if cache_key in cache: return cache[cache_key] try: rr = image.reduceRegion( reducer=ee.Reducer.mean(), geometry=region, scale=int(scale), maxPixels=int(max_pixels) ) val = rr.get(band) if val is None: cache[cache_key] = default return default v = val.getInfo() if v is None: cache[cache_key] = default return default got = float(v) cache[cache_key] = got return got except Exception as e: # log to session for debugging if desired st.session_state.setdefault("_ee_errors", []).append(str(e)) cache[cache_key] = default return default def safe_reduce_histogram(region, image, band, scale=1000, max_pixels=int(1e7)): """Return a frequency histogram dict or {}.""" cache_key = f"hist::{region.toGeoJSONString()[:200]}::{str(image)}::{band}::{scale}" cache = st.session_state.setdefault("_ee_cache", {}) if cache_key in cache: return cache[cache_key] try: rr = image.reduceRegion( reducer=ee.Reducer.frequencyHistogram(), geometry=region, scale=int(scale), maxPixels=int(max_pixels) ) val = rr.get(band) if val is None: cache[cache_key] = {} return {} hist = val.getInfo() if hist is None: cache[cache_key] = {} return {} cache[cache_key] = hist return hist except Exception as e: st.session_state.setdefault("_ee_errors", []).append(str(e)) cache[cache_key] = {} return {} def safe_time_series(region, collection, band, start, end, reducer=ee.Reducer.mean(), scale=1000, max_pixels=int(1e7)): """Return simple timeseries list of (date, value) for a collection.""" try: # reduce each image over region, map to list def per_image(img): date = img.date().format("YYYY-MM-dd") val = img.reduceRegion(reducer=reducer, geometry=region, scale=int(scale), maxPixels=int(max_pixels)).get(band) return ee.Feature(None, {"date": date, "val": val}) feats = collection.filterDate(start, end).map(per_image).filter(ee.Filter.notNull(["val"])).getInfo() # feats is a dict with 'features' list points = [] for f in feats.get("features", []): props = f.get("properties", {}) date = props.get("date") val = props.get("val") if val is not None: try: points.append((date, float(val))) except Exception: pass return points except Exception as e: st.session_state.setdefault("_ee_errors", []).append(str(e)) return [] # ---------------------------- # Map setup # ---------------------------- m = geemap.Map(center=[28.0, 72.0], zoom=5, draw_export=True) # enriched basemaps list (many options; geemap will ignore unknown) basemaps = [ "HYBRID", "ROADMAP", "TERRAIN", "SATELLITE", "Esri.WorldImagery", "Esri.WorldTopoMap", "Esri.WorldShadedRelief", "Esri.NatGeoWorldMap", "Esri.OceanBasemap", "CartoDB.Positron", "CartoDB.DarkMatter", "Stamen.Terrain", "Stamen.Watercolor", "OpenStreetMap", "Esri.WorldGrayCanvas", "Esri.WorldStreetMap" ] for b in basemaps: try: m.add_basemap(b) except Exception: pass # ---------------------------- # Datasets (choose stable EE catalog IDs) # Terrain (DEM) -> prefer NASADEM then SRTM fallback # ---------------------------- # Try NASADEM (higher quality), fallback to SRTM: try: dem = ee.Image("NASA/NASADEM_HGT/001") # NASADEM if available dem_band_name = "elevation" except Exception: dem = ee.Image("USGS/SRTMGL1_003") dem_band_name = "elevation" # ---------------------------- # Soil -> use OpenLandMap clay fraction (v02) which provides multiple bands like b0,b10,b30,... and is available # (fallback to SoilGrids if OpenLandMap missing) # ---------------------------- soil_img = None soil_band = "b200" # default deep band try: soil_img = ee.Image("OpenLandMap/SOL/SOL_CLAY-WFRACTION_USDA-3A1A1A_M/v02") # pick default band b200 (100-200cm). Offer depth selection below in UI. except Exception: # fallback to SoilGrids if available try: soil_img = ee.Image("projects/soilgrids-isric/clay_mean") # SoilGrids band names like 'clay_0-5cm_mean' etc โ€” we will pick a default after UI selection. soil_band = "clay_0-5cm_mean" except Exception: soil_img = None # ---------------------------- # Seismic -> attempt SEDAC GSHAP, then GEM # ---------------------------- seismic_img = None try: seismic_img = ee.Image("SEDAC/GSHAPSeismicHazard") # band "gshap" seismic_band = "gshap" except Exception: try: seismic_img = ee.Image("GEM/2015/GlobalSeismicHazard") # may or may not exist # many GEM products use band b0 etc. seismic_band = "b0" except Exception: seismic_img = None seismic_band = None # ---------------------------- # Flood -> JRC Global Surface Water # ---------------------------- try: water = ee.Image("JRC/GSW1_4/GlobalSurfaceWater") water_band = "occurrence" except Exception: water = None water_band = None # ---------------------------- # Landcover -> ESA WorldCover v200 (10 m) - "Map" band # ---------------------------- try: landcover = ee.Image("ESA/WorldCover/v200") lc_band = "Map" except Exception: landcover = None lc_band = None # ---------------------------- # NDVI collection (MODIS 1km as robust global product) # ---------------------------- try: ndvi_col = ee.ImageCollection("MODIS/061/MOD13A2").select("NDVI") except Exception: ndvi_col = None # ---------------------------- # Add layers to map (visuals) # ---------------------------- # Add DEM visualization try: m.addLayer(dem, {"min": 0, "max": 4000, "palette": ["blue", "green", "brown", "white"]}, "DEM / Topography") except Exception: pass # Add soil view (if available) โ€” user will pick depth later if soil_img: try: # show the dataset (choose one band to display; if band missing, geemap will raise โ€” catch it) # prefer b0 or b200 if available; if both missing, show first band available_bands = soil_img.bandNames().getInfo() # find a band to display display_band = soil_band if soil_band in available_bands else available_bands[0] m.addLayer(soil_img.select(display_band), {"min": 0.0, "max": 0.6, "palette": ["#ffffcc","#c2e699","#78c679","#31a354"]}, f"Soil Clay ({display_band})") except Exception: pass # Seismic if seismic_img: try: # for SEDAC band "gshap" map 0-1 m.addLayer(seismic_img, {"min": 0, "max": 1, "palette": ["white", "yellow", "red"]}, "Seismic Hazard") except Exception: pass # Flood if water is not None: try: m.addLayer(water.select(water_band), {"min":0,"max":100,"palette":["white","blue"]}, "Water Occurrence (JRC)") except Exception: pass # Landcover if landcover is not None: try: m.addLayer(landcover, {"min":10,"max":100,"palette":["#006400","#ffbb22","#ffff4c","#f096ff","#fa0000","#b4b4b4","#f0f0f0","#0064c8","#0096a0","#00cf75"]}, "Landcover (WorldCover)") except Exception: pass # Add country boundaries & graticule try: countries = ee.FeatureCollection("USDOS/LSIB_SIMPLE/2017") m.addLayer(countries.style(**{"color": "black", "fillColor": "00000000", "width": 1}), {}, "Country Boundaries") except Exception: pass try: m.addLayer(geemap.latlon_grid(5.0, region=ee.Geometry.Rectangle([-180, -90, 180, 90])).style(**{"color":"gray","width":0.5}), {}, "Lat/Lon Grid") except Exception: pass # Enable drawing m.add_draw_control(polyline=False, circle=False, circlemarker=False, rectangle=True, polygon=True) m.to_streamlit(height=700, responsive=True) st.markdown("๐Ÿ‘‰ Draw a polygon/rectangle on the map (draw tool). After drawing click **Compute Summaries**.") # Button to compute summaries (we'll fetch ROI from map object) if "compute_button" not in st.session_state: st.session_state["compute_button"] = False if st.button("Compute Summaries"): st.session_state["compute_button"] = True # get ROI helper (try multiple attributes) def get_roi_from_map(m): # preferred property candidates = [ getattr(m, "user_roi", None), getattr(m, "last_drawn_geojson", None), getattr(m, "draw_last_feature", None), getattr(m, "draw_features", None), getattr(m, "user_drawn_features", None), getattr(m, "drawn_features", None), ] for c in candidates: if c: # if it's already an ee.Geometry if isinstance(c, ee.Geometry): return c # if it's a GeoJSON dict try: if isinstance(c, dict): return ee.Geometry(c) # sometimes geemap keeps features list if isinstance(c, list) and len(c) > 0 and isinstance(c[0], dict): return ee.Geometry(c[0]) # string that is GeoJSON if isinstance(c, str): import json js = json.loads(c) if isinstance(js, dict): return ee.Geometry(js) except Exception: pass # try Map.draw_last_feature property try: s = getattr(m, "draw_last_feature", None) if s: import json if isinstance(s, str): return ee.Geometry(json.loads(s)) except Exception: pass return None # If user pressed compute, extract ROI and compute if st.session_state.get("compute_button", False): roi = get_roi_from_map(m) if roi is None: st.error("No drawn ROI found. Please draw a polygon/rectangle on the map using the draw tool and press Compute Summaries again.") return # ensure ROI is ee.Geometry try: if not isinstance(roi, ee.Geometry): roi = ee.Geometry(roi) except Exception: st.error("Failed to parse ROI as Earth Engine geometry.") return st.success("Polygon found โ€” computing (this may take a few seconds)...") # choose soil depth via UI (0-200cm). If soil_img is OpenLandMap provide depth selection; if SoilGrids, show its bands chosen_soil_band = None if soil_img is not None: try: bands = soil_img.bandNames().getInfo() # if OpenLandMap bands like b0,b10,... show friendly selector depth_choice = st.selectbox("Soil depth / band to analyze", options=bands, index=bands.index(soil_band) if soil_band in bands else 0) chosen_soil_band = depth_choice except Exception: chosen_soil_band = None # compute using safe helpers with conservative maxPixels and reasonable scales # scales: soil 250-1000m, dem 90-1000m, seismic coarse 5000m, flood 30-250m, landcover 30m soil_val = None if soil_img is not None and chosen_soil_band is not None: soil_val = safe_get_reduce(roi, soil_img.select(chosen_soil_band), chosen_soil_band, scale=1000, default=None, max_pixels=int(5e7)) elev_val = safe_get_reduce(roi, dem, dem_band_name if dem_band_name else "elevation", scale=1000, default=None, max_pixels=int(5e7)) seismic_val = None if seismic_img is not None and seismic_band is not None: seismic_val = safe_get_reduce(roi, seismic_img, seismic_band, scale=5000, default=None, max_pixels=int(5e7)) flood_val = None if water is not None and water_band is not None: flood_val = safe_get_reduce(roi, water.select(water_band), water_band, scale=30, default=None, max_pixels=int(5e7)) # landcover histogram lc_stats = {} if landcover is not None and lc_band is not None: lc_stats = safe_reduce_histogram(roi, landcover, lc_band, scale=30, max_pixels=int(5e7)) # NDVI timeseries (last 2 years) โ€” provide small chart ndvi_ts = [] if ndvi_col is not None: end = datetime.utcnow().strftime("%Y-%m-%d") start = (datetime.utcnow().replace(year=datetime.utcnow().year - 2)).strftime("%Y-%m-%d") ndvi_ts = safe_time_series(roi, ndvi_col, "NDVI", start, end, reducer=ee.Reducer.mean(), scale=1000, max_pixels=int(5e7)) # ---------------------------- # UI: display numeric summary # ---------------------------- def pretty(x, fmt="{:.2f}"): return "N/A" if x is None else fmt.format(x) st.subheader("๐Ÿ“Š Regional Data Summary") st.write(f"**Soil ({chosen_soil_band}):** {pretty(soil_val)}") st.write(f"**Average Elevation:** {pretty(elev_val, '{:.1f}')} m") st.write(f"**Seismic (mean):** {pretty(seismic_val)}") st.write(f"**Flood occurrence (mean %):** {pretty(flood_val)}") # Landcover pie chart (colored) if lc_stats: # convert keys into ints if possible (WorldCover class codes) labels = [] values = [] for k, v in lc_stats.items(): labels.append(str(k)) values.append(v) fig1, ax1 = plt.subplots(figsize=(6,4)) ax1.pie(values, labels=labels, autopct="%1.1f%%", startangle=90) ax1.set_title("Landcover Distribution (class codes)") st.pyplot(fig1) else: st.info("No landcover histogram available.") # NDVI timeseries plot if ndvi_ts: dates = [d for d, v in ndvi_ts] vals = [v for d, v in ndvi_ts] fig2, ax2 = plt.subplots(figsize=(8,3)) ax2.plot(dates, vals, marker="o") ax2.set_title("NDVI (mean) โ€” last 2 years") ax2.set_xlabel("Date") ax2.set_ylabel("NDVI (scaled)") plt.xticks(rotation=45) st.pyplot(fig2) else: st.info("NDVI timeseries not available or too sparse.") # Soil histogram (if available) soil_hist = None try: soil_hist = soil_img.reduceRegion( reducer=ee.Reducer.histogram(maxBuckets=20), geometry=roi, scale=1000, maxPixels=int(5e7) ).get(chosen_soil_band).getInfo() if (soil_img is not None and chosen_soil_band) else None except Exception: soil_hist = None if soil_hist and isinstance(soil_hist, dict) and "bucketMeans" in soil_hist: fig3, ax3 = plt.subplots(figsize=(6,4)) ax3.bar(soil_hist["bucketMeans"], soil_hist["histogram"], width= (soil_hist["bucketMeans"][1]-soil_hist["bucketMeans"][0]) if len(soil_hist["bucketMeans"])>1 else 1, color="saddlebrown") ax3.set_title(f"Soil histogram ({chosen_soil_band})") ax3.set_xlabel("Clay fraction (kg/kg)") ax3.set_ylabel("Pixel count") st.pyplot(fig3) # ---------------------------- # Save results to session_state for reports # ---------------------------- # Ensure sites and active site exist if "sites" not in st.session_state or "active_site" not in st.session_state: # just store soil_json if site structure not present st.session_state["soil_json"] = { "Soil": None if soil_val is None else float(soil_val), "Soil Band": chosen_soil_band, "Elevation": None if elev_val is None else float(elev_val), "Seismic": None if seismic_val is None else float(seismic_val), "Flood": None if flood_val is None else float(flood_val), "Landcover Stats": lc_stats or {}, "NDVI TS": ndvi_ts or [] } st.success("Saved results to st.session_state['soil_json']. (No active site present.)") else: # Save into active site JSON fields (keeping your field names unchanged) active = st.session_state["active_site"] try: site_obj = st.session_state["sites"][active] except Exception: # if your sites is a list of dicts with Site ID matching idx, try to match try: site_obj = st.session_state["sites"][int(active)] except Exception: site_obj = None # fallback: if site_obj is None just write soil_json and exit if site_obj is None: st.session_state["soil_json"] = { "Soil": None if soil_val is None else float(soil_val), "Soil Band": chosen_soil_band, "Elevation": None if elev_val is None else float(elev_val), "Seismic": None if seismic_val is None else float(seismic_val), "Flood": None if flood_val is None else float(flood_val), "Landcover Stats": lc_stats or {}, "NDVI TS": ndvi_ts or [] } st.success("Saved results to st.session_state['soil_json']. (Could not find active site object.)") else: # update the exact fields you requested (names unchanged) site_obj["Soil Profile"] = f"{round(soil_val,3)} ({chosen_soil_band})" if soil_val is not None else "No data" site_obj["Topo Data"] = f"{round(elev_val,2)} m (mean)" if elev_val is not None else "No data" site_obj["Seismic Data"] = f"{round(seismic_val,4)}" if seismic_val is not None else "No data" site_obj["Flood Data"] = f"{round(flood_val,2)} %" if flood_val is not None else "No data" # Environmental Data: combine landcover summary + NDVI basic stats env_summary = { "Landcover Histogram": lc_stats or {}, "NDVI_timeseries_points": ndvi_ts or [] } site_obj["Environmental Data"] = env_summary # Save drawn polygon GeoJSON for future map restore and report inclusion try: # fetch GeoJSON from ROI geojson = roi.toGeoJSON() if hasattr(roi, "toGeoJSON") else ee.Geometry(roi).getInfo() site_obj["drawn_polygon"] = geojson except Exception: site_obj["drawn_polygon"] = None # Save to soil_json as well (for report block that expects it) st.session_state["soil_json"] = { "Soil": None if soil_val is None else float(soil_val), "Soil Band": chosen_soil_band, "Elevation": None if elev_val is None else float(elev_val), "Seismic": None if seismic_val is None else float(seismic_val), "Flood": None if flood_val is None else float(flood_val), "Landcover Stats": lc_stats or {}, "NDVI TS": ndvi_ts or [] } st.success("๐Ÿ“‘ Results saved to active site and st.session_state['soil_json'] for report integration.") # Snapshot map as HTML and save path into site object (map snapshot - small HTML content) try: snap_html = m.to_html(None) # returns HTML string if path None # store minimal snapshot content into site or soil_json (may be large; consider storing link) if "sites" in st.session_state and site_obj is not None: site_obj["map_snapshot"] = snap_html # caution: large string st.session_state["last_map_snapshot"] = snap_html except Exception: pass # end of Compute Summaries block # end locator_page() # GeoMate Ask (RAG) โ€” simple chat with memory per site and auto-extract numeric values def rag_page(): st.header("๐Ÿค– GeoMate Ask (RAG + Groq)") site = st.session_state["sites"][st.session_state["active_site"]] st.markdown("Chat with GeoMate. The LLM has memory per site for this session; any engineering values provided will be parsed and saved.") if "rag_history" not in st.session_state: st.session_state["rag_history"] = {i: [] for i in range(len(st.session_state["sites"]))} hist = st.session_state["rag_history"].get(site["Site ID"], []) for entry in hist: who, text = entry.get("who"), entry.get("text") if who == "bot": st.markdown(f"
{text}
", unsafe_allow_html=True) else: st.markdown(f"
{text}
", unsafe_allow_html=True) user_msg = st.text_input("You:", key="rag_input") if st.button("Send", key="rag_send"): if not user_msg.strip(): st.warning("Enter a message.") else: # Save user msg st.session_state["rag_history"][site["Site ID"]].append({"who":"user","text":user_msg}) # Build prompt including site context context = { "site": {k:v for k,v in site.items() if k in ["Site Name","lat","lon","USCS","AASHTO","GI","Load Bearing Capacity","Soil Profile","Flood Data","Seismic Data"]}, "chat_history": st.session_state["rag_history"][site["Site ID"]] } prompt = f"You are GeoMate AI. Site context: {json.dumps(context)}. User: {user_msg}\nRespond professionally and concisely. If user provides numeric engineering values, return them in the format: [[FIELD: value unit]]." resp = groq_generate(prompt, model=st.session_state["llm_model"], max_tokens=400) # Save bot reply st.session_state["rag_history"][site["Site ID"]].append({"who":"bot","text":resp}) # Display st.markdown(f"
{resp}
", unsafe_allow_html=True) # Try to extract bracketed fields like [[Load bearing capacity: 2000 psf]] import re matches = re.findall(r"\[\[([A-Za-z0-9 _/-]+):\s*([0-9.+-eE]+)\s*([A-Za-z%\/]*)\]\]", resp) for m in matches: field = m[0].strip() val = m[1].strip() unit = m[2].strip() # Map common fields to site keys if "bearing" in field.lower(): site["Load Bearing Capacity"] = f"{val} {unit}" elif "skin" in field.lower(): site["Skin Shear Strength"] = f"{val} {unit}" elif "compaction" in field.lower(): site["Relative Compaction"] = f"{val} {unit}" st.success("Response saved and any recognized numeric fields auto-stored in the site data.") # Reports page โ€” conversational missing-parameter bot & PDF generation REPORT_FIELDS = [ ("Load Bearing Capacity","kPa or psf"), ("Skin Shear Strength","kPa"), ("Relative Compaction","%"), ("Rate of Consolidation","mm/yr or days"), ("Nature of Construction","text"), ("Borehole Count","number"), ("Max Depth (m)","m"), ("SPT N (avg)","blows/ft"), ("CBR (%)","%"), ("Allowable Bearing (kPa)","kPa") ] def reports_page(): st.header("๐Ÿ“‘ Reports โ€” Classification & Full Geotechnical") site = st.session_state["sites"][st.session_state["active_site"]] st.subheader("Classification-only report") if site.get("classifier_decision"): st.markdown("You have a saved classification for this site.") if st.button("Generate Classification PDF"): fname = f"classification_{site['Site Name'].replace(' ','_')}.pdf" # simple PDF buffer = io.BytesIO() doc = SimpleDocTemplate(buffer, pagesize=A4) elems = [] elems.append(Paragraph("Soil Classification Report", getSampleStyleSheet()['Title'])) elems.append(Spacer(1,6)) elems.append(Paragraph(f"Site: {site.get('Site Name')}", getSampleStyleSheet()['Normal'])) elems.append(Spacer(1,6)) elems.append(Paragraph("Classification result:", getSampleStyleSheet()['Heading2'])) elems.append(Paragraph(site.get("classifier_decision","-"), getSampleStyleSheet()['BodyText'])) doc.build(elems) buffer.seek(0) st.download_button("Download Classification PDF", buffer, file_name=fname, mime="application/pdf") else: st.info("No classification saved for this site yet. Use the Classifier page.") st.markdown("---") st.subheader("Full Geotechnical Report (chatbot will gather missing fields)") if st.button("Start Report Chatbot"): st.session_state["sites"][st.session_state["active_site"]]["report_convo_state"] = 0 st.rerun() # Conversational data collection state = site.get("report_convo_state", 0) if site.get("report_convo_state") is not None: st.markdown("Chatbot will ask for missing fields. You can answer or type 'skip' to leave blank.") # Show current known fields st.write("Current key parameters (live):") show_table = [] for f,_ in REPORT_FIELDS: show_table.append((f, site.get(f, "Not provided"))) st.table(show_table) # continue conversation step-by-step if state < len(REPORT_FIELDS): field, unit = REPORT_FIELDS[state] ans = st.text_input(f"GeoMate โ€” Please provide '{field}' ({unit})", key=f"report_in_{state}") c1, c2 = st.columns([1,1]) with c1: if st.button("Submit", key=f"report_submit_{state}"): if ans.strip().lower() in ("skip","don't know","dont know","na","n/a",""): site[field] = "Not provided" else: site[field] = ans.strip() site["report_convo_state"] = state + 1 st.rerun() with c2: if st.button("Skip", key=f"report_skip_{state}"): site[field] = "Not provided" site["report_convo_state"] = state + 1 st.rerun() else: st.success("All report questions asked. You can generate the full report now.") if st.button("Generate Full Geotechnical Report PDF"): # Prepare ext_refs ext_ref_text = st.text_area("Optional: External references (one per line)", value="") ext_refs = [r.strip() for r in ext_ref_text.splitlines() if r.strip()] # Build PDF using reportlab builder outname = f"Full_Geotech_Report_{site.get('Site Name','site')}.pdf" # include map image bytes if available mapimg = site.get("map_snapshot") build_full_geotech_pdf(site, outname, include_map_image=mapimg, ext_refs=ext_refs) with open(outname, "rb") as f: st.download_button("Download Full Geotechnical Report", f, file_name=outname, mime="application/pdf") # 8) Page router if "page" not in st.session_state: st.session_state["page"] = "Home" page = st.session_state["page"] # Option menu top (main nav) selected = option_menu( None, ["Home","Soil recognizer","Classifier","GSD","OCR","Locator","RAG","Reports"], icons=["house","chart","journal-code","bar-chart","camera","geo-alt","robot","file-earmark-text"], menu_icon="cast", default_index=["Home","Soil recognizer","Classifier","GSD","OCR","Locator","RAG","Reports"].index(page) if page in ["Home","Soil recognizer","Classifier","GSD","OCR","Locator","RAG","Reports"] else 0, orientation="horizontal", styles={ "container": {"padding":"0px","background-color":"#0b0b0b"}, "nav-link": {"font-size":"14px","color":"#cfcfcf"}, "nav-link-selected": {"background-color":"#FF7A00","color":"white"}, } ) st.session_state["page"] = selected page = selected # Display page content if page == "Home": landing_page() elif page == "Classifier": soil_classifier_page() elif page == "GSD": gsd_page() elif page == "OCR": ocr_page() elif page == "Locator": locator_page() elif page == "RAG": rag_page() elif page == "Reports": reports_page() elif page == "Soil recognizer": soil_recognizer_page() else: landing_page() # Footer st.markdown("
", unsafe_allow_html=True) st.markdown("
GeoMate V2 โ€ข AI geotechnical copilot โ€ข Built for HF Spaces
", unsafe_allow_html=True)