Spaces:
Sleeping
Sleeping
| 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, "<div class='declining'>ERROR: API Key Missing</div>", "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, "<div class='stable'>Missing Images</div>", "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.6-preview", | |
| contents=[prompt, past_img, present_img] | |
| ) | |
| text_out = response.text | |
| badge_html = "<div class='stable'>➖ Trajectory: STABLE</div>" | |
| if "IMPROVING" in text_out.upper(): | |
| badge_html = "<div class='improving'>📈 Trajectory: IMPROVING</div>" | |
| elif "DECLINING" in text_out.upper(): | |
| badge_html = "<div class='declining'>📉 Trajectory: DECLINING</div>" | |
| 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, "<div class='declining'>Analysis Failed</div>", 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) |