import gradio as gr from google import genai import os import re from PIL import Image, ImageDraw import requests from io import BytesIO # --- 1. Custom CSS --- custom_css = """ .improving { background-color: #dcf2dc; color: #155724; padding: 12px; border-radius: 8px; font-weight: bold; font-size: 1.2em; text-align: center; border: 1px solid #c3e6cb;} .declining { background-color: #f8d7da; color: #721c24; padding: 12px; border-radius: 8px; font-weight: bold; font-size: 1.2em; text-align: center; border: 1px solid #f5c6cb;} .stable { background-color: #fff3cd; color: #856404; padding: 12px; border-radius: 8px; font-weight: bold; font-size: 1.2em; text-align: center; border: 1px solid #ffeeba;} """ # --- 2. Google Maps API Fetcher --- def fetch_streetview(address): """Geocodes an address and fetches the most recent Street View image.""" gmaps_api_key = os.environ.get("GMAPS_API_KEY") if not gmaps_api_key: raise gr.Error("ERROR: GMAPS_API_KEY not found in Secrets.") if not address: raise gr.Error("Please enter a valid address.") gr.Info("Geocoding address...") # 1. Geocoding API geo_url = "https://maps.googleapis.com/maps/api/geocode/json" geo_params = {"address": address, "key": gmaps_api_key} geo_data = requests.get(geo_url, params=geo_params).json() if geo_data['status'] != 'OK': error_detail = geo_data.get('error_message', geo_data['status']) raise gr.Error(f"Maps API Error: {error_detail}") lat = geo_data['results'][0]['geometry']['location']['lat'] lng = geo_data['results'][0]['geometry']['location']['lng'] gr.Info("Fetching Street View imagery...") # 2. Street View Static API sv_url = "https://maps.googleapis.com/maps/api/streetview" sv_params = { "size": "640x640", "location": f"{lat},{lng}", "fov": "80", "pitch": "0", "key": gmaps_api_key } sv_res = requests.get(sv_url, params=sv_params) # --- ALIGNED IF/ELSE BLOCK --- if sv_res.status_code == 200: gr.Info("Image successfully loaded!") return Image.open(BytesIO(sv_res.content)) else: error_msg = sv_res.text if sv_res.text else "No reason provided." raise gr.Error(f"Maps 403 Error: {error_msg}") # --- 3. Spatial Drawing Engine --- def draw_forensic_targets(image, report_text): annotated_img = image.copy() draw = ImageDraw.Draw(annotated_img) width, height = annotated_img.size coords = re.findall(r'\[(\d+),\s*(\d+)\]', report_text) for y_str, x_str in coords: y = int((int(y_str) / 1000) * height) x = int((int(x_str) / 1000) * width) r = 15 draw.ellipse((x - r, y - r, x + r, y + r), outline="red", width=5) return annotated_img # --- 4. Robotics-ER Core Engine --- def ground_truth_engine(past_img, present_img, audit_level, progress=gr.Progress()): progress(0, desc="Waking Spatial Sentinel...") api_key = os.environ.get("GOOGLE_API_KEY") if not api_key: return None, "
ERROR: API Key Missing
", "Set GOOGLE_API_KEY in Secrets." client = genai.Client(api_key=api_key) if past_img is None or present_img is None: return None, "
Missing Images
", "Please upload both images." audit_directives = { "Standard Audit": "Perform a structural comparison of roofing, siding, and landscaping.", "Deep Forensic": """Perform an exhaustive spatial audit. Identify subtle wear (shingles, foundation, paint) and biological encroachment. MANDATORY: Return center points for top 3 changes as [y, x] normalized coordinates (0-1000).""" } directive = audit_directives.get(audit_level) progress(0.2, desc=f"Running {audit_level}...") prompt = f""" SYSTEM INSTRUCTION: You are a structural forensic agent. DIRECTIVE: {directive} TASK: Compare 'Past Condition' vs 'Current Condition'. Return a structured report. YOU MUST CONCLUDE with exactly one of these phrases: 'Maintenance Trajectory: IMPROVING', 'Maintenance Trajectory: STABLE', or 'Maintenance Trajectory: DECLINING'. """ try: progress(0.5, desc="Robotics-ER is scanning for spatial anomalies...") response = client.models.generate_content( model="gemini-robotics-er-1.5-preview", contents=[prompt, past_img, present_img] ) text_out = response.text badge_html = "
➖ Trajectory: STABLE
" if "IMPROVING" in text_out.upper(): badge_html = "
📈 Trajectory: IMPROVING
" elif "DECLINING" in text_out.upper(): badge_html = "
📉 Trajectory: DECLINING
" progress(0.8, desc="Annotating image...") final_present_img = present_img if "Deep Forensic" in audit_level: final_present_img = draw_forensic_targets(present_img, text_out) progress(1.0, desc="Audit Finalized.") return (past_img, final_present_img), badge_html, text_out except Exception as e: return None, "
Analysis Failed
", str(e) # --- UI LAYOUT --- with gr.Blocks(theme=gr.themes.Soft(), css=custom_css) as demo: gr.Markdown("# 🏠 GroundTruth AI: Spatial Property Sentinel") # --- NEW: Address Search Bar --- with gr.Row(): address_input = gr.Textbox(label="Property Address", placeholder="e.g., 1600 Pennsylvania Avenue NW, Washington, DC", scale=4) fetch_btn = gr.Button("🌍 Fetch Current Street View", variant="secondary", scale=1) # --- TOP ROW: The "Input Console" --- with gr.Row(): with gr.Column(scale=1): p_img = gr.Image(label="Past Condition (Manual Upload)", type="pil") with gr.Column(scale=1): c_img = gr.Image(label="Current Condition (Auto-Fetched)", type="pil") with gr.Column(scale=1): audit_mode = gr.Radio( choices=["Standard Audit", "Deep Forensic"], value="Standard Audit", label="Select Audit Precision" ) submit = gr.Button("Analyze Property Trajectory", variant="primary", size="lg") # --- MIDDLE ROW: The "Hero Output" --- with gr.Row(): with gr.Column(scale=1): status_badge = gr.HTML() slider_out = gr.ImageSlider(label="Spatial Delta Comparison", type="pil", height=600) # --- BOTTOM ROW: The "Forensic Data" --- with gr.Row(): with gr.Column(scale=1): with gr.Accordion("View Detailed Forensic Data", open=False): report_out = gr.Markdown() # --- EVENTS --- # 1. Wire the Fetch Button to the Maps API and output to the Current Image box fetch_btn.click( fn=fetch_streetview, inputs=[address_input], outputs=[c_img] ) # 2. Wire the Analyze Button to the Robotics-ER Engine submit.click( fn=ground_truth_engine, inputs=[p_img, c_img, audit_mode], outputs=[slider_out, status_badge, report_out], show_progress="minimal" ) if __name__ == "__main__": demo.launch(server_name="0.0.0.0", server_port=7860)