Upload 3 files

app.py

@@ -1589,22 +1589,27 @@ def _select_best_polygon_row(
     return rank.iloc[0], float("nan"), alerts
 
 
-def _infer_polygon_context_by_numbloco(
+def _select_polygon_row_by_numbloco(
     polygons: PolygonContext,
     numbloco: Any,
     x: float | None = None,
     y: float | None = None,
-) -> tuple[dict[str, Any], dict[str, str], list[str]]:
+) -> tuple[pd.Series | None, float, list[str], int]:
     if not polygons.has_data or not polygons.numbloco_key_index:
-        return {}, {}, []
+        return None, float("nan"), [], 0
 
     key = _normalize_numbloco_key(numbloco)
     if not key:
-        return {}, {}, []
+        return None, float("nan"), [], 0
 
     candidate_idx = polygons.numbloco_key_index.get(key)
     if not candidate_idx:
-        return {}, {}, [f"NUMBLOCO `{_safe_text(numbloco, '')}` nao encontrado no geoparquet enriquecido; usada busca espacial."]
+        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)
@@ -1620,11 +1625,29 @@ def _infer_polygon_context_by_numbloco(
     matches = polygons.gdf.loc[candidate_idx]
     row, polygon_distance, alerts = _select_best_polygon_row(polygons, matches, point)
     if row is None:
-        return {}, {}, [f"NUMBLOCO `{_safe_text(numbloco, '')}` sem linha utilizavel no geoparquet; usada busca espacial."]
+        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
 
@@ -2492,6 +2515,75 @@ def initialize_app_state(base_dir: Path = BASE_DIR) -> AppState:
 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,
@@ -2845,6 +2937,11 @@ with gr.Blocks(title=APP_TITLE) as demo:
         ],
         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],