hidroweb-python

Biblioteca Python completa e profissional para consumo da API HidroWebService da ANA (Agencia Nacional de Aguas e Saneamento Basico).

Funcionalidades

API Completa

• Cliente Python sincrono e assincrono
• CLI poderosa para uso via terminal
• Cache inteligente (memoria, disco, Redis)
• Export de dados (CSV, Excel, JSON, Parquet)
• Retry automatico com backoff exponencial
• Paginacao automatica para grandes datasets
• Type hints completos
• Logging estruturado

Endpoints Disponiveis

• Autenticacao OAuth
• Series telemetricas adotadas
• Series telemetricas detalhadas
• Inventario de estacoes
• Filtros avancados por UF, bacia, tipo

Instalacao

Via uv (recomendado)

uv add hidroweb

Via pip

pip install hidroweb

Instalacao completa (com todas as dependencias opcionais)

pip install hidroweb[full]
# Inclui: cache redis, export parquet

Pre-requisitos

Para usar esta biblioteca, voce precisa solicitar credenciais de acesso a ANA:

1. Envie um e-mail para: hidro@ana.gov.br
2. Assunto: "[SEU CPF/CNPJ] - Solicitacao de acesso a API HidroWebService"
3. Inclua no corpo:
   • Nome completo e instituicao
   • CPF ou CNPJ (sera seu usuario)
   • Endereco de e-mail
   • Breve explicacao do motivo

Uso Basico

API Python

from hidroweb import ClienteHidroweb

# Inicializar cliente
cliente = ClienteHidroweb(
    identificador="SEU_CPF_OU_CNPJ",
    senha="SUA_SENHA"
)

# Listar estacoes ativas de Sao Paulo
estacoes = cliente.inventario.listar_estacoes(
    uf="SP",
    operando=True
)

# Obter serie telemetrica
medicoes = cliente.series.obter_serie_adotada(
    codigo_estacao="15400000",
    range_intervalo="DIAS_30"
)

# Exibir dados
for medicao in medicoes[:5]:
    print(f"{medicao.Data_Hora_Medicao}: Cota={medicao.Cota_Adotada}cm")

Com Cache

from hidroweb import ClienteHidroweb
from hidroweb.cache.disco import CacheDisco
from datetime import timedelta

# Cliente com cache em disco
cache = CacheDisco(ttl_padrao=timedelta(hours=24))
cliente = ClienteHidroweb(
    identificador="SEU_CPF",
    senha="SUA_SENHA",
    cache=cache
)

# Primeira busca: da API
medicoes = cliente.series.obter_serie_adotada("15400000")

# Segunda busca: do cache (instantaneo!)
medicoes = cliente.series.obter_serie_adotada("15400000")

Export de Dados

from hidroweb import ClienteHidroweb
from hidroweb.export import GerenciadorExport
from pathlib import Path

cliente = ClienteHidroweb(identificador="CPF", senha="SENHA")

# Buscar dados
medicoes = cliente.series.obter_serie_adotada("15400000")

# Exportar para CSV
GerenciadorExport.exportar(medicoes, Path("dados.csv"), formato="csv")

# Exportar para Excel
GerenciadorExport.exportar(medicoes, Path("dados.xlsx"), formato="xlsx", nome_aba="Medicoes")

# Exportar para JSON
GerenciadorExport.exportar(medicoes, Path("dados.json"), formato="json", indent=2)

# Exportar para Parquet (ideal para big data)
GerenciadorExport.exportar(medicoes, Path("dados.parquet"), formato="parquet")

API Assincrona

import asyncio
from hidroweb import ClienteHidrowebAsync

async def main():
    async with ClienteHidrowebAsync(
        identificador="CPF",
        senha="SENHA"
    ) as cliente:
        # Buscar multiplas estacoes em paralelo
        codigos = ["15400000", "15410000", "15420000"]
        resultados = await cliente.buscar_multiplas_estacoes(codigos)

        for codigo, medicoes in resultados.items():
            print(f"Estacao {codigo}: {len(medicoes)} medicoes")

asyncio.run(main())

CLI (Command Line Interface)

# Autenticar
hidroweb auth --identificador SEU_CPF --senha SUA_SENHA

# Listar estacoes
hidroweb estacoes listar --uf SP --operando --limite 10

# Ver informacoes de uma estacao
hidroweb estacoes info 15400000

# Obter serie telemetrica
hidroweb series obter 15400000 --intervalo DIAS_30 --formato tabela

# Exportar dados
hidroweb export series 15400000 dados.csv --intervalo DIAS_30 --formato csv
hidroweb export series 15400000 dados.xlsx --formato xlsx
hidroweb export series 15400000 dados.json --formato json

Funcionalidades Avancadas

Retry Automatico

from hidroweb.retry import com_retry, EstrategiaRetry

# Usar decorador
@com_retry(max_tentativas=5, backoff_base=2.0)
def buscar_dados_criticos():
    return cliente.series.obter_serie_adotada("15400000")

# Ou usar diretamente
estrategia = EstrategiaRetry(
    max_tentativas=5,
    backoff_base=1.0,
    backoff_max=60.0
)
resultado = estrategia.executar(buscar_dados_criticos)

Paginacao

from hidroweb.paginacao import Paginador

# Criar paginador
paginador = Paginador(
    funcao_busca=lambda limit, offset: cliente.inventario.listar_estacoes(),
    tamanho_pagina=100
)

# Iterar por paginas
for item in paginador.iterar_todas():
    processar(item)

# Ou obter todas de uma vez
todas = paginador.obter_todas()

Cache Redis (opcional)

from hidroweb import ClienteHidroweb
from hidroweb.cache.redis import CacheRedis
from datetime import timedelta

# Requer: pip install hidroweb[redis]
cache = CacheRedis(
    host='localhost',
    port=6379,
    ttl_padrao=timedelta(hours=1)
)

cliente = ClienteHidroweb(
    identificador="CPF",
    senha="SENHA",
    cache=cache
)

Documentacao Completa

• Guia de Inicio Rapido
• Autenticacao
• Sistema de Cache
• Export de Dados
• CLI Completo
• API Assincrona
• Referencia da API

Exemplos Praticos

#	Arquivo	Descricao
01	01_autenticacao.py	Autenticacao basica
02	02_listar_estacoes.py	Listar estacoes por UF
03	03_buscar_series.py	Buscar series telemetricas
04	04_uso_cache.py	Cache em memoria
05	05_export_csv.py	Exportar para CSV
06	06_async_basico.py	API assincrona basica
07	07_export_excel.py	Exportar para Excel
08	08_series_detalhadas.py	Series detalhadas por data
09	09_async_multiplas_estacoes.py	Async com multiplas estacoes
10	10_uso_avancado.py	Pipeline com cache, retry e export

Desenvolvimento

# Clonar repositorio
git clone https://github.com/seu-usuario/hidroweb-python
cd hidroweb-python

# Instalar com dependencias de desenvolvimento
uv pip install -e ".[dev]"

# Rodar testes
pytest

# Rodar testes com coverage
pytest --cov=src/hidroweb --cov-report=html

# Linting
ruff check src/ tests/
black --check src/ tests/

# Type checking
mypy src/

Contribuindo

Contribuicoes sao bem-vindas! Veja CONTRIBUTING.md

Status do Projeto

• Core API implementado
• Cache completo (memoria, disco, Redis)
• Export (CSV, Excel, JSON, Parquet)
• CLI funcional
• API assincrona
• Retry e paginacao
• Testes completos
• Documentacao completa

Licenca

MIT License - veja LICENSE

Links Uteis

• Swagger da API
• Portal SNIRH
• ANA - Monitoramento
• Documentacao Oficial

Exemplos de Uso

Analise de Dados

import pandas as pd
from hidroweb import ClienteHidroweb

cliente = ClienteHidroweb(identificador="CPF", senha="SENHA")

# Buscar dados de multiplas estacoes
estacoes_sp = cliente.inventario.listar_estacoes(uf="SP", operando=True)

# Coletar series de todas
dados_completos = []
for estacao in estacoes_sp[:10]:  # Primeiras 10
    medicoes = cliente.series.obter_serie_adotada(estacao.codigoestacao)
    dados_completos.extend([m.model_dump() for m in medicoes])

# Criar DataFrame para analise
df = pd.DataFrame(dados_completos)
print(df.describe())

Pipeline de Dados

from hidroweb import ClienteHidroweb
from hidroweb.cache.disco import CacheDisco
from hidroweb.export import GerenciadorExport
from pathlib import Path

# Setup
cache = CacheDisco()
cliente = ClienteHidroweb("CPF", "SENHA", cache=cache)

# Pipeline: buscar -> processar -> exportar
def pipeline_dados(uf: str, output_dir: Path):
    # 1. Buscar estacoes
    estacoes = cliente.inventario.listar_estacoes(uf=uf, operando=True)

    # 2. Buscar series (com cache automatico)
    for estacao in estacoes:
        medicoes = cliente.series.obter_serie_adotada(estacao.codigoestacao)

        # 3. Exportar cada estacao
        arquivo = output_dir / f"{estacao.codigoestacao}.parquet"
        GerenciadorExport.exportar(medicoes, arquivo, formato="parquet")

        print(f"  {estacao.Estacao_Nome}: {len(medicoes)} medicoes")

# Executar
pipeline_dados("MG", Path("dados_mg"))

Agradecimentos

• ANA (Agencia Nacional de Aguas e Saneamento Basico) pela disponibilizacao da API
• Comunidade Python Brasil

---

Desenvolvido para a comunidade hidrologica brasileira