USER
está correto o código ?
# -*- coding: utf-8 -*-
import overpy
import json
import os
import requests
import json
import shapely.geometry
import math
from shapely.geometry import Point, shape
from shapely.geometry import shape, Point
from shapely.ops import unary_union
from math import sqrt
import time
# --- Funções Utilitárias ---
def baixar_dados_overpass(bbox):
query = f"""
[out:json][timeout:25];
(
way["highway"]({bbox});
);
out body;
>;
out skel qt;
"""
url = "http://overpass-api.de/api/interpreter"
response = requests.post(url, data=query)
data = response.json()
return data
def processar_dados(dados):
nodes = {}
ways = []
for elemento in dados["elements"]:
if elemento["type"] == "node":
nodes[elemento["id"]] = elemento
elif elemento["type"] == "way":
ways.append(elemento)
return nodes, ways
def encontrar_intersecoes(nodes, ways):
node_count = {}
ways_dict = {}
for way in ways:
for node_id in way["nodes"]:
node_count[node_id] = node_count.get(node_id, 0) + 1
if node_id not in ways_dict:
ways_dict[node_id] = []
ways_dict[node_id].append(way)
intersecoes = {node_id: ways_dict[node_id] for node_id, count in node_count.items() if count > 1}
return intersecoes
def angulo_entre_vetores(p1, p2, p3):
"""Calcula o ângulo entre 3 pontos: p1->p2->p3"""
a = (p1[0] - p2[0], p1[1] - p2[1])
b = (p3[0] - p2[0], p3[1] - p2[1])
dot = a[0]*b[0] + a[1]*b[1]
norm_a = math.hypot(*a)
norm_b = math.hypot(*b)
cos_theta = dot / (norm_a * norm_b + 1e-9)
angle = math.degrees(math.acos(max(min(cos_theta, 1), -1)))
return angle
def refinar_filtrar_intersecoes_v5(intersecoes, nodes, geo):
intersecoes_validas = []
for node_id, vias_conectadas in intersecoes.items():
node = nodes[node_id]
ponto = Point(node["lon"], node["lat"])
if not geo.contains(ponto):
continue
nomes_vias = set()
vias_info = []
for via in vias_conectadas:
tags = via.get("tags", {})
tipo = tags.get("highway", "")
nome = tags.get("name")
junta = tags.get("junction", "")
if tipo in ["footway", "path", "track", "service", "pedestrian"]:
continue
if junta == "roundabout":
nome = f"rotatoria_{via['id']}"
if nome:
nomes_vias.add(nome)
vias_info.append((via["id"], tipo, nome))
if len(vias_info) < 2:
continue
if len(nomes_vias) <= 1:
pontos_ao_redor = []
for via in vias_conectadas:
if node_id in via["nodes"]:
idx = via["nodes"].index(node_id)
nodes_via = via["nodes"]
if idx > 0 and idx < len(nodes_via) - 1:
p1 = nodes[nodes_via[idx - 1]]
p2 = nodes[node_id]
p3 = nodes[nodes_via[idx + 1]]
pontos_ao_redor.append((p1, p2, p3))
angulos = []
for p1, p2, p3 in pontos_ao_redor:
ponto1 = (p1["lon"], p1["lat"])
ponto2 = (p2["lon"], p2["lat"])
ponto3 = (p3["lon"], p3["lat"])
angulo = angulo_entre_vetores(ponto1, ponto2, ponto3)
angulos.append(angulo)
if all(angulo < 30 for angulo in angulos):
continue
intersecoes_validas.append((node_id, ponto))
return intersecoes_validas
def main(bbox, geojson_area):
print("🧪 Baixando dados da API Overpass...")
dados = baixar_dados_overpass(bbox)
print("✅ Dados recebidos. Processando...")
nodes, ways = processar_dados(dados)
print(f"🌐 Nós totais: {len(nodes)} | Vias totais: {len(ways)}")
intersecoes = encontrar_intersecoes(nodes, ways)
print(f"🔍 Interseções brutas encontradas: {len(intersecoes)}")
area = shape(geojson_area["features"][0]["geometry"])
intersecoes_refinadas = refinar_filtrar_intersecoes_v5(intersecoes, nodes, area)
print(f"✅ Interseções REAIS refinadas: {len(intersecoes_refinadas)}")
return intersecoes_refinadas
def arredondar_coord(lat, lon, casas=4):
try: return (round(float(lat), casas), round(float(lon), casas))
except: return None
def salvar_json(data, filepath):
try:
os.makedirs(os.path.dirname(filepath), exist_ok=True)
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
print(f"Arquivo JSON salvo: '{filepath}'")
return True
except Exception as e:
print(f"Erro salvando JSON '{filepath}': {e}")
return False
def listar_e_filtrar_geojson(geojson_path, distrito_nomes_desejados):
print(f"\nAbrindo GeoJSON: {os.path.abspath(geojson_path)}")
try:
with open(geojson_path, 'r', encoding='utf-8') as f:
geojson_data = json.load(f)
except FileNotFoundError:
print(f"*** ERRO: GeoJSON NÃO ENCONTRADO '{geojson_path}' ***")
return None, None, None
except Exception as e:
print(f"ERRO lendo/parseando GeoJSON: {e}")
return None, None, None
if not isinstance(geojson_data, dict) or geojson_data.get("type") != "FeatureCollection":
return None, None, None
nomes_prop_ds_nome = set(
f.get("properties", {}).get("ds_nome")
for f in geojson_data.get("features", [])
if f.get("properties", {}).get("ds_nome")
)
print("\n--- Nomes Distrito GeoJSON ('ds_nome') ---")
[print(f"- {n}") for n in sorted(list(nomes_prop_ds_nome)) if n] or print("Nenhum nome encontrado.")
print("----\n")
nomes_encontrados_upper = {n.upper() for n in nomes_prop_ds_nome}
distritos_desejados_set_upper = {name.upper() for name in distrito_nomes_desejados}
features_filtradas = []
geometrias_shapely = []
distritos_processados = set()
print(f"Filtrando por {len(distritos_desejados_set_upper)} distritos desejados...")
for feature in geojson_data.get("features", []):
nome = feature.get("properties", {}).get("ds_nome")
if nome and nome.upper() in distritos_desejados_set_upper:
distritos_processados.add(nome)
features_filtradas.append(feature)
try:
geom = shape(feature['geometry'])
geom = geom if geom.is_valid else geom.buffer(0)
if geom.is_valid and not geom.is_empty:
geometrias_shapely.append(geom)
except Exception as e:
print(f"Aviso proc geom '{nome}': {e}")
nomes_desejados_nao_encontrados = {
name for name in distrito_nomes_desejados if name.upper() not in nomes_encontrados_upper
}
if nomes_desejados_nao_encontrados:
print("\n!!!! ATENÇÃO: Distritos NÃO encontrados: !!!!\n - " + "\n - ".join(sorted(list(nomes_desejados_nao_encontrados))) + "\n-------------------------------------------\n")
if not features_filtradas or not geometrias_shapely:
print("ERRO CRÍTICO: Nenhum distrito processado.")
return None, None, nomes_prop_ds_nome
print(f"Processados {len(features_filtradas)} distritos com {len(geometrias_shapely)} geometrias.")
geojson_filtrado = {
"type": "FeatureCollection",
"name": "distritos-filtrados",
"crs": geojson_data.get("crs"),
"features": features_filtradas
}
return geojson_filtrado, geometrias_shapely, nomes_prop_ds_nome
def calcular_bbox_geral(geometrias_shapely):
if not geometrias_shapely:
return None
try:
valid_geoms = [g for g in geometrias_shapely if g.is_valid and not g.is_empty]
if not valid_geoms:
return None
uniao = unary_union(valid_geoms)
margin = 0.0001
if uniao.is_empty:
return None
min_lon, min_lat, max_lon, max_lat = uniao.bounds
return (min_lon - margin, min_lat - margin, max_lon + margin, max_lat + margin)
except Exception as e:
print(f"Erro BBOX: {e}.")
return None
def dividir_bbox(bbox, linhas=2, colunas=2):
min_lon, min_lat, max_lon, max_lat = bbox
h_step = (max_lat - min_lat) / linhas
w_step = (max_lon - min_lon) / colunas
sub_bboxes = []
for i in range(linhas):
s = min_lat + i * h_step
n = min_lat + (i + 1) * h_step
for j in range(colunas):
w = min_lon + j * w_step
e = min_lon + (j + 1) * w_step
sub_bboxes.append((w, s, e, n))
return sub_bboxes
def buscar_dados_overpass_sub_bbox(sub_bbox_coords, delay_segundos=5):
bbox_overpass = f"{sub_bbox_coords[1]},{sub_bbox_coords[0]},{sub_bbox_coords[3]},{sub_bbox_coords[2]}"
api = overpy.Overpass()
highway_filter = '["highway"~"^(motorway|trunk|primary|secondary|tertiary|unclassified|residential|living_street)$"]'
# ✅ QUERY MULTILINHA correta
query = f"""
[out:json][timeout:480];
(
way({bbox_overpass}){highway_filter};
);
out body;
>;
out skel qt;
"""
print(f"\n Consultando sub-BBOX: {bbox_overpass}")
try:
print(f" Aguardando {delay_segundos}s...")
time.sleep(delay_segundos)
result = api.query(query)
print(f" Consulta sub-BBOX OK. Nós: {len(result.nodes)}, Ways: {len(result.ways)}.")
for way in result.ways:
for node in way.nodes:
if not hasattr(node, 'way_ids'):
node.way_ids = []
node.way_ids.append(str(way.id))
return result
except overpy.exception.OverpassError as e:
print(f" ERRO Overpass Query: {e}")
[print(f" -> API: {m}") for m in getattr(e, 'msgs', [])]
return None
except Exception as e:
print(f" ERRO Inesperado Overpass: {e}")
return None
def refinar_filtrar_intersecoes_v5(nodes_candidatos, ways_dict, geometrias_validas):
if not nodes_candidatos or not ways_dict or not geometrias_validas:
return []
intersecoes_refinadas = []
print(f"\n[🔍] Refinando {len(nodes_candidatos)} nós candidatos (Lógica v5)...")
try:
area_unificada = unary_union([g for g in geometrias_validas if g.is_valid and not g.is_empty])
except Exception as e:
print(f"[⚠️] Erro unindo geometrias: {e}")
return []
stats = {'ok': 0, 'fora': 0, 'sem_nome': 0, 'rotatoria': 0, 'poucas_ruas': 0}
for node in nodes_candidatos:
try:
ponto = Point(float(node.lon), float(node.lat))
if not area_unificada.contains(ponto):
stats['fora'] += 1
continue
if not hasattr(node, 'way_ids') or not node.way_ids:
stats['poucas_ruas'] += 1
continue
nomes_ruas = set()
ignorar = False
for wid in node.way_ids:
way = ways_dict.get(int(wid))
if not way:
continue
if way.tags.get("junction") == "roundabout":
ignorar = True
break
nome = way.tags.get("name")
if nome:
nomes_ruas.add(nome.strip())
if ignorar:
stats['rotatoria'] += 1
continue
if len(nomes_ruas) < 2:
stats['sem_nome'] += 1
continue
intersecoes_refinadas.append(node)
stats['ok'] += 1
except Exception as e:
print(f"[❌] Erro processando nó {getattr(node, 'id', '?')}: {e}")
print("\n[✅] Estatísticas final do refino v5:")
for k, v in stats.items():
print(f" - {k.capitalize()}: {v}")
return intersecoes_refinadas
def formatar_e_deduplicar_final(nodes_refinados, casas_decimais=4, distancia_minima=0.0003, offset_val=0.00015):
if not nodes_refinados:
return []
finais = []
marcadas = set()
ultima_coord = None
alternador = True # True = deslocamento positivo, False = negativo
print(f"\nFormatando e deduplicando {len(nodes_refinados)} interseções com offset alternado...")
for node in nodes_refinados:
try:
lat, lon = float(node.lat), float(node.lon)
latlon_original = arredondar_coord(lat, lon, casas=casas_decimais)
if latlon_original is None or latlon_original in marcadas:
continue
if ultima_coord:
dist = sqrt((lat - ultima_coord[0])**2 + (lon - ultima_coord[1])**2)
if dist < distancia_minima:
continue # muito próximo, pula
# Alterna offset para evitar sobreposição visual
if alternador:
lat += offset_val
lon -= offset_val
else:
lat -= offset_val
lon += offset_val
alternador = not alternador
latlon_offset = arredondar_coord(lat, lon, casas=casas_decimais)
if latlon_offset in marcadas:
continue
finais.append({
"lat": lat,
"lng": lon,
"osm_node_id": int(node.id)
})
marcadas.add(latlon_offset)
ultima_coord = (lat, lon)
except Exception as e:
print(f"Erro format/dedup nó {node.id}: {e}")
print(f"Deduplicação com offset concluída. {len(finais)} interseções únicas.")
return finais
# --- Execução Principal ---
INPUT_GEOJSON_PATH = "saopaulo_distritos_poligonos.geojson"
DISTRITOS_ZONA_LESTE = [
"ARICANDUVA", "ARTUR ALVIM", "CANGAIBA", "CARRAO", "CIDADE LIDER", "CIDADE TIRADENTES",
"ERMELINO MATARAZZO", "GUAIANASES", "IGUATEMI", "ITAIM PAULISTA", "ITAQUERA",
"JARDIM HELENA", "JOSE BONIFACIO", "LAJEADO", "MOOCA", "PARQUE DO CARMO", "PENHA",
"PONTE RASA", "SAPOPEMBA", "SAO LUCAS", "SAO MATEUS", "SAO MIGUEL", "SAO RAFAEL",
"TATUAPE", "VILA CURUCA", "VILA FORMOSA", "VILA JACUI", "VILA MATILDE", "VILA PRUDENTE",
]
print(f"INFO: Usando {len(DISTRITOS_ZONA_LESTE)} distritos ZL.")
CASAS_DECIMAIS_DEDUP = 4
GRADE_LINHAS = 2
GRADE_COLUNAS = 2
DELAY_CONSULTAS = 5
OUTPUT_DIR = "outputs_refinado_v11"
OUTPUT_GEOJSON_FILTRADO_PATH = os.path.join(OUTPUT_DIR, "distritos_zona_leste_usados.geojson")
OUTPUT_INTERSECOES_PATH = os.path.join(OUTPUT_DIR, "intersecoes_zona_leste_final_v11.json")
print(f"Iniciando processo v11 (Query Fixa, Refino v4, Dedup={CASAS_DECIMAIS_DEDUP})...")
if not os.path.exists(OUTPUT_DIR):
try:
os.makedirs(OUTPUT_DIR)
print(f"Diretório '{OUTPUT_DIR}' criado.")
except OSError as e:
print(f"ERRO CRÍTICO: Criar diretório '{OUTPUT_DIR}': {e}")
exit()
geojson_zl_filtrado, geometrias_zl_shapely, _ = listar_e_filtrar_geojson(INPUT_GEOJSON_PATH, DISTRITOS_ZONA_LESTE)
if geometrias_zl_shapely:
if geojson_zl_filtrado:
salvar_json(geojson_zl_filtrado, OUTPUT_GEOJSON_FILTRADO_PATH)
bbox_geral_zl = calcular_bbox_geral(geometrias_zl_shapely)
if bbox_geral_zl:
print(f"\nBBOX geral: {bbox_geral_zl}")
sub_bboxes = dividir_bbox(bbox_geral_zl, linhas=GRADE_LINHAS, colunas=GRADE_COLUNAS)
print(f"Dividido em {len(sub_bboxes)} sub-BBOXes ({GRADE_LINHAS}x{GRADE_COLUNAS}).")
todos_nos_brutos = {}
todas_ways_brutas = {}
for i, sub_bbox in enumerate(sub_bboxes):
print(f"\nProcessando sub-BBOX {i+1}/{len(sub_bboxes)}...")
sub_result = buscar_dados_overpass_sub_bbox(sub_bbox, delay_segundos=DELAY_CONSULTAS)
if sub_result and sub_result.nodes:
for node in sub_result.nodes:
todos_nos_brutos[node.id] = node
if hasattr(sub_result, 'ways') and sub_result.ways:
for way in sub_result.ways:
todas_ways_brutas[way.id] = way
else:
print(f"*** AVISO: Sub-BBOX {i+1} falhou ou retornou vazio. ***")
if todos_nos_brutos:
print(f"\nConsultas concluídas. {len(todos_nos_brutos)} nós únicos, {len(todas_ways_brutas)} ways únicas.")
intersecoes_refinadas = refinar_filtrar_intersecoes_v5(
list(todos_nos_brutos.values()), todas_ways_brutas, geometrias_zl_shapely
)
intersecoes_finais = formatar_e_deduplicar_final(
intersecoes_refinadas, casas_decimais=CASAS_DECIMAIS_DEDUP
)
if intersecoes_finais:
if salvar_json(intersecoes_finais, OUTPUT_INTERSECOES_PATH):
print(f"\n===> TOTAL FINAL: {len(intersecoes_finais)} interseções salvas em '{OUTPUT_INTERSECOES_PATH}'.")
else:
print("\nERRO ao salvar JSON final.")
else:
print("\nNenhuma interseção final encontrada após refinamento/deduplicação.")
else:
print("\nNenhum nó encontrado nas consultas bem-sucedidas.")
else:
print("\nERRO CRÍTICO: Não calculou BBOX.")
else:
print("\n!!! INTERROMPIDO: Falha ao processar GeoJSON !!!")
print("\nProcesso concluído.")ASSISTANT
O código está bastante completo e bem estruturado para o propósito de baixar dados da API Overpass, processar e filtrar interseções de ruas dentro de uma área geográfica dada (zona leste de São Paulo, no seu exemplo).
De forma geral, sim, o código está correto e aparentemente funcional, com boas práticas como tratamento de exceção, uso do `shapely` para geometria e controle de taxa de query ao Overpass via delay. Porém, alguns pontos merecem destaque ou poderiam ser melhorados para maior clareza, eficiência e robustez:
---
### Pontos positivos:
- Uso de funções específicas e bem nomeadas que separam responsabilidades (download, processamento, filtragem, deduplicação).
- Uso do pacote `overpy` para query e parsing dos dados OSM, que é mais robusto do que tratar manualmente JSON.
- Verificação de inválidos e limpezas geométricas com `buffer(0)`.
- Controle de sobreposição geográfica (refinamento com ângulos entre vias).
- Log detalhado com prints informativos.
- Deduplicação com deslocamento alternado para evitar sobreposição visual.
- Verificação da existência do diretório e criação quando necessário.
- Tratamento para evitar ruas tipo "footway" e rotatórias (junction=roundabout).
---
### Pontos para revisão / melhorias potenciais:
1. **Importações repetidas**:
```python
import json
from shapely.geometry import Point, shape
from shapely.geometry import shape, Point
```
- `json` e as importações do mesmo módulo repetidas podem ser removidas para clareza.
- Importações duplicadas de `Point` e `shape` podem ser unificadas numa linha só.
2. **Uso inconsistente de variáveis de coordenadas**:
- Em algumas funções (exemplo: `refinar_filtrar_intersecoes_v5`), as coordenadas são passadas / usadas como `(lon, lat)`, em outras como `(lat, lon)`. Fique atento para manter um padrão (geralmente, OSM e shapely usam `(lon, lat)`).
3. **Comentários e documentação**:
- Algumas funções poderiam ter docstrings detalhando parâmetros e retornos.
- Comentários são úteis para indicar propósito de passos complexos (ex: o why do offset alternado).
4. **Tratamento de exceções genéricas**:
- Vários blocos `try-except Exception as e:` capturam erros genéricos e continuam. Embora importante para robustez, pode esconder erros sutis.
5. **Uso do `1e-9` para evitar divisão por zero**:
```python
cos_theta = dot / (norm_a * norm_b + 1e-9)
```
- OK para evitar ZeroDivisionError, mas pode ser melhorável (ex: ignorar se vetores muito curtos).
6. **Naming: nomes mistos entre português e inglês**:
- Ex: `baixar_dados_overpass`, `refinar_filtrar_intersecoes_v5` (português), mas depois `formatar_e_deduplicar_final` misturando português e inglês.
- Consistência de idioma torna manutenção mais fácil.
7. **Reuso de função `refinar_filtrar_intersecoes_v5` em 2 versões diferentes no código**:
- Aparece uma função `refinar_filtrar_intersecoes_v5(intersecoes, nodes, geo)` (que processa dict de interseções) e depois outra `refinar_filtrar_intersecoes_v5(nodes_candidatos, ways_dict, geometrias_validas)` (que processa objetos do `overpy`). Nomes iguais em escopos diferentes, pode causar confusão (funções definidas duas vezes acabam sobrescrevendo).
---
### Resumo
- O código deve funcionar como esperado, considerando que:
- As dependências (`overpy`, `shapely`, `requests`) estejam instaladas.
- O arquivo GeoJSON esteja corretamente formatado e presente.
- O Overpass API responda dentro do timeout configurado.
- Recomendo:
- Remover importações duplicadas.
- Renomear as funções `refinar_filtrar_intersecoes_v5` para nomes distintos para evitar sobrescrita/confusão.
- Padronizar o uso de coordenadas `(lon, lat)` onde possível.
- Adicionar mais logs ou debug prints caso precise debugar.
- Testar com pequenas áreas para validar fluxo.
Se desejar, posso ajudar a ajustar os pontos mencionados ou montar uma versão revisada e comentada!
---
Quer ajuda para fazer alguma dessas melhorias?