from __future__ import annotations
import base64
import json
import os
import inspect
from pathlib import Path
from typing import Any
import gradio as gr
from src.board_export import create_board_artifacts
from src.cad_parser import candidate_to_selection, parse_dxf
from src.climate import fetch_climate_bundle
from src.diagrams import create_diagrams
from src.earth_reference import build_earth_reference
from src.evidence import evidence_to_html_table, make_evidence, reset_evidence_counter
from src.export import write_markdown_export
from src.geocoding import reverse_geocode_site
from src.geojson_parser import parse_geojson_file
from src.geometry import normalize_map_state, selection_from_lat_lon
from src.kml_parser import parse_kml_file
from src.location import extract_coordinates_from_text
from src.models import BoundaryCandidate, ReportBundle, SiteSelection
from src.osm_context import fetch_osm_context
from src.report import build_board_bundle, build_markdown_report
from src.sample_data import apply_chorwad_sample_fallbacks
from src.small_model_assistant import build_assistant_brief
from src.soil import fetch_soil
from src.sun import summarize_sun_wind
from src.topography import fetch_topography
from src.upload_limits import (
MAX_BOARD_PREVIEW_MB,
MAX_DXF_MB,
MAX_GEOJSON_MB,
MAX_KML_MB,
MAX_PDF_REFERENCE_MB,
validate_upload,
)
ROOT = Path(__file__).parent
ASSETS = ROOT / "assets"
def read_asset(name: str) -> str:
path = ASSETS / name
return path.read_text(encoding="utf-8") if path.exists() else ""
def image_preview_html(path: str | None) -> str:
if not path:
return """
Waiting for selected landPresentation board will appear here
Draw a site boundary, drop a pin radius, or upload CAD/KML/GeoJSON. Optional project fields can stay blank.
Boundary DeskTrace a site, drop a radius pin, or anchor uploaded CAD/KML/GeoJSON before generating the board pack.
Approximate trace
Map loading. Drawn boundaries are approximate; upload CAD/KML/GeoJSON for better accuracy.
"""
DELIVERABLES_HTML = """
Board PNG/PDFsheet-style artifact
6 diagramsclimate, context, edge, matrix
Evidencesource / limit / verification
Reporteditable workbook text
"""
SAMPLE_MAP_STATE = json.dumps(
{
"mode": "polygon",
"geometry": {
"type": "Polygon",
"coordinates": [
[
[70.24495, 21.00195],
[70.24605, 21.00210],
[70.24578, 21.00302],
[70.24470, 21.00286],
[70.24495, 21.00195],
]
],
},
}
)
def load_sample_site():
return (
"Chorwad Coastal Thesis Sample",
"Chorwad site - sample boundary",
"Architecture thesis / resort concept",
"Sample drawn boundary; replace with faculty CAD, KML/KMZ, GeoJSON, or real site survey for actual work.",
"21.00245, 70.24550",
(
"Use as validation notes only: check fishing/tourism activity, seasonal visitor movement, local material language, "
"coastal humidity, service access, and public/private edge conditions during site visit."
),
(
"Before treating the sample as a real submission, compare with Google Earth/KML or faculty CAD. Manually mark visible trees, "
"water edge, road approach, surrounding structures, open ground, drainage signs, and any construction activity."
),
SAMPLE_MAP_STATE,
)
def build_head_html() -> str:
shell_css = """
"""
leaflet_css = read_asset("leaflet.css")
leaflet_js = read_asset("leaflet.js")
if leaflet_css and leaflet_js:
return f"{shell_css}\n\n"
return shell_css + """
"""
HEAD_HTML = build_head_html()
def parse_dxf_candidates(dxf_file: Any):
if not dxf_file:
return gr.update(choices=[], value=None), "Upload a DXF to inspect boundary candidates.", None
try:
path = validate_upload(
_file_path(dxf_file),
allowed_suffixes={".dxf"},
max_mb=MAX_DXF_MB,
label="DXF",
)
parsed = parse_dxf(path)
candidates: list[BoundaryCandidate] = parsed["boundary_candidates"] # type: ignore[assignment]
choices = [(f"{c.id} | {c.layer_name} | {c.area_sqm:,.1f} sqm | {c.vertex_count} vertices", c.id) for c in candidates]
context = parsed.get("context_layers", {})
layers = parsed.get("layers", [])
summary = [
f"**DXF:** `{path.name}`",
f"**Detected unit:** {parsed.get('unit')}",
f"**Layers found:** {len(layers)}",
f"**Boundary candidates:** {len(candidates)}",
]
if context:
summary.append("**Context layers:** " + ", ".join(f"{k}: {v}" for k, v in context.items()))
if not candidates:
summary.append("No boundary candidate was found. Try drawing the boundary or exporting a simpler DXF with a site boundary layer.")
return gr.update(choices=choices, value=choices[0][1] if choices else None), "\n\n".join(summary), parsed
except Exception as exc: # noqa: BLE001
return gr.update(choices=[], value=None), f"DXF parse failed: {type(exc).__name__}: {exc}", None
def generate_site_analysis(
project_name: str,
site_name: str,
project_type: str,
boundary_source: str,
culture_notes: str,
earth_observation_notes: str,
map_state: str,
manual_location: str,
dxf_file: Any,
selected_candidate_id: str,
dxf_state: dict[str, Any] | None,
geojson_file: Any,
kml_file: Any,
pdf_file: Any,
):
reset_evidence_counter()
warnings: list[str] = []
try:
selection, setup_evidence, setup_warnings = _build_selection(
map_state=map_state,
manual_location=manual_location,
dxf_file=dxf_file,
selected_candidate_id=selected_candidate_id,
dxf_state=dxf_state,
geojson_file=geojson_file,
kml_file=kml_file,
)
warnings.extend(setup_warnings)
except Exception as exc: # noqa: BLE001
message = f"### Input Needed\n\n{exc}"
return None, None, None, message, None, None, None, None, None, None, None, "", message, ""
if pdf_file:
try:
validate_upload(
_file_path(pdf_file),
allowed_suffixes={".pdf"},
max_mb=MAX_PDF_REFERENCE_MB,
label="PDF reference",
)
except Exception as exc: # noqa: BLE001
warnings.append(f"PDF reference ignored: {type(exc).__name__}: {exc}")
warnings.append("PDF upload is reference-only in this MVP; geometry is not extracted from PDFs.")
evidence = list(setup_evidence)
evidence.append(
make_evidence(
category="Soil / ground",
finding="Soil and foundation are treated as site-visit/professional verification items in this MVP.",
source_name="Safety rule",
source_url="",
source_type="application policy",
resolution_or_scope="project-wide",
confidence="high",
limitation="No geotechnical survey or final foundation recommendation is generated.",
design_implication="Use only as a prompt to request soil reports or professional input.",
verification_needed="Obtain geotechnical/professional verification before foundation decisions.",
output_label="professional_verification_required",
)
)
if culture_notes:
evidence.append(
make_evidence(
category="Region / culture",
finding="User supplied local/culture/site-observation notes.",
source_name="User note",
source_url="",
source_type="user-provided",
resolution_or_scope="editable observation",
confidence="medium",
limitation="Not independently verified by the app.",
design_implication="Use as a prompt for site visit and studio discussion.",
verification_needed="Verify with site visit, local interviews, or faculty-approved sources.",
output_label="user_input",
)
)
earth_reference, earth_evidence = build_earth_reference(selection, earth_observation_notes or "")
evidence.extend(earth_evidence)
lat, lon = selection.anchor_lat, selection.anchor_lon
climate: dict[str, Any] = {"forecast": None, "recent_historical": None, "climate_normal": None}
osm_context: dict[str, Any] = {"counts": {}, "features": []}
topography: dict[str, Any] = {}
soil: dict[str, Any] = {}
site_identity: dict[str, Any] | None = None
sun_summary: dict[str, str] = {
"orientation_note": "Sun/orientation summary unavailable because no anchor coordinate is available.",
"east_west_note": "",
"wind_note": "",
"limitation": "Provide a real-world anchor coordinate for public-data analysis.",
}
if lat is not None and lon is not None:
site_identity, identity_evidence = reverse_geocode_site(lat, lon)
evidence.extend(identity_evidence)
climate, climate_evidence = fetch_climate_bundle(lat, lon)
evidence.extend(climate_evidence)
wind_dir = _current_wind_direction(climate)
sun_summary, sun_evidence = summarize_sun_wind(lat, lon, wind_dir)
evidence.extend(sun_evidence)
osm_context, osm_evidence = fetch_osm_context(selection)
evidence.extend(osm_evidence)
topography, topography_evidence = fetch_topography(selection)
evidence.extend(topography_evidence)
soil, soil_evidence = fetch_soil(selection)
evidence.extend(soil_evidence)
else:
warnings.append("No anchor coordinate was available, so climate, sun, OSM, terrain, and soil public-data layers were skipped.")
(
site_identity,
climate,
osm_context,
topography,
soil,
sample_evidence,
sample_warnings,
) = apply_chorwad_sample_fallbacks(
project_name=project_name,
site_name=site_name,
boundary_source=boundary_source,
site_identity=site_identity,
climate=climate,
osm_context=osm_context,
topography=topography,
soil=soil,
anchor_lat=selection.anchor_lat,
anchor_lon=selection.anchor_lon,
)
evidence.extend(sample_evidence)
warnings.extend(sample_warnings)
(
project_name,
site_name,
project_type,
boundary_source,
auto_label_warnings,
) = _resolve_project_fields(
project_name=project_name,
site_name=site_name,
project_type=project_type,
boundary_source=boundary_source,
selection=selection,
site_identity=site_identity,
)
warnings.extend(auto_label_warnings)
diagram_paths = create_diagrams(
selection,
climate,
osm_context,
sun_summary,
topography=topography,
soil=soil,
)
if len(diagram_paths) < 3:
warnings.append(
"One or more diagrams could not be generated; use the Markdown report and evidence table, then retry if needed."
)
assistant_md = build_assistant_brief(
selection=selection,
evidence_rows=evidence,
warnings=warnings,
project_type=project_type,
)
bundle = build_board_bundle(
project_name=project_name,
site_name=site_name,
selection=selection,
climate=climate,
osm_context=osm_context,
sun_summary=sun_summary,
site_identity=site_identity,
evidence_rows=evidence,
diagram_paths=diagram_paths,
topography=topography,
soil=soil,
warnings=warnings,
)
if boundary_source:
bundle.board_markdown += f"\n\n### Boundary Source\n\n{boundary_source}\n"
if culture_notes:
bundle.board_markdown += f"\n\n### Editable Local / Culture Notes\n\n{culture_notes}\n"
bundle.board_markdown += "\n\n" + earth_reference["markdown"]
bundle.board_markdown += "\n\n### Small Model Assistant / Template Brief\n\n" + assistant_md
report = build_markdown_report(project_name, site_name, selection, bundle)
export_path = write_markdown_export(report, project_name or site_name or "site-intelligence")
board_artifacts = create_board_artifacts(
project_name=project_name,
site_name=site_name,
project_type=project_type,
boundary_source=boundary_source,
culture_notes=culture_notes,
selection=selection,
climate=climate,
osm_context=osm_context,
sun_summary=sun_summary,
site_identity=site_identity,
evidence_rows=evidence,
diagram_paths=diagram_paths,
warnings=warnings,
topography=topography,
soil=soil,
)
evidence_html = evidence_to_html_table(evidence)
padded_paths = diagram_paths + [None] * (6 - len(diagram_paths))
warning_md = "### Warnings / Limits\n\n" + ("\n".join(f"- {w}" for w in warnings) if warnings else "- No additional warnings.")
return (
image_preview_html(board_artifacts["png"]),
board_artifacts["png"],
board_artifacts["pdf"],
report,
export_path,
padded_paths[0],
padded_paths[1],
padded_paths[2],
padded_paths[3],
padded_paths[4],
padded_paths[5],
evidence_html,
warning_md,
assistant_md,
)
def _build_selection(
*,
map_state: str,
manual_location: str,
dxf_file: Any,
selected_candidate_id: str,
dxf_state: dict[str, Any] | None,
geojson_file: Any,
kml_file: Any,
) -> tuple[SiteSelection, list[Any], list[str]]:
evidence = []
warnings = []
anchor = extract_coordinates_from_text(manual_location)
if dxf_file and selected_candidate_id and dxf_state:
validate_upload(
_file_path(dxf_file),
allowed_suffixes={".dxf"},
max_mb=MAX_DXF_MB,
label="DXF",
)
candidates: list[BoundaryCandidate] = dxf_state.get("boundary_candidates", [])
selected = next((c for c in candidates if c.id == selected_candidate_id), None)
if not selected:
raise ValueError("Select a DXF boundary candidate before generating.")
if not anchor:
anchor = _anchor_from_map_state(map_state)
if not anchor:
raise ValueError("CAD/DXF boundaries need an anchor lat/lon or map pin before public data can run.")
selection = candidate_to_selection(selected, anchor[0], anchor[1])
evidence.append(
make_evidence(
category="Boundary",
finding=f"DXF boundary candidate `{selected.id}` was selected from layer `{selected.layer_name}`.",
source_name=selected.source_file,
source_url="",
source_type="uploaded DXF",
resolution_or_scope=f"{selected.unit}; local CAD geometry",
confidence="medium",
limitation="CAD may use local coordinates and is not legal/cadastral verification.",
design_implication="Use for preliminary board area/perimeter and site-form understanding.",
verification_needed="Confirm with faculty drawing, survey, or official project documents.",
output_label="cad_derived",
)
)
context = dxf_state.get("context_layers") or {}
if context:
evidence.append(
make_evidence(
category="Existing conditions",
finding="DXF contains context layers: " + ", ".join(f"{k}: {v}" for k, v in context.items()),
source_name=selected.source_file,
source_url="",
source_type="uploaded DXF",
resolution_or_scope="CAD layer names and entity counts",
confidence="medium",
limitation="Layer names are user/CAD-authored and may be inconsistent.",
design_implication="Use layers as prompts for roads, water, vegetation, built-up, and contour checks.",
verification_needed="Visually inspect CAD and verify conditions on site.",
output_label="cad_derived",
)
)
return selection, evidence, warnings
if geojson_file:
geojson_path = validate_upload(
_file_path(geojson_file),
allowed_suffixes={".geojson", ".json"},
max_mb=MAX_GEOJSON_MB,
label="GeoJSON",
)
selection = parse_geojson_file(geojson_path)
evidence.append(
make_evidence(
category="Boundary",
finding="GeoJSON boundary was uploaded and treated as WGS84 geometry.",
source_name=geojson_path.name,
source_url="",
source_type="uploaded GeoJSON",
resolution_or_scope="uploaded polygon",
confidence="high",
limitation="Boundary is only as reliable as the uploaded GeoJSON source.",
design_implication="Use for boundary-based first-pass site analysis.",
verification_needed="Confirm with original source or site survey.",
output_label="user_input",
)
)
return selection, evidence, warnings
if kml_file:
kml_path = validate_upload(
_file_path(kml_file),
allowed_suffixes={".kml", ".kmz"},
max_mb=MAX_KML_MB,
label="KML/KMZ",
)
selection = parse_kml_file(kml_path)
evidence.append(
make_evidence(
category="Boundary",
finding="KML/KMZ boundary was uploaded and treated as Google Earth / WGS84 geometry.",
source_name=kml_path.name,
source_url="",
source_type="uploaded KML/KMZ",
resolution_or_scope="uploaded polygon; Google Earth/GIS-style coordinates",
confidence="medium",
limitation="Boundary is only as reliable as the exported KML/KMZ source and is not legal/cadastral verification.",
design_implication="Use for boundary-based first-pass site analysis and board generation.",
verification_needed="Confirm with faculty CAD, survey, or official project documents.",
output_label="user_input",
)
)
return selection, evidence, warnings
try:
if map_state and map_state.strip() and map_state.strip() != "{}":
selection = normalize_map_state(map_state)
evidence.append(
make_evidence(
category="Boundary",
finding=f"Map selection was created using `{selection.selection_type}` mode.",
source_name="User map input",
source_url="",
source_type="drawn map / pin",
resolution_or_scope=selection.accuracy_label,
confidence="low" if selection.selection_type == "pin_radius" else "medium",
limitation="Drawn and pin-radius boundaries are approximate.",
design_implication="Use for early studio context and verification planning.",
verification_needed="Verify with CAD, KML, GeoJSON, faculty drawing, or site survey.",
output_label="user_input",
)
)
return selection, evidence, warnings
except Exception as exc: # noqa: BLE001
warnings.append(f"Map selection could not be used: {exc}")
if anchor:
selection = selection_from_lat_lon(anchor[0], anchor[1], radius_m=250)
evidence.append(
make_evidence(
category="Boundary",
finding="Manual coordinate input was converted to pin-radius context analysis.",
source_name="User coordinate input",
source_url="",
source_type="manual lat/lon or parsed URL",
resolution_or_scope="250 m radius",
confidence="low",
limitation="Point-radius mode is not exact boundary analysis.",
design_implication="Use for neighborhood/context analysis only.",
verification_needed="Add a drawn or uploaded boundary for plot-level analysis.",
output_label="user_input",
)
)
return selection, evidence, warnings
raise ValueError("Draw, pin, upload GeoJSON/DXF, or enter coordinates first.")
def _anchor_from_map_state(map_state: str) -> tuple[float, float] | None:
try:
selection = normalize_map_state(map_state)
if selection.anchor_lat is not None and selection.anchor_lon is not None:
return selection.anchor_lat, selection.anchor_lon
except Exception:
return None
return None
def _current_wind_direction(climate: dict[str, Any]) -> float | None:
forecast = climate.get("forecast") or {}
value = forecast.get("current_wind_direction_deg")
try:
return float(value) if value is not None else None
except (TypeError, ValueError):
return None
def _resolve_project_fields(
*,
project_name: str,
site_name: str,
project_type: str,
boundary_source: str,
selection: SiteSelection,
site_identity: dict[str, Any] | None,
) -> tuple[str, str, str, str, list[str]]:
warnings: list[str] = []
project = (project_name or "").strip()
site = (site_name or "").strip()
ptype = (project_type or "").strip()
source = (boundary_source or "").strip()
if not project:
project = "Preliminary Site Analysis"
warnings.append("Project name was not provided; the app used a generic title.")
if not site or site.lower() in {"untitled site", "untitled", "site"}:
site = _auto_site_name(selection, site_identity)
warnings.append("Site name was not provided; the app generated a label from the selected area.")
if not ptype:
ptype = "Early-stage architecture site analysis"
if not source:
source = _auto_boundary_source(selection)
warnings.append("Boundary source was not provided; the report labels the selection from the input mode.")
return project, site, ptype, source, warnings
def _auto_site_name(selection: SiteSelection, site_identity: dict[str, Any] | None) -> str:
if site_identity:
for key in ("city", "town", "village", "district", "state"):
value = site_identity.get(key)
if value:
return f"{value} selected site"
display = site_identity.get("display_name")
if display:
return f"{str(display).split(',')[0]} selected site"
if selection.anchor_lat is not None and selection.anchor_lon is not None:
return f"Selected site {selection.anchor_lat:.5f}, {selection.anchor_lon:.5f}"
return "Selected site area"
def _auto_boundary_source(selection: SiteSelection) -> str:
if selection.selection_type in {"drawn_polygon", "rectangle"}:
return "User-drawn boundary on OpenStreetMap base map; approximate and not legal/cadastral."
if selection.selection_type == "pin_radius":
return "User-selected pin and radius; approximate context analysis, not exact boundary."
if selection.selection_type == "dxf_boundary":
return "Uploaded DXF boundary candidate selected by user."
if selection.selection_type == "geojson_boundary":
return "Uploaded GeoJSON boundary selected by user."
if selection.selection_type == "kml_boundary":
return "Uploaded KML/KMZ boundary selected by user."
return "User-provided site selection."
def _file_path(file_obj: Any) -> str:
return str(getattr(file_obj, "name", file_obj))
def build_app() -> gr.Blocks:
map_js = read_asset("map.js")
map_js_callback = gradio_js_callback(map_js)
block_kwargs: dict[str, Any] = {"title": "Site Intelligence Studio"}
block_parameters = inspect.signature(gr.Blocks).parameters
if "css" in block_parameters:
block_kwargs["css"] = read_asset("style.css")
if "head" in block_parameters:
block_kwargs["head"] = HEAD_HTML
if "js" in block_parameters:
block_kwargs["js"] = map_js_callback
with gr.Blocks(**block_kwargs) as demo:
gr.HTML(
"""
Build Small Hackathon / Backyard AI
Site Intelligence Studio
Boundary firstOpen dataField verify
Start with land, not a prompt.Draw a boundary, drop a radius pin, or upload CAD/KML/GeoJSON. The app turns that selection into a sourced site-analysis board pack.
Only the site is required.Project fields are optional. Blank names are auto-labelled from coordinates or approximate address.
Output stays cautious.Climate, terrain, soil, OSM, and AI text are labelled by source, confidence, and verification need.
For studentsfaster studio sheets
For reviewerssources and limits visible
For safetyno final foundation or legal claims
For demoworks from polygon only
""" + DELIVERABLES_HTML + """
"""
)
with gr.Row(elem_classes=["studio-workbench"]):
with gr.Column(scale=3, elem_classes=["brief-pane"]):
gr.HTML(
"""
01
Site briefOptional context for cleaner board copy
Minimum inputDraw a polygon, draw a rectangle, drop a pin radius, or paste coordinates. Everything else can be blank.
"""
)
with gr.Accordion("Optional labels and manual anchor", open=False, elem_classes=["brief-accordion"]):
project_name = gr.Textbox(
label="Project name (optional)",
placeholder="Auto: Preliminary Site Analysis",
info="Leave blank if you only want to draw/select a site and generate analysis.",
)
site_name = gr.Textbox(
label="Site name (optional)",
placeholder="Auto-filled from approximate address or coordinates",
info="If blank, the app labels the selected polygon/pin from reverse geocoding or coordinates.",
)
project_type = gr.Textbox(
label="Project type (optional)",
placeholder="Thesis, resort, housing, institute...",
)
boundary_source = gr.Textbox(
label="Boundary source (optional)",
placeholder="Auto-labelled from map draw, pin radius, CAD, KML/KMZ, or GeoJSON",
)
manual_location = gr.Textbox(
label="Lat/lon or Google Maps URL",
placeholder="21.002, 70.245 or a Google Maps URL with coordinates",
info="Used as fallback input and as CAD anchor. Drawn polygons use their centroid.",
)
culture_notes = gr.Textbox(
label="Editable region / culture / activity notes",
lines=4,
placeholder="Local activity, user groups, culture, materials, typology, observed movement patterns...",
info="This remains user-provided evidence, not AI-invented demographic analysis.",
)
with gr.Column(scale=6, elem_classes=["canvas-pane"]):
gr.HTML(
"""
02
Boundary canvasTrace land first. Files and forms are secondary.
Boundary accuracy: map tracing is useful for early context only. Upload CAD/DXF, KML, or GeoJSON when available. Public climate and map layers use the selected anchor or centroid, not legal plot data.
"
)
with gr.Column(scale=3, elem_classes=["source-pane"]):
gr.HTML(
"""
03
Source intakeOptional files, sample data, and final run
No upload requiredUploads improve accuracy, but map selection alone still creates a full preliminary board pack.
"""
)
with gr.Accordion("Optional Google Earth / field notes", open=False, elem_classes=["brief-accordion"]):
earth_observation_notes = gr.Textbox(
label="Satellite / Google Earth / site-observation notes",
lines=4,
placeholder="Visible trees, water edge, existing structures, access roads, open ground, slope/drainage signs, construction nearby...",
info="Optional. These are user observations and visual-reference notes, not app-verified satellite facts.",
)
sample_btn = gr.Button("Use Chorwad sample site", variant="secondary")
with gr.Accordion("Boundary and reference uploads", open=False, elem_classes=["upload-panel"]):
dxf_file = gr.File(label="DXF upload", file_types=[".dxf"], type="filepath")
geojson_file = gr.File(label="GeoJSON upload", file_types=[".geojson", ".json"], type="filepath")
kml_file = gr.File(label="Google Earth KML/KMZ upload", file_types=[".kml", ".kmz"], type="filepath")
pdf_file = gr.File(label="PDF reference only", file_types=[".pdf"], type="filepath")
dxf_candidate = gr.Dropdown(label="DXF boundary candidate", choices=[], interactive=True)
dxf_summary = gr.Markdown("Upload a DXF to inspect boundary candidates.")
gr.HTML(
"""
Studio diagram pack.The first row is the core prototype output. The second row is the submission upgrade: edge/access mapping, climate strategy, and a constraints-verification matrix.