geospatial-scripts/process_sismigra_historico.py at master · yan-fab/geospatial-scripts · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
import pandas as pd
import geopandas as gpd
import requests
import unicodedata
import re
import os
import sys

# Corrige warning do PROJ e GDAL no QGIS Python
os.environ['PROJ_DATA'] = r'C:\Program Files\QGIS 3.44.7\share\proj'
os.environ['GDAL_DATA'] = r'C:\Program Files\QGIS 3.44.7\share\gdal'

def normalize_name(name):
    if pd.isna(name):
        return ""
    name = str(name).upper()
    name = ''.join(c for c in unicodedata.normalize('NFD', name) if unicodedata.category(c) != 'Mn')
    name = re.sub(r'[^A-Z0-9]', '', name)
    return name

print("Lendo CSV Histórico de Bolivianos...")
try:
    df = pd.read_csv(r"C:\Users\yanju\Downloads\SISMIGRA_Historico\SISMIGRA_Historico_Bolivianos.csv", sep=";", encoding="latin-1")
except Exception as e:
    print(f"Erro ao ler CSV: {e}")
    sys.exit(1)

print("Padronizando e contando totais...")
df['nome_norm'] = df['MUNICIPIO_RESIDENCIA'].apply(normalize_name)
df_grouped = df.groupby('nome_norm').size().reset_index(name='TOTAL_BOLIVIANOS')

print("Baixando dicionário de municipios de SP do IBGE...")
url_loc = "https://servicodados.ibge.gov.br/api/v1/localidades/estados/SP/municipios"
res_loc = requests.get(url_loc)
res_loc.raise_for_status()
locs = res_loc.json()

loc_data = []
for loc in locs:
    codarea = str(loc['id'])[:7] # The malha v2 uses 7 digits
    nome = loc['nome']
    loc_data.append({'codarea': codarea, 'NOME_OFICIAL': nome, 'nome_norm': normalize_name(nome)})

df_ibge = pd.DataFrame(loc_data)

print("Cruzando totais com o dicionario do IBGE (Left Join)...")
df_ibge_com_totais = df_ibge.merge(df_grouped, on='nome_norm', how='left')

# Preenche -1 para os municipios oficiais que não tiveram match no CSV
df_ibge_com_totais['TOTAL_BOLIVIANOS'] = df_ibge_com_totais['TOTAL_BOLIVIANOS'].fillna(-1).astype(int)

print("Baixando malha espacial de SP V2 (IBGE)...")
try:
    url_malha = "https://servicodados.ibge.gov.br/api/v2/malhas/35?resolucao=5&formato=application/vnd.geo+json"
    res_malha = requests.get(url_malha)
    res_malha.raise_for_status()
    malha_path = r"C:\Users\yanju\.gemini\antigravity\scratch\geo_workspace\malha_sp_municipios.geojson"
    with open(malha_path, "w", encoding="utf-8") as f:
        f.write(res_malha.text)

    gdf_malha = gpd.read_file(malha_path)
    gdf_malha['codarea'] = gdf_malha['codarea'].astype(str)
except Exception as e:
    print(f"Erro ao baixar a malha: {e}")
    sys.exit(1)

print("Cruzando todos os dados com a malha espacial...")
gdf_final = gdf_malha.merge(df_ibge_com_totais, on='codarea', how='inner') # inner join pra manter apenas o que tem geometria em SP

# Remove colunas tecnicas redundantes
gdf_final = gdf_final.drop(columns=['codarea', 'nome_norm'])

output_path = r"C:\Users\yanju\.gemini\antigravity\scratch\geo_workspace\camada_sismigra_bolivianos_totais.gpkg"
print(f"Salvando GeoPackage (Cidades vazias = -1)...")
try:
    gdf_final.to_file(output_path, driver="GPKG", layer="totais_bolivianos")
    print(f"Processamento concluído e salvo em: {output_path}")
except Exception as e:
    print(f"Erro ao salvar: {e}")
    sys.exit(1)