"""Streamlit frontend - Optimized One-Page UI for Land Redistribution Algorithm. Single-page design with: - Left: Configuration + Input - Center: Action + Status - Right: Results + Visualization """ import streamlit as st import requests import json import plotly.graph_objects as go from plotly.subplots import make_subplots import pandas as pd from typing import Dict, Any import matplotlib.pyplot as plt from shapely.geometry import shape, Polygon import numpy as np from plotly.subplots import make_subplots import pandas as pd from typing import Dict, Any import os from dotenv import load_dotenv # Load environment variables load_dotenv() # Configuration - Support both local and production deployment API_URL = os.getenv("API_URL", "http://localhost:8000") # Page config - Wide layout for one-page design st.set_page_config( page_title="Land Redistribution Optimizer", page_icon="🏘️", layout="wide", initial_sidebar_state="collapsed" ) # Custom CSS for better styling st.markdown(""" """, unsafe_allow_html=True) # Header st.markdown("""

🏘️ Land Redistribution Optimizer

NSGA-II Grid Optimization + OR-Tools Block Subdivision

""", unsafe_allow_html=True) # Initialize session state if 'land_plot' not in st.session_state: st.session_state.land_plot = None if 'result' not in st.session_state: st.session_state.result = None if 'status' not in st.session_state: st.session_state.status = 'ready' # Main layout: 3 columns col_config, col_action, col_result = st.columns([1.2, 1, 2]) # ==================== COLUMN 1: Configuration ==================== with col_config: st.markdown("### ⚙️ Configuration") # Quick Presets with st.expander("🎯 Quick Presets", expanded=True): preset = st.selectbox( "Choose a preset:", ["Custom", "🚀 Fastest", "⚖️ Balanced", "🏆 Best Quality"], help="Select a preset or use Custom to set your own values" ) # Apply preset values if preset == "🚀 Fastest": default_pop = 20 default_gen = 50 default_ort = 0.5 elif preset == "⚖️ Balanced": default_pop = 50 default_gen = 75 default_ort = 5.0 elif preset == "🏆 Best Quality": default_pop = 150 default_gen = 150 default_ort = 15.0 else: # Custom default_pop = 50 default_gen = 50 default_ort = 5.0 # Grid Optimization Parameters with st.expander("🔲 Grid Optimization", expanded=True): st.markdown("**Spacing (meters):**") c1, c2 = st.columns(2) with c1: spacing_min = st.number_input( "Min", min_value=30.0, max_value=150.0, value=50.0, step=5.0, help="Minimum grid spacing" ) with c2: spacing_max = st.number_input( "Max", min_value=30.0, max_value=200.0, value=100.0, step=5.0, help="Maximum grid spacing" ) st.markdown("**Rotation Angle (degrees):**") c1, c2 = st.columns(2) with c1: angle_min = st.number_input( "Min Angle", min_value=0.0, max_value=90.0, value=0.0, step=1.0, help="Minimum rotation angle" ) with c2: angle_max = st.number_input( "Max Angle", min_value=0.0, max_value=90.0, value=90.0, step=1.0, help="Maximum rotation angle" ) # Subdivision Parameters with st.expander("📐 Lot Subdivision", expanded=True): st.markdown("**Lot Width (meters):**") c1, c2, c3 = st.columns(3) with c1: min_lot_width = st.number_input( "Min", min_value=10.0, max_value=40.0, value=20.0, step=1.0, help="Minimum lot width" ) with c2: target_lot_width = st.number_input( "Target", min_value=20.0, max_value=100.0, value=40.0, step=5.0, help="Target lot width" ) with c3: max_lot_width = st.number_input( "Max", min_value=40.0, max_value=120.0, value=80.0, step=5.0, help="Maximum lot width" ) # Optimization Parameters with st.expander("⚡ Optimization", expanded=False): st.markdown("**NSGA-II Genetic Algorithm:**") c1, c2 = st.columns(2) with c1: population_size = st.number_input( "Population Size", min_value=20, max_value=200, value=default_pop, step=10, help="Number of solutions per generation" ) with c2: generations = st.number_input( "Generations", min_value=50, max_value=500, value=default_gen, step=10, help="Number of evolution iterations" ) st.markdown("**OR-Tools Solver:**") ortools_time_limit = st.number_input( "Time per Block (seconds)", min_value=0.1, max_value=60.0, value=default_ort, step=0.1, help="Maximum time for solving each block" ) # Show time estimate est_time = (population_size * generations) / 50 if est_time > 60: st.info(f"⏱️ Estimated time: ~{est_time//60:.0f} minutes") else: st.info(f"⏱️ Estimated time: ~{est_time:.0f} seconds") if est_time > 600: st.warning("⚠️ May timeout (>10 min). Consider reducing parameters.") # Infrastructure Parameters with st.expander("🏗️ Infrastructure", expanded=False): road_width = st.number_input( "Road Width (m)", min_value=3.0, max_value=10.0, value=6.0, step=0.5, help="Width of roads between blocks" ) block_depth = st.number_input( "Block Depth (m)", min_value=30.0, max_value=100.0, value=50.0, step=5.0, help="Depth of each block" ) # ==================== COLUMN 2: Input & Action ==================== with col_action: st.markdown("### 📍 Land Plot") # Input method selection input_method = st.radio( "Input method:", ["Sample", "DXF Upload", "GeoJSON Upload", "Manual"], horizontal=False ) if input_method == "Sample": # Predefined sample sample_type = st.selectbox( "Sample type:", ["Rectangle 100x100", "L-Shape", "Irregular", "Large Site"] ) if sample_type == "Rectangle 100x100": coords = [[[0, 0], [100, 0], [100, 100], [0, 100], [0, 0]]] elif sample_type == "L-Shape": coords = [[[0, 0], [60, 0], [60, 40], [40, 40], [40, 100], [0, 100], [0, 0]]] elif sample_type == "Irregular": coords = [[[0, 0], [80, 10], [100, 50], [90, 100], [20, 90], [0, 0]]] else: # Large Site coords = [[ [0, 0], [950, 50], [1000, 800], [400, 1100], [100, 900], [-50, 400], [0, 0] ]] st.session_state.land_plot = { "type": "Polygon", "coordinates": coords, "properties": {"name": sample_type} } elif input_method == "DXF Upload": st.info("📐 Upload DXF file containing site boundary (closed polyline)") uploaded = st.file_uploader( "DXF file", type=['dxf'], key="dxf_upload", help="File should contain closed LWPOLYLINE or POLYLINE for site boundary" ) if uploaded: with st.spinner("⏳ Parsing DXF..."): try: # Upload to backend API files = {"file": (uploaded.name, uploaded.getvalue(), "application/dxf")} response = requests.post(f"{API_URL}/api/upload-dxf", files=files) if response.status_code == 200: data = response.json() st.session_state.land_plot = data['polygon'] st.success(f"✅ {data['message']}") st.info(f"📊 Area: {data['area']:.2f} m²") else: st.error(f"Failed to parse DXF: {response.text}") st.session_state.land_plot = None except Exception as e: st.error(f"Error uploading DXF: {str(e)}") st.session_state.land_plot = None elif input_method == "GeoJSON Upload": uploaded = st.file_uploader("GeoJSON file", type=['json', 'geojson'], key="geojson_upload") if uploaded: try: data = json.load(uploaded) if data['type'] == 'FeatureCollection': st.session_state.land_plot = data['features'][0]['geometry'] else: st.session_state.land_plot = data st.success(f"✅ Loaded {uploaded.name}") except Exception as e: st.error(f"Invalid file: {e}") st.session_state.land_plot = None else: # Manual coords_input = st.text_area( "Coordinates (JSON):", '''[ [0, 0], [950, 50], [1000, 800], [400, 1100], [100, 900], [-50, 400], [0, 0] ]''', height=150 ) try: coords = json.loads(coords_input) st.session_state.land_plot = { "type": "Polygon", "coordinates": [coords], "properties": {} } except: st.error("Invalid JSON") # Preview if st.session_state.land_plot: with st.expander("📋 Preview", expanded=False): st.json(st.session_state.land_plot, expanded=False) st.markdown("---") # Status & Action st.markdown("### 🚀 Execute") # Status indicator status = st.session_state.status if status == 'ready': st.success("✅ Ready to optimize") elif status == 'running': st.warning("⏳ Processing...") elif status == 'complete': st.success("✅ Complete!") else: st.error("❌ Error occurred") # Run button if st.button("🚀 Run Optimization", type="primary", use_container_width=True, disabled=st.session_state.land_plot is None): st.session_state.status = 'running' config = { "spacing_min": spacing_min, "spacing_max": spacing_max, "angle_min": angle_min, "angle_max": angle_max, "min_lot_width": min_lot_width, "max_lot_width": max_lot_width, "target_lot_width": target_lot_width, "road_width": road_width, "block_depth": block_depth, "population_size": population_size, "generations": generations, "ortools_time_limit": ortools_time_limit } with st.spinner("Running NSGA-II + OR-Tools..."): try: # Show progress information progress_text = st.empty() progress_text.info(f"🔄 Starting optimization with {population_size} population × {generations} generations...") response = requests.post( f"{API_URL}/api/optimize", json={ "config": config, "land_plots": [st.session_state.land_plot] }, timeout=600 # Increased to 10 minutes ) progress_text.empty() if response.status_code == 200: st.session_state.result = response.json() st.session_state.status = 'complete' st.rerun() else: st.session_state.status = 'error' st.error(f"API Error: {response.text[:200]}") except requests.exceptions.Timeout: st.session_state.status = 'error' st.error(f"⏱️ Optimization timed out after 10 minutes. Try reducing Population ({population_size}) or Generations ({generations}).") except requests.exceptions.ConnectionError: st.session_state.status = 'error' st.error("Cannot connect to API. Is backend running on port 8000?") except Exception as e: st.session_state.status = 'error' st.error(f"Error: {str(e)}") # Reset button if st.session_state.result: if st.button("🔄 Reset", use_container_width=True): st.session_state.result = None st.session_state.status = 'ready' st.rerun() # ==================== COLUMN 3: Results ==================== with col_result: st.markdown("### 📊 Results") if st.session_state.result is None: # Show placeholder with input preview st.info("Run optimization to see results here") # Show input polygon preview if st.session_state.land_plot: coords = st.session_state.land_plot['coordinates'][0] xs = [c[0] for c in coords] ys = [c[1] for c in coords] fig = go.Figure() fig.add_trace(go.Scatter( x=xs, y=ys, fill='toself', fillcolor='rgba(100, 126, 234, 0.2)', line=dict(color='#667eea', width=2), name='Input Land' )) fig.update_layout( height=400, margin=dict(l=20, r=20, t=40, b=20), title="Input Land Plot", showlegend=False ) fig.update_yaxes(scaleanchor="x", scaleratio=1) st.plotly_chart(fig, use_container_width=True) else: result = st.session_state.result stats = result.get('statistics', {}) # Metrics row m1, m2, m3, m4 = st.columns(4) with m1: st.metric("🔲 Blocks", stats.get('total_blocks', 0)) with m2: st.metric("🏠 Lots", stats.get('total_lots', 0)) with m3: st.metric("🌳 Parks", stats.get('total_parks', 0)) with m4: st.metric("📏 Avg Width", f"{stats.get('avg_lot_width', 0):.1f}m") # Optimized parameters st.markdown("**Optimized Parameters:**") p1, p2 = st.columns(2) with p1: st.info(f"🔲 Spacing: **{stats.get('optimal_spacing', 0):.1f}m**") with p2: st.info(f"📐 Angle: **{stats.get('optimal_angle', 0):.1f}°**") p1, p2 = st.columns(2) with p1: st.info(f"🔲 Spacing: **{stats.get('optimal_spacing', 0):.1f}m**") with p2: st.info(f"📐 Angle: **{stats.get('optimal_angle', 0):.1f}°**") # === Notebook-Style Visualization (Matplotlib) === st.markdown("### 🗺️ Master Plan Visualization") def plot_notebook_style(result_data): """ Replicate the Detailed 1/500 Planning Plot. Includes: Roads, Setbacks, Zoning, Loop Network, Transformers, Drainage. """ try: def plot_geometry(geom, **kwargs): """Helper to plot Polygon or MultiPolygon.""" if geom.geom_type == 'Polygon': xs, ys = geom.exterior.xy ax.fill(xs, ys, **kwargs) elif geom.geom_type == 'MultiPolygon': for poly in geom.geoms: xs, ys = poly.exterior.xy ax.fill(xs, ys, **kwargs) def plot_outline(geom, **kwargs): """Helper to plot outline of Polygon or MultiPolygon.""" if geom.geom_type == 'Polygon': xs, ys = geom.exterior.xy ax.plot(xs, ys, **kwargs) elif geom.geom_type == 'MultiPolygon': for poly in geom.geoms: xs, ys = poly.exterior.xy ax.plot(xs, ys, **kwargs) # Setup figure fig, ax = plt.subplots(figsize=(12, 12)) ax.set_aspect('equal') ax.set_facecolor('#f0f0f0') # Retrieve features from final layout (Stage 3 includes everything) features = result_data.get('final_layout', {}).get('features', []) # 1. Draw Roads & Sidewalks (Layer 0) for f in features: if f['properties'].get('type') == 'road_network': geom = shape(f['geometry']) if not geom.is_empty: plot_geometry(geom, color='#607d8b', alpha=0.3, label='Hạ tầng giao thông') # 2. Draw Commercial Lots & Setbacks (Layer 1) for f in features: props = f['properties'] ftype = props.get('type') if ftype == 'lot': geom = shape(f['geometry']) plot_outline(geom, color='black', linewidth=0.5) plot_geometry(geom, color='#fff9c4', alpha=0.5) elif ftype == 'setback': geom = shape(f['geometry']) plot_outline(geom, color='red', linestyle='--', linewidth=0.8, alpha=0.7) # 3. Draw Service / Technical Areas (Layer 2) for f in features: props = f['properties'] ftype = props.get('type') geom = shape(f['geometry']) if ftype == 'xlnt': plot_geometry(geom, color='#b2dfdb', alpha=0.9) ax.text(geom.centroid.x, geom.centroid.y, "XLNT", ha='center', fontsize=8, color='black', weight='bold') elif ftype == 'service': plot_geometry(geom, color='#d1c4e9', alpha=0.9) ax.text(geom.centroid.x, geom.centroid.y, "Điều hành", ha='center', fontsize=8, color='black', weight='bold') elif ftype == 'park': plot_geometry(geom, color='#f6ffed', alpha=0.5) plot_outline(geom, color='green', linewidth=0.5, linestyle=':') # 4. Draw Electrical Infrastructure (Loop) for f in features: if f['properties'].get('type') == 'connection': line = shape(f['geometry']) xs, ys = line.xy ax.plot(xs, ys, color='blue', linestyle='-', linewidth=0.5, alpha=0.4) # 5. Draw Transformers for f in features: if f['properties'].get('type') == 'transformer': pt = shape(f['geometry']) ax.scatter(pt.x, pt.y, c='red', marker='^', s=100, zorder=10) # 6. Draw Drainage (Arrows) for i, f in enumerate([feat for feat in features if feat['properties'].get('type') == 'drainage']): if i % 3 == 0: # Sample to avoid clutter line = shape(f['geometry']) # Shapely LineString to Arrow start = line.coords[0] end = line.coords[1] dx = end[0] - start[0] dy = end[1] - start[1] ax.arrow(start[0], start[1], dx, dy, head_width=5, head_length=5, fc='cyan', ec='cyan', alpha=0.6) # Title ax.set_title("QUY HOẠCH CHI TIẾT 1/500 (PRODUCTION READY)\n" "Bao gồm: Đường phân cấp, Vạt góc, Chỉ giới XD, Điện mạch vòng, Thoát nước tự chảy", fontsize=14) # Custom Legend from matplotlib.lines import Line2D custom_lines = [Line2D([0], [0], color='#fff9c4', lw=4), Line2D([0], [0], color='red', linestyle='--', lw=1), Line2D([0], [0], color='#607d8b', lw=4), Line2D([0], [0], color='blue', lw=1), Line2D([0], [0], marker='^', color='w', markerfacecolor='red', markersize=10), Line2D([0], [0], color='cyan', lw=1, marker='>')] ax.legend(custom_lines, ['Đất CN', 'Chỉ giới XD (Setback)', 'Đường giao thông', 'Cáp điện ngầm (Loop)', 'Trạm biến áp', 'Hướng thoát nước'], loc='lower right') plt.tight_layout() return fig except Exception as e: st.error(f"Plotting error: {e}") return None # Display Plot fig = plot_notebook_style(result) if fig: st.pyplot(fig) # Visualization (Plotly) stages = result.get('stages', []) if len(stages) >= 2: fig = make_subplots( rows=1, cols=2, subplot_titles=('Stage 1: Grid Optimization', 'Stage 2: Subdivision'), horizontal_spacing=0.05 ) # Stage 1: Grid blocks for feature in stages[0]['geometry']['features']: coords = feature['geometry']['coordinates'][0] xs = [c[0] for c in coords] ys = [c[1] for c in coords] fig.add_trace(go.Scatter( x=xs, y=ys, fill='toself', fillcolor='rgba(100, 126, 234, 0.5)', line=dict(color='#667eea', width=1), showlegend=False, hoverinfo='skip' ), row=1, col=1) # Stage 2: Lots and parks for feature in stages[1]['geometry']['features']: coords = feature['geometry']['coordinates'][0] xs = [c[0] for c in coords] ys = [c[1] for c in coords] ftype = feature['properties'].get('type', 'lot') color = 'rgba(255, 152, 0, 0.7)' if ftype == 'lot' else 'rgba(76, 175, 80, 0.7)' line_color = '#ff9800' if ftype == 'lot' else '#4caf50' fig.add_trace(go.Scatter( x=xs, y=ys, fill='toself', fillcolor=color, line=dict(color=line_color, width=1), showlegend=False, hoverinfo='text', text=ftype.title() ), row=1, col=2) fig.update_layout( height=450, margin=dict(l=20, r=20, t=40, b=20), showlegend=False ) fig.update_xaxes(scaleanchor="y", scaleratio=1) fig.update_yaxes(scaleanchor="x", scaleratio=1) st.plotly_chart(fig, use_container_width=True) # Legend st.markdown("""
Grid Blocks
Residential Lots
Parks
""", unsafe_allow_html=True) # Download section st.markdown("---") st.markdown("**📥 Download Results:**") d1, d2, d3 = st.columns(3) with d1: if result.get('final_layout'): st.download_button( "📄 GeoJSON", data=json.dumps(result['final_layout'], indent=2), file_name="layout.geojson", mime="application/json", use_container_width=True ) with d2: st.download_button( "📊 Full Report", data=json.dumps(result, indent=2), file_name="report.json", mime="application/json", use_container_width=True ) with d3: # DXF Export button if st.button("📐 Export DXF", use_container_width=True, key="export_dxf"): with st.spinner("Generating DXF..."): try: response = requests.post( f"{API_URL}/api/export-dxf", json={"result": result} ) if response.status_code == 200: st.download_button( "⬇️ Download DXF", data=response.content, file_name="land_redistribution.dxf", mime="application/dxf", use_container_width=True, key="download_dxf" ) else: st.error("Failed to generate DXF") except Exception as e: st.error(f"DXF export error: {str(e)}")