dexteredep's picture
update readme
52d0a6b
"""
Gradio 6 User Interface for Disaster Risk Construction Planner
Main application entry point
"""
import os
from pathlib import Path
from dotenv import load_dotenv
# Load environment variables from .env file
env_path = Path(__file__).parent / '.env'
load_dotenv(dotenv_path=env_path)
import gradio as gr
import asyncio
import logging
import json
from typing import Optional, Tuple
from orchestrator_client import get_orchestrator_client
from components import (
format_risk_display,
format_recommendations_display,
format_costs_display,
format_facilities_map,
format_visualization_display,
format_house_specs_display
)
from models import Coordinates
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def process_construction_plan(
building_type: str,
latitude: float,
longitude: float,
building_area: Optional[float] = None,
progress=gr.Progress()
) -> Tuple[str, str, str, str, str, str, str, str, str]:
"""
Handler function to process construction plan request
Args:
building_type: Selected building type
latitude: Latitude coordinate
longitude: Longitude coordinate
building_area: Optional building area in square meters
progress: Gradio progress tracker
Returns:
Tuple of formatted results for each tab (status, risk, house_specs, recommendations, costs, facilities, visualization, visualization_image, visualization_download)
"""
try:
# Update progress
progress(0, desc="Initializing...")
# Get orchestrator client
client = get_orchestrator_client()
# Progress callback
def update_progress(message: str):
logger.info(message)
progress(0.5, desc=message)
# Call orchestrator agent
progress(0.2, desc="Connecting to Orchestrator Agent...")
result = asyncio.run(
client.generate_construction_plan(
building_type=building_type,
latitude=latitude,
longitude=longitude,
building_area=building_area,
progress_callback=update_progress
)
)
progress(0.9, desc="Formatting results...")
# Check for errors
if not result.get('success', False):
error = result.get('error', {})
error_message = f"""
<div style="padding: 20px; background: #fef2f2; border-radius: 8px; border: 2px solid #dc2626;">
<h2 style="color: #dc2626; margin-top: 0;">❌ Error</h2>
<p><strong>Code:</strong> {error.get('code', 'UNKNOWN')}</p>
<p><strong>Message:</strong> {error.get('message', 'An unknown error occurred')}</p>
{f"<p><strong>Retry Possible:</strong> {'Yes' if error.get('retry_possible', False) else 'No'}</p>" if 'retry_possible' in error else ''}
</div>
"""
return (
error_message, # status
error_message, # risk
error_message, # house_specs
error_message, # recommendations
error_message, # costs
error_message, # facilities
error_message, # visualization
None, # visualization_image
None # visualization_download
)
# Extract construction plan (keep as dict for easier handling)
construction_plan = result.get('construction_plan')
if not construction_plan:
error_message = """
<div style="padding: 20px; background: #fef2f2; border-radius: 8px;">
<h2 style="color: #dc2626;">❌ No Data</h2>
<p>Construction plan data is missing from the response.</p>
</div>
"""
return (error_message,) * 7 + (None, None)
# Format status/executive summary
status_html = _format_status(construction_plan)
# Extract data (work with dicts directly)
risk_assessment = construction_plan.get('risk_assessment', {})
house_specs = construction_plan.get('house_specifications', {})
recommendations = construction_plan.get('construction_recommendations', {})
costs = construction_plan.get('material_costs', {})
facilities = construction_plan.get('critical_facilities', {})
visualization = construction_plan.get('visualization', {})
# Debug logging
logger.info(f"Construction plan type: {type(construction_plan)}")
logger.info(f"Risk assessment type: {type(risk_assessment)}")
logger.info(f"Risk assessment keys: {risk_assessment.keys() if isinstance(risk_assessment, dict) else 'not a dict'}")
logger.info(f"Risk assessment success field: {risk_assessment.get('success') if isinstance(risk_assessment, dict) else 'N/A'}")
# Format risk display
risk_html = format_risk_display(risk_assessment)
# Format house specifications
try:
house_specs_html = format_house_specs_display(house_specs)
except Exception as e:
logger.error(f"Error formatting house specifications: {e}")
house_specs_html = "<div style='padding: 20px;'><h3>πŸ—οΈ House Specifications</h3><p>House specifications data unavailable</p></div>"
# Format recommendations
try:
recommendations_html = format_recommendations_display(recommendations)
except Exception as e:
logger.error(f"Error formatting recommendations: {e}")
recommendations_html = f"<div style='padding: 20px;'><h3>πŸ“‹ Recommendations</h3><pre>{str(recommendations)[:1000]}</pre></div>"
# Format costs
try:
costs_html = format_costs_display(costs)
except Exception as e:
logger.error(f"Error formatting costs: {e}")
costs_html = f"<div style='padding: 20px;'><h3>πŸ’° Costs</h3><pre>{str(costs)[:1000]}</pre></div>"
# Format facilities
try:
site_coords = Coordinates(latitude=latitude, longitude=longitude)
facilities_html, map_data = format_facilities_map(facilities, site_coords)
except Exception as e:
logger.error(f"Error formatting facilities: {e}")
facilities_html = f"<div style='padding: 20px;'><h3>πŸ“ Facilities</h3><pre>{str(facilities)[:1000]}</pre></div>"
map_data = ""
# Add map data to facilities HTML if available
if map_data:
facilities_html += map_data
# Format visualization
try:
visualization_html, visualization_image, visualization_download = format_visualization_display(visualization)
except Exception as e:
logger.error(f"Error formatting visualization: {e}")
visualization_html = "<div style='padding: 20px;'><h3>🎨 Visualization</h3><p>Visualization data unavailable</p></div>"
visualization_image = None
visualization_download = None
progress(1.0, desc="Complete!")
return (
status_html,
risk_html,
house_specs_html,
recommendations_html,
costs_html,
facilities_html,
visualization_html,
visualization_image,
visualization_download
)
except Exception as e:
logger.error(f"Error processing construction plan: {str(e)}", exc_info=True)
error_message = f"""
<div style="padding: 20px; background: #fef2f2; border-radius: 8px; border: 2px solid #dc2626;">
<h2 style="color: #dc2626; margin-top: 0;">❌ Unexpected Error</h2>
<p><strong>Message:</strong> {str(e)}</p>
<p>Please check the logs for more details.</p>
</div>
"""
return (error_message,) * 7 + (None, None)
def _format_status(construction_plan) -> str:
"""Format executive summary and status"""
# Handle both dict and object formats
if isinstance(construction_plan, dict):
metadata = construction_plan.get('metadata', {})
summary = construction_plan.get('executive_summary', {})
else:
metadata = construction_plan.metadata if hasattr(construction_plan, 'metadata') else {}
summary = construction_plan.executive_summary if hasattr(construction_plan, 'executive_summary') else {}
# Extract values safely
building_type = metadata.get('building_type', 'Unknown') if isinstance(metadata, dict) else getattr(metadata, 'building_type', 'Unknown')
location = metadata.get('location', {}) if isinstance(metadata, dict) else getattr(metadata, 'location', {})
location_name = location.get('name', 'Unknown') if isinstance(location, dict) else getattr(location, 'name', 'Unknown')
coordinates = metadata.get('coordinates', {}) if isinstance(metadata, dict) else getattr(metadata, 'coordinates', {})
lat = coordinates.get('latitude', 0) if isinstance(coordinates, dict) else getattr(coordinates, 'latitude', 0)
lon = coordinates.get('longitude', 0) if isinstance(coordinates, dict) else getattr(coordinates, 'longitude', 0)
building_area = metadata.get('building_area') if isinstance(metadata, dict) else getattr(metadata, 'building_area', None)
generated_at = metadata.get('generated_at', 'Unknown') if isinstance(metadata, dict) else getattr(metadata, 'generated_at', 'Unknown')
overall_risk = summary.get('overall_risk', 'Unknown') if isinstance(summary, dict) else getattr(summary, 'overall_risk', 'Unknown')
html = f"""
<div style="padding: 20px;">
<h2 style="margin-top: 0; color: #ffffff !important; font-weight: 700;">πŸ“Š Construction Plan Summary</h2>
<div style="background: white; padding: 20px; border-radius: 8px; margin-bottom: 20px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); border: 1px solid #e2e8f0;">
<h3 style="color: #1e40af !important; margin-top: 0; font-weight: 600;">πŸ“‹ Project Information</h3>
<p style="color: #0f172a !important; margin: 8px 0;"><strong style="color: #0f172a !important;">Building Type:</strong> {building_type.replace('_', ' ').title()}</p>
<p style="color: #0f172a !important; margin: 8px 0;"><strong style="color: #0f172a !important;">Location:</strong> {location_name}</p>
<p style="color: #0f172a !important; margin: 8px 0;"><strong style="color: #0f172a !important;">Coordinates:</strong> {lat:.4f}Β°N, {lon:.4f}Β°E</p>
{f"<p style='color: #0f172a !important; margin: 8px 0;'><strong style='color: #0f172a !important;'>Building Area:</strong> {building_area:.2f} sq meters</p>" if building_area else ""}
<p style="color: #0f172a !important; margin: 8px 0;"><strong style="color: #0f172a !important;">Generated:</strong> {generated_at}</p>
</div>
<div style="background: white; padding: 20px; border-radius: 8px; margin-bottom: 20px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); border: 1px solid #e2e8f0;">
<h3 style="color: #b91c1c !important; margin-top: 0; font-weight: 600;">πŸ›‘οΈ Overall Risk Assessment</h3>
<p style="font-size: 1.1em; line-height: 1.6; color: #0f172a !important;">{overall_risk}</p>
</div>
"""
# Critical concerns
critical_concerns = summary.get('critical_concerns', []) if isinstance(summary, dict) else getattr(summary, 'critical_concerns', [])
if critical_concerns:
html += """
<div style="background: white; padding: 20px; border-radius: 8px; margin-bottom: 20px; border-left: 5px solid #dc2626; box-shadow: 0 2px 4px rgba(0,0,0,0.1); border: 1px solid #fecaca;">
<h3 style="color: #dc2626 !important; margin-top: 0; font-weight: 600;">⚠️ Critical Concerns</h3>
<ul style="line-height: 1.8; color: #0f172a !important;">
"""
for concern in critical_concerns:
html += f"<li style='color: #0f172a !important;'><strong style='color: #dc2626 !important;'>{concern}</strong></li>"
html += """
</ul>
</div>
"""
# Key recommendations
key_recommendations = summary.get('key_recommendations', []) if isinstance(summary, dict) else getattr(summary, 'key_recommendations', [])
if key_recommendations:
html += """
<div style="background: white; padding: 20px; border-radius: 8px; margin-bottom: 20px; border-left: 5px solid #f59e0b; box-shadow: 0 2px 4px rgba(0,0,0,0.1); border: 1px solid #fde68a;">
<h3 style="color: #d97706 !important; margin-top: 0; font-weight: 600;">🎯 Key Recommendations</h3>
<ol style="line-height: 1.8; font-weight: 500; color: #0f172a !important;">
"""
for rec in key_recommendations:
html += f"<li style='color: #0f172a !important;'>{rec}</li>"
html += """
</ol>
</div>
"""
# Building-specific notes
building_specific_notes = summary.get('building_specific_notes', []) if isinstance(summary, dict) else getattr(summary, 'building_specific_notes', [])
if building_specific_notes:
html += """
<div style="background: white; padding: 20px; border-radius: 8px; margin-bottom: 20px; border-left: 5px solid #3b82f6; box-shadow: 0 2px 4px rgba(0,0,0,0.1); border: 1px solid #bfdbfe;">
<h3 style="color: #1e40af !important; margin-top: 0; font-weight: 600;">πŸ“ Building-Specific Notes</h3>
<ul style="line-height: 1.8; color: #0f172a !important;">
"""
for note in building_specific_notes:
html += f"<li style='color: #0f172a !important;'>{note}</li>"
html += """
</ul>
</div>
"""
html += """
</div>
"""
return html
# Create Gradio interface
def create_interface():
"""Create and configure Gradio interface with tabbed output"""
with gr.Blocks(
title="Filo - DRM focused Construction Planner"
) as app:
gr.Markdown(
"""
# πŸ—οΈ Filo - DRM focused Construction Planner
### Plan disaster-resistant construction in the Philippines
This tool helps you assess disaster risks, get construction recommendations,
estimate material costs, and locate critical facilities for your building project.
"""
)
# Input section at the top
gr.Markdown("### πŸ“ Project Details")
with gr.Row(equal_height=True):
with gr.Column(scale=2):
building_type = gr.Dropdown(
label="Building Type",
choices=[
"Residential - Single Family Home",
"Residential - Multi-Family Housing",
"Residential - High-Rise Apartment",
"Commercial - Office Building",
"Commercial - Retail/Shopping Center",
"Industrial - Warehouse/Factory",
"Institutional - School",
"Institutional - Hospital/Healthcare",
"Infrastructure - Bridge/Road",
"Mixed-Use Development"
],
value="Residential - Single Family Home",
info="Select the type of building you plan to construct"
)
with gr.Column(scale=1):
latitude = gr.Number(
label="Latitude",
minimum=4.0,
maximum=21.0,
value=14.5995,
info="4Β°N to 21Β°N"
)
with gr.Column(scale=1):
longitude = gr.Number(
label="Longitude",
minimum=116.0,
maximum=127.0,
value=120.9842,
info="116Β°E to 127Β°E"
)
with gr.Column(scale=1):
building_area = gr.Number(
label="Building Area (sq m)",
minimum=1,
value=100,
info="Optional for cost calculations"
)
with gr.Row():
with gr.Column(scale=1):
submit_btn = gr.Button(
"πŸš€ Generate Construction Plan",
variant="primary",
size="lg"
)
with gr.Column(scale=4):
pass # Empty space for alignment
gr.Markdown(
"""
**Note:** Coordinates must be within the Philippines.
Example locations: Manila (14.5995Β°N, 120.9842Β°E) β€’ Cebu (10.3157Β°N, 123.8854Β°E) β€’ Davao (7.1907Β°N, 125.4553Β°E)
"""
)
# Results section below
gr.Markdown("### πŸ“‹ Construction Plan Results")
with gr.Tabs() as tabs:
with gr.Tab("πŸ“Š Summary"):
status_output = gr.HTML(
label="Executive Summary",
value="<p style='text-align: center; color: #6b7280; padding: 40px;'>Click 'Generate Construction Plan' to begin</p>"
)
with gr.Tab("πŸ›‘οΈ Risk Assessment"):
risk_output = gr.HTML(
label="Risk Assessment",
value="<p style='text-align: center; color: #6b7280; padding: 40px;'>Risk assessment will appear here</p>"
)
with gr.Tab("🎨 Visualization"):
with gr.Column():
visualization_image_output = gr.Image(
label="Architectural Sketch",
type="filepath",
show_label=True,
height=512
)
visualization_download_output = gr.File(
label="Download Visualization",
visible=False
)
visualization_output = gr.HTML(
label="Visualization Details",
value="<p style='text-align: center; color: #6b7280; padding: 40px;'>Architectural visualization will appear here</p>"
)
with gr.Tab("πŸ—οΈ House Specifications"):
house_specs_output = gr.HTML(
label="House Specifications",
value="<p style='text-align: center; color: #6b7280; padding: 40px;'>House specifications will appear here</p>"
)
with gr.Tab("πŸ“‹ Recommendations"):
recommendations_output = gr.HTML(
label="Construction Recommendations",
value="<p style='text-align: center; color: #6b7280; padding: 40px;'>Construction recommendations will appear here</p>"
)
with gr.Tab("πŸ’° Costs"):
costs_output = gr.HTML(
label="Material Costs",
value="<p style='text-align: center; color: #6b7280; padding: 40px;'>Material cost estimates will appear here</p>"
)
with gr.Tab("πŸ“ Facilities"):
facilities_output = gr.HTML(
label="Critical Facilities",
value="<p style='text-align: center; color: #6b7280; padding: 40px;'>Critical facilities information will appear here</p>"
)
# Connect button to handler
submit_btn.click(
fn=process_construction_plan,
inputs=[building_type, latitude, longitude, building_area],
outputs=[
status_output,
risk_output,
house_specs_output,
recommendations_output,
costs_output,
facilities_output,
visualization_output,
visualization_image_output,
visualization_download_output
]
).then(
# Show download button if visualization is available
lambda download_path: gr.update(visible=download_path is not None, value=download_path),
inputs=[visualization_download_output],
outputs=[visualization_download_output]
)
gr.Markdown(
"""
---
### About This Tool
The Disaster Risk Construction Planner uses AI agents to:
- Assess disaster risks using Philippines Disaster Risk data
- Research construction best practices and guidelines
- Estimate material costs for disaster-resistant construction
- Locate critical facilities near your building site
**Disclaimer:** This tool provides guidance only. Always consult with licensed
engineers, architects, and local authorities for actual construction projects.
"""
)
return app
if __name__ == "__main__":
import os
# Get port from environment variable or use default
port = int(os.getenv("GRADIO_SERVER_PORT", "7860"))
app = create_interface()
try:
app.launch(
server_name="0.0.0.0",
server_port=port,
share=False,
theme=gr.themes.Soft()
)
except OSError as e:
if "Cannot find empty port" in str(e):
print(f"\n❌ Port {port} is already in use!")
print(f"\nTo use a different port, either:")
print(f" 1. Set environment variable: export GRADIO_SERVER_PORT=7861")
print(f" 2. Kill the existing process: kill $(lsof -t -i:{port})")
print(f" 3. Run with a different port: GRADIO_SERVER_PORT=7861 python app.py")
raise