Create terrenos_2026.py

terrenos_2026.py

@@ -0,0 +1,196 @@
+import glob
+import os
+
+import geopandas as gpd
+import gradio as gr
+import numpy as np
+import pandas as pd
+import xgboost as xgb
+from pyproj import Transformer
+from shapely import wkb
+from shapely.geometry import Point
+
+# Configuracao de recursos
+BASE_DIR = os.path.dirname(os.path.abspath(__file__))
+PARQUET_PATH = os.path.join(BASE_DIR, "base_referencia_final.parquet")
+
+# Modelo final do notebook TERRITORIAL_2026_SHAP_VTOTAL_Final_Ouro.ipynb
+MODEL_PATTERN = os.path.join(BASE_DIR, "TERRITORIAL_2026_SHAP_VTOTAL_20260407*.json")
+model_candidates = sorted(glob.glob(MODEL_PATTERN))
+if not model_candidates:
+    raise FileNotFoundError(
+        "Nao foi encontrado modelo com padrao "
+        "'TERRITORIAL_2026_SHAP_VTOTAL_20260407*.json'."
+    )
+MODEL_PATH = model_candidates[-1]
+
+# Features exigidas pelo modelo final salvo no JSON
+FEATURE_NAMES = [
+    "FONTE",
+    "AREA",
+    "TESTADA",
+    "IAPOND",
+    "taxa_ocupacao",
+    "avg_bldg_footprint",
+    "dist_to_main_road",
+    "dist_to_park",
+    "RH_x_Ano_Dado",
+]
+
+# Variaveis espaciais derivadas por proximidade (nao informadas pelo usuario)
+CONTEXT_FEATURES = [
+    "taxa_ocupacao",
+    "avg_bldg_footprint",
+    "dist_to_main_road",
+    "dist_to_park",
+]
+
+print("Carregando modelo e base de referencia...")
+model = xgb.Booster()
+model.load_model(MODEL_PATH)
+
+model_features = model.feature_names
+if model_features is not None and model_features != FEATURE_NAMES:
+    raise ValueError(
+        "Features do JSON diferem das configuradas no app.\n"
+        f"Modelo: {model_features}\n"
+        f"App:    {FEATURE_NAMES}"
+    )
+
+try:
+    # Tentativa direta como GeoParquet
+    gdf_ref = gpd.read_parquet(PARQUET_PATH)
+except Exception:
+    # Fallback para parquet comum com geometria em WKB
+    df_temp = pd.read_parquet(PARQUET_PATH)
+    if "geometry" not in df_temp.columns:
+        raise ValueError("A base de referencia nao possui coluna 'geometry'.")
+
+    df_temp["geometry"] = df_temp["geometry"].apply(
+        lambda x: wkb.loads(bytes(x)) if isinstance(x, (bytes, bytearray, memoryview)) else x
+    )
+    gdf_ref = gpd.GeoDataFrame(df_temp, geometry="geometry", crs="EPSG:31982")
+
+if gdf_ref.crs is None:
+    gdf_ref = gdf_ref.set_crs("EPSG:31982", allow_override=True)
+
+missing_context = [c for c in CONTEXT_FEATURES if c not in gdf_ref.columns]
+if missing_context:
+    raise ValueError(
+        "A base de referencia nao possui as colunas espaciais exigidas: "
+        + ", ".join(missing_context)
+    )
+
+df_ref = pd.DataFrame(gdf_ref[CONTEXT_FEATURES].copy()).astype(float)
+
+# Converte coordenadas de entrada (WGS84) para o CRS da base de referencia
+transformer = Transformer.from_crs("EPSG:4326", gdf_ref.crs, always_xy=True)
+
+
+def predict_terreno(
+    area_val,
+    testada_val,
+    rh_val,
+    iapond_val,
+    lat_val,
+    lon_val,
+    ano_val,
+    fonte_str,
+):
+    try:
+        area_val = float(area_val)
+        testada_val = float(testada_val)
+        rh_val = float(rh_val)
+        iapond_val = float(iapond_val)
+        lat_val = float(lat_val)
+        lon_val = float(lon_val)
+        ano_int = int(ano_val)
+
+        if area_val <= 0:
+            raise ValueError("Area total deve ser maior que zero.")
+        if testada_val <= 0:
+            raise ValueError("Testada principal deve ser maior que zero.")
+        if rh_val < 0:
+            raise ValueError("RH nao pode ser negativo.")
+
+        # Engenharia de variaveis conforme treino no notebook
+        rh_x_ano = rh_val * ano_int
+
+        # Busca do vizinho mais proximo por distancia euclidiana (nos bastidores)
+        pt_x, pt_y = transformer.transform(lon_val, lat_val)
+        distancias = gdf_ref.geometry.distance(Point(pt_x, pt_y))
+        idx_vizinho = distancias.idxmin()
+        urb_vars = df_ref.loc[idx_vizinho]
+
+        features = {
+            "FONTE": 1.0 if str(fonte_str).strip().lower() == "oferta" else 0.0,
+            "AREA": float(np.log1p(area_val)),
+            "TESTADA": float(np.log1p(testada_val)),
+            "IAPOND": iapond_val,
+            "taxa_ocupacao": float(urb_vars["taxa_ocupacao"]),
+            "avg_bldg_footprint": float(urb_vars["avg_bldg_footprint"]),
+            "dist_to_main_road": float(np.log1p(max(urb_vars["dist_to_main_road"], 0.0))),
+            "dist_to_park": float(urb_vars["dist_to_park"]),
+            "RH_x_Ano_Dado": float(rh_x_ano),
+        }
+
+        X_input = pd.DataFrame([features], columns=FEATURE_NAMES)
+        dmatrix = xgb.DMatrix(X_input, feature_names=FEATURE_NAMES)
+
+        # Target do treino foi log1p(VTOTAL)
+        pred_log_vtotal = float(model.predict(dmatrix)[0])
+        vtotal = max(float(np.expm1(pred_log_vtotal)), 0.0)
+        vunit = vtotal / area_val
+
+        return f"R$ {vunit:,.2f} / m2", f"R$ {vtotal:,.2f}"
+
+    except Exception as e:
+        return f"Erro: {str(e)}", "Revise os valores informados."
+
+
+with gr.Blocks(title="Territorial XGBoost 2026") as demo:
+    gr.Markdown("# Territorial XGBoost 2026")
+    gr.Markdown(
+        "Modelo XGBoost com Spatial K-Fold, otimizado com Optuna.
+        "Variaveis espaciais de contexto são calculadas automaticamente por proximidade do ponto informado (lat/lon)."
+    )
+
+    with gr.Row():
+        with gr.Column():
+            with gr.Group():
+                gr.Markdown("### Dimensões, Potencial Construtivo e Região Homogênea")
+                area = gr.Number(label="Area Total (m2)", value=300)
+                testada = gr.Number(label="Testada Principal (m)", value=10)
+                iapond = gr.Number(label="IA Ponderado", value=1.3)
+                rh = gr.Number(label="RH (Indice de Valorizacao)", value=50)
+
+        with gr.Column():
+            with gr.Group():
+                gr.Markdown("### Localizacao, Tempo e Fonte")
+                lat = gr.Number(label="Latitude", value=-30.0346)
+                lon = gr.Number(label="Longitude", value=-51.2177)
+                ano = gr.Dropdown(
+                    choices=[str(y) for y in range(2016, 2027)],
+                    label="Ano do Dado",
+                    value="2026",
+                )
+                fonte = gr.Radio(
+                    ["Venda", "Oferta"],
+                    label="Tipo de Dado (Fonte)",
+                    value="Venda",
+                )
+
+            gr.Markdown("---")
+            btn = gr.Button("CALCULAR VALOR ESTIMADO", variant="primary")
+            vunit_out = gr.Textbox(label="Valor Unitario Estimado (R$/m2)")
+            vtotal_out = gr.Textbox(label="Valor Total Estimado (R$)")
+
+    btn.click(
+        predict_terreno,
+        inputs=[area, testada, rh, iapond, lat, lon, ano, fonte],
+        outputs=[vunit_out, vtotal_out],
+    )
+
+
+if __name__ == "__main__":
+    demo.launch()