Recursos Hídricos — ANA (Agência Nacional de Águas)

Base URL: https://dadosabertos.ana.gov.br/api/3/action/

import requests
import pandas as pd
 
# Consultar estações fluviométricas via HidroWeb
url = "https://www.snirh.gov.br/hidroweb/rest/api/estacao/"
 
params = {
    "uf": "AM",  # Amazonas
    "tipoEstacao": 1,  # Fluviométrica
}
 
response = requests.get(url, params=params, timeout=60)
response.raise_for_status()
estacoes = response.json()
 
df_estacoes = pd.DataFrame(estacoes)
print(f"Estações fluviométricas no AM: {len(df_estacoes):,}")
print(df_estacoes[["codigo", "nome", "bacia", "rio", "municipio"]].head(10))

import requests
import pandas as pd
from io import StringIO
 
# Baixar série histórica de vazão de uma estação
# Exemplo: estação 15030000 (Rio Amazonas em Óbidos)
COD_ESTACAO = "15030000"
 
url = f"https://www.snirh.gov.br/hidroweb/rest/api/documento/convencionais"
params = {
    "tipo": 3,  # Tipo 3 = vazão
    "codEstacao": COD_ESTACAO,
    "dataInicio": "2020-01-01",
    "dataFim": "2024-12-31",
}
 
response = requests.get(url, params=params, timeout=120)
response.raise_for_status()
 
# O retorno pode ser um arquivo CSV ou JSON dependendo do endpoint
# Processar conforme o formato retornado
print(f"Dados de vazão para estação {COD_ESTACAO}")
print(f"Tamanho da resposta: {len(response.content):,} bytes")

import requests
import pandas as pd
 
# API de dados abertos da ANA — consultar outorgas
BASE_URL = "https://dadosabertos.ana.gov.br/api/3/action"
 
# Primeiro, identificar o ID do recurso de outorgas
resp = requests.get(
    f"{BASE_URL}/package_show",
    params={"id": "outorgas"},
    timeout=30,
)
 
if resp.status_code == 200:
    dados = resp.json()
    recursos = dados.get("result", {}).get("resources", [])
    print("Recursos de outorgas disponíveis:")
    for r in recursos:
        print(f"  - {r['name']}: {r['format']} ({r['id']})")
 
 
# Consultar dados de outorgas via datastore
def consultar_outorgas(uf: str = None, finalidade: str = None, limite: int = 100):
    """
    Consulta outorgas de uso de recursos hídricos.
 
    Args:
        uf: Sigla da UF
        finalidade: Finalidade de uso (irrigação, abastecimento, etc.)
        limite: Número máximo de registros
 
    Returns:
        DataFrame com outorgas
    """
    url = f"{BASE_URL}/datastore_search"
    params = {
        "resource_id": "RESOURCE_ID_OUTORGAS",  # Substituir pelo ID real
        "limit": limite,
    }
 
    filtros = {}
    if uf:
        filtros["uf"] = uf
    if finalidade:
        filtros["finalidade"] = finalidade
 
    if filtros:
        import json
        params["filters"] = json.dumps(filtros)
 
    response = requests.get(url, params=params, timeout=30)
    response.raise_for_status()
    resultado = response.json()
 
    registros = resultado.get("result", {}).get("records", [])
    return pd.DataFrame(registros)
 
 
# Exemplo: outorgas no Mato Grosso
df_outorgas = consultar_outorgas(uf="MT")
print(f"Outorgas encontradas: {len(df_outorgas)}")

import requests
import pandas as pd
 
# Sistema de Acompanhamento de Reservatórios
SAR_URL = "https://www.snirh.gov.br/sar/"
 
# Consultar nível dos principais reservatórios
# Os dados são publicados diariamente
 
 
def consultar_reservatorios():
    """
    Consulta o nível atual dos reservatórios monitorados pela ANA.
 
    Returns:
        DataFrame com dados dos reservatórios
    """
    url = f"{SAR_URL}rest/api/reservatorio"
    response = requests.get(url, timeout=30)
    response.raise_for_status()
    return pd.DataFrame(response.json())
 
 
df_reservatorios = consultar_reservatorios()
print(f"Reservatórios monitorados: {len(df_reservatorios):,}")
print(df_reservatorios[["nome", "bacia", "volume_util_pct"]].head(10))
 
# Reservatórios em estado crítico (volume < 30%)
criticos = df_reservatorios[
    pd.to_numeric(df_reservatorios["volume_util_pct"], errors="coerce") < 30
]
print(f"\nReservatórios em estado crítico (<30%): {len(criticos)}")
print(criticos[["nome", "bacia", "volume_util_pct"]].to_string(index=False))

import geopandas as gpd
 
# Carregar divisão hidrográfica do Brasil
gdf_bacias = gpd.read_file("bacias_hidrograficas_brasil.shp")
 
print(f"Total de bacias: {len(gdf_bacias):,}")
print(f"Colunas: {list(gdf_bacias.columns)}")
 
# Área por bacia de nível 1
por_bacia = (
    gdf_bacias.groupby("nome_bacia")
    .agg(area_km2=("area_km2", "sum"))
    .sort_values("area_km2", ascending=False)
)
 
print("\nBacias hidrográficas por área:")
print(por_bacia.to_string())

import geopandas as gpd
import pandas as pd
 
# Verificar se áreas com outorgas de irrigação apresentam desmatamento recente
 
# 1. Carregar outorgas com coordenadas
df_outorgas = pd.read_csv("outorgas_ana.csv", sep=";", encoding="utf-8")
df_outorgas = df_outorgas.dropna(subset=["latitude", "longitude"])
 
gdf_outorgas = gpd.GeoDataFrame(
    df_outorgas,
    geometry=gpd.points_from_xy(df_outorgas["longitude"], df_outorgas["latitude"]),
    crs="EPSG:4674",
)
 
# 2. Carregar polígonos PRODES
gdf_prodes = gpd.read_file("prodes_amazonia_2023.shp").to_crs("EPSG:4674")
 
# 3. Buffer de 5km ao redor dos pontos de outorga
gdf_outorgas_buffer = gdf_outorgas.copy()
gdf_outorgas_buffer["geometry"] = gdf_outorgas_buffer.geometry.buffer(0.05)  # ~5km
 
# 4. Cruzamento espacial
outorgas_com_desmat = gpd.sjoin(
    gdf_outorgas_buffer, gdf_prodes, how="inner", predicate="intersects"
)
 
print(f"Outorgas próximas a áreas desmatadas: {len(outorgas_com_desmat):,}")
print(f"Outorgados afetados: {outorgas_com_desmat['cpf_cnpj'].nunique():,}")