""" 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"""

āŒ Error

Code: {error.get('code', 'UNKNOWN')}

Message: {error.get('message', 'An unknown error occurred')}

{f"

Retry Possible: {'Yes' if error.get('retry_possible', False) else 'No'}

" if 'retry_possible' in error else ''}
""" 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 = """

āŒ No Data

Construction plan data is missing from the response.

""" 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 = "

šŸ—ļø House Specifications

House specifications data unavailable

" # Format recommendations try: recommendations_html = format_recommendations_display(recommendations) except Exception as e: logger.error(f"Error formatting recommendations: {e}") recommendations_html = f"

šŸ“‹ Recommendations

{str(recommendations)[:1000]}
" # Format costs try: costs_html = format_costs_display(costs) except Exception as e: logger.error(f"Error formatting costs: {e}") costs_html = f"

šŸ’° Costs

{str(costs)[:1000]}
" # 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"

šŸ“ Facilities

{str(facilities)[:1000]}
" 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 = "

šŸŽØ Visualization

Visualization data unavailable

" 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"""

āŒ Unexpected Error

Message: {str(e)}

Please check the logs for more details.

""" 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"""

šŸ“Š Construction Plan Summary

šŸ“‹ Project Information

Building Type: {building_type.replace('_', ' ').title()}

Location: {location_name}

Coordinates: {lat:.4f}°N, {lon:.4f}°E

{f"

Building Area: {building_area:.2f} sq meters

" if building_area else ""}

Generated: {generated_at}

šŸ›”ļø Overall Risk Assessment

{overall_risk}

""" # Critical concerns critical_concerns = summary.get('critical_concerns', []) if isinstance(summary, dict) else getattr(summary, 'critical_concerns', []) if critical_concerns: html += """

āš ļø Critical Concerns

""" # Key recommendations key_recommendations = summary.get('key_recommendations', []) if isinstance(summary, dict) else getattr(summary, 'key_recommendations', []) if key_recommendations: html += """

šŸŽÆ Key Recommendations

    """ for rec in key_recommendations: html += f"
  1. {rec}
  2. " html += """
""" # 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 += """

šŸ“ Building-Specific Notes

""" html += """
""" 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="

Click 'Generate Construction Plan' to begin

" ) with gr.Tab("šŸ›”ļø Risk Assessment"): risk_output = gr.HTML( label="Risk Assessment", value="

Risk assessment will appear here

" ) 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="

Architectural visualization will appear here

" ) with gr.Tab("šŸ—ļø House Specifications"): house_specs_output = gr.HTML( label="House Specifications", value="

House specifications will appear here

" ) with gr.Tab("šŸ“‹ Recommendations"): recommendations_output = gr.HTML( label="Construction Recommendations", value="

Construction recommendations will appear here

" ) with gr.Tab("šŸ’° Costs"): costs_output = gr.HTML( label="Material Costs", value="

Material cost estimates will appear here

" ) with gr.Tab("šŸ“ Facilities"): facilities_output = gr.HTML( label="Critical Facilities", value="

Critical facilities information will appear here

" ) # 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