|
|
""" |
|
|
Gradio 6 User Interface for Disaster Risk Construction Planner |
|
|
Main application entry point |
|
|
""" |
|
|
|
|
|
import os |
|
|
from pathlib import Path |
|
|
from dotenv import load_dotenv |
|
|
|
|
|
|
|
|
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 |
|
|
|
|
|
|
|
|
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: |
|
|
|
|
|
progress(0, desc="Initializing...") |
|
|
|
|
|
|
|
|
client = get_orchestrator_client() |
|
|
|
|
|
|
|
|
def update_progress(message: str): |
|
|
logger.info(message) |
|
|
progress(0.5, desc=message) |
|
|
|
|
|
|
|
|
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...") |
|
|
|
|
|
|
|
|
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, |
|
|
error_message, |
|
|
error_message, |
|
|
error_message, |
|
|
error_message, |
|
|
error_message, |
|
|
error_message, |
|
|
None, |
|
|
None |
|
|
) |
|
|
|
|
|
|
|
|
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) |
|
|
|
|
|
|
|
|
status_html = _format_status(construction_plan) |
|
|
|
|
|
|
|
|
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', {}) |
|
|
|
|
|
|
|
|
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'}") |
|
|
|
|
|
|
|
|
risk_html = format_risk_display(risk_assessment) |
|
|
|
|
|
|
|
|
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>" |
|
|
|
|
|
|
|
|
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>" |
|
|
|
|
|
|
|
|
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>" |
|
|
|
|
|
|
|
|
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 = "" |
|
|
|
|
|
|
|
|
if map_data: |
|
|
facilities_html += map_data |
|
|
|
|
|
|
|
|
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""" |
|
|
|
|
|
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 {} |
|
|
|
|
|
|
|
|
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 = 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 = 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 = 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 |
|
|
|
|
|
|
|
|
|
|
|
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. |
|
|
""" |
|
|
) |
|
|
|
|
|
|
|
|
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 |
|
|
|
|
|
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) |
|
|
""" |
|
|
) |
|
|
|
|
|
|
|
|
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>" |
|
|
) |
|
|
|
|
|
|
|
|
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( |
|
|
|
|
|
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 |
|
|
|
|
|
|
|
|
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 |
|
|
|