from __future__ import annotations import json import logging import math import os import re import unicodedata from dataclasses import dataclass, field from pathlib import Path from typing import Any import gradio as gr import numpy as np import pandas as pd import xgboost as xgb from pyproj import Transformer try: from .reporting import build_metric_summary, build_pdf_report, split_prediction_frames except ImportError: from reporting import build_metric_summary, build_pdf_report, split_prediction_frames LOGGER = logging.getLogger("avm_gradio_app") if not LOGGER.handlers: logging.basicConfig( level=os.getenv("APP_LOG_LEVEL", "INFO").upper(), format="%(asctime)s | %(levelname)s | %(name)s | %(message)s", ) BASE_DIR = Path(__file__).resolve().parent WGS84_TO_WEBMERC = Transformer.from_crs("EPSG:4326", "EPSG:3857", always_xy=True) ZONEAMENTO_SHP_ENV = "ZONEAMENTO_SHP_PATH" IAPOND_MAP_ENV = "IAPOND_MAP_PATH" IAPOND_PDF_ENV = "IAPOND_PDF_PATH" POLYGON_CONTEXT_ENV = "POLYGON_CONTEXT_PATH" RH_NUMBLOCO_CSV_ENV = "RH_NUMBLOCO_CSV_PATH" TESTADA_NUMBLOCO_CSV_ENV = "TESTADA_NUMBLOCO_CSV_PATH" IAPOND_DEFAULT = 1.0 AUTO_GLEBA_AREA_THRESHOLD = 3000.0 POLYGON_FALLBACK_MAX_DISTANCE_M = 250.0 POLYGON_TARGET_CRS = "EPSG:31982" IAPOND_BY_INDICE = { "1": 1.0, "2A": 1.0, "2B": 1.0, "3": 1.3, "4": 1.3, "4A": 1.3, "5": 1.3, "6": 1.3, "7": 1.3, "9": 1.3, "11": 1.6, "13": 1.6, "15": 1.9, "17": 1.9, "19": 2.4, "21": 0.65, "23": 1.0, "25": 1.0, "31": 0.1, "33": 0.1, "35": 0.2, "37": 0.5, "39": 1.0, "41": 1.0, } RH_BINS = [-np.inf, 13.5, 31.5, 59.5, 155.0, np.inf] RH_LABELS = ["RH_muito_baixo", "RH_baixo", "RH_medio", "RH_alto", "RH_muito_alto"] DEFAULTS_NUMERIC = { "RH": 50.0, "IAPOND": 1.0, "APP": 0.0, "CP": 0.0, "LOTPOS": 0.0, "area_simpson": 0.0, "taxa_ocupacao": 0.0, "bldg_density": 0.0, "avg_bldg_footprint": 0.0, "circularity": 0.0, "dist_to_main_road": 0.0, "dist_to_park": 0.0, "Dist_weighted_Density": 0.0, } DEFAULTS_CATEGORICAL = { "FONTE": "0", "Ano_Dado": "2026", "faixa_rh": "RH_alto", "faixa_area_modelo": "desconhecido", "FINALIDADE": "desconhecido", "MZ": "desconhecido", "UEU": "desconhecido", "SUBUNIDADE": "desconhecido", "DENSIDADE": "desconhecido", "ATIVIDADE": "desconhecido", "INDICE": "desconhecido", "VOLUMETRIA": "desconhecido", } REQUESTED_MODEL_NOTEBOOKS = { "terreno": "TERRITORIAL_2026_TERRENO", "gleba": "TERRITORIAL_2026_GLEBA", } RELIABLE_MODEL_CONTEXT_SOURCES = { "rh_numbloco_csv", "testada_numbloco_csv", "poligono_numbloco", "poligono_intersect", "poligono_proximo", "zoneamento_shp", "zoneamento_shp_ia", } ZONEAMENTO_FIELDS = [ "MZ", "UEU", "SUBUNIDADE", "DENSIDADE", "ATIVIDADE", "INDICE", "VOLUMETRIA", ] ZONEAMENTO_COLUMN_ALIASES: dict[str, list[str]] = { "MZ": ["MZ", "MACROZONA"], "UEU": ["UEU"], "SUBUNIDADE": ["SUBUNIDADE", "SUBUNID", "SUB_UNIDADE", "SUBUNIDADE1"], "DENSIDADE": ["DENSIDADE", "DENSID"], "ATIVIDADE": ["ATIVIDADE", "ATIVID"], "INDICE": ["INDICE", "IND_APR", "INDAPROV", "INDICEAPR"], "VOLUMETRIA": ["VOLUMETRIA", "VOLUM"], "IA": ["IA", "IAPOND", "IA_POND", "INDICEAPROVEITAMENTO"], "CODIGO": ["CODIGO", "COD", "CODINDICE", "COD_INDICE", "INDICECOD"], } CONTEXT_FIELDS = sorted( { "FONTE", "Ano_Dado", "AREA", "AREA_bruta", "TESTADA", "TESTADA_bruta", "RH", "IAPOND", "APP", "CP", "LOTPOS", "FINALIDADE", "BAIRRO", "COD_BAIRRO", "area_simpson", "taxa_ocupacao", "bldg_density", "avg_bldg_footprint", "circularity", "dist_to_main_road", "dist_to_park", "MZ", "UEU", "SUBUNIDADE", "DENSIDADE", "ATIVIDADE", "INDICE", "VOLUMETRIA", } ) POLYGON_CONTEXT_FIELDS = [ "NUMBLOCO", "AREA", "TESTADA", "RH", "faixa_rh", "IAPOND", "APP", "PE", "CP", "log_area_modelo", "log_testada_modelo", "area_simpson", "taxa_ocupacao", "bldg_density", "avg_bldg_footprint", "circularity", "dist_to_main_road", "dist_to_park", "Dist_weighted_Density", "intersection_density", "MZ", "UEU", "SUBUNIDADE", "DENSIDADE", "ATIVIDADE", "INDICE", "VOLUMETRIA", ] def _is_missing(value: Any) -> bool: if value is None: return True if isinstance(value, str) and value.strip() == "": return True try: return bool(pd.isna(value)) except Exception: return False def _to_float(value: Any, default: float | None = None) -> float | None: series = pd.to_numeric(pd.Series([value]), errors="coerce") number = series.iloc[0] if pd.isna(number): return default return float(number) def _to_int(value: Any, default: int | None = None) -> int | None: number = _to_float(value) if number is None: return default return int(round(number)) def _normalize_token(value: Any) -> str: if _is_missing(value): return "" text = str(value).strip() text = re.sub(r"\s+", " ", text) return text.upper() def _normalize_ascii_token(value: Any) -> str: token = _normalize_token(value) token = unicodedata.normalize("NFKD", token) token = "".join(ch for ch in token if not unicodedata.combining(ch)) return token def _normalize_column_token(value: Any) -> str: token = _normalize_ascii_token(value) return re.sub(r"[^A-Z0-9]", "", token) def _normalize_code_token(value: Any) -> str: if _is_missing(value): return "" text = _normalize_ascii_token(value) text = text.replace(",", ".") text = re.sub(r"\s+", "", text).strip() if text == "": return "" code_match = re.fullmatch(r"0*([0-9]+)([A-Z]?)", text) if code_match: return f"{code_match.group(1)}{code_match.group(2)}" return text def _normalize_numbloco_key(value: Any) -> str: if _is_missing(value): return "" text = _normalize_ascii_token(value) text = re.sub(r"\.0$", "", text) text = re.sub(r"[^A-Z0-9]", "", text) if text.isdigit(): return text.lstrip("0") or "0" return text def _find_column_by_alias(columns: list[str], aliases: list[str]) -> str | None: normalized = {_normalize_column_token(col): str(col) for col in columns} for alias in aliases: key = _normalize_column_token(alias) if key in normalized: return normalized[key] for col in columns: col_key = _normalize_column_token(col) for alias in aliases: alias_key = _normalize_column_token(alias) if alias_key and (col_key.startswith(alias_key) or alias_key.startswith(col_key)): return str(col) return None def _safe_text(value: Any, default: str = "desconhecido") -> str: if _is_missing(value): return default text = str(value).strip() return text if text else default def _currency_brl(value: float) -> str: if not np.isfinite(value): return "n/a" return f"R$ {value:,.2f}" def _jsonable(value: Any) -> Any: if isinstance(value, (np.floating,)): return float(value) if isinstance(value, (np.integer,)): return int(value) if isinstance(value, (np.bool_,)): return bool(value) if isinstance(value, (pd.Timestamp,)): return value.isoformat() if isinstance(value, dict): return {str(k): _jsonable(v) for k, v in value.items()} if isinstance(value, (list, tuple)): return [_jsonable(v) for v in value] return value def _read_json(path: Path) -> dict[str, Any]: with path.open("r", encoding="utf-8") as f: return json.load(f) def _read_parquet_optional(path: Path | None) -> pd.DataFrame: if path is None or not path.exists(): return pd.DataFrame() try: return pd.read_parquet(path) except Exception as exc: LOGGER.warning("Falha ao ler parquet %s: %s", path, exc) return pd.DataFrame() def _read_csv_optional(path: Path | None) -> pd.DataFrame: if path is None or not path.exists(): return pd.DataFrame() try: return pd.read_csv(path) except Exception as exc: LOGGER.warning("Falha ao ler csv %s: %s", path, exc) return pd.DataFrame() def _find_first_file(directory: Path, patterns: list[str]) -> Path | None: for pattern in patterns: found = sorted(directory.glob(pattern)) if found: return found[0] return None def _candidate_base_dirs(base_dir: Path) -> list[Path]: candidates = [base_dir, base_dir.parent] unique: list[Path] = [] seen: set[str] = set() for path in candidates: try: key = str(path.resolve()) except Exception: key = str(path) if key in seen: continue seen.add(key) unique.append(path) return unique def _candidate_rh_numbloco_paths(base_dir: Path) -> list[Path]: candidates: list[Path] = [] env_path = os.getenv(RH_NUMBLOCO_CSV_ENV, "").strip() if env_path: candidates.append(Path(env_path).expanduser()) for root in _candidate_base_dirs(base_dir): candidates.extend( [ root / "data" / "reference" / "numbloco_rh.csv", root / "numbloco_rh.csv", ] ) unique: list[Path] = [] seen: set[str] = set() for path in candidates: try: key = str(path.resolve()) except Exception: key = str(path) if key in seen: continue seen.add(key) unique.append(path) return unique def _candidate_testada_numbloco_paths(base_dir: Path) -> list[Path]: candidates: list[Path] = [] env_path = os.getenv(TESTADA_NUMBLOCO_CSV_ENV, "").strip() if env_path: candidates.append(Path(env_path).expanduser()) for root in _candidate_base_dirs(base_dir): candidates.extend( [ root / "data" / "reference" / "numbloco_testada.csv", root / "numbloco_testada.csv", ] ) unique: list[Path] = [] seen: set[str] = set() for path in candidates: try: key = str(path.resolve()) except Exception: key = str(path) if key in seen: continue seen.add(key) unique.append(path) return unique def load_rh_numbloco_context(base_dir: Path = BASE_DIR) -> RhNumblocoContext: for path in _candidate_rh_numbloco_paths(base_dir): if not path.exists(): continue try: df = pd.read_csv(path) except Exception as exc: LOGGER.warning("Falha ao ler CSV RH por NUMBLOCO %s: %s", path, exc) continue numbloco_col = _find_column_by_alias(list(df.columns), ["NUMBLOCO", "NUM_BLOCO", "NUM BLOCO"]) rh_col = _find_column_by_alias(list(df.columns), ["RHVALOR", "RH_VALOR", "RH", "VALOR_RH"]) if numbloco_col is None or rh_col is None: LOGGER.warning("CSV %s sem colunas NUMBLOCO/RH reconheciveis.", path) continue work = pd.DataFrame( { "__numbloco_key": df[numbloco_col].map(_normalize_numbloco_key), "RH": pd.to_numeric(df[rh_col], errors="coerce"), } ) work = work[(work["__numbloco_key"] != "") & work["RH"].notna()] if work.empty: LOGGER.warning("CSV %s nao contem pares NUMBLOCO/RH validos.", path) continue key_counts = work.groupby("__numbloco_key").size() duplicate_key_count = int((key_counts > 1).sum()) conflicting_key_count = int(work.groupby("__numbloco_key")["RH"].nunique().gt(1).sum()) rh_by_numbloco = work.groupby("__numbloco_key")["RH"].median().astype(float).to_dict() LOGGER.info( "RH por NUMBLOCO carregado: %s (linhas=%s, chaves=%s, duplicadas=%s, conflitos=%s)", path.name, len(work), len(rh_by_numbloco), duplicate_key_count, conflicting_key_count, ) if conflicting_key_count: LOGGER.warning( "CSV RH por NUMBLOCO possui %s chaves com valores conflitantes; usando mediana por NUMBLOCO.", conflicting_key_count, ) return RhNumblocoContext( path=path, rh_by_numbloco=rh_by_numbloco, duplicate_key_count=duplicate_key_count, conflicting_key_count=conflicting_key_count, ) message = "CSV RH por NUMBLOCO nao encontrado." LOGGER.warning(message) return RhNumblocoContext(error=message) def load_testada_numbloco_context(base_dir: Path = BASE_DIR) -> TestadaNumblocoContext: for path in _candidate_testada_numbloco_paths(base_dir): if not path.exists(): continue try: df = pd.read_csv(path) except Exception as exc: LOGGER.warning("Falha ao ler CSV TESTADA por NUMBLOCO %s: %s", path, exc) continue numbloco_col = _find_column_by_alias(list(df.columns), ["NUMBLOCO", "NUM_BLOCO", "NUM BLOCO"]) testada_col = _find_column_by_alias( list(df.columns), ["MTR_TESTADA", "TESTADA", "TESTADA_BRUTA", "FRENTE", "METROS_TESTADA"], ) if numbloco_col is None or testada_col is None: LOGGER.warning("CSV %s sem colunas NUMBLOCO/TESTADA reconheciveis.", path) continue work = pd.DataFrame( { "__numbloco_key": df[numbloco_col].map(_normalize_numbloco_key), "TESTADA": pd.to_numeric(df[testada_col], errors="coerce"), } ) work = work[(work["__numbloco_key"] != "") & work["TESTADA"].notna() & (work["TESTADA"] > 0)] if work.empty: LOGGER.warning("CSV %s nao contem pares NUMBLOCO/TESTADA validos.", path) continue key_counts = work.groupby("__numbloco_key").size() duplicate_key_count = int((key_counts > 1).sum()) conflicting_key_count = int(work.groupby("__numbloco_key")["TESTADA"].nunique().gt(1).sum()) testada_by_numbloco = work.groupby("__numbloco_key")["TESTADA"].median().astype(float).to_dict() LOGGER.info( "TESTADA por NUMBLOCO carregada: %s (linhas=%s, chaves=%s, duplicadas=%s, conflitos=%s)", path.name, len(work), len(testada_by_numbloco), duplicate_key_count, conflicting_key_count, ) if conflicting_key_count: LOGGER.warning( "CSV TESTADA por NUMBLOCO possui %s chaves com valores conflitantes; usando mediana por NUMBLOCO.", conflicting_key_count, ) return TestadaNumblocoContext( path=path, testada_by_numbloco=testada_by_numbloco, duplicate_key_count=duplicate_key_count, conflicting_key_count=conflicting_key_count, ) message = "CSV TESTADA por NUMBLOCO nao encontrado." LOGGER.warning(message) return TestadaNumblocoContext(error=message) def _parse_model_schema(model_path: Path) -> tuple[list[str], dict[str, str]]: payload = _read_json(model_path) learner = payload.get("learner", {}) names = learner.get("feature_names") or [] types = learner.get("feature_types") or [] feature_types: dict[str, str] = {} for idx, name in enumerate(names): feature_types[str(name)] = str(types[idx]) if idx < len(types) else "float" return [str(name) for name in names], feature_types def _parse_area_bins(raw_bins: list[Any]) -> list[float]: bins: list[float] = [] for item in raw_bins: if item is None: bins.append(float("inf")) continue if isinstance(item, (int, float)): bins.append(float(item)) continue text = str(item).strip().lower() if text in {"inf", "infinity", "+inf", "+infinity"}: bins.append(float("inf")) elif text in {"-inf", "-infinity"}: bins.append(float("-inf")) else: value = _to_float(item) bins.append(float(value) if value is not None else float("nan")) return bins def _derive_ano_dado(df: pd.DataFrame) -> pd.DataFrame: if "Ano_Dado" in df.columns: return df out = df.copy() if "ANO" in out.columns: out["Ano_Dado"] = pd.to_numeric(out["ANO"], errors="coerce").astype("Int64") return out ano_cols = [col for col in out.columns if str(col).startswith("ANO_")] if not ano_cols: return out idx_max = out[ano_cols].idxmax(axis=1) out["Ano_Dado"] = ( idx_max.astype("string") .str.replace("ANO_", "", regex=False) .astype("Int64") ) return out def _summarize_series(series: pd.Series, prefer_mode: bool) -> Any: clean = series.dropna() if clean.empty: return None if prefer_mode: mode = clean.astype("string").mode(dropna=True) return None if mode.empty else str(mode.iloc[0]) as_num = pd.to_numeric(clean, errors="coerce") if float(as_num.notna().mean()) >= 0.8: return float(as_num.median()) mode = clean.astype("string").mode(dropna=True) return None if mode.empty else str(mode.iloc[0]) def _normalize_category_values(values: list[str]) -> list[str]: out = sorted({str(v) for v in values if str(v).strip() != ""}) if "desconhecido" not in out: out.append("desconhecido") return out def _normalize_fonte(value: Any, categories: list[str]) -> str: if _is_missing(value): value = DEFAULTS_CATEGORICAL["FONTE"] text = str(value).strip().lower() if text in {"venda", "sale"}: if "0" in categories and "Venda" not in categories: return "0" return "Venda" if "Venda" in categories else "0" if text in {"oferta", "offer"}: if "1" in categories and "Oferta" not in categories: return "1" return "Oferta" if "Oferta" in categories else "1" if text in {"0", "1"}: return text return str(value).strip() or DEFAULTS_CATEGORICAL["FONTE"] def _is_reliable_model_context_source(source: str) -> bool: return source in RELIABLE_MODEL_CONTEXT_SOURCES or str(source).startswith("mapa_ia:") def _categorize_area(area: float, bins: list[float], labels: list[str]) -> str: if not bins or not labels: return "desconhecido" series = pd.Series([area], dtype="float64") cat = pd.cut(series, bins=bins, labels=labels, include_lowest=True).astype("string").iloc[0] if _is_missing(cat): return labels[-1] return str(cat) def _categorize_rh(rh: float) -> str: series = pd.Series([rh], dtype="float64") cat = pd.cut(series, bins=RH_BINS, labels=RH_LABELS).astype("string").iloc[0] if _is_missing(cat): return RH_LABELS[-1] return str(cat) def _compute_grid_id(x: float, y: float, cell_size: float) -> str: if not np.isfinite(x) or not np.isfinite(y) or cell_size <= 0: return "" gx = int(math.floor(x / cell_size)) gy = int(math.floor(y / cell_size)) return f"{gx}_{gy}" def _infer_testada(area: float, local_context: dict[str, Any]) -> tuple[float, str]: for key in ("TESTADA", "TESTADA_bruta", "testada"): candidate = _to_float(local_context.get(key)) if candidate is not None and candidate > 0: return float(candidate), "inferido_coord" # Heuristica geométrica simples para reduzir input manual quando testada não estiver disponível. # Assume razão profundidade/testada ~ 2.5 para lotes urbanos. heuristic = math.sqrt(max(area, 1.0) / 2.5) heuristic = min(max(heuristic, 5.0), 120.0) return float(heuristic), "heuristica_area" def _derive_cp(area: float, testada: float) -> float: if area <= 0 or testada <= 0: raise ValueError("AREA e TESTADA devem ser positivas para calcular CP.") pe = area / testada if area > 5000 and pe > 90: return float(4.8 * (testada**0.2) * (area**-0.4)) if pe < 20: return float(math.sqrt(pe / 20.0)) if pe <= 33: return 1.0 if pe < 90: return float(math.sqrt(33.0 / pe)) return 0.6 def _parse_overrides_json(text: str) -> dict[str, Any]: if _is_missing(text): return {} try: payload = json.loads(str(text)) except json.JSONDecodeError as exc: raise ValueError(f"JSON invalido em overrides: {exc.msg}") from exc if not isinstance(payload, dict): raise ValueError("Overrides deve ser um objeto JSON.") return {str(k).strip().lower(): v for k, v in payload.items()} @dataclass class ArtifactBundle: key: str label: str artifact_dir: Path metadata_path: Path model_path: Path metadata: dict[str, Any] model: xgb.Booster feature_names: list[str] feature_types: dict[str, str] area_bins: list[float] area_labels: list[str] grid_cell_size: float tempo_index_df: pd.DataFrame grid_surface_df: pd.DataFrame segment_surface_df: pd.DataFrame surface_global_df: pd.DataFrame tier_map_raw: dict[str, str] tier_map_norm: dict[str, str] search_results_df: pd.DataFrame holdout_predictions_df: pd.DataFrame oof_predictions_df: pd.DataFrame notebook_name: str grid_value_map: dict[tuple[str, str], float] = field(default_factory=dict) segment_value_map: dict[str, float] = field(default_factory=dict) global_surface_value: float | None = None tempo_index_map: dict[str, float] = field(default_factory=dict) reference_categories: dict[str, list[str]] = field(default_factory=dict) @dataclass class ReferenceContext: path: Path | None = None df: pd.DataFrame = field(default_factory=pd.DataFrame) x: np.ndarray = field(default_factory=lambda: np.array([], dtype=float)) y: np.ndarray = field(default_factory=lambda: np.array([], dtype=float)) @property def has_data(self) -> bool: return not self.df.empty and self.x.size > 0 and self.y.size > 0 @dataclass class ZoneamentoContext: shp_path: Path | None = None gdf: pd.DataFrame = field(default_factory=pd.DataFrame) column_map: dict[str, str] = field(default_factory=dict) iapond_map: dict[str, float] = field(default_factory=dict) iapond_map_source: str = "fallback_default" error: str | None = None @property def has_data(self) -> bool: return self.shp_path is not None and not self.gdf.empty @dataclass class PolygonContext: path: Path | None = None gdf: pd.DataFrame = field(default_factory=pd.DataFrame) available_fields: list[str] = field(default_factory=list) numbloco_key_index: dict[str, list[int]] = field(default_factory=dict) error: str | None = None fallback_max_distance_m: float = POLYGON_FALLBACK_MAX_DISTANCE_M to_base_crs: Transformer | None = None @property def has_data(self) -> bool: return self.path is not None and not self.gdf.empty and "geometry" in self.gdf.columns @dataclass class RhNumblocoContext: path: Path | None = None rh_by_numbloco: dict[str, float] = field(default_factory=dict) duplicate_key_count: int = 0 conflicting_key_count: int = 0 error: str | None = None @property def has_data(self) -> bool: return self.path is not None and bool(self.rh_by_numbloco) @dataclass class TestadaNumblocoContext: path: Path | None = None testada_by_numbloco: dict[str, float] = field(default_factory=dict) duplicate_key_count: int = 0 conflicting_key_count: int = 0 error: str | None = None @property def has_data(self) -> bool: return self.path is not None and bool(self.testada_by_numbloco) @dataclass class AppState: bundles: dict[str, ArtifactBundle] = field(default_factory=dict) reference: ReferenceContext = field(default_factory=ReferenceContext) zoneamento: ZoneamentoContext = field(default_factory=ZoneamentoContext) polygon_context: PolygonContext = field(default_factory=PolygonContext) rh_numbloco: RhNumblocoContext = field(default_factory=RhNumblocoContext) testada_numbloco: TestadaNumblocoContext = field(default_factory=TestadaNumblocoContext) error: str | None = None k_neighbors: int = 100 def _build_grid_and_temporal_maps(bundle: ArtifactBundle) -> None: if not bundle.grid_surface_df.empty and {"segmento_area", "grid_id", "valor"}.issubset(bundle.grid_surface_df.columns): for _, row in bundle.grid_surface_df.iterrows(): seg_key = _normalize_token(row["segmento_area"]) gid = str(row["grid_id"]).strip() val = _to_float(row["valor"]) if seg_key and gid and val is not None: bundle.grid_value_map[(seg_key, gid)] = float(val) if not bundle.segment_surface_df.empty and {"segmento_area", "valor"}.issubset(bundle.segment_surface_df.columns): for _, row in bundle.segment_surface_df.iterrows(): seg_key = _normalize_token(row["segmento_area"]) val = _to_float(row["valor"]) if seg_key and val is not None: bundle.segment_value_map[seg_key] = float(val) if not bundle.surface_global_df.empty and "mediana_global" in bundle.surface_global_df.columns: val = _to_float(bundle.surface_global_df["mediana_global"].iloc[0]) bundle.global_surface_value = float(val) if val is not None else None if not bundle.tempo_index_df.empty and {"periodo_tempo", "indice_temporal_interno"}.issubset(bundle.tempo_index_df.columns): for _, row in bundle.tempo_index_df.iterrows(): periodo = _safe_text(row["periodo_tempo"], "") indice = _to_float(row["indice_temporal_interno"]) if periodo and indice is not None: bundle.tempo_index_map[periodo] = float(indice) def _preferred_artifact_dirs(base_dir: Path) -> dict[str, list[Path]]: return { "gleba": [ base_dir / "artifacts_gleba_cadastro", base_dir.parent / "artifacts_gleba_cadastro", ], "terreno": [ base_dir / "artifacts_terreno_cadastro", base_dir.parent / "pddua_shp" / "artifacts_terreno_cadastro", base_dir.parent / "artifacts_terreno_cadastro", ], } def _artifact_prediction_paths(artifact_dir: Path, key: str) -> tuple[Path | None, Path | None]: if key == "gleba": holdout = _find_first_file(artifact_dir, ["predictions_holdout_gleba_vunit.csv", "*predictions_holdout*.csv"]) oof = _find_first_file(artifact_dir, ["predictions_oof_gleba_vunit.csv", "*predictions_oof*.csv"]) else: holdout = _find_first_file(artifact_dir, ["predictions_holdout.csv", "*predictions_holdout*.csv"]) oof = _find_first_file(artifact_dir, ["predictions_oof.csv", "*predictions_oof*.csv"]) return holdout, oof def load_artifacts(base_dir: Path = BASE_DIR) -> dict[str, ArtifactBundle]: bundles: dict[str, ArtifactBundle] = {} preferred_dirs = _preferred_artifact_dirs(base_dir) for key, candidates in preferred_dirs.items(): artifact_dir = next((path for path in candidates if path.exists()), None) if artifact_dir is None: LOGGER.warning("Artefatos solicitados nao encontrados para %s.", key) continue metadata_path = _find_first_file(artifact_dir, ["*_metadata.json"]) if metadata_path is None: LOGGER.warning("Sem metadata em %s", artifact_dir) continue metadata = _read_json(metadata_path) label = "Terreno" if key == "terreno" else "Gleba" model_path = _find_first_file( artifact_dir, [f"{key}_xgb_model.json", "*_xgb_model.json"], ) if model_path is None: LOGGER.warning("Sem modelo em %s", artifact_dir) continue feature_names, feature_types = _parse_model_schema(model_path) model = xgb.Booster() model.load_model(str(model_path)) area_bins = _parse_area_bins(list(metadata.get("area_bins", []))) area_labels = [str(label_item) for label_item in metadata.get("area_labels", [])] grid_cell_size = float(_to_float(metadata.get("grid_cell_size"), 1000.0) or 1000.0) tempo_index_df = _read_parquet_optional(_find_first_file(artifact_dir, [f"{key}_tempo_index.parquet", "*_tempo_index.parquet"])) grid_surface_df = _read_parquet_optional(_find_first_file(artifact_dir, [f"{key}_grid_surface.parquet", "*_grid_surface.parquet"])) segment_surface_df = _read_parquet_optional(_find_first_file(artifact_dir, [f"{key}_segment_surface.parquet", "*_segment_surface.parquet"])) surface_global_df = _read_parquet_optional(_find_first_file(artifact_dir, [f"{key}_surface_global.parquet", "*_surface_global.parquet"])) search_results_df = _read_csv_optional(_find_first_file(artifact_dir, [f"{key}_search_results.csv", "*_search_results.csv"])) holdout_path, oof_path = _artifact_prediction_paths(artifact_dir, key) holdout_predictions_df = _read_csv_optional(holdout_path) oof_predictions_df = _read_csv_optional(oof_path) tier_map_path = _find_first_file(artifact_dir, [f"{key}_tier_map.json", "*_tier_map.json"]) tier_map_raw = _read_json(tier_map_path) if tier_map_path else {} tier_map_norm = {_normalize_token(name): str(tier) for name, tier in tier_map_raw.items()} bundle = ArtifactBundle( key=key, label=label, artifact_dir=artifact_dir, metadata_path=metadata_path, model_path=model_path, metadata=metadata, model=model, feature_names=feature_names, feature_types=feature_types, area_bins=area_bins, area_labels=area_labels, grid_cell_size=grid_cell_size, tempo_index_df=tempo_index_df, grid_surface_df=grid_surface_df, segment_surface_df=segment_surface_df, surface_global_df=surface_global_df, tier_map_raw=tier_map_raw, tier_map_norm=tier_map_norm, search_results_df=search_results_df, holdout_predictions_df=holdout_predictions_df, oof_predictions_df=oof_predictions_df, notebook_name=REQUESTED_MODEL_NOTEBOOKS[key], ) _build_grid_and_temporal_maps(bundle) bundles[key] = bundle LOGGER.info("Artefatos carregados: %s (%s)", label, artifact_dir) if not bundles: raise RuntimeError("Nao foi possivel carregar artefatos de nenhum modelo.") return bundles def load_reference_data(base_dir: Path = BASE_DIR) -> ReferenceContext: env_path = os.getenv("REFERENCE_DATA_PATH", "").strip() candidates: list[Path] = [] if env_path: candidates.append(Path(env_path).expanduser()) for root in _candidate_base_dirs(base_dir): candidates.extend( [ root / "imoveis_v2_xgboost.parquet", root / "data" / "reference" / "base_referencia_final.parquet", root / "base_completa_quarteiroes.parquet", ] ) for path in candidates: if not path.exists(): continue try: df = pd.read_parquet(path) except Exception as exc: LOGGER.warning("Falha ao ler referencia %s: %s", path, exc) continue if df.empty: LOGGER.warning("Base de referencia vazia: %s", path) continue df = df.copy() df.columns = [str(col) for col in df.columns] df = _derive_ano_dado(df) if {"lat", "lon"}.issubset(df.columns): lon = pd.to_numeric(df["lon"], errors="coerce").to_numpy(dtype=float) lat = pd.to_numeric(df["lat"], errors="coerce").to_numpy(dtype=float) x, y = WGS84_TO_WEBMERC.transform(lon, lat) df["x"] = x df["y"] = y elif {"x", "y"}.issubset(df.columns): df["x"] = pd.to_numeric(df["x"], errors="coerce") df["y"] = pd.to_numeric(df["y"], errors="coerce") else: LOGGER.warning("Base %s sem lat/lon ou x/y. Inferencia por vizinho sera desativada.", path) return ReferenceContext(path=path, df=pd.DataFrame()) mask = df["x"].notna() & df["y"].notna() df = df.loc[mask].reset_index(drop=True) if df.empty: LOGGER.warning("Base %s sem coordenadas validas apos limpeza.", path) return ReferenceContext(path=path, df=pd.DataFrame()) ref = ReferenceContext( path=path, df=df, x=df["x"].to_numpy(dtype=float), y=df["y"].to_numpy(dtype=float), ) LOGGER.info("Base de referencia carregada: %s (n=%s)", path.name, len(df)) return ref LOGGER.warning("Nenhuma base de referencia disponivel. Sera usado fallback por defaults.") return ReferenceContext(path=None, df=pd.DataFrame()) def _candidate_polygon_paths(base_dir: Path) -> list[Path]: candidates: list[Path] = [] env_path = os.getenv(POLYGON_CONTEXT_ENV, "").strip() if env_path: candidates.append(Path(env_path).expanduser()) enriched_candidates: list[Path] = [] legacy_candidates: list[Path] = [] for root in _candidate_base_dirs(base_dir): enriched_candidates.extend( [ root / "poligonos_enriquecidos_260317_com_features_modelos.parquet", root / "data" / "reference" / "poligonos_enriquecidos_260317_com_features_modelos.parquet", root / "data" / "geospatial" / "poligonos_enriquecidos_260317_com_features_modelos.parquet", ] ) legacy_candidates.extend( [ root / "poligonos_enriquecidos_260317.parquet", root / "data" / "reference" / "poligonos_enriquecidos_260317.parquet", root / "data" / "geospatial" / "poligonos_enriquecidos_260317.parquet", root / "data" / "reference" / "poligonos_enriquecidos.parquet", root / "data" / "geospatial" / "poligonos_enriquecidos.parquet", ] ) candidates.extend(enriched_candidates) candidates.extend(legacy_candidates) unique: list[Path] = [] seen: set[str] = set() for path in candidates: try: key = str(path.resolve()) except Exception: key = str(path) if key in seen: continue seen.add(key) unique.append(path) return unique def load_polygon_context(base_dir: Path = BASE_DIR) -> PolygonContext: context = PolygonContext() try: import geopandas as gpd except Exception as exc: context.error = f"geopandas indisponivel para geoparquet enriquecido: {exc}" LOGGER.warning("Camada de poligonos desativada: %s", context.error) return context for path in _candidate_polygon_paths(base_dir): if not path.exists(): continue try: gdf = gpd.read_parquet(path) except Exception as exc: LOGGER.warning("Falha ao ler geoparquet %s: %s", path, exc) continue if gdf.empty: LOGGER.warning("Geoparquet vazio: %s", path) continue if gdf.crs is None: LOGGER.warning("Geoparquet sem CRS: %s", path) continue try: gdf = gdf.to_crs(POLYGON_TARGET_CRS) except Exception as exc: LOGGER.warning("Falha ao reprojetar geoparquet %s para %s: %s", path, POLYGON_TARGET_CRS, exc) continue available_fields = [field for field in POLYGON_CONTEXT_FIELDS if field in gdf.columns] model_context_fields = [field for field in available_fields if field != "NUMBLOCO"] if not model_context_fields: LOGGER.warning("Geoparquet %s sem campos espaciais esperados.", path) continue geometry_col = gdf.geometry.name work = gdf[available_fields + [geometry_col]].copy().reset_index(drop=True) work["__geom_area"] = work.geometry.area numbloco_key_index: dict[str, list[int]] = {} if "NUMBLOCO" in work.columns: work["__numbloco_key"] = work["NUMBLOCO"].map(_normalize_numbloco_key) for idx, key in work["__numbloco_key"].items(): if key: numbloco_key_index.setdefault(str(key), []).append(int(idx)) LOGGER.info( "Geoparquet enriquecido carregado: %s (n=%s, campos=%s, NUMBLOCO=%s chaves)", path.name, len(work), ",".join(available_fields), len(numbloco_key_index), ) return PolygonContext( path=path, gdf=work, available_fields=available_fields, numbloco_key_index=numbloco_key_index, fallback_max_distance_m=float( _to_float( os.getenv("POLYGON_FALLBACK_MAX_DISTANCE_M", POLYGON_FALLBACK_MAX_DISTANCE_M), POLYGON_FALLBACK_MAX_DISTANCE_M, ) or POLYGON_FALLBACK_MAX_DISTANCE_M ), to_base_crs=Transformer.from_crs("EPSG:4326", gdf.crs, always_xy=True), ) context.error = "Nenhum geoparquet enriquecido encontrado." LOGGER.warning(context.error) return context def _candidate_zoneamento_paths(base_dir: Path) -> list[Path]: candidates: list[Path] = [] env_path = os.getenv(ZONEAMENTO_SHP_ENV, "").strip() if env_path: path = Path(env_path).expanduser() if path.suffix.lower() == ".shp": candidates.append(path) search_roots: list[Path] = [] for root in _candidate_base_dirs(base_dir): search_roots.extend([root / "pddua_shp", root]) patterns = ["*SUBUNIDADE*.shp", "*subunidade*.shp", "*ZONEAMENTO*.shp", "*.shp"] for root in search_roots: if not root.exists(): continue for pattern in patterns: for shp in sorted(root.glob(pattern)): if shp.is_file() and shp not in candidates: candidates.append(shp) return candidates def _candidate_iapond_shapefile_paths(base_dir: Path, exclude: Path | None = None) -> list[Path]: candidates: list[Path] = [] for parent in _candidate_base_dirs(base_dir): for root in [parent / "pddua_shp", parent]: if not root.exists(): continue for pattern in ["*ANEXO6*.shp", "*IA*.shp", "*.shp"]: for shp in sorted(root.glob(pattern)): if not shp.is_file(): continue if exclude is not None and shp.resolve() == exclude.resolve(): continue if shp not in candidates: candidates.append(shp) return candidates def _build_iapond_map_from_dataframe(df: pd.DataFrame) -> dict[str, float]: if df.empty: return {} columns = [str(col) for col in df.columns] ia_col = _find_column_by_alias(columns, ZONEAMENTO_COLUMN_ALIASES["IA"]) if ia_col is None: return {} code_columns: list[str] = [] for canonical in ("CODIGO", "INDICE"): code_col = _find_column_by_alias(columns, ZONEAMENTO_COLUMN_ALIASES[canonical]) if code_col and code_col not in code_columns: code_columns.append(code_col) if not code_columns: return {} out: dict[str, float] = {} for code_col in code_columns: codes = df[code_col].tolist() ia_values = df[ia_col].tolist() for code_raw, ia_raw in zip(codes, ia_values): code = _normalize_code_token(code_raw) ia_number = _to_float(ia_raw) if code == "" or ia_number is None: continue out[code] = float(ia_number) return out def _load_iapond_map_from_json(path: Path) -> dict[str, float]: payload = _read_json(path) out: dict[str, float] = {} if isinstance(payload, dict): for key, value in payload.items(): code = _normalize_code_token(key) ia_number = _to_float(value) if code and ia_number is not None: out[code] = float(ia_number) return out if isinstance(payload, list): try: frame = pd.DataFrame(payload) except Exception: return {} return _build_iapond_map_from_dataframe(frame) return {} def _load_iapond_map_from_pdf(path: Path) -> dict[str, float]: try: from pypdf import PdfReader except Exception as exc: LOGGER.warning("pypdf indisponivel para leitura de %s: %s", path, exc) return {} try: reader = PdfReader(str(path)) except Exception as exc: LOGGER.warning("Falha ao abrir PDF de mapeamento IA %s: %s", path, exc) return {} out: dict[str, float] = {} saw_header = False header_pattern = re.compile(r"\bCODIGO\b.*\bIA\b") row_pattern = re.compile(r"^\s*([0-9]+(?:[.,][0-9]+)*)\s*[\|;,:\t ]+\s*([0-9]+(?:[.,][0-9]+)?)\s*$") for page in reader.pages: text = page.extract_text() or "" for line in text.splitlines(): ascii_line = _normalize_ascii_token(line) if header_pattern.search(ascii_line): saw_header = True continue if not saw_header: continue match = row_pattern.match(line.strip()) if not match: continue code = _normalize_code_token(match.group(1)) ia_number = _to_float(match.group(2)) if code and ia_number is not None: out[code] = float(ia_number) if not out: LOGGER.warning("PDF %s nao trouxe linhas CODIGO->IA em formato tabular reconhecivel.", path.name) return out def _candidate_iapond_map_paths(base_dir: Path) -> list[Path]: candidates: list[Path] = [] env_map = os.getenv(IAPOND_MAP_ENV, "").strip() if env_map: candidates.append(Path(env_map).expanduser()) for parent in _candidate_base_dirs(base_dir): defaults = [ parent / "pddua_shp" / "iapond_map.csv", parent / "pddua_shp" / "iapond_map.parquet", parent / "pddua_shp" / "iapond_map.json", parent / "iapond_map.csv", parent / "iapond_map.parquet", parent / "iapond_map.json", parent / "data" / "reference" / "iapond_map.csv", parent / "data" / "reference" / "iapond_map.parquet", parent / "data" / "reference" / "iapond_map.json", ] for path in defaults: if path not in candidates: candidates.append(path) return candidates def _candidate_iapond_pdf_paths(base_dir: Path) -> list[Path]: candidates: list[Path] = [] env_pdf = os.getenv(IAPOND_PDF_ENV, "").strip() if env_pdf: candidates.append(Path(env_pdf).expanduser()) for parent in _candidate_base_dirs(base_dir): defaults = [ parent / "pddua_layout_anexo6.pdf", parent / "pddua_shp" / "pddua_layout_anexo6.pdf", parent / "docs" / "manuals" / "pddua_layout_anexo6.pdf", parent / "docs" / "technical" / "pddua_layout_anexo6.pdf", ] for path in defaults: if path not in candidates: candidates.append(path) return candidates def _load_iapond_map_from_path(path: Path) -> dict[str, float]: suffix = path.suffix.lower() if suffix == ".json": return _load_iapond_map_from_json(path) if suffix in {".csv", ".txt"}: frame = pd.read_csv(path) return _build_iapond_map_from_dataframe(frame) if suffix == ".parquet": frame = pd.read_parquet(path) return _build_iapond_map_from_dataframe(frame) if suffix == ".pdf": return _load_iapond_map_from_pdf(path) return {} def _load_iapond_map_from_shapefile(path: Path, gpd_module: Any) -> dict[str, float]: try: frame = gpd_module.read_file(path, ignore_geometry=True) except TypeError: frame = gpd_module.read_file(path) except Exception as exc: LOGGER.warning("Falha ao ler shapefile de IA %s: %s", path, exc) return {} frame_df = pd.DataFrame(frame) if "geometry" in frame_df.columns: frame_df = frame_df.drop(columns=["geometry"]) return _build_iapond_map_from_dataframe(frame_df) def _resolve_iapond_from_zone_row(row: pd.Series, zoneamento: ZoneamentoContext) -> tuple[float | None, str]: if "IA" in row.index: ia_number = _to_float(row.get("IA")) if ia_number is not None: return float(max(ia_number, 1e-6)), "zoneamento_shp_ia" code_candidates = [ _normalize_code_token(row.get("INDICE")), _normalize_code_token(row.get("CODIGO")), ] for code in code_candidates: if code and code in IAPOND_BY_INDICE: return float(IAPOND_BY_INDICE[code]), "indice_anexo6" if code and code in zoneamento.iapond_map: return float(max(zoneamento.iapond_map[code], 1e-6)), f"mapa_ia:{zoneamento.iapond_map_source}" return None, "nao_disponivel" def load_zoneamento_data(base_dir: Path = BASE_DIR) -> ZoneamentoContext: context = ZoneamentoContext() try: import geopandas as gpd except Exception as exc: context.error = f"geopandas indisponivel: {exc}" LOGGER.warning("Zoneamento desativado: %s", context.error) return context candidates = _candidate_zoneamento_paths(base_dir) if not candidates: context.error = "Nenhum shapefile de zoneamento encontrado." LOGGER.warning(context.error) return context best_score = -1 selected_path: Path | None = None selected_gdf: pd.DataFrame | None = None selected_column_map: dict[str, str] = {} for path in candidates: if not path.exists(): continue try: gdf = gpd.read_file(path) except Exception as exc: LOGGER.warning("Falha ao ler shapefile %s: %s", path, exc) continue if gdf.empty: continue columns = [str(col) for col in gdf.columns] canonical_map: dict[str, str] = {} used_source_columns: set[str] = set() for canonical, aliases in ZONEAMENTO_COLUMN_ALIASES.items(): col = _find_column_by_alias(columns, aliases + [canonical]) if col and col not in used_source_columns: canonical_map[canonical] = col used_source_columns.add(col) score = len([field for field in ZONEAMENTO_FIELDS if field in canonical_map]) if score > best_score: best_score = score selected_path = path selected_gdf = gdf selected_column_map = canonical_map if score == len(ZONEAMENTO_FIELDS): break if selected_path is None or selected_gdf is None or best_score < 3: context.error = "Nenhum shapefile com colunas suficientes para zoneamento (MZ/UEU/SUBUNIDADE/...)" LOGGER.warning(context.error) return context if selected_gdf.crs is None: context.error = f"Shapefile sem CRS: {selected_path}" LOGGER.warning(context.error) return context try: selected_gdf = selected_gdf.to_crs(epsg=3857) except Exception as exc: context.error = f"Falha ao reprojetar shapefile {selected_path.name} para EPSG:3857: {exc}" LOGGER.warning(context.error) return context keep_columns = [selected_column_map[field] for field in selected_column_map if selected_column_map[field] in selected_gdf.columns] keep_columns = list(dict.fromkeys(keep_columns)) geometry_col = selected_gdf.geometry.name work = selected_gdf[keep_columns + [geometry_col]].copy() rename_map = {source: canonical for canonical, source in selected_column_map.items() if source in work.columns} work = work.rename(columns=rename_map) work = work.reset_index(drop=True) work["__geom_area"] = work.geometry.area iapond_map = _build_iapond_map_from_dataframe(work.drop(columns=[geometry_col], errors="ignore")) iapond_source = "shapefile_ia" if not iapond_map: for shp_map_path in _candidate_iapond_shapefile_paths(base_dir, exclude=selected_path): external_map = _load_iapond_map_from_shapefile(shp_map_path, gpd) if external_map: iapond_map = external_map iapond_source = str(shp_map_path.name) break if not iapond_map: for map_path in _candidate_iapond_map_paths(base_dir): if not map_path.exists(): continue try: external_map = _load_iapond_map_from_path(map_path) except Exception as exc: LOGGER.warning("Falha ao ler mapeamento IA %s: %s", map_path, exc) continue if external_map: iapond_map = external_map iapond_source = str(map_path.name) break if not iapond_map: for pdf_path in _candidate_iapond_pdf_paths(base_dir): if not pdf_path.exists(): continue try: external_map = _load_iapond_map_from_path(pdf_path) except Exception as exc: LOGGER.warning("Falha ao ler IA de PDF %s: %s", pdf_path, exc) continue if external_map: iapond_map = external_map iapond_source = str(pdf_path.name) break context.shp_path = selected_path context.gdf = work context.column_map = selected_column_map context.iapond_map = iapond_map context.iapond_map_source = iapond_source if iapond_map else "fallback_default" LOGGER.info( "Zoneamento carregado de %s (n=%s, campos_zoneamento=%s, mapa_ia=%s)", selected_path.name, len(work), best_score, len(iapond_map), ) if not iapond_map: LOGGER.warning("Mapeamento IA nao encontrado; IAPOND permanecera manual quando necessario.") return context def _infer_zoneamento_context(zoneamento: ZoneamentoContext, x: float, y: float) -> tuple[dict[str, Any], dict[str, str], list[str]]: if not zoneamento.has_data: return {}, {}, [] try: from shapely.geometry import Point except Exception as exc: LOGGER.warning("Shapely indisponivel para inferencia de zoneamento: %s", exc) return {}, {}, [] point = Point(float(x), float(y)) alerts: list[str] = [] try: idx = zoneamento.gdf.sindex.query(point, predicate="intersects") idx_list = list(idx) except Exception: mask = zoneamento.gdf.geometry.intersects(point) idx_list = zoneamento.gdf.index[mask].tolist() if not idx_list: alerts.append("Coordenada fora do shapefile de zoneamento; campos de zoneamento permanecem manuais.") return {}, {}, alerts matches = zoneamento.gdf.iloc[idx_list] if "__geom_area" in matches.columns: row = matches.sort_values("__geom_area", ascending=True).iloc[0] else: row = matches.iloc[0] values: dict[str, Any] = {} sources: dict[str, str] = {} for field in ZONEAMENTO_FIELDS: if field in row.index and not _is_missing(row.get(field)): values[field] = row.get(field) sources[field] = "zoneamento_shp" iapond_value, iapond_source = _resolve_iapond_from_zone_row(row, zoneamento) if iapond_value is not None: values["IAPOND"] = float(iapond_value) sources["IAPOND"] = iapond_source else: alerts.append("INDICE sem IA mapeado; IAPOND permanece manual.") return values, sources, alerts def _polygon_row_to_context( polygons: PolygonContext, row: pd.Series, source_label: str, polygon_distance: float, ) -> tuple[dict[str, Any], dict[str, str], list[str]]: values: dict[str, Any] = {} sources: dict[str, str] = {} for field in polygons.available_fields: if field in row.index and not _is_missing(row.get(field)): values[field] = row.get(field) sources[field] = source_label values["_polygon_distance_m"] = polygon_distance values["_polygon_lookup_mode"] = source_label if "NUMBLOCO" in row.index and not _is_missing(row.get("NUMBLOCO")): values["_polygon_numbloco"] = row.get("NUMBLOCO") return values, sources, [] def _select_best_polygon_row( polygons: PolygonContext, matches: pd.DataFrame, point: Any | None = None, ) -> tuple[pd.Series | None, float, list[str]]: if matches.empty: return None, float("nan"), [] alerts: list[str] = [] if point is not None and "geometry" in matches.columns: try: intersecting = matches.loc[matches.geometry.intersects(point)] except Exception: intersecting = pd.DataFrame() if not intersecting.empty: if "__geom_area" in intersecting.columns: return intersecting.sort_values("__geom_area", ascending=True).iloc[0], 0.0, alerts return intersecting.iloc[0], 0.0, alerts try: distances = matches.geometry.distance(point) nearest_idx = distances.idxmin() nearest_distance = _to_float(distances.loc[nearest_idx], float("nan")) except Exception: nearest_idx = None nearest_distance = float("nan") if nearest_idx is not None and np.isfinite(nearest_distance): alerts.append( "NUMBLOCO encontrado, mas a coordenada nao intersecta seus poligonos; " f"selecionado registro do mesmo NUMBLOCO mais proximo a {nearest_distance:.1f} m." ) return matches.loc[nearest_idx], float(nearest_distance), alerts rank = matches.copy() feature_cols = [field for field in matches.columns if field in polygons.available_fields] rank["__non_null_count"] = rank[feature_cols].notna().sum(axis=1) if feature_cols else 0 if "__geom_area" not in rank.columns: rank["__geom_area"] = float("inf") rank = rank.sort_values(["__non_null_count", "__geom_area"], ascending=[False, True]) return rank.iloc[0], float("nan"), alerts def _select_polygon_row_by_numbloco( polygons: PolygonContext, numbloco: Any, x: float | None = None, y: float | None = None, ) -> tuple[pd.Series | None, float, list[str], int]: if not polygons.has_data or not polygons.numbloco_key_index: return None, float("nan"), [], 0 key = _normalize_numbloco_key(numbloco) if not key: return None, float("nan"), [], 0 candidate_idx = polygons.numbloco_key_index.get(key) if not candidate_idx: return ( None, float("nan"), [f"NUMBLOCO `{_safe_text(numbloco, '')}` nao encontrado no geoparquet enriquecido; usada busca espacial."], 0, ) point = None x_number = _to_float(x) y_number = _to_float(y) if x_number is not None and y_number is not None and np.isfinite(x_number) and np.isfinite(y_number): try: from shapely.geometry import Point point = Point(x_number, y_number) except Exception as exc: LOGGER.warning("Shapely indisponivel para desempate por NUMBLOCO: %s", exc) matches = polygons.gdf.loc[candidate_idx] row, polygon_distance, alerts = _select_best_polygon_row(polygons, matches, point) if row is None: return ( None, float("nan"), [f"NUMBLOCO `{_safe_text(numbloco, '')}` sem linha utilizavel no geoparquet; usada busca espacial."], len(candidate_idx), ) if len(candidate_idx) > 1: alerts.append(f"NUMBLOCO encontrado em {len(candidate_idx)} poligonos; aplicado desempate deterministico.") return row, polygon_distance, alerts, len(candidate_idx) def _infer_polygon_context_by_numbloco( polygons: PolygonContext, numbloco: Any, x: float | None = None, y: float | None = None, ) -> tuple[dict[str, Any], dict[str, str], list[str]]: row, polygon_distance, alerts, _ = _select_polygon_row_by_numbloco(polygons, numbloco, x, y) if row is None: return {}, {}, alerts values, sources, _ = _polygon_row_to_context(polygons, row, "poligono_numbloco", polygon_distance) return values, sources, alerts def _infer_polygon_context(polygons: PolygonContext, x: float, y: float) -> tuple[dict[str, Any], dict[str, str], list[str]]: if not polygons.has_data: return {}, {}, [] try: from shapely.geometry import Point except Exception as exc: LOGGER.warning("Shapely indisponivel para geoparquet enriquecido: %s", exc) return {}, {}, [] point = Point(float(x), float(y)) alerts: list[str] = [] try: idx = polygons.gdf.sindex.query(point, predicate="intersects") idx_list = list(idx) except Exception: mask = polygons.gdf.geometry.intersects(point) idx_list = polygons.gdf.index[mask].tolist() source_label = "poligono_intersect" row = None polygon_distance = 0.0 if idx_list: matches = polygons.gdf.iloc[idx_list] row, polygon_distance, _ = _select_best_polygon_row(polygons, matches, point) else: try: distances = polygons.gdf.geometry.distance(point) nearest_idx = distances.idxmin() nearest_distance = _to_float(distances.loc[nearest_idx], float("nan")) except Exception: nearest_idx = None nearest_distance = float("nan") if nearest_idx is not None and np.isfinite(nearest_distance): row = polygons.gdf.loc[nearest_idx] source_label = "poligono_proximo" polygon_distance = float(nearest_distance) if nearest_distance > polygons.fallback_max_distance_m: alerts.append( f"Coordenada sem interseccao exata com a malha enriquecida; " f"usado poligono mais proximo a {nearest_distance:.1f} m " "(baixa confiabilidade espacial)." ) else: alerts.append( f"Coordenada sem interseccao exata com a malha enriquecida; " f"usado poligono mais proximo a {nearest_distance:.1f} m." ) else: alerts.append( "Camada de poligonos enriquecidos sem interseccao para a coordenada; " "mantido fallback por base de referencia." ) return {}, {}, alerts if row is None: return {}, {}, alerts values, sources, _ = _polygon_row_to_context(polygons, row, source_label, polygon_distance) return values, sources, alerts def _infer_rh_context_by_numbloco(rh_context: RhNumblocoContext, numbloco: Any) -> tuple[dict[str, Any], dict[str, str], list[str]]: if not rh_context.has_data: return {}, {}, [] key = _normalize_numbloco_key(numbloco) if not key: return {}, {}, [] rh_value = rh_context.rh_by_numbloco.get(key) if rh_value is None or not np.isfinite(float(rh_value)): return {}, {}, [f"NUMBLOCO `{_safe_text(numbloco, '')}` sem RH no CSV; usado fallback espacial quando disponivel."] return {"RH": float(rh_value)}, {"RH": "rh_numbloco_csv"}, [] def _infer_testada_context_by_numbloco( testada_context: TestadaNumblocoContext, numbloco: Any, ) -> tuple[dict[str, Any], dict[str, str], list[str]]: if not testada_context.has_data: return {}, {}, [] key = _normalize_numbloco_key(numbloco) if not key: return {}, {}, [] testada_value = testada_context.testada_by_numbloco.get(key) if testada_value is None or not np.isfinite(float(testada_value)): return {}, {}, [f"NUMBLOCO `{_safe_text(numbloco, '')}` sem TESTADA no CSV; usado fallback atual."] return ( {"TESTADA": float(testada_value), "TESTADA_bruta": float(testada_value)}, {"TESTADA": "testada_numbloco_csv", "TESTADA_bruta": "testada_numbloco_csv"}, [], ) def _build_reference_categories(bundle: ArtifactBundle, reference: ReferenceContext) -> dict[str, list[str]]: maps: dict[str, list[str]] = {} for feature in bundle.feature_names: if str(bundle.feature_types.get(feature, "")).lower() != "c": continue values: list[str] = [] if reference.has_data and feature in reference.df.columns: values.extend( _safe_text(value, "") for value in reference.df[feature].dropna().tolist() if _safe_text(value, "") != "" ) if feature == "Ano_Dado" and reference.has_data: if "Ano_Dado" in reference.df.columns: values.extend( str(int(value)) for value in pd.to_numeric(reference.df["Ano_Dado"], errors="coerce").dropna().tolist() ) elif "ANO" in reference.df.columns: values.extend( str(int(value)) for value in pd.to_numeric(reference.df["ANO"], errors="coerce").dropna().tolist() ) if feature == "faixa_area_modelo": values.extend(bundle.area_labels) if feature == "faixa_rh": values.extend(RH_LABELS) if feature == "FONTE": values.extend(["0", "1", "Venda", "Oferta"]) maps[feature] = _normalize_category_values(values) return maps def _infer_local_context(reference: ReferenceContext, x: float, y: float, k_neighbors: int) -> dict[str, Any]: if not reference.has_data: return {} dx = reference.x - x dy = reference.y - y dist2 = dx * dx + dy * dy if dist2.size == 0: return {} k = max(1, min(k_neighbors, dist2.size)) idx = np.argpartition(dist2, k - 1)[:k] idx = idx[np.argsort(dist2[idx])] local = reference.df.iloc[idx] context: dict[str, Any] = {} for field in CONTEXT_FIELDS: if field not in local.columns: continue prefer_mode = field in { "FONTE", "Ano_Dado", "BAIRRO", "COD_BAIRRO", "FINALIDADE", "MZ", "UEU", "SUBUNIDADE", "DENSIDADE", "ATIVIDADE", "INDICE", "VOLUMETRIA", } context[field] = _summarize_series(local[field], prefer_mode=prefer_mode) context["_nearest_distance_m"] = float(np.sqrt(dist2[idx[0]])) return context def select_model( tipo_imovel: str, bundles: dict[str, ArtifactBundle], area_m2: float | None = None, ) -> tuple[ArtifactBundle, str]: token = _normalize_token(tipo_imovel) if token.startswith("AUTO"): if area_m2 is None or not np.isfinite(area_m2) or area_m2 <= 0: raise ValueError("Area valida e obrigatoria para roteamento automatico.") key = "terreno" if float(area_m2) < AUTO_GLEBA_AREA_THRESHOLD else "gleba" source = f"auto_area_{AUTO_GLEBA_AREA_THRESHOLD:.0f}" elif token.startswith("TER"): key = "terreno" source = "manual" elif token.startswith("GLE"): key = "gleba" source = "manual" else: raise ValueError("Tipo de imovel invalido. Use Auto (area), Terreno ou Gleba.") if key not in bundles: raise ValueError(f"Artefatos nao encontrados para {key}.") return bundles[key], source def _pick_value( field: str, overrides_ci: dict[str, Any], direct_inputs: dict[str, Any], local_context: dict[str, Any], default: Any, aliases: list[str] | None = None, context_sources: dict[str, str] | None = None, reliable_only: bool = False, ) -> tuple[Any, str]: aliases = aliases or [] context_sources = context_sources or {} keys = [field.lower(), *(alias.lower() for alias in aliases)] candidate_names = [field, *aliases] for key in keys: if key in overrides_ci and not _is_missing(overrides_ci[key]): return overrides_ci[key], "override_json" for name in candidate_names: if name in direct_inputs and not _is_missing(direct_inputs[name]): return direct_inputs[name], "input_ui" for name in candidate_names: if name in local_context and not _is_missing(local_context[name]): source = context_sources.get(name, context_sources.get(field, "inferido_coord")) if reliable_only and not _is_reliable_model_context_source(source): continue return local_context[name], source return default, "default" def _pick_testada_value( overrides_ci: dict[str, Any], direct_inputs: dict[str, Any], local_context: dict[str, Any], context_sources: dict[str, str], ) -> tuple[Any, str]: keys = ["testada_bruta", "testada"] for key in keys: if key in overrides_ci and not _is_missing(overrides_ci[key]): return overrides_ci[key], "override_json" for name in ("TESTADA_bruta", "TESTADA"): if name in local_context and not _is_missing(local_context[name]): source = context_sources.get(name, context_sources.get("TESTADA_bruta", "inferido_coord")) if source == "testada_numbloco_csv": return local_context[name], source return _pick_value( "TESTADA_bruta", overrides_ci, direct_inputs, local_context, None, aliases=["TESTADA", "testada", "testada_bruta"], context_sources=context_sources, reliable_only=True, ) def _resolve_temporal_index(bundle: ArtifactBundle, ano_dado: int) -> tuple[float, str | None, str]: if not bundle.tempo_index_map: return 1.0, None, "sem_artifato" periodos = sorted(bundle.tempo_index_map.keys()) candidatos = [periodo for periodo in periodos if str(ano_dado) in periodo] if candidatos: periodo = sorted(candidatos)[-1] return float(bundle.tempo_index_map[periodo]), periodo, "ano_match" if not bundle.tempo_index_df.empty and {"periodo_tempo", "n_obs_periodo"}.issubset(bundle.tempo_index_df.columns): tmp = bundle.tempo_index_df.copy() tmp["n_obs_periodo"] = pd.to_numeric(tmp["n_obs_periodo"], errors="coerce") tmp = tmp.dropna(subset=["n_obs_periodo"]) if not tmp.empty: row = tmp.sort_values("n_obs_periodo", ascending=False).iloc[0] periodo = _safe_text(row["periodo_tempo"], "") if periodo in bundle.tempo_index_map: return float(bundle.tempo_index_map[periodo]), periodo, "maior_n_obs" periodo = periodos[-1] return float(bundle.tempo_index_map[periodo]), periodo, "fallback_ultimo" def _lookup_spatial_surface(bundle: ArtifactBundle, segmento_area: str, grid_id: str) -> tuple[float | None, str]: seg_key = _normalize_token(segmento_area) if seg_key and grid_id: val = bundle.grid_value_map.get((seg_key, grid_id)) if val is not None: return float(val), "grid" if seg_key: val = bundle.segment_value_map.get(seg_key) if val is not None: return float(val), "segmento" if bundle.global_surface_value is not None: return float(bundle.global_surface_value), "global" return None, "sem_artifato" def apply_preprocessing( bundle: ArtifactBundle, payload: dict[str, Any], local_context: dict[str, Any], overrides_ci: dict[str, Any], context_sources: dict[str, str] | None = None, ) -> tuple[dict[str, Any], dict[str, str], list[str]]: alerts: list[str] = [] source_map: dict[str, str] = {} context_sources = context_sources or {} area = _to_float(payload.get("area")) if area is None or area <= 0: raise ValueError("Area deve ser numerica e maior que zero.") lat = _to_float(payload.get("lat")) lon = _to_float(payload.get("lon")) if lat is None or lon is None: raise ValueError("Latitude e longitude sao obrigatorias.") if not (-90.0 <= lat <= 90.0 and -180.0 <= lon <= 180.0): raise ValueError("Latitude/longitude fora da faixa valida.") x, y = WGS84_TO_WEBMERC.transform(lon, lat) nearest_distance = _to_float(local_context.get("_nearest_distance_m"), float("nan")) if np.isfinite(nearest_distance): threshold = bundle.grid_cell_size * 8.0 if nearest_distance > threshold: alerts.append( f"Coordenada distante da base de referencia ({nearest_distance:.0f} m). " "Revisar localizacao e dados cadastrais." ) direct_inputs = { "BAIRRO": payload.get("bairro"), "Ano_Dado": payload.get("ano_dado"), "IAPOND": payload.get("iapond"), "APP": payload.get("app"), "LOTPOS": payload.get("lotpos"), "TESTADA": payload.get("testada"), "TESTADA_bruta": payload.get("testada"), } mercado_default = "TERRENO" if bundle.key == "terreno" else "GLEBA" rh_raw, source_map["RH"] = _pick_value( "RH", overrides_ci, direct_inputs, local_context, None, context_sources=context_sources, reliable_only=True, ) iapond_raw, source_map["IAPOND"] = _pick_value( "IAPOND", overrides_ci, direct_inputs, local_context, None, context_sources=context_sources, reliable_only=True, ) app_raw, source_map["APP"] = _pick_value( "APP", overrides_ci, direct_inputs, local_context, None, context_sources=context_sources, reliable_only=True, ) cp_raw, source_map["CP"] = _pick_value( "CP", overrides_ci, direct_inputs, local_context, None, context_sources=context_sources, reliable_only=True, ) lotpos_raw, source_map["LOTPOS"] = _pick_value( "LOTPOS", overrides_ci, direct_inputs, local_context, None, context_sources=context_sources, reliable_only=True, ) ano_raw, source_map["Ano_Dado"] = _pick_value( "Ano_Dado", overrides_ci, direct_inputs, local_context, None, aliases=["ano", "ano_dado"], context_sources=context_sources, reliable_only=True, ) fonte_raw = DEFAULTS_CATEGORICAL["FONTE"] source_map["FONTE"] = "fixo_zero" bairro_raw, source_map["BAIRRO"] = _pick_value( "BAIRRO", overrides_ci, direct_inputs, local_context, "", aliases=["bairro"], context_sources=context_sources, ) finalidade_raw, source_map["FINALIDADE"] = _pick_value( "FINALIDADE", overrides_ci, direct_inputs, local_context, mercado_default, aliases=["finalidade"], context_sources=context_sources, ) testada_raw, source_map["TESTADA_bruta"] = _pick_testada_value( overrides_ci, direct_inputs, local_context, context_sources, ) testada = _to_float(testada_raw) if testada is None or testada <= 0: raise ValueError("TESTADA ausente; informe manualmente ou use uma coordenada com TESTADA disponivel no poligono.") rh = _to_float(rh_raw) if rh is not None: rh = max(rh, 1e-6) iapond = _to_float(iapond_raw) if iapond is not None: iapond = max(iapond, 1e-6) app = _to_float(app_raw) if app is not None: app = min(max(app, 0.0), 1.0) cp_override = _to_float(cp_raw) if source_map.get("CP") == "override_json" and cp_override is not None: cp = cp_override else: cp = _derive_cp(area=area, testada=testada) source_map["CP"] = "regra_area_testada" pe = area / testada source_map["PE"] = "regra_area_testada" lotpos = _to_float(lotpos_raw) ano_dado = _to_int(ano_raw) if ano_dado is not None: ano_dado = min(max(ano_dado, 2010), 2100) area_simpson_raw, source_map["area_simpson"] = _pick_value( "area_simpson", overrides_ci, direct_inputs, local_context, None, context_sources=context_sources, reliable_only=True, ) taxa_ocupacao_raw, source_map["taxa_ocupacao"] = _pick_value( "taxa_ocupacao", overrides_ci, direct_inputs, local_context, None, context_sources=context_sources, reliable_only=True, ) bldg_density_raw, source_map["bldg_density"] = _pick_value( "bldg_density", overrides_ci, direct_inputs, local_context, None, context_sources=context_sources, reliable_only=True, ) avg_bldg_raw, source_map["avg_bldg_footprint"] = _pick_value( "avg_bldg_footprint", overrides_ci, direct_inputs, local_context, None, context_sources=context_sources, reliable_only=True, ) circularity_raw, source_map["circularity"] = _pick_value( "circularity", overrides_ci, direct_inputs, local_context, None, context_sources=context_sources, reliable_only=True, ) dist_road_raw, source_map["dist_to_main_road"] = _pick_value( "dist_to_main_road", overrides_ci, direct_inputs, local_context, None, context_sources=context_sources, reliable_only=True, ) dist_park_raw, source_map["dist_to_park"] = _pick_value( "dist_to_park", overrides_ci, direct_inputs, local_context, None, context_sources=context_sources, reliable_only=True, ) area_simpson = _to_float(area_simpson_raw) taxa_ocupacao = _to_float(taxa_ocupacao_raw) bldg_density = _to_float(bldg_density_raw) avg_bldg = _to_float(avg_bldg_raw) circularity = _to_float(circularity_raw) dist_road = _to_float(dist_road_raw) if dist_road is not None: dist_road = max(dist_road, 0.0) dist_park = _to_float(dist_park_raw) if dist_park is not None: dist_park = max(dist_park, 0.0) dist_weighted_density = None if bldg_density is not None and dist_road is not None: dist_weighted_density = bldg_density / (dist_road + 1.0) source_map["Dist_weighted_Density"] = "calculado" faixa_area = _categorize_area(area, bundle.area_bins, bundle.area_labels) faixa_rh = _categorize_rh(rh) if rh is not None else None area_total_construivel = None if app is not None and iapond is not None: area_total_construivel = max(area * (1.0 - app) * iapond, 1e-6) log_area = math.log1p(max(area, 0.0)) log_testada = math.log1p(max(testada, 0.0)) finite_bins = [value for value in bundle.area_bins if np.isfinite(value)] if finite_bins: area_min = min(finite_bins) area_max = max(finite_bins) if area < area_min or area > area_max: alerts.append( f"Area informada ({area:.2f} m2) fora da faixa observada nos bins do modelo " f"({area_min:.2f} a {area_max:.2f} m2)." ) grid_id = _compute_grid_id(x, y, bundle.grid_cell_size) spatial_idx, spatial_source = _lookup_spatial_surface(bundle, faixa_area, grid_id) source_map["indice_espacial_grid"] = spatial_source if spatial_idx is None: spatial_idx = float("nan") temporal_idx, periodo_tempo, temporal_source = _resolve_temporal_index(bundle, ano_dado) source_map["indice_temporal_interno"] = temporal_source bairro_text = _safe_text(bairro_raw, "") bairro_key = _normalize_token(bairro_text) tier_localizacao = bundle.tier_map_norm.get(bairro_key, "localizacao_desconhecida") source_map["tier_localizacao"] = "tier_map" if bairro_key in bundle.tier_map_norm else "fallback" zone_values: dict[str, str] = {} for field in ZONEAMENTO_FIELDS: value, source = _pick_value( field, overrides_ci, direct_inputs, local_context, None, aliases=[field.lower()], context_sources=context_sources, reliable_only=True, ) zone_values[field] = None if _is_missing(value) else _safe_text(value) source_map[field] = source if bundle.key == "gleba" and zone_values[field] is None: alerts.append(f"{field} nao inferido automaticamente; informe manualmente via override quando necessario.") prepared = { "AREA_bruta": area, "TESTADA_bruta": testada, "lat": lat, "lon": lon, "x": float(x), "y": float(y), "RH": rh, "IAPOND": iapond, "APP": app, "PE": pe, "CP": cp, "LOTPOS": lotpos, "Ano_Dado": str(ano_dado), "FONTE": fonte_raw, "FINALIDADE": _safe_text(finalidade_raw, mercado_default), "BAIRRO": bairro_text, "faixa_area_modelo": faixa_area, "faixa_rh": faixa_rh, "log_area_modelo": log_area, "log_testada_modelo": log_testada, "Area_Total_Construivel": area_total_construivel, "area_simpson": area_simpson, "taxa_ocupacao": taxa_ocupacao, "bldg_density": bldg_density, "avg_bldg_footprint": avg_bldg, "circularity": circularity, "dist_to_main_road": dist_road, "dist_to_park": dist_park, "Dist_weighted_Density": dist_weighted_density, "indice_espacial_grid": spatial_idx, "periodo_tempo": periodo_tempo or "", "indice_temporal_interno": temporal_idx, "tier_localizacao": tier_localizacao, } prepared.update(zone_values) if not _is_missing(prepared["FONTE"]): fonte_categories = bundle.reference_categories.get("FONTE", ["0", "1", "desconhecido"]) prepared["FONTE"] = _normalize_fonte(prepared["FONTE"], fonte_categories) return prepared, source_map, alerts def _default_for_feature(feature: str, feature_type: str) -> Any: if feature in DEFAULTS_NUMERIC: return DEFAULTS_NUMERIC[feature] if feature in DEFAULTS_CATEGORICAL: return DEFAULTS_CATEGORICAL[feature] if "int" in feature_type: return 0 if feature_type == "c": return "desconhecido" return 0.0 def _prepare_categorical_value(feature: str, value: Any, categories: list[str]) -> str: if feature == "FONTE": return _normalize_fonte(value, categories) if _is_missing(value): return "desconhecido" number = _to_float(value) if number is not None and float(number).is_integer(): text = str(int(number)) else: text = _safe_text(value, "desconhecido") return text def _missing_required_features(bundle: ArtifactBundle, prepared_values: dict[str, Any]) -> list[str]: return [ feature for feature in bundle.feature_names if feature not in prepared_values or _is_missing(prepared_values.get(feature)) ] def build_features(bundle: ArtifactBundle, prepared_values: dict[str, Any]) -> pd.DataFrame: missing = _missing_required_features(bundle, prepared_values) if missing: raise ValueError( "Features obrigatorias sem preenchimento confiavel: " f"{', '.join(missing)}. Informe manualmente os campos disponiveis ou use overrides JSON." ) data: dict[str, Any] = {} for feature in bundle.feature_names: feature_type = str(bundle.feature_types.get(feature, "float")).lower() value = prepared_values[feature] if feature_type == "c": categories = bundle.reference_categories.get(feature, ["desconhecido"]) cat_value = _prepare_categorical_value(feature, value, categories) if cat_value not in categories: categories = categories + [cat_value] data[feature] = pd.Series(pd.Categorical([cat_value], categories=categories)) continue if "int" in feature_type: number = _to_float(value, 0.0) or 0.0 data[feature] = [int(round(number))] continue number = _to_float(value, 0.0) or 0.0 data[feature] = [float(number)] return pd.DataFrame(data, columns=bundle.feature_names) def predict_value(bundle: ArtifactBundle, x_input: pd.DataFrame, area_m2: float) -> dict[str, Any]: dmatrix = xgb.DMatrix( x_input, enable_categorical=True, feature_names=bundle.feature_names, ) pred_raw = float(bundle.model.predict(dmatrix)[0]) use_log_backtransform = bool(bundle.metadata.get("model_target_log_col")) pred_level = math.expm1(pred_raw) if use_log_backtransform else pred_raw pred_level = max(pred_level, 0.0) target_col = str(bundle.metadata.get("model_target_col", "")).upper() if "VUNIT" in target_col and "VTOTAL" not in target_col: vunit = pred_level vtotal = pred_level * area_m2 target_mode = "VUNIT" else: vtotal = pred_level vunit = vtotal / area_m2 if area_m2 > 0 else float("nan") target_mode = "VTOTAL" return { "pred_raw_model": pred_raw, "pred_level": pred_level, "target_mode": target_mode, "valor_total": float(vtotal), "valor_unitario": float(vunit), } def format_output( bundle: ArtifactBundle, prediction: dict[str, Any], prepared_values: dict[str, Any], source_map: dict[str, str], alerts: list[str], local_context: dict[str, Any], routing_source: str, ) -> tuple[str, dict[str, Any]]: resultado_lines = [ "### Resultado de Inferencia", f"- **Modelo aplicado:** {bundle.label} (`{bundle.model_path.name}`)", f"- **Roteamento macro:** `{routing_source}`", f"- **Escala do alvo no treino:** `{bundle.metadata.get('model_target_log_col', 'sem_log')}`", f"- **Valor unitario estimado:** **{_currency_brl(prediction['valor_unitario'])} / m2**", f"- **Valor total estimado:** **{_currency_brl(prediction['valor_total'])}**", f"- **Modo da saida:** `{prediction['target_mode']}`", ] if alerts: resultado_lines.append("- **Alertas:**") for msg in sorted(set(alerts)): resultado_lines.append(f" - {msg}") resumo = { "status": "ok", "modelo": { "tipo": bundle.label, "roteamento_macro": routing_source, "notebook_origem": bundle.notebook_name, "artifact_dir": str(bundle.artifact_dir), "model_file": bundle.model_path.name, "metadata_file": bundle.metadata_path.name, "n_features": len(bundle.feature_names), }, "predicao": prediction, "variaveis_principais": { "area_m2": prepared_values.get("AREA_bruta"), "testada_m": prepared_values.get("TESTADA_bruta"), "lat": prepared_values.get("lat"), "lon": prepared_values.get("lon"), "x": prepared_values.get("x"), "y": prepared_values.get("y"), "RH": prepared_values.get("RH"), "IAPOND": prepared_values.get("IAPOND"), "APP": prepared_values.get("APP"), "PE": prepared_values.get("PE"), "CP": prepared_values.get("CP"), "Area_Total_Construivel": prepared_values.get("Area_Total_Construivel"), "faixa_area_modelo": prepared_values.get("faixa_area_modelo"), "faixa_rh": prepared_values.get("faixa_rh"), "FONTE": prepared_values.get("FONTE"), "Ano_Dado": prepared_values.get("Ano_Dado"), "FINALIDADE": prepared_values.get("FINALIDADE"), "MZ": prepared_values.get("MZ"), "UEU": prepared_values.get("UEU"), "SUBUNIDADE": prepared_values.get("SUBUNIDADE"), "DENSIDADE": prepared_values.get("DENSIDADE"), "ATIVIDADE": prepared_values.get("ATIVIDADE"), "INDICE": prepared_values.get("INDICE"), "VOLUMETRIA": prepared_values.get("VOLUMETRIA"), "indice_espacial_grid": prepared_values.get("indice_espacial_grid"), "periodo_tempo": prepared_values.get("periodo_tempo"), "indice_temporal_interno": prepared_values.get("indice_temporal_interno"), "tier_localizacao": prepared_values.get("tier_localizacao"), }, "origem_variaveis": source_map, "diagnostico": { "distancia_vizinho_m": _to_float(local_context.get("_nearest_distance_m")), "distancia_poligono_m": _to_float(local_context.get("_polygon_distance_m")), "modo_busca_poligono": local_context.get("_polygon_lookup_mode"), "numbloco_poligono": local_context.get("_polygon_numbloco"), "alerts": sorted(set(alerts)), }, } return "\n".join(resultado_lines), _jsonable(resumo) def _build_report_limitations( bundle: ArtifactBundle, alerts: list[str], train_metrics: dict[str, float] | None, ) -> list[str]: limitations = [ "A estimativa depende da qualidade das coordenadas informadas e da cobertura da base espacial de apoio.", "A busca espacial usa o poligono intersectante; fora da malha, usa o poligono mais proximo e registra a distancia.", "A inferencia aplica artefatos congelados; o app nao reestima nem retreina modelos.", ] if train_metrics is None: limitations.append( "Predicoes de treino nao foram exportadas para este modelo; por isso R2 (log) e grafico de treino aparecem como n/d." ) metadata_warnings = bundle.metadata.get("warnings") if isinstance(metadata_warnings, list): limitations.extend( str(item) for item in metadata_warnings if "cod" not in str(item).lower() and "prd" not in str(item).lower() ) limitations.extend(alerts) return list(dict.fromkeys(limitations)) def initialize_app_state(base_dir: Path = BASE_DIR) -> AppState: state = AppState() try: state.k_neighbors = max(1, int(os.getenv("K_NEIGHBORS", "120"))) except Exception: state.k_neighbors = 120 try: state.reference = load_reference_data(base_dir) state.bundles = load_artifacts(base_dir) state.rh_numbloco = load_rh_numbloco_context(base_dir) state.testada_numbloco = load_testada_numbloco_context(base_dir) state.polygon_context = load_polygon_context(base_dir) state.zoneamento = load_zoneamento_data(base_dir) for bundle in state.bundles.values(): bundle.reference_categories = _build_reference_categories(bundle, state.reference) except Exception as exc: LOGGER.exception("Falha na inicializacao da aplicacao") state.error = str(exc) return state APP_STATE = initialize_app_state(BASE_DIR) def obter_lat_lon_centroide_por_numbloco( numbloco: Any, lat_atual: Any = None, lon_atual: Any = None, ) -> tuple[float, float] | None: polygon_x = None polygon_y = None lon_number = _to_float(lon_atual) lat_number = _to_float(lat_atual) if ( APP_STATE.polygon_context.to_base_crs is not None and lon_number is not None and lat_number is not None and -180.0 <= lon_number <= 180.0 and -90.0 <= lat_number <= 90.0 ): try: polygon_x, polygon_y = APP_STATE.polygon_context.to_base_crs.transform(lon_number, lat_number) except Exception as exc: LOGGER.warning("Falha ao transformar coordenada atual para desempate por NUMBLOCO: %s", exc) row, _, _, _ = _select_polygon_row_by_numbloco(APP_STATE.polygon_context, numbloco, polygon_x, polygon_y) if row is None: return None geometry = row.get("geometry") if geometry is None or getattr(geometry, "is_empty", True): return None if hasattr(geometry, "is_valid") and not geometry.is_valid: return None source_crs = getattr(APP_STATE.polygon_context.gdf, "crs", None) if source_crs is None: return None try: centroid = geometry.centroid if centroid is None or getattr(centroid, "is_empty", True): return None lon_centroid, lat_centroid = Transformer.from_crs(source_crs, "EPSG:4326", always_xy=True).transform( float(centroid.x), float(centroid.y), ) except Exception as exc: LOGGER.warning("Falha ao calcular centroide por NUMBLOCO: %s", exc) return None lat_result = _to_float(lat_centroid) lon_result = _to_float(lon_centroid) if ( lat_result is None or lon_result is None or not np.isfinite(lat_result) or not np.isfinite(lon_result) or not (-90.0 <= lat_result <= 90.0 and -180.0 <= lon_result <= 180.0) ): return None return float(lat_result), float(lon_result) def atualizar_lat_lon_por_numbloco(numbloco: str, lat_atual: Any, lon_atual: Any) -> tuple[Any, Any]: centroide = obter_lat_lon_centroide_por_numbloco(numbloco, lat_atual, lon_atual) if centroide is None: return gr.update(), gr.update() lat, lon = centroide return gr.update(value=lat), gr.update(value=lon) def predict( area: float, numbloco: str, testada: float | None, lat: float, lon: float, bairro: str, ano_dado: str, iapond: float | None, app: float | None, lotpos: float | None, overrides_json: str, ) -> tuple[str, dict[str, Any], str | None]: if APP_STATE.error: message = f"Erro de inicializacao: {APP_STATE.error}" return message, {"status": "erro", "mensagem": APP_STATE.error}, None try: area_value = _to_float(area) bundle, routing_source = select_model("Auto (area)", APP_STATE.bundles, area_m2=area_value) payload = { "area": area, "numbloco": None if numbloco is None or str(numbloco).strip() == "" else numbloco, "testada": testada, "lat": lat, "lon": lon, "bairro": None if bairro is None or str(bairro).strip() == "" else bairro, "ano_dado": None if ano_dado in {None, ""} else ano_dado, "fonte": "0", "iapond": iapond, "app": app, "lotpos": lotpos, } overrides_ci = _parse_overrides_json(overrides_json) lon_number = _to_float(payload["lon"]) lat_number = _to_float(payload["lat"]) if lon_number is None or lat_number is None: raise ValueError("Latitude/longitude invalidas.") x, y = WGS84_TO_WEBMERC.transform(lon_number, lat_number) local_context = _infer_local_context(APP_STATE.reference, x, y, APP_STATE.k_neighbors) context_sources = { key: "inferido_coord" for key in local_context.keys() if not str(key).startswith("_") } if APP_STATE.polygon_context.to_base_crs is not None: polygon_x, polygon_y = APP_STATE.polygon_context.to_base_crs.transform(lon_number, lat_number) numbloco_lookup_value = payload.get("numbloco") if _is_missing(numbloco_lookup_value): numbloco_lookup_value = overrides_ci.get("numbloco") polygon_context, polygon_sources, polygon_alerts = _infer_polygon_context_by_numbloco( APP_STATE.polygon_context, numbloco_lookup_value, polygon_x, polygon_y, ) if not polygon_context: spatial_context, spatial_sources, spatial_alerts = _infer_polygon_context( APP_STATE.polygon_context, polygon_x, polygon_y, ) polygon_context = spatial_context polygon_sources = spatial_sources polygon_alerts.extend(spatial_alerts) else: polygon_context, polygon_sources, polygon_alerts = {}, {}, [] zone_context, zone_sources, zone_alerts = _infer_zoneamento_context(APP_STATE.zoneamento, x, y) rh_context, rh_sources, rh_alerts = _infer_rh_context_by_numbloco( APP_STATE.rh_numbloco, payload.get("numbloco") if not _is_missing(payload.get("numbloco")) else overrides_ci.get("numbloco"), ) testada_context, testada_sources, testada_alerts = _infer_testada_context_by_numbloco( APP_STATE.testada_numbloco, payload.get("numbloco") if not _is_missing(payload.get("numbloco")) else overrides_ci.get("numbloco"), ) merged_context = dict(local_context) merged_context.update(zone_context) merged_context.update(polygon_context) merged_context.update(rh_context) merged_context.update(testada_context) context_sources.update(zone_sources) context_sources.update(polygon_sources) context_sources.update(rh_sources) context_sources.update(testada_sources) prepared, source_map, alerts = apply_preprocessing( bundle=bundle, payload=payload, local_context=merged_context, overrides_ci=overrides_ci, context_sources=context_sources, ) alerts.extend(polygon_alerts) alerts.extend(zone_alerts) alerts.extend(rh_alerts) alerts.extend(testada_alerts) x_input = build_features(bundle, prepared) prediction = predict_value(bundle, x_input, area_m2=float(prepared["AREA_bruta"])) markdown, details = format_output( bundle=bundle, prediction=prediction, prepared_values=prepared, source_map=source_map, alerts=alerts, local_context=merged_context, routing_source=routing_source, ) train_predictions, test_predictions = split_prediction_frames(bundle.holdout_predictions_df) metrics = build_metric_summary(bundle.holdout_predictions_df, bundle.metadata) report_path = build_pdf_report( model_label=bundle.label.upper(), model_file=bundle.model_path.name, user_inputs={ "Area territorial (m2)": area, "NUMBLOCO": payload.get("numbloco"), "Latitude": lat, "Longitude": lon, "Ano_Dado": payload["ano_dado"], "FON": "0", "LOTPOS": payload["lotpos"], "Testada manual": testada, }, prepared_values={feature: prepared.get(feature) for feature in bundle.feature_names}, source_map={feature: source_map.get(feature, "calculado/derivado") for feature in bundle.feature_names}, prediction=prediction, metrics=metrics, train_predictions=train_predictions, test_predictions=test_predictions, limitations=_build_report_limitations(bundle, alerts, metrics.get("train")), ) details["relatorio_pdf"] = report_path details["metricas_relatorio"] = _jsonable(metrics) return markdown, details, report_path except Exception as exc: LOGGER.exception("Falha na predicao") return ( f"Erro na inferencia: {exc}", { "status": "erro", "mensagem": str(exc), "recomendacao": ( "Verifique area/lat/lon (testada e opcional) e, para casos de Gleba com falta " "de dados cadastrais, informe overrides JSON." ), }, None, ) def preencher_por_coordenadas( numbloco: str, lat: float, lon: float, ) -> tuple[Any, Any, Any, Any, Any, dict[str, Any]]: if APP_STATE.error: return gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), { "status": "erro", "mensagem": APP_STATE.error, } if APP_STATE.polygon_context.to_base_crs is None: return gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), { "status": "indisponivel", "mensagem": "Camada de poligonos enriquecidos indisponivel.", } lon_number = _to_float(lon) lat_number = _to_float(lat) if lon_number is None or lat_number is None: return gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), { "status": "erro", "mensagem": "Latitude/longitude invalidas.", } polygon_x, polygon_y = APP_STATE.polygon_context.to_base_crs.transform(lon_number, lat_number) values, sources, alerts = _infer_polygon_context_by_numbloco( APP_STATE.polygon_context, numbloco, polygon_x, polygon_y, ) if not values: spatial_values, spatial_sources, spatial_alerts = _infer_polygon_context(APP_STATE.polygon_context, polygon_x, polygon_y) values = spatial_values sources = spatial_sources alerts.extend(spatial_alerts) rh_context, rh_sources, rh_alerts = _infer_rh_context_by_numbloco(APP_STATE.rh_numbloco, numbloco) testada_context, testada_sources, testada_alerts = _infer_testada_context_by_numbloco(APP_STATE.testada_numbloco, numbloco) values.update(rh_context) values.update(testada_context) sources.update(rh_sources) sources.update(testada_sources) alerts.extend(rh_alerts) alerts.extend(testada_alerts) area_value = _to_float(values.get("AREA")) testada_value = _to_float(values.get("TESTADA")) rh_value = _to_float(values.get("RH")) iapond_value = _to_float(values.get("IAPOND")) app_value = _to_float(values.get("APP")) lotpos_value = _to_float(values.get("LOTPOS")) payload = { "status": "ok" if values else "sem_resultado", "area_referencia_poligono": area_value, "testada": testada_value, "RH": rh_value, "IAPOND": iapond_value, "APP": app_value, "LOTPOS": lotpos_value, "NUMBLOCO_informado": None if _is_missing(numbloco) else _safe_text(numbloco, ""), "NUMBLOCO_poligono": values.get("_polygon_numbloco"), "distancia_poligono_m": _to_float(values.get("_polygon_distance_m")), "modo_busca_poligono": values.get("_polygon_lookup_mode"), "fontes": sources, "alerts": alerts, } return ( gr.update(value=testada_value) if testada_value is not None else gr.update(), gr.update(value=rh_value) if rh_value is not None else gr.update(), gr.update(value=iapond_value) if iapond_value is not None else gr.update(), gr.update(value=app_value) if app_value is not None else gr.update(), gr.update(value=lotpos_value) if lotpos_value is not None else gr.update(), _jsonable(payload), ) def _loaded_models_text(state: AppState) -> str: if state.error: return f"**Falha ao iniciar app:** `{state.error}`" lines = ["**Modelos carregados:**"] for key in sorted(state.bundles.keys()): bundle = state.bundles[key] lines.append( f"- `{bundle.label}` -> `{bundle.notebook_name}` / `{bundle.model_path.name}` " f"em `{bundle.artifact_dir.name}` " f"({len(bundle.feature_names)} features)" ) if state.reference.path: lines.append(f"**Base de referencia:** `{state.reference.path.name}`") else: lines.append("**Base de referencia:** nao encontrada (fallback por defaults)") if state.rh_numbloco.has_data and state.rh_numbloco.path is not None: lines.append( f"**RH por NUMBLOCO:** `{state.rh_numbloco.path.name}` " f"({len(state.rh_numbloco.rh_by_numbloco)} chaves; " f"duplicadas: {state.rh_numbloco.duplicate_key_count}; " f"conflitos: {state.rh_numbloco.conflicting_key_count})" ) elif state.rh_numbloco.error: lines.append(f"**RH por NUMBLOCO:** indisponivel ({state.rh_numbloco.error})") if state.testada_numbloco.has_data and state.testada_numbloco.path is not None: lines.append( f"**TESTADA por NUMBLOCO:** `{state.testada_numbloco.path.name}` " f"({len(state.testada_numbloco.testada_by_numbloco)} chaves; " f"duplicadas: {state.testada_numbloco.duplicate_key_count}; " f"conflitos: {state.testada_numbloco.conflicting_key_count})" ) elif state.testada_numbloco.error: lines.append(f"**TESTADA por NUMBLOCO:** indisponivel ({state.testada_numbloco.error})") if state.polygon_context.has_data and state.polygon_context.path is not None: lines.append( f"**Poligonos enriquecidos:** `{state.polygon_context.path.name}` " f"(campos: {', '.join(state.polygon_context.available_fields)}; " f"NUMBLOCO: {len(state.polygon_context.numbloco_key_index)} chaves)" ) elif state.polygon_context.error: lines.append(f"**Poligonos enriquecidos:** indisponivel ({state.polygon_context.error})") else: lines.append("**Poligonos enriquecidos:** nao configurado") if state.zoneamento.has_data and state.zoneamento.shp_path is not None: lines.append( f"**Zoneamento:** `{state.zoneamento.shp_path.name}` " f"(mapa IA: {len(state.zoneamento.iapond_map)} chaves, fonte: `{state.zoneamento.iapond_map_source}`)" ) elif state.zoneamento.error: lines.append(f"**Zoneamento:** indisponivel ({state.zoneamento.error})") else: lines.append("**Zoneamento:** nao configurado (fallback por contexto local)") return "\n".join(lines) APP_TITLE = "AVM Territorial 2026 - Inferencia XGBoost (Terreno/Gleba)" with gr.Blocks(title=APP_TITLE) as demo: gr.Markdown(f"# {APP_TITLE}") gr.Markdown( "Informe area, NUMBLOCO quando disponivel, e coordenadas do imovel. O app seleciona automaticamente " "`TERRENO` para area menor que `3000 m2` e `GLEBA` para area maior ou igual a `3000 m2`." ) gr.Markdown(_loaded_models_text(APP_STATE)) with gr.Row(): area_input = gr.Number(label="Area territorial (m2)", value=405.0, precision=6) numbloco_input = gr.Textbox( label="NUMBLOCO", value="", placeholder="Opcional; usado antes da busca espacial", ) with gr.Row(): lat_input = gr.Number(label="Latitude", value=-30.0315, precision=8) lon_input = gr.Number(label="Longitude", value=-51.1645, precision=8) autofill_button = gr.Button("Buscar atributos por NUMBLOCO/coordenada") autofill_json = gr.JSON(label="Diagnostico do autopreenchimento espacial") with gr.Accordion("Opcional - ajustes e overrides de variaveis inferidas", open=False): testada_input = gr.Textbox( label="Testada (m)", value="", placeholder="Opcional", ) bairro_input = gr.Textbox(label="Bairro (opcional)", placeholder="Ex.: PETROPOLIS") ano_input = gr.Dropdown( label="Ano_Dado", choices=[str(year) for year in range(2016, 2031)], value="2026", ) with gr.Row(): rh_input = gr.Textbox(label="RH", value="", placeholder="Preenchido por NUMBLOCO", interactive=False) iapond_input = gr.Textbox(label="IAPOND", value="", placeholder="Opcional") app_input = gr.Textbox(label="APP", value="", placeholder="Opcional") lotpos_input = gr.Number(label="LOTPOS", value=0.0, precision=6) overrides_input = gr.Code( label="Overrides JSON (opcional, prioridade maxima)", language="json", value='{\n "RH": null,\n "APP": null,\n "CP": null,\n "LOTPOS": null,\n "FINALIDADE": null\n}', lines=9, ) predict_button = gr.Button("Calcular estimativa", variant="primary") result_markdown = gr.Markdown() details_json = gr.JSON(label="Resumo tecnico da inferencia") report_file = gr.File(label="Relatorio PDF") predict_button.click( fn=predict, inputs=[ area_input, numbloco_input, testada_input, lat_input, lon_input, bairro_input, ano_input, iapond_input, app_input, lotpos_input, overrides_input, ], outputs=[result_markdown, details_json, report_file], ) numbloco_input.change( fn=atualizar_lat_lon_por_numbloco, inputs=[numbloco_input, lat_input, lon_input], outputs=[lat_input, lon_input], ) autofill_button.click( fn=preencher_por_coordenadas, inputs=[numbloco_input, lat_input, lon_input], outputs=[ testada_input, rh_input, iapond_input, app_input, lotpos_input, autofill_json, ], ) if __name__ == "__main__": port = int(os.getenv("PORT", "7860")) demo.launch(server_name="0.0.0.0", server_port=port)