Muller's Notes

Neste Post criaremos scripts em python para coletar e processar dados geoespaciais de cobertura das Operadoras disponibilizados publicamente pela Anatel.

Baixaremos todos os arquivos das operadoras e consolidaremos em um único arquivo geoespacial, acrescentaremos algumas colunas para enriquecer os dados iniciais.

A Anatel disponibiliza os dados de cobertura das operadoras neste link aqui.

Carregando bibliotecas

Utilizaremos as bibliotecas Request e BeautifulSoup para coletar os dados que baixaremos via html pois precisaremos buscar por tags e classes no HTML

Utilizaremos a biblioteca geopandas para analisar e salvar os dados processados.

from datetime import datetimeimport reimport requestsfrom bs4 import BeautifulSoupfrom urllib.request import Request, urlopenimport pandas as pdimport geopandas as gpd

Baixando página da Anatel

BASE_URL="http://sistemas.anatel.gov.br"req = Request(BASE_URL+ "/se/public/cmap.php")html_page = urlopen(req)soup = BeautifulSoup(html_page, "lxml")

Descobrindo links

Iremos armazenar os dados dos links em um Dataframe identificando operadora, tecnologia e a última data de atualização conforme informado pela Anatel.

columns=['texto','operadora','tecnologia','last_update','url','arquivo']df = pd.DataFrame(columns=columns)#Loop que encontra todas as Tags <li> da página da Anatelfor link in soup.find_all("li"):	#Coleta do nome do arquivo detectando a div entidade        texto = link.find("div",{"class":"tec_entidade"})        #Coleta da data de atualização pela tag small    last_update = link.find("small")        #Coleta do link do arquivo para download via tag <div> e <a>    url = link.find("div",{"class":"download_shp"}).find("a")        #limpando e dividindo texto    if texto:        texto = texto.getText()        tecnologia = texto.split("_")[0].upper()        operadora = texto.split("_")[1].upper()            #limpando o campo data    if last_update:        last_update = last_update.getText().split(":")[1].split(")")[0].strip()        last_update = datetime.strptime(last_update, "%d/%m/%Y")        #tratando URL    if url:        url = link.find("div",{"class":"download_shp"}).find("a").get("href")    row = [texto,operadora,tecnologia,last_update,url,None]        df.loc[len(df)] = row

Download dos Links

#filtrar apenas linhas cuja url não é vaziafor index,row in df[df['url'].notna()].iterrows():        #construindo nome do arquivo para salvar em disco    arquivo = row['operadora']+"_"+row['tecnologia']+".zip"    df.loc[index,'arquivo'] = arquivo        #construindo url de download    remote_url = BASE_URL + row['url']        #download do arquivo    print("Coletando: "+remote_url)    data = requests.get(remote_url)    with open("./inputs/"+arquivo, 'wb') as file:        file.write(data.content)

Abrindo arquivos e consolidando dados num arquivo único

listdf = []for index,row in df[df['url'].notna()].iterrows():    gdf = geopandas.read_file('zip://inputs/'+row['arquivo'])    gdf= gdf.explode()    gdf['operadora'] = row['operadora']    gdf['tecnologia'] = row['tecnologia']    gdf['last_update'] = row['last_update']    gdf['tecnologia'] = row['tecnologia']    listdf.append(gdf.copy())    rdf = gpd.GeoDataFrame(pd.concat( listdf, ignore_index=True))rdf.to_file("cobertura.geojson", driver='GeoJSON')