reboucassr/dataset
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

V.1.0

Browse Source
This commit is contained in:
kolmeasi
2026-06-01 14:21:31 -03:00
parent 77ec04cc43
commit e5aa85fe9f
27 changed files with 1589 additions and 1908885 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
softwares/test/meteorologia_aeroportos/_apps/__pycache__/_diag_outliers.cpython-313.pyc Normal file
View File

Binary file not shown.

BIN
softwares/test/meteorologia_aeroportos/_apps/__pycache__/concat_meteorologia.cpython-313.pyc Normal file
View File

Binary file not shown.

BIN
softwares/test/meteorologia_aeroportos/_apps/__pycache__/db.cpython-313.pyc Normal file
View File

Binary file not shown.

141
softwares/test/meteorologia_aeroportos/_apps/_diag_outliers.py Normal file
View File

@@ -0,0 +1,141 @@
"""
Command-line diagnostic tool for the surface meteorology SQLite database.
Reports values outside the physical limits defined in :mod:`db`, showing
per-variable outlier counts and basic summary statistics.
Usage:
python _diag_outliers.py
python _diag_outliers.py --db path/to/met.db
python _diag_outliers.py --top 20
"""
import argparse
import sys
from pathlib import Path
import pandas as pd
# Resolved relative to this file so the script works from any directory.
_APPS_DIR = Path(__file__).resolve().parent # .../meteorologia_aeroportos/_apps/
_BASE_DIR = _APPS_DIR.parent # .../meteorologia_aeroportos/
_DEFAULT_DB = _BASE_DIR / "db" / "met.db"
def print_outliers(
db_path: Path,
top_n: int = 10,
) -> None:
"""Queries and prints outlier statistics for all analytic variables.
Args:
db_path: Path to the SQLite database file.
top_n: Number of extreme values to display per variable.
"""
import db as _db
if not db_path.exists():
print(f"[erro] Banco não encontrado: {db_path}", file=sys.stderr)
sys.exit(1)
conn = _db.get_connection(db_path)
_db.ensure_schema(conn)
total: int = conn.execute(
"SELECT COUNT(*) FROM observations"
).fetchone()[0]
print(f"=== Outliers por variável (fora dos limites físicos) ===\n")
print(f"Total de observações: {total:,}\n")
for col, (lo, hi) in _db.PHYSICAL_LIMITS.items():
rows = conn.execute(
f"""
SELECT aerodrome, dt, {col}
FROM observations
WHERE {col} IS NOT NULL
AND ({col} < ? OR {col} > ?)
ORDER BY ABS({col}) DESC
LIMIT ?
""",
(lo, hi, top_n),
).fetchall()
if not rows:
continue
cnt: int = conn.execute(
f"SELECT COUNT(*) FROM observations "
f"WHERE {col} < ? OR {col} > ?",
(lo, hi),
).fetchone()[0]
print(f"--- {col} (limite: {lo} a {hi}) ---")
for r in rows:
print(f" {r[0]} {r[1]} {col}={r[2]}")
print(f" Total fora do limite: {cnt:,} ({cnt / total * 100:.2f}%)\n")
conn.close()
def print_stats(db_path: Path) -> None:
"""Prints min/max/mean for every analytic variable.
Args:
db_path: Path to the SQLite database file.
"""
import db as _db
conn = _db.get_connection(db_path)
_db.ensure_schema(conn)
cols_expr = ", ".join(
f"MIN({c}) AS min_{c}, MAX({c}) AS max_{c}, AVG({c}) AS avg_{c}"
for c in _db.ANL_COLS
)
row = conn.execute(f"SELECT {cols_expr} FROM observations").fetchone()
keys = [
f"{fn}_{c}"
for c in _db.ANL_COLS
for fn in ("min", "max", "avg")
]
print("\n=== Estatísticas básicas ===\n")
for k, v in zip(keys, row):
if v is not None:
print(f" {k}: {v:.2f}")
conn.close()
def main() -> None:
"""Entry point: parse arguments and run diagnostics."""
parser = argparse.ArgumentParser(
description="Outlier diagnostics for the meteorology SQLite database",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
examples:
python _diag_outliers.py
python _diag_outliers.py --db /path/to/met.db --top 20
""",
)
parser.add_argument(
"--db",
default=str(_DEFAULT_DB),
metavar="PATH",
help=f"Path to the SQLite database (default: {_DEFAULT_DB})",
)
parser.add_argument(
"--top",
type=int,
default=10,
metavar="N",
help="Number of extreme values to display per variable (default: 10)",
)
args = parser.parse_args()
print_outliers(Path(args.db), top_n=args.top)
print_stats(Path(args.db))
if __name__ == "__main__":
main()

521
softwares/test/meteorologia_aeroportos/_apps/concat_meteorologia.py Normal file
View File

@@ -0,0 +1,521 @@
"""
CSV aggregator for ICEA/DECEA surface meteorology data.
Merges the per-tab, per-quarter CSV files produced by the scraper into a single
wide table per aerodrome. Includes spot-check validation and optional cleanup
of the quarterly source files.
Standalone usage:
python concat_meteorologia.py [--dados-dir <dir>] [--validate] [--n-samples 20] [--cleanup]
Importable by the pipeline:
from concat_meteorologia import build_aerodrome_table, validate_sample, cleanup_source_files
"""
import argparse
import csv
import io
import random
import re
import sys
from pathlib import Path
from typing import Callable, Optional
import pandas as pd
# ── Paths (resolved relative to this file, independent of CWD) ───────────────
_APPS_DIR = Path(__file__).resolve().parent # .../meteorologia_aeroportos/_apps/
_BASE_DIR = _APPS_DIR.parent # .../meteorologia_aeroportos/
_REPO_ROOT = _BASE_DIR.parents[2] # dataset/
DADOS_DIR = _BASE_DIR / "db" / "dados" # temporary quarterly CSVs
PREPROC_DIR = (
_REPO_ROOT / "tabelas" / "preproc" / "meteorologia_aeroportos"
) # permanent compiled-CSV output (CLI mode)
DATETIME_COL = "Data e HoraObservação"
DATETIME_FMT = "%d/%m/%Y - %H:%M:%S"
# Prefixo curto + coluna pivô (None = sem pivô, 1 linha por timestamp)
TAB_CONFIG: dict[str, tuple[str, Optional[str]]] = {
"CGT": ("cgt", None),
"Nuvem": ("nuv", None),
"Precipitação": ("prec", None),
"Pressão": ("pres", None),
"RVR": ("rvr", "Cabeceira"),
"Temperatura": ("temp", "Pista"),
"Teto": ("teto", "Pista"),
"Vento": ("vent", "Cabeceira"),
"Visibilidade": ("visib", None),
}
_FNAME_RE = re.compile(
r"Dados de Superfície (.+?) - Localidade (\w+) - .+ - Período (\d{8})\s+-\s+(\d{8})\.csv"
)
# ---------------------------------------------------------------------------
# Utilitários
# ---------------------------------------------------------------------------
def _clean(name: str) -> str:
"""Normalises a column name to a safe identifier fragment.
Removes parenthesised content, unit symbols, and replaces whitespace/
punctuation runs with a single underscore.
Args:
name: Raw column name from a source CSV.
Returns:
Cleaned string suitable for use as part of a column identifier.
"""
name = re.sub(r"\(.*?\)", "", name)
name = re.sub(r"[ºµ%°]", "", name)
name = re.sub(r"[\s/\\,;]+", "_", name.strip())
return name.strip("_")
def parse_filename(
fname: str,
) -> Optional[tuple[str, str, pd.Timestamp, pd.Timestamp]]:
"""Extracts tab name, aerodrome code, and date range from a CSV filename.
Expected pattern:
``Dados de Superfície <tab> - Localidade <ICAO> - Período <DDMMYYYY> - <DDMMYYYY>.csv``
Args:
fname: Bare filename (no directory component).
Returns:
``(tab_name, aerodrome, start_ts, end_ts)`` tuple, or ``None`` if the
filename does not match the expected pattern.
"""
m = _FNAME_RE.match(fname)
if not m:
return None
tab = m.group(1).strip()
aero = m.group(2).strip()
s = pd.to_datetime(m.group(3), format="%d%m%Y")
e = pd.to_datetime(m.group(4), format="%d%m%Y")
return tab, aero, s, e
# ---------------------------------------------------------------------------
# Carregamento e pivotagem
# ---------------------------------------------------------------------------
def load_tab(path: Path, tab_name: str) -> pd.DataFrame:
"""
Carrega CSV de uma aba, normaliza coluna de data/hora e aplica pivô se necessário.
Retorna DataFrame com colunas prefixadas (ex: cgt_CGT_1, temp_10R_Bulbo_Seco).
"""
prefix, pivot_col = TAB_CONFIG.get(tab_name, (_clean(tab_name[:5].lower()), None))
df = pd.read_csv(path, encoding="utf-8-sig")
dt_candidates = [c for c in df.columns if "Data" in c and "Hora" in c]
if dt_candidates and dt_candidates[0] != DATETIME_COL:
df = df.rename(columns={dt_candidates[0]: DATETIME_COL})
if DATETIME_COL not in df.columns:
return df
if pivot_col and pivot_col in df.columns:
value_cols = [c for c in df.columns if c not in (DATETIME_COL, pivot_col)]
pivoted = df.pivot_table(
index=DATETIME_COL, columns=pivot_col, values=value_cols, aggfunc="first"
)
pivoted.columns = [
f"{prefix}_{_clean(str(piv))}_{_clean(col)}"
for col, piv in pivoted.columns
]
return pivoted.reset_index()
# Múltiplas linhas por timestamp sem chave nomeada (ex: camadas de nuvem)
if df.duplicated(subset=[DATETIME_COL], keep=False).any():
df = df.copy()
df["_camada"] = df.groupby(DATETIME_COL).cumcount() + 1
value_cols = [c for c in df.columns if c not in (DATETIME_COL, "_camada")]
pivoted = df.pivot_table(
index=DATETIME_COL, columns="_camada", values=value_cols, aggfunc="first"
)
pivoted.columns = [
f"{prefix}_c{int(cam)}_{_clean(col)}"
for col, cam in pivoted.columns
]
return pivoted.reset_index()
rename = {c: f"{prefix}_{_clean(c)}" for c in df.columns if c != DATETIME_COL}
return df.rename(columns=rename)
# ---------------------------------------------------------------------------
# Construção da tabela por aeródromo
# ---------------------------------------------------------------------------
def build_aerodrome_table(
files: list[Path],
extra_df: Optional[pd.DataFrame] = None,
log: Callable[[str], None] = print,
) -> tuple[str, pd.DataFrame]:
"""
Agrupa CSVs por aba, concatena no tempo e mescla todas as abas.
Se extra_df for fornecido (arquivo compilado anterior), é incluído no merge final.
Retorna (aerodrome_code, DataFrame_mesclado).
"""
tab_frames: dict[str, list[pd.DataFrame]] = {}
aerodrome = ""
for f in sorted(files):
parsed = parse_filename(f.name)
if not parsed:
log(f" [skip] nome nao reconhecido: {f.name}")
continue
tab_name, aero, *_ = parsed
aerodrome = aero
df = load_tab(f, tab_name)
tab_frames.setdefault(tab_name, []).append(df)
if not tab_frames and extra_df is None:
return aerodrome, pd.DataFrame()
per_tab: list[pd.DataFrame] = []
for tab_name in TAB_CONFIG:
if tab_name not in tab_frames:
continue
combined = pd.concat(tab_frames[tab_name], ignore_index=True)
combined = combined.sort_values(DATETIME_COL).reset_index(drop=True)
per_tab.append(combined)
if not per_tab:
return aerodrome, extra_df if extra_df is not None else pd.DataFrame()
result = per_tab[0]
for df in per_tab[1:]:
result = pd.merge(result, df, on=DATETIME_COL, how="outer")
result = result.sort_values(DATETIME_COL).reset_index(drop=True)
# Inclui dados já compilados anteriormente
if extra_df is not None and not extra_df.empty:
result = pd.concat([extra_df, result], ignore_index=True)
result = result.sort_values(DATETIME_COL).drop_duplicates(
subset=[DATETIME_COL], keep="last"
).reset_index(drop=True)
return aerodrome, result
def date_range_from_data(df: pd.DataFrame) -> tuple[str, str]:
"""Returns the formatted ``(start, end)`` date strings from a merged DataFrame.
Args:
df: DataFrame that contains the :data:`DATETIME_COL` column.
Returns:
``("YYYY_MM_DD", "YYYY_MM_DD")`` strings for the earliest and latest
parsed timestamps.
"""
dates = pd.to_datetime(df[DATETIME_COL], format=DATETIME_FMT, errors="coerce")
return dates.min().strftime("%Y_%m_%d"), dates.max().strftime("%Y_%m_%d")
# ---------------------------------------------------------------------------
# Validação amostral
# ---------------------------------------------------------------------------
def validate_sample(
merged_df: pd.DataFrame,
source_files: list[Path],
n: int = 20,
log: Callable[[str], None] = print,
) -> tuple[int, int, list[str]]:
"""
Valida a compilação sorteando n linhas do DataFrame mesclado e comparando
com os arquivos trimestrais originais.
Retorna (n_ok, n_fail, lista_erros).
"""
if not source_files or merged_df.empty:
return 0, 0, []
# Índice: (tab_name, periodo_start, periodo_end) -> Path
file_index: dict[tuple[str, str, str], Path] = {}
for f in source_files:
parsed = parse_filename(f.name)
if parsed:
tab_name, _, s, e = parsed
key = (tab_name, s.strftime("%d%m%Y"), e.strftime("%d%m%Y"))
file_index[key] = f
if not file_index:
return 0, 0, ["Nenhum arquivo trimestral indexado para validação"]
# Tabs simples (sem pivô) são mais fáceis de validar diretamente
validatable_tabs = [
tab for tab, (pfx, piv) in TAB_CONFIG.items() if piv is None
]
# Filtra tabs que realmente existem nos arquivos disponíveis
available_tabs = {parse_filename(f.name)[0] for f in source_files if parse_filename(f.name)}
validatable_tabs = [t for t in validatable_tabs if t in available_tabs]
if not validatable_tabs:
return 0, 0, ["Nenhuma aba sem pivô disponível para validar"]
# Mapeia colunas mescladas → (tab, coluna_original) para tabs sem pivô
col_map: dict[str, tuple[str, str]] = {}
for tab_name in validatable_tabs:
prefix, _ = TAB_CONFIG[tab_name]
# Pega colunas do merged que começam com esse prefixo
merged_cols = [c for c in merged_df.columns if c.startswith(f"{prefix}_")]
for mc in merged_cols:
col_map[mc] = (tab_name, mc) # guardaremos o prefixo para busca
sample_size = min(n, len(merged_df))
sample_idx = random.sample(range(len(merged_df)), sample_size)
n_ok = 0
n_fail = 0
errors = []
for idx in sample_idx:
row = merged_df.iloc[idx]
timestamp = row[DATETIME_COL]
# Escolhe uma aba aleatória para validar nesta linha
tab_name = random.choice(validatable_tabs)
prefix, _ = TAB_CONFIG[tab_name]
merged_cols = [c for c in merged_df.columns if c.startswith(f"{prefix}_")]
if not merged_cols:
continue
# Localiza o arquivo trimestral correspondente ao timestamp
try:
ts = pd.to_datetime(timestamp, format=DATETIME_FMT, errors="coerce")
except Exception:
continue
if pd.isna(ts):
continue
source_path = None
for (tab, s_str, e_str), fpath in file_index.items():
if tab != tab_name:
continue
s_dt = pd.to_datetime(s_str, format="%d%m%Y")
e_dt = pd.to_datetime(e_str, format="%d%m%Y")
if s_dt <= ts <= e_dt + pd.Timedelta(days=1):
source_path = fpath
break
if source_path is None:
continue # timestamp fora do range de arquivos disponíveis — pula
# Carrega o CSV original e procura o timestamp
try:
src_df = pd.read_csv(source_path, encoding="utf-8-sig")
dt_col = next((c for c in src_df.columns if "Data" in c and "Hora" in c), None)
if dt_col is None:
continue
src_row = src_df[src_df[dt_col] == timestamp]
if src_row.empty:
continue
# Compara as colunas (pega a primeira coluna de dados não-data)
val_cols = [c for c in src_df.columns if c != dt_col]
if not val_cols:
continue
check_col = val_cols[0]
src_val = str(src_row.iloc[0][check_col]).strip()
# Mapeia para o nome mesclado
merged_col_name = f"{prefix}_{_clean(check_col)}"
if merged_col_name not in merged_df.columns:
continue
merged_val = str(row.get(merged_col_name, "")).strip()
if src_val == merged_val or (src_val == "-" and merged_val in ("-", "nan", "")):
n_ok += 1
else:
n_fail += 1
errors.append(
f" MISMATCH @ {timestamp} | aba={tab_name} | col={check_col} "
f"| src={src_val!r} != merged={merged_val!r}"
)
except Exception as exc:
errors.append(f" ERRO ao validar {timestamp}: {exc}")
return n_ok, n_fail, errors
# ---------------------------------------------------------------------------
# Limpeza de arquivos trimestrais
# ---------------------------------------------------------------------------
def cleanup_source_files(
files: list[Path],
log: Callable[[str], None] = print,
) -> int:
"""Apaga os arquivos trimestrais. Retorna quantos foram deletados."""
deleted = 0
for f in files:
try:
f.unlink()
deleted += 1
except Exception as exc:
log(f" [warn] nao foi possivel apagar {f.name}: {exc}")
log(f" {deleted} arquivo(s) intermediario(s) removidos.")
return deleted
# ---------------------------------------------------------------------------
# CLI standalone
# ---------------------------------------------------------------------------
def main(
dados_dir: Path = DADOS_DIR,
compiled_dir: Path = PREPROC_DIR,
do_validate: bool = True,
n_samples: int = 20,
do_cleanup: bool = False,
log: Callable[[str], None] = print,
) -> Optional[pd.DataFrame]:
"""Aggregates all quarterly CSVs found in *dados_dir* into per-aerodrome tables.
Args:
dados_dir: Directory containing ``Dados de Superfície*.csv`` files.
compiled_dir: Root output directory for compiled aerodrome CSVs.
Each aerodrome gets a sub-directory: ``<compiled_dir>/<ICAO>/``.
do_validate: Run spot-check validation against the source files.
n_samples: Number of rows to spot-check.
do_cleanup: Remove quarterly CSVs after successful validation.
log: Logging callback.
Returns:
The last compiled DataFrame, or ``None`` if no data was found.
"""
base = Path(dados_dir)
files = [f for f in base.glob("Dados de Superfície*.csv") if f.is_file()]
if not files:
log(f"Nenhum CSV encontrado em {base.resolve()}")
return None
# Agrupa por aeródromo
aero_files: dict[str, list[Path]] = {}
for f in files:
parsed = parse_filename(f.name)
if parsed:
_, aero, *_ = parsed
aero_files.setdefault(aero, []).append(f)
final_df = None
for aero, afiles in aero_files.items():
log(f"\n{'=' * 64}")
log(f"Aeródromo : {aero} ({len(afiles)} arquivo(s))")
log(f"{'=' * 64}")
# Check for an existing compiled file (incremental append)
aero_compiled = Path(compiled_dir) / aero
extra_df: Optional[pd.DataFrame] = None
existing = sorted(aero_compiled.glob(f"{aero}_*.csv")) if aero_compiled.is_dir() else []
if existing:
log(f" Arquivo compilado existente: {existing[-1].name} — será atualizado")
extra_df = pd.read_csv(existing[-1], encoding="utf-8-sig")
aerodrome, df = build_aerodrome_table(afiles, extra_df=extra_df, log=log)
if df.empty:
log(" Sem dados.")
continue
start_str, end_str = date_range_from_data(df)
aero_compiled.mkdir(parents=True, exist_ok=True)
out_name = f"{aero}_{start_str}_{end_str}.csv"
out_path = aero_compiled / out_name
df.to_csv(out_path, index=False, encoding="utf-8-sig", quoting=csv.QUOTE_ALL)
log(f" Linhas : {len(df)}")
log(f" Colunas : {len(df.columns)}")
log(f" Periodo : {start_str.replace('_', '-')} -> {end_str.replace('_', '-')}")
log(f" Arquivo : {out_path}")
for old in existing:
if old.name != out_name:
old.unlink()
log(f" [removido arquivo anterior: {old.name}]")
# Validação amostral
validated = True
if do_validate and afiles:
log(f"\n Validando {n_samples} amostras...")
n_ok, n_fail, errs = validate_sample(df, afiles, n=n_samples, log=log)
log(f" Validacao: {n_ok} OK | {n_fail} divergencias")
for e in errs:
log(e)
if n_fail > 0:
validated = False
log(" [ATENCAO] Divergencias encontradas — arquivos intermediarios NAO removidos.")
# Limpeza dos trimestrais
if do_cleanup:
if validated:
log(f"\n Removendo {len(afiles)} arquivo(s) trimestral(is)...")
cleanup_source_files(afiles, log=log)
else:
log(" [skip cleanup] Validação com falhas — arquivos mantidos para inspeção.")
final_df = df
return final_df
if __name__ == "__main__":
try:
if hasattr(sys.stdout, "buffer"):
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
if hasattr(sys.stderr, "buffer"):
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")
except Exception:
pass
ap = argparse.ArgumentParser(
description="Aggregates surface meteorology CSVs into per-aerodrome tables"
)
ap.add_argument("--dados-dir", default=str(DADOS_DIR), metavar="DIR",
help="Directory containing quarterly CSV files")
ap.add_argument("--compiled-dir", default=str(PREPROC_DIR), metavar="DIR",
help="Root output directory for compiled CSVs")
ap.add_argument("--validate", action="store_true", default=True)
ap.add_argument("--no-validate", dest="validate", action="store_false")
ap.add_argument("--n-samples", type=int, default=20, metavar="N")
ap.add_argument("--cleanup", action="store_true",
help="Remove quarterly CSVs after successful validation")
ap.add_argument("--db", action="store_true",
help="Also upsert analytics into the SQLite database")
_args = ap.parse_args()
result_df = main(
dados_dir = Path(_args.dados_dir),
compiled_dir = Path(_args.compiled_dir),
do_validate = _args.validate,
n_samples = _args.n_samples,
do_cleanup = _args.cleanup,
)
if _args.db and result_df is not None:
import db as _db
from pipeline import build_analytics as _build
db_path = DB_PATH
conn = _db.get_connection(db_path)
_db.ensure_schema(conn)
anl = _build(result_df)
_RE2 = re.compile(r"Dados de Superfície .+ - Localidade (\w+) -")
aeros: set[str] = set()
for f in Path(_args.dados_dir).glob("Dados de Superfície*.csv"):
m2 = _RE2.match(f.name)
if m2:
aeros.add(m2.group(1))
aerodrome = next(iter(aeros), "UNKN")
n = _db.upsert_analytics(conn, aerodrome, anl)
conn.close()
print(f"[db] {n} linhas upsertadas em {db_path} para {aerodrome}")

1446
softwares/test/meteorologia_aeroportos/_apps/dashboard.py Normal file
View File

File diff suppressed because it is too large Load Diff

493
softwares/test/meteorologia_aeroportos/_apps/db.py Normal file
View File

@@ -0,0 +1,493 @@
"""
SQLite storage layer for aerodrome meteorological analytics.
Tables:
observations — Hourly analytic time-series (10 float variables per timestamp).
outlier_log — Audit trail of values treated by physical-limit enforcement.
aerodromes — Catalog of aerodromes available on the ICEA/DECEA website.
Example:
>>> from db import get_connection, ensure_schema, upsert_analytics, query_analytics
>>> conn = get_connection(Path("met.db"))
>>> ensure_schema(conn)
>>> upsert_analytics(conn, "SBGR", anl_df)
>>> df = query_analytics(conn, "SBGR", "2024-01-01", "2024-12-31")
"""
import sqlite3
from datetime import date, datetime, timezone
from pathlib import Path
from typing import Optional
import pandas as pd
# ── Analytic columns ──────────────────────────────────────────────────────────
ANL_COLS: list[str] = [
"T", "Td", "UR", "QNH", "WS", "WG", "WD", "VIS", "TETO", "PREC"
]
# ── Physical limits per variable ──────────────────────────────────────────────
PHYSICAL_LIMITS: dict[str, tuple[float, float]] = {
"T": (-25.0, 55.0), # °C — Brazilian climate extremes
"Td": (-30.0, 40.0), # °C — dew point
"UR": ( 0.0, 100.0), # %
"QNH": (940.0, 1060.0), # hPa — absolute world records
"WS": ( 0.0, 200.0), # kt
"WG": ( 0.0, 250.0), # kt — Cat-5 hurricane gust ≈ 175 kt
"WD": ( 0.0, 360.0), # °
"VIS": ( 0.0, 9999.0), # dam
"TETO": ( 0.0, 9999.0), # dam
"PREC": ( 0.0, 500.0), # mm — world record hourly ≈ 300 mm/h
}
# ── SQL schema ────────────────────────────────────────────────────────────────
_SCHEMA = """
CREATE TABLE IF NOT EXISTS observations (
aerodrome TEXT NOT NULL,
dt TEXT NOT NULL,
T REAL, Td REAL, UR REAL, QNH REAL,
WS REAL, WG REAL, WD REAL,
VIS REAL, TETO REAL, PREC REAL,
PRIMARY KEY (aerodrome, dt)
);
CREATE INDEX IF NOT EXISTS idx_aero_dt ON observations(aerodrome, dt);
CREATE TABLE IF NOT EXISTS outlier_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
aerodrome TEXT NOT NULL,
dt TEXT NOT NULL,
variable TEXT NOT NULL,
orig_value REAL NOT NULL,
treatment TEXT NOT NULL DEFAULT 'SET_NULL',
reason TEXT,
applied_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_outlier_aero ON outlier_log(aerodrome, variable);
CREATE TABLE IF NOT EXISTS aerodromes (
icao TEXT PRIMARY KEY,
nome TEXT,
uf TEXT,
updated_at TEXT
);
"""
# ── Connection ────────────────────────────────────────────────────────────────
def get_connection(db_path: Path) -> sqlite3.Connection:
"""Opens (or creates) a SQLite database with WAL mode enabled.
Args:
db_path: Filesystem path to the ``.db`` file. Parent directories are
created automatically if they do not exist.
Returns:
An open :class:`sqlite3.Connection` configured with WAL journal mode,
NORMAL synchronous writes, and a 32 MB page cache.
"""
db_path.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(str(db_path), check_same_thread=False)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA synchronous=NORMAL")
conn.execute("PRAGMA cache_size=-32000")
return conn
def ensure_schema(conn: sqlite3.Connection) -> None:
"""Creates all tables and indices if they do not already exist.
Args:
conn: Open SQLite connection.
"""
conn.executescript(_SCHEMA)
conn.commit()
# ── Physical-limit clipping ───────────────────────────────────────────────────
def _clip_to_limits(df: pd.DataFrame) -> pd.DataFrame:
"""Sets values outside physical limits to NaN (no audit log entry).
Args:
df: DataFrame whose columns may include any subset of ANL_COLS.
Returns:
Copy of *df* with out-of-range values replaced by ``float("nan")``.
"""
df = df.copy()
for col, (lo, hi) in PHYSICAL_LIMITS.items():
if col in df.columns:
mask = df[col].notna() & ((df[col] < lo) | (df[col] > hi))
df.loc[mask, col] = float("nan")
return df
# ── Upsert analytics ──────────────────────────────────────────────────────────
def upsert_analytics(
conn: sqlite3.Connection,
aerodrome: str,
anl_df: pd.DataFrame,
) -> int:
"""Inserts or replaces rows in the observations table.
Physical limits are applied automatically before insertion; out-of-range
values are silently set to NULL.
Args:
conn: Open SQLite connection with schema already applied.
aerodrome: ICAO code of the aerodrome (e.g. ``"SBGR"``).
anl_df: DataFrame with a ``_dt`` column (datetime-like) and any subset
of :data:`ANL_COLS` as float columns.
Returns:
Number of rows processed (inserted or replaced).
"""
if anl_df.empty:
return 0
anl_df = _clip_to_limits(anl_df)
cols_present = [c for c in ANL_COLS if c in anl_df.columns]
placeholders = ", ".join(["?"] * (2 + len(cols_present)))
col_names = ", ".join(["aerodrome", "dt"] + cols_present)
sql = (
f"INSERT OR REPLACE INTO observations ({col_names}) VALUES ({placeholders})"
)
def _dt_str(v: object) -> str:
if isinstance(v, (datetime, pd.Timestamp)):
return v.strftime("%Y-%m-%d %H:%M:%S")
return str(v)[:19]
rows = []
for _, row in anl_df.iterrows():
vals: list = [aerodrome, _dt_str(row["_dt"])]
for c in cols_present:
v = row[c]
vals.append(None if pd.isna(v) else float(v))
rows.append(tuple(vals))
with conn:
conn.executemany(sql, rows)
return len(rows)
# ── Physical-limit repair on existing rows ────────────────────────────────────
def apply_physical_limits(
conn: sqlite3.Connection,
aerodrome: Optional[str] = None,
) -> int:
"""Audits and corrects out-of-range values already stored in the database.
Every corrected value is logged in ``outlier_log`` before being set to
NULL. Pass ``aerodrome=None`` to process all aerodromes.
Args:
conn: Open SQLite connection with schema already applied.
aerodrome: ICAO code to restrict processing, or ``None`` for all.
Returns:
Total number of outlier records created and corrected.
"""
now = datetime.now(timezone.utc).isoformat()
total = 0
aero_filter = "AND aerodrome=?" if aerodrome else ""
aero_params_suffix = (aerodrome,) if aerodrome else ()
for col, (lo, hi) in PHYSICAL_LIMITS.items():
rows = conn.execute(
f"SELECT aerodrome, dt, {col} FROM observations "
f"WHERE {col} IS NOT NULL AND {col} < ? {aero_filter}",
(lo,) + aero_params_suffix,
).fetchall()
if rows:
conn.executemany(
"INSERT INTO outlier_log"
"(aerodrome,dt,variable,orig_value,treatment,reason,applied_at)"
" VALUES(?,?,?,?,?,?,?)",
[(r[0], r[1], col, r[2], "SET_NULL", f"below_limit:{lo}", now)
for r in rows],
)
total += len(rows)
rows = conn.execute(
f"SELECT aerodrome, dt, {col} FROM observations "
f"WHERE {col} IS NOT NULL AND {col} > ? {aero_filter}",
(hi,) + aero_params_suffix,
).fetchall()
if rows:
conn.executemany(
"INSERT INTO outlier_log"
"(aerodrome,dt,variable,orig_value,treatment,reason,applied_at)"
" VALUES(?,?,?,?,?,?,?)",
[(r[0], r[1], col, r[2], "SET_NULL", f"above_limit:{hi}", now)
for r in rows],
)
total += len(rows)
where_clause = f"WHERE aerodrome='{aerodrome}'" if aerodrome else ""
conn.execute(f"""
UPDATE observations SET
T = CASE WHEN T IS NOT NULL AND (T < -25 OR T > 55 ) THEN NULL ELSE T END,
Td = CASE WHEN Td IS NOT NULL AND (Td < -30 OR Td > 40 ) THEN NULL ELSE Td END,
UR = CASE WHEN UR IS NOT NULL AND (UR < 0 OR UR > 100 ) THEN NULL ELSE UR END,
QNH = CASE WHEN QNH IS NOT NULL AND (QNH < 940 OR QNH > 1060) THEN NULL ELSE QNH END,
WS = CASE WHEN WS IS NOT NULL AND (WS < 0 OR WS > 200 ) THEN NULL ELSE WS END,
WG = CASE WHEN WG IS NOT NULL AND (WG < 0 OR WG > 250 ) THEN NULL ELSE WG END,
WD = CASE WHEN WD IS NOT NULL AND (WD < 0 OR WD > 360 ) THEN NULL ELSE WD END,
VIS = CASE WHEN VIS IS NOT NULL AND (VIS < 0 OR VIS > 9999) THEN NULL ELSE VIS END,
TETO = CASE WHEN TETO IS NOT NULL AND (TETO < 0 OR TETO > 9999) THEN NULL ELSE TETO END,
PREC = CASE WHEN PREC IS NOT NULL AND (PREC < 0 OR PREC > 500 ) THEN NULL ELSE PREC END
{where_clause}
""")
conn.commit()
return total
# ── Outlier audit queries ─────────────────────────────────────────────────────
def get_outlier_summary(
conn: sqlite3.Connection,
aerodrome: Optional[str] = None,
) -> pd.DataFrame:
"""Returns per-variable outlier counts and statistics from the audit log.
Args:
conn: Open SQLite connection.
aerodrome: Filter to a single ICAO code, or ``None`` for all.
Returns:
DataFrame with columns ``variable``, ``aerodrome``, ``n_outliers``,
``min_orig``, ``max_orig``, ``reason``, ``last_applied``.
"""
aero_filter = "WHERE aerodrome=?" if aerodrome else ""
params = (aerodrome,) if aerodrome else ()
return pd.read_sql_query(
f"""
SELECT
variable,
aerodrome,
COUNT(*) AS n_outliers,
MIN(orig_value) AS min_orig,
MAX(orig_value) AS max_orig,
reason,
MAX(applied_at) AS last_applied
FROM outlier_log
{aero_filter}
GROUP BY variable, aerodrome, reason
ORDER BY variable, aerodrome
""",
conn, params=params,
)
def get_outlier_detail(
conn: sqlite3.Connection,
aerodrome: Optional[str] = None,
variable: Optional[str] = None,
limit: int = 500,
) -> pd.DataFrame:
"""Returns row-level outlier records from the audit log.
Args:
conn: Open SQLite connection.
aerodrome: Filter by ICAO code, or ``None`` for all.
variable: Filter by variable name (e.g. ``"T"``), or ``None`` for all.
limit: Maximum number of rows to return.
Returns:
DataFrame with columns ``aerodrome``, ``dt``, ``variable``,
``orig_value``, ``reason``, ``applied_at``.
"""
conditions: list[str] = []
params: list[object] = []
if aerodrome:
conditions.append("aerodrome=?")
params.append(aerodrome)
if variable:
conditions.append("variable=?")
params.append(variable)
where = ("WHERE " + " AND ".join(conditions)) if conditions else ""
return pd.read_sql_query(
f"SELECT aerodrome, dt, variable, orig_value, reason, applied_at "
f"FROM outlier_log {where} ORDER BY applied_at DESC, aerodrome, dt "
f"LIMIT {limit}",
conn, params=params,
)
def has_outlier_log(conn: sqlite3.Connection) -> bool:
"""Returns True if at least one outlier record exists in the audit log.
Args:
conn: Open SQLite connection.
"""
n = conn.execute("SELECT COUNT(*) FROM outlier_log").fetchone()[0]
return n > 0
# ── Aerodrome catalog ─────────────────────────────────────────────────────────
def upsert_aerodromes(
conn: sqlite3.Connection,
aerodromes: list[dict],
) -> int:
"""Inserts or updates aerodrome catalog entries.
Args:
conn: Open SQLite connection with schema already applied.
aerodromes: List of dicts, each with keys ``icao``, ``nome``, ``uf``.
Returns:
Number of records processed.
"""
if not aerodromes:
return 0
now = datetime.now(timezone.utc).isoformat()
with conn:
conn.executemany(
"INSERT OR REPLACE INTO aerodromes(icao, nome, uf, updated_at)"
" VALUES(?, ?, ?, ?)",
[(a["icao"], a.get("nome", ""), a.get("uf", ""), now)
for a in aerodromes if a.get("icao")],
)
return len(aerodromes)
def list_all_aerodromes(conn: sqlite3.Connection) -> list[dict]:
"""Returns all entries from the aerodrome catalog table.
Args:
conn: Open SQLite connection.
Returns:
List of ``{"icao": str, "nome": str, "uf": str}`` dicts ordered by
ICAO code. Returns an empty list if the catalog is empty.
"""
rows = conn.execute(
"SELECT icao, nome, uf FROM aerodromes ORDER BY icao"
).fetchall()
return [{"icao": r[0], "nome": r[1] or "", "uf": r[2] or ""} for r in rows]
def list_aerodromes_with_data(conn: sqlite3.Connection) -> list[dict]:
"""Returns catalog entries for aerodromes that have rows in observations.
Joins the ``observations`` table with the ``aerodromes`` catalog so the
caller receives full display metadata (name, state) for each code that
actually has data.
Args:
conn: Open SQLite connection with schema already applied.
Returns:
List of ``{"icao": str, "nome": str, "uf": str}`` dicts ordered by
ICAO code. Returns an empty list when no observations exist yet.
"""
rows = conn.execute(
"""SELECT DISTINCT o.aerodrome,
COALESCE(a.nome, ''),
COALESCE(a.uf, '')
FROM observations o
LEFT JOIN aerodromes a ON a.icao = o.aerodrome
ORDER BY o.aerodrome"""
).fetchall()
return [{"icao": r[0], "nome": r[1], "uf": r[2]} for r in rows]
# ── General queries ───────────────────────────────────────────────────────────
def query_analytics(
conn: sqlite3.Connection,
aerodrome: str,
start_dt: Optional[str] = None,
end_dt: Optional[str] = None,
) -> pd.DataFrame:
"""Queries analytic observations for one aerodrome over an optional date range.
Args:
conn: Open SQLite connection.
aerodrome: ICAO code to query.
start_dt: Inclusive start date/datetime string (``"YYYY-MM-DD"`` or
``"YYYY-MM-DD HH:MM:SS"``). No lower bound if ``None``.
end_dt: Inclusive end date (``"YYYY-MM-DD"``). When only a date is
given, the comparison extends to ``23:59:59`` of that day.
Returns:
DataFrame with columns ``_dt`` (``datetime64``) plus the ANL_COLS
floats. Returns an empty DataFrame if no rows match.
"""
sql = (
"SELECT dt, T, Td, UR, QNH, WS, WG, WD, VIS, TETO, PREC "
"FROM observations WHERE aerodrome=?"
)
params: list[object] = [aerodrome]
if start_dt:
sql += " AND dt >= ?"
params.append(start_dt)
if end_dt:
sql += " AND dt <= ?"
params.append(end_dt + " 23:59:59" if len(end_dt) == 10 else end_dt)
sql += " ORDER BY dt"
df = pd.read_sql_query(sql, conn, params=params)
if df.empty:
return df
df["_dt"] = pd.to_datetime(df["dt"], format="%Y-%m-%d %H:%M:%S", errors="coerce")
return df.drop(columns=["dt"]).dropna(subset=["_dt"]).reset_index(drop=True)
def get_coverage(
conn: sqlite3.Connection,
aerodrome: str,
) -> Optional[tuple[date, date]]:
"""Returns the earliest and latest observation dates for one aerodrome.
Args:
conn: Open SQLite connection.
aerodrome: ICAO code to inspect.
Returns:
``(min_date, max_date)`` tuple, or ``None`` if no observations exist.
"""
row = conn.execute(
"SELECT MIN(dt), MAX(dt) FROM observations WHERE aerodrome=?",
(aerodrome,),
).fetchone()
if not row or row[0] is None:
return None
return date.fromisoformat(row[0][:10]), date.fromisoformat(row[1][:10])
def list_aerodromes(conn: sqlite3.Connection) -> list[str]:
"""Returns ICAO codes of all aerodromes that have at least one observation.
Args:
conn: Open SQLite connection.
Returns:
Sorted list of ICAO code strings.
"""
rows = conn.execute(
"SELECT DISTINCT aerodrome FROM observations ORDER BY aerodrome"
).fetchall()
return [r[0] for r in rows]
def aerodrome_stats(conn: sqlite3.Connection, aerodrome: str) -> dict:
"""Returns basic row-count statistics for one aerodrome.
Args:
conn: Open SQLite connection.
aerodrome: ICAO code to inspect.
Returns:
Dict with keys ``min_dt``, ``max_dt``, ``n_obs``, ``n_T``, ``n_VIS``.
Returns an empty dict if no observations exist.
"""
row = conn.execute(
"SELECT MIN(dt), MAX(dt), COUNT(*), COUNT(T), COUNT(VIS) "
"FROM observations WHERE aerodrome=?",
(aerodrome,),
).fetchone()
if not row or row[0] is None:
return {}
return {
"min_dt": row[0], "max_dt": row[1],
"n_obs": row[2], "n_T": row[3], "n_VIS": row[4],
}

537
softwares/test/meteorologia_aeroportos/_apps/pipeline.py Normal file
View File

@@ -0,0 +1,537 @@
"""
End-to-end pipeline orchestrator for ICEA/DECEA surface meteorology.
Workflow:
1. Check existing coverage in SQLite (``db.get_coverage``).
2. Forward pass — download new data after ``ex_max`` (update).
3. Backward pass — download history before ``ex_min`` (auto-stop on empty years).
4. Compute analytics from the quarterly CSV files.
5. Upsert into SQLite and write the analytics CSV backup to PREPROC_DIR.
6. Spot-check validation against the quarterly source files.
7. Remove quarterly intermediate CSVs (if validation passed and cleanup enabled).
Example:
python pipeline.py --aerodrome SBGR --all-years
python pipeline.py --aerodrome SBGR --start-year 2020 --end-year 2025 --no-cleanup
"""
import argparse
import io
import sys
import unicodedata
from datetime import date
from pathlib import Path
from typing import Callable, Optional
import pandas as pd
from selenium.webdriver.support.ui import WebDriverWait
import db as _db
from scraper_meteorologia import (
SITE_MIN_DATE,
STOP_EMPTY_YEARS,
make_driver,
scrape_year,
)
from concat_meteorologia import (
build_aerodrome_table,
cleanup_source_files,
validate_sample,
)
# ── Paths (resolved relative to this file, independent of CWD) ───────────────
_APPS_DIR = Path(__file__).resolve().parent # .../meteorologia_aeroportos/_apps/
_BASE_DIR = _APPS_DIR.parent # .../meteorologia_aeroportos/
_REPO_ROOT = _BASE_DIR.parents[2] # dataset/
DADOS_DIR = _BASE_DIR / "db" / "dados" # temporary CSV files
DB_PATH = _BASE_DIR / "db" / "met.db" # SQLite database
PREPROC_DIR = (
_REPO_ROOT / "tabelas" / "preproc" / "meteorologia_aeroportos"
) # permanent analytics CSV backups
# ── Progress milestones ───────────────────────────────────────────────────────
_PROG_START = 0.05
_PROG_FORWARD_DONE = 0.35
_PROG_ANALYTICS = 0.72
_PROG_UPSERT = 0.80
_PROG_VALIDATE = 0.88
_PROG_CLEANUP = 0.95
_PROG_DONE = 1.00
# ---------------------------------------------------------------------------
# Analytics helpers (standalone, no dependency on dashboard.py)
# ---------------------------------------------------------------------------
def _norm(s: str) -> str:
"""Returns *s* lowercased, ASCII-only, with diacritics stripped."""
return unicodedata.normalize("NFKD", s.lower()).encode("ascii", "ignore").decode("ascii")
def _to_num(s: pd.Series) -> pd.Series:
"""Coerces *s* to float, treating ``'-'`` and blank strings as NaN.
Args:
s: A pandas Series of raw string or numeric values.
Returns:
Float Series with unparseable values replaced by ``NaN``.
"""
if pd.api.types.is_numeric_dtype(s):
return s.astype(float)
v = (
s.astype(str)
.str.strip()
.replace({"-": None, "nan": None, "NaN": None, "None": None, "": None})
)
v = v.str.replace(",", ".", regex=False)
return pd.to_numeric(v, errors="coerce")
def _agg(
df: pd.DataFrame,
prefix: str,
kw: str,
fn: str = "mean",
) -> pd.Series:
"""Aggregates columns matching ``<prefix>_*<kw>*``, preferring generic ones.
Columns whose name contains ``_-_`` are treated as generic (station-wide)
and preferred over runway/heading-specific columns. When multiple specific
columns are present, they are combined using *fn*.
Args:
df: Merged wide DataFrame with prefixed column names.
prefix: Column prefix to filter on (e.g. ``"temp"``).
kw: Keyword fragment to match in normalised column names.
fn: Aggregation function: ``"mean"``, ``"min"``, or ``"max"``.
Returns:
Float Series aligned with *df*'s index.
"""
kw_n = _norm(kw)
cols = [c for c in df.columns if c.startswith(f"{prefix}_") and kw_n in _norm(c)]
if not cols:
return pd.Series(dtype=float, index=df.index)
gen = [c for c in cols if "_-_" in c]
spc = [c for c in cols if "_-_" not in c]
if gen:
s = _to_num(df[gen[0]])
if s.notna().mean() > 0.05:
return s
if spc:
mat = pd.DataFrame({c: _to_num(df[c]) for c in spc})
return {"mean": mat.mean, "min": mat.min, "max": mat.max}[fn](axis=1)
return _to_num(df[cols[0]])
def build_analytics(merged: pd.DataFrame) -> pd.DataFrame:
"""Derives the 10 clean analytic variables from the 84-column merged CSV.
Args:
merged: Raw merged DataFrame produced by
:func:`concat_meteorologia.build_aerodrome_table`.
Returns:
DataFrame with columns ``_dt`` (``datetime64``) plus
``T``, ``Td``, ``UR``, ``QNH``, ``WS``, ``WG``, ``WD``,
``VIS``, ``TETO``, ``PREC`` (all float).
"""
dt_col = next((c for c in merged.columns if "Data" in c and "Hora" in c), None)
dt = (
pd.to_datetime(merged[dt_col], format="%d/%m/%Y - %H:%M:%S", errors="coerce")
if dt_col
else pd.Series(dtype="datetime64[ns]")
)
qnh = (
_to_num(merged["pres_QNH"])
if "pres_QNH" in merged.columns
else _agg(merged, "pres", "QNH")
)
vis = next(
(
_to_num(merged[c])
for c in merged.columns
if "visib" in c.lower() and "predominante" in _norm(c)
),
pd.Series(dtype=float, index=merged.index),
)
prec = next(
(
_to_num(merged[c])
for c in merged.columns
if "prec_precipita" in c.lower() and "dura" not in c.lower()
),
pd.Series(dtype=float, index=merged.index),
)
return pd.DataFrame({
"_dt": dt.values,
"T": _agg(merged, "temp", "Bulbo_Seco").values,
"Td": _agg(merged, "temp", "Orvalho").values,
"UR": _agg(merged, "temp", "Umidade").values,
"QNH": qnh.values,
"WS": _agg(merged, "vent", "Velocidade").values,
"WG": _agg(merged, "vent", "Rajada", "max").values,
"WD": _agg(merged, "vent", "Dire").values,
"VIS": vis.values,
"TETO": _agg(merged, "teto", "Teto", "min").values,
"PREC": prec.values,
})
# ---------------------------------------------------------------------------
# Two-pass scraping
# ---------------------------------------------------------------------------
def _run_forward(
driver: object,
wait: WebDriverWait,
aerodrome: str,
ex_max: date,
dados_dir: str,
log: Callable[[str], None],
) -> int:
"""Downloads new observations after *ex_max* (update pass).
Args:
driver: Selenium WebDriver instance.
wait: Configured WebDriverWait bound to *driver*.
aerodrome: ICAO code to scrape.
ex_max: Latest date already stored; scraping starts just after this.
dados_dir: Directory where quarterly CSVs are written.
log: Logging callback.
Returns:
Number of non-empty quarters downloaded.
"""
today = date.today()
if ex_max >= today:
log("[pipeline] Dados já atualizados até hoje.")
return 0
log(f"\n[pipeline] === Passe 1: atualização {ex_max.year} → {today.year} ===")
total = 0
for year in range(today.year, ex_max.year - 1, -1):
log(f"\n{'=' * 55}\nAno {year} — {aerodrome} (forward)\n{'=' * 55}")
total += scrape_year(driver, wait, aerodrome, year, dados_dir,
stop_before=ex_max, log=log)
return total
def _run_backward(
driver: object,
wait: WebDriverWait,
aerodrome: str,
ex_min: date,
all_years: bool,
dados_dir: str,
log: Callable[[str], None],
progress: Callable[[float, str], None],
) -> int:
"""Downloads historical data before *ex_min* (backward pass).
Stops automatically after :data:`STOP_EMPTY_YEARS` consecutive empty years
when *all_years* is ``True``.
Args:
driver: Selenium WebDriver instance.
wait: Configured WebDriverWait bound to *driver*.
aerodrome: ICAO code to scrape.
ex_min: Earliest date already stored; scraping goes back from here.
all_years: When ``True``, enable the consecutive-empty-year stop logic.
dados_dir: Directory where quarterly CSVs are written.
log: Logging callback.
progress: Progress callback ``(fraction, message)``.
Returns:
Number of non-empty quarters downloaded.
"""
start_year = ex_min.year - 1
if start_year < SITE_MIN_DATE.year:
log("[pipeline] Histórico já vai até o limite do site.")
return 0
log(f"\n[pipeline] === Passe 2: histórico {start_year} → {SITE_MIN_DATE.year} ===")
total = 0
consecutive_empty = 0
years = list(range(start_year, SITE_MIN_DATE.year - 1, -1))
for i, year in enumerate(years):
pct = _PROG_FORWARD_DONE + (_PROG_ANALYTICS - _PROG_FORWARD_DONE) * (
i / max(len(years), 1)
)
progress(pct, f"Histórico {aerodrome} {year}")
log(f"\n{'=' * 55}\nAno {year} — {aerodrome} (backward)\n{'=' * 55}")
n = scrape_year(driver, wait, aerodrome, year, dados_dir,
stop_before=None, log=log)
total += n
if n == 0:
consecutive_empty += 1
log(f" [sem dados] ({consecutive_empty}/{STOP_EMPTY_YEARS})")
if all_years and consecutive_empty >= STOP_EMPTY_YEARS:
log(f"\n{STOP_EMPTY_YEARS} anos consecutivos sem dados — parando.")
break
else:
consecutive_empty = 0
return total
# ---------------------------------------------------------------------------
# Main pipeline
# ---------------------------------------------------------------------------
def run_pipeline(
aerodrome: str,
dados_dir: Path = DADOS_DIR,
db_path: Path = DB_PATH,
preproc_dir: Path = PREPROC_DIR,
all_years: bool = True,
start_year: Optional[int] = None,
end_year: Optional[int] = None,
headless: bool = True,
n_samples: int = 20,
do_validate: bool = True,
do_cleanup: bool = True,
log: Callable[[str], None] = print,
progress: Callable[[float, str], None] = lambda pct, msg: None,
) -> dict:
"""Runs the full scrape → analytics → SQLite → validate → cleanup pipeline.
Args:
aerodrome: ICAO code to process (e.g. ``"SBGR"``).
dados_dir: Directory for temporary quarterly CSV files.
db_path: Path to the SQLite database file.
preproc_dir: Root directory for permanent analytics CSV backups.
Each aerodrome gets a sub-directory: ``<preproc_dir>/<aerodrome>/``.
all_years: When ``True``, scrape the full available history.
start_year: First year to scrape (used when *all_years* is ``False``).
end_year: Last year to scrape (used when *all_years* is ``False``).
headless: Run Chrome in headless mode (``True``) or visible (``False``).
n_samples: Number of spot-check samples for validation.
do_validate: Enable spot-check validation against source CSVs.
do_cleanup: Remove quarterly CSVs after a successful validation.
log: Logging callback receiving a single string.
progress: Progress callback receiving ``(fraction: float, message: str)``.
Returns:
Dict with keys ``rows``, ``period_start``, ``period_end``,
``n_ok``, ``n_fail``, ``errors``, ``db_path``.
"""
base = Path(dados_dir)
base.mkdir(parents=True, exist_ok=True)
db_path = Path(db_path)
# 1. Existing coverage ────────────────────────────────────────────────────
conn = _db.get_connection(db_path)
_db.ensure_schema(conn)
coverage = _db.get_coverage(conn, aerodrome)
if coverage:
ex_min, ex_max = coverage
log(f"\n[pipeline] Cobertura existente: {ex_min} → {ex_max}")
else:
log(f"\n[pipeline] Sem dados anteriores para {aerodrome}.")
ex_min = ex_max = None
progress(_PROG_START, f"Iniciando scraping {aerodrome}")
# 2. Scraping ─────────────────────────────────────────────────────────────
driver = make_driver(headless=headless)
wait = WebDriverWait(driver, 60)
try:
if coverage:
_run_forward(driver, wait, aerodrome, ex_max, str(base), log)
progress(_PROG_FORWARD_DONE, f"Histórico {aerodrome}")
_run_backward(driver, wait, aerodrome, ex_min, all_years,
str(base), log, progress)
else:
log(f"\n[pipeline] === Coleta inicial: {date.today().year} → {SITE_MIN_DATE.year} ===")
end_yr = end_year if not all_years else date.today().year
start_yr = start_year if not all_years else SITE_MIN_DATE.year
consec = 0
years = list(range(end_yr, start_yr - 1, -1))
for i, year in enumerate(years):
pct = _PROG_START + (_PROG_FORWARD_DONE + 0.30) * (i / max(len(years), 1))
progress(pct, f"Scraping {aerodrome} {year}")
log(f"\n{'=' * 55}\nAno {year} — {aerodrome}\n{'=' * 55}")
n = scrape_year(driver, wait, aerodrome, year, str(base), log=log)
if n == 0:
consec += 1
log(f" [sem dados] ({consec}/{STOP_EMPTY_YEARS})")
if all_years and consec >= STOP_EMPTY_YEARS:
log(f"\n{STOP_EMPTY_YEARS} anos consecutivos — parando.")
break
else:
consec = 0
except KeyboardInterrupt:
log("\n[pipeline] Interrompido.")
finally:
driver.quit()
progress(_PROG_ANALYTICS, "Construindo analytics…")
# 3. Concat → analytics ───────────────────────────────────────────────────
new_files = [f for f in base.glob("Dados de Superfície*.csv") if f.is_file()]
if not new_files:
cov = _db.get_coverage(conn, aerodrome)
if cov:
s, e = str(cov[0]), str(cov[1])
stats = _db.aerodrome_stats(conn, aerodrome)
progress(_PROG_DONE, "Concluído")
return dict(rows=stats.get("n_obs", 0), period_start=s, period_end=e,
n_ok=0, n_fail=0, errors=[], db_path=str(db_path))
log("[pipeline] Nenhum dado disponível.")
return dict(rows=0, period_start="", period_end="",
n_ok=0, n_fail=0, errors=[], db_path=str(db_path))
_, merged = build_aerodrome_table(new_files, log=log)
if merged.empty:
log("[pipeline] Merged vazio.")
progress(_PROG_DONE, "Concluído")
return dict(rows=0, period_start="", period_end="",
n_ok=0, n_fail=0, errors=[], db_path=str(db_path))
anl = build_analytics(merged)
anl = anl.dropna(subset=["_dt"]).reset_index(drop=True)
progress(_PROG_UPSERT, "Inserindo no banco SQLite…")
# 4. Upsert into SQLite ───────────────────────────────────────────────────
n_upserted = _db.upsert_analytics(conn, aerodrome, anl)
log(f"\n[pipeline] {n_upserted} linhas upsertadas no SQLite ({db_path.name})")
# Write analytics CSV backup to tabelas/preproc/meteorologia_aeroportos/<ICAO>/
anl_dir = Path(preproc_dir) / aerodrome
anl_dir.mkdir(parents=True, exist_ok=True)
cov_final = _db.get_coverage(conn, aerodrome)
if cov_final:
s_str = str(cov_final[0]).replace("-", "_")
e_str = str(cov_final[1]).replace("-", "_")
csv_backup = anl_dir / f"{aerodrome}_{s_str}_{e_str}.csv"
for old in anl_dir.glob(f"{aerodrome}_*.csv"):
old.unlink()
anl_out = anl.copy()
anl_out["_dt"] = anl_out["_dt"].dt.strftime("%Y-%m-%d %H:%M:%S")
anl_out.to_csv(csv_backup, index=False, encoding="utf-8-sig")
log(f"[pipeline] Backup CSV: {csv_backup}")
progress(_PROG_VALIDATE, "Validando…")
# 5. Validation ───────────────────────────────────────────────────────────
n_ok = n_fail = 0
val_errors: list[str] = []
validated = True
if do_validate and new_files:
log(f"\n[pipeline] Validando {n_samples} amostras…")
n_ok, n_fail, val_errors = validate_sample(merged, new_files, n=n_samples, log=log)
log(f"[pipeline] Validação: {n_ok} OK | {n_fail} divergências")
for e in val_errors:
log(e)
if n_fail > 0:
validated = False
progress(_PROG_CLEANUP, "Cleanup…")
# 6. Cleanup ──────────────────────────────────────────────────────────────
if do_cleanup and new_files:
if validated:
log(f"\n[pipeline] Removendo {len(new_files)} CSVs trimestrais…")
cleanup_source_files(new_files, log=log)
else:
log("[pipeline] Validação com falhas — CSVs trimestrais mantidos.")
cov_now = _db.get_coverage(conn, aerodrome)
stats = _db.aerodrome_stats(conn, aerodrome)
conn.close()
progress(_PROG_DONE, "Concluído")
log("\n[pipeline] Pipeline finalizado.")
return dict(
rows = stats.get("n_obs", len(anl)),
period_start = str(cov_now[0]) if cov_now else "",
period_end = str(cov_now[1]) if cov_now else "",
n_ok = n_ok,
n_fail = n_fail,
errors = val_errors,
db_path = str(db_path),
)
# ---------------------------------------------------------------------------
# CLI entry point
# ---------------------------------------------------------------------------
def main() -> None:
"""Parses CLI arguments and runs the pipeline."""
try:
if hasattr(sys.stdout, "buffer"):
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
if hasattr(sys.stderr, "buffer"):
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")
except Exception:
pass
parser = argparse.ArgumentParser(
description="Pipeline meteorologia ICEA/DECEA → SQLite",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
exemplos:
python pipeline.py --aerodrome SBGR --all-years
python pipeline.py --aerodrome SBGR --start-year 2020 --end-year 2025
python pipeline.py --aerodrome SBSP --all-years --no-headless --no-cleanup
""",
)
parser.add_argument("--aerodrome", default="SBGR", metavar="ICAO")
parser.add_argument("--all-years", action="store_true")
parser.add_argument("--start-year", type=int, metavar="ANO")
parser.add_argument("--end-year", type=int, metavar="ANO")
parser.add_argument("--dados-dir", default=str(DADOS_DIR),metavar="DIR",
help="Directory for temporary quarterly CSVs")
parser.add_argument("--db-path", default=str(DB_PATH), metavar="PATH",
help="Path to the SQLite database file")
parser.add_argument("--preproc-dir", default=str(PREPROC_DIR), metavar="DIR",
help="Root directory for permanent analytics CSV backups")
parser.add_argument("--no-headless", action="store_true")
parser.add_argument("--no-validate", action="store_true")
parser.add_argument("--n-samples", type=int, default=20)
parser.add_argument("--no-cleanup", action="store_true")
args = parser.parse_args()
if not args.all_years and (args.start_year is None or args.end_year is None):
parser.error("Use --all-years ou informe --start-year e --end-year")
result = run_pipeline(
aerodrome = args.aerodrome,
dados_dir = Path(args.dados_dir),
db_path = Path(args.db_path),
preproc_dir = Path(args.preproc_dir),
all_years = args.all_years,
start_year = args.start_year,
end_year = args.end_year,
headless = not args.no_headless,
n_samples = args.n_samples,
do_validate = not args.no_validate,
do_cleanup = not args.no_cleanup,
)
print(f"\n{'=' * 55}")
print(f" Linhas : {result['rows']}")
print(f" Período : {result['period_start']} → {result['period_end']}")
print(f" Validação : {result['n_ok']} OK | {result['n_fail']} divergências")
print(f" Banco : {result['db_path']}")
if __name__ == "__main__":
main()

605
softwares/test/meteorologia_aeroportos/_apps/scraper_meteorologia.py Normal file
View File

@@ -0,0 +1,605 @@
"""
Surface meteorology web scraper for the ICEA/DECEA data portal.
Source: https://pesquisa.icea.decea.mil.br/superficie_list/
Standalone usage:
python scraper_meteorologia.py --aerodrome SBGR --all-years
python scraper_meteorologia.py --aerodrome SBGR --start-year 2020 --end-year 2025
python scraper_meteorologia.py --aerodrome SBSP --start-year 2023 --end-year 2023 --no-headless
python scraper_meteorologia.py --fetch-catalog # writes aerodrome catalog to SQLite
Importable by the pipeline:
from scraper_meteorologia import make_driver, scrape_year, fetch_aerodrome_catalog, SITE_MIN_DATE
"""
import argparse
import calendar
import csv
import io
import os
import re
import sys
import time
from datetime import date
from pathlib import Path
from typing import Callable, Optional
import pandas as pd
from bs4 import BeautifulSoup
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait
from webdriver_manager.chrome import ChromeDriverManager
# ── Site constants ────────────────────────────────────────────────────────────
BASE_URL = "https://pesquisa.icea.decea.mil.br/superficie_list/"
SITE_MIN_DATE = date(1947, 12, 1)
SITE_MAX_DATE = date.today() # dynamic: never beyond today
CHUNK_MONTHS = 3
STOP_EMPTY_YEARS = 2
# ── Paths (resolved relative to this file, independent of CWD) ───────────────
_APPS_DIR = Path(__file__).resolve().parent # .../meteorologia_aeroportos/_apps/
_BASE_DIR = _APPS_DIR.parent # .../meteorologia_aeroportos/
DADOS_DIR = _BASE_DIR / "db" / "dados" # temporary CSV output
DB_PATH = _BASE_DIR / "db" / "met.db" # SQLite database
def fetch_site_max_date(driver: webdriver.Chrome) -> date:
"""
Lê o atributo data-date-end-date do campo de período no site e atualiza
SITE_MAX_DATE globalmente. Retorna date.today() como fallback.
"""
global SITE_MAX_DATE
try:
val = driver.execute_script(
"var el = document.getElementById('dtPeriodo');"
"return el ? el.getAttribute('data-date-end-date') : null;"
)
if val:
d, m, y = val.strip().split("/")
SITE_MAX_DATE = date(int(y), int(m), int(d))
return SITE_MAX_DATE
except Exception:
pass
return SITE_MAX_DATE
TABS = [
("tab_cgt", "CGT"),
("tab_nuvem", "Nuvem"),
("tab_prec", "Precipitação"),
("tab_pres", "Pressão"),
("tab_rvr", "RVR"),
("tab_temp", "Temperatura"),
("tab_teto", "Teto"),
("tab_vent", "Vento"),
("tab_visib", "Visibilidade"),
]
# ---------------------------------------------------------------------------
# WebDriver
# ---------------------------------------------------------------------------
def make_driver(headless: bool = True) -> webdriver.Chrome:
"""Creates a configured Chrome WebDriver instance.
Downloads the matching ChromeDriver automatically via ``webdriver-manager``
if it is not already cached.
Args:
headless: Run Chrome without a visible window (``True``) or visibly
(``False``). Set to ``False`` for interactive debugging.
Returns:
A :class:`selenium.webdriver.Chrome` instance ready for use.
"""
opts = webdriver.ChromeOptions()
if headless:
opts.add_argument("--headless=new")
opts.add_argument("--no-sandbox")
opts.add_argument("--disable-dev-shm-usage")
opts.add_argument("--window-size=1920,1080")
opts.add_argument("--disable-blink-features=AutomationControlled")
opts.add_experimental_option("excludeSwitches", ["enable-automation"])
service = Service(ChromeDriverManager().install())
driver = webdriver.Chrome(service=service, options=opts)
driver.set_page_load_timeout(300)
driver.set_script_timeout(120)
try:
driver.command_executor.client_config.timeout = 300
except Exception:
pass
return driver
# ---------------------------------------------------------------------------
# Formulário
# ---------------------------------------------------------------------------
def select_aerodrome(driver: webdriver.Chrome, wait: WebDriverWait, code: str) -> str:
"""Seleciona aeródromo no Bootstrap Select (JS-first, fallback visual)."""
result = driver.execute_script(f"""
var candidates = document.querySelectorAll(
'select.selectpicker, select[name*="localidade"], select[name*="aerodrome"], select'
);
for (var s = 0; s < candidates.length; s++) {{
var sel = candidates[s];
for (var i = 0; i < sel.options.length; i++) {{
if (sel.options[i].text.indexOf('{code}') !== -1) {{
sel.selectedIndex = i;
var txt = sel.options[i].text;
if (typeof jQuery !== 'undefined') {{
try {{ jQuery(sel).selectpicker('val', sel.options[i].value); }} catch(e) {{}}
jQuery(sel).trigger('change');
}} else {{
sel.dispatchEvent(new Event('change', {{bubbles: true}}));
}}
return txt;
}}
}}
}}
return null;
""")
if result:
return result.strip()
btn = wait.until(EC.element_to_be_clickable(
(By.CSS_SELECTOR, "div.bootstrap-select > button.dropdown-toggle")
))
btn.click()
time.sleep(0.4)
try:
search = driver.find_element(By.CSS_SELECTOR, ".bootstrap-select .bs-searchbox input")
search.clear()
search.send_keys(code)
time.sleep(0.5)
except Exception:
pass
option_link = wait.until(EC.element_to_be_clickable((
By.XPATH,
f"//ul[contains(@class,'dropdown-menu inner')]"
f"//li[not(contains(@class,'hidden')) and not(contains(@class,'d-none'))]"
f"//span[contains(text(),'{code}')]/..",
)))
try:
text = option_link.find_element(By.CSS_SELECTOR, "span.text").text.strip()
except Exception:
text = code
option_link.click()
time.sleep(0.3)
return text
def fetch_aerodrome_catalog(headless: bool = True) -> list[dict]:
"""
Acessa o site ICEA e extrai a lista completa de aeródromos disponíveis.
Retorna lista de dicts com chaves 'icao', 'nome', 'uf'.
"""
driver = make_driver(headless=headless)
try:
driver.get(BASE_URL)
WebDriverWait(driver, 30).until(
EC.presence_of_element_located(
(By.CSS_SELECTOR, "select.selectpicker, select")
)
)
time.sleep(1.5) # aguarda Bootstrap Select renderizar todas as opções
raw_options = driver.execute_script("""
var candidates = document.querySelectorAll(
'select.selectpicker, select[name*="localidade"], select[name*="aerodrome"], select'
);
var result = [];
for (var s = 0; s < candidates.length; s++) {
var sel = candidates[s];
if (sel.options.length > 5) {
for (var i = 0; i < sel.options.length; i++) {
var txt = sel.options[i].text.trim();
if (txt) result.push(txt);
}
break;
}
}
return result;
""")
finally:
driver.quit()
catalog = []
for text in (raw_options or []):
if not text or len(text) < 4:
continue
icao = text[:4].upper()
if not re.match(r"[A-Z]{4}", icao):
continue
# formato típico: "SBGR - GUARULHOS / CUMBICA (SP)"
nome = text[4:].lstrip(" -–—").strip()
uf_match = re.search(r"\(([A-Z]{2})\)\s*$", nome)
uf = uf_match.group(1) if uf_match else ""
if uf_match:
nome = nome[: uf_match.start()].strip().rstrip("(),- ")
catalog.append({"icao": icao, "nome": nome, "uf": uf})
return catalog
def set_period(driver: webdriver.Chrome, start: date, end: date) -> None:
"""Sets the date-range picker on the ICEA portal using JavaScript.
Args:
driver: Active Chrome WebDriver instance on the ICEA portal page.
start: First day of the desired period.
end: Last day of the desired period.
"""
s = start.strftime("%d/%m/%Y")
e = end.strftime("%d/%m/%Y")
driver.execute_script(f"""
var el = document.getElementById('dtPeriodo');
if (!el) return;
if (typeof jQuery !== 'undefined' && jQuery(el).data('daterangepicker')) {{
jQuery(el).data('daterangepicker').setStartDate('{s}');
jQuery(el).data('daterangepicker').setEndDate('{e}');
}}
el.value = '{s} - {e}';
el.dispatchEvent(new Event('change', {{bubbles: true}}));
""")
# ---------------------------------------------------------------------------
# Extração de dados
# ---------------------------------------------------------------------------
def show_all_rows(driver: webdriver.Chrome, tab_id: str) -> None:
"""Expands the DataTable in *tab_id* to show all rows at once.
Tries the DataTables JS API first; falls back to the ``<select>`` length
control if jQuery/DataTables is not available.
Args:
driver: Active Chrome WebDriver instance.
tab_id: HTML ``id`` attribute of the tab ``<div>`` element.
"""
driver.execute_script(f"""
try {{
var dt = jQuery('#{tab_id} table').DataTable();
dt.page.len(-1).draw();
}} catch(e) {{
var sel = document.querySelector('#{tab_id} select[name$="_length"]');
if (sel) {{
var best = 0;
for (var i = 0; i < sel.options.length; i++) {{
var v = parseInt(sel.options[i].value);
if (v === -1 || v > parseInt(sel.options[best].value)) best = i;
}}
sel.selectedIndex = best;
sel.dispatchEvent(new Event('change', {{bubbles: true}}));
}}
}}
""")
time.sleep(1.2)
def parse_html_table(html: str) -> Optional[pd.DataFrame]:
"""Parses the first HTML ``<table>`` found in *html* into a DataFrame.
Args:
html: Raw HTML string containing a ``<table>`` element.
Returns:
DataFrame with headers from ``<thead>`` and rows from ``<tbody>``,
or ``None`` if no table or no data rows are found.
"""
soup = BeautifulSoup(html, "lxml")
table = soup.find("table")
if not table:
return None
headers = []
thead = table.find("thead")
if thead:
headers = [th.get_text(strip=True) for th in thead.find_all(["th", "td"])]
rows = []
tbody = table.find("tbody")
if tbody:
for tr in tbody.find_all("tr"):
cells = [td.get_text(strip=True) for td in tr.find_all(["th", "td"])]
if any(c.strip() for c in cells):
rows.append(cells)
if not rows:
return None
return pd.DataFrame(rows, columns=headers if headers else None)
def get_full_table(driver: webdriver.Chrome, tab_id: str) -> Optional[pd.DataFrame]:
"""Collects all DataTable pages from a tab and returns the concatenated DataFrame.
Handles pagination by iterating the "next" button until it is disabled.
Deduplicates rows that appear on multiple pages.
Args:
driver: Active Chrome WebDriver instance.
tab_id: HTML ``id`` of the tab ``<div>`` to read.
Returns:
Deduplicated DataFrame, or ``None`` if the tab contains no data rows.
"""
frames: list[pd.DataFrame] = []
seen: set[str] = set()
while True:
tab_elem = driver.find_element(By.ID, tab_id)
html = tab_elem.get_attribute("innerHTML")
fp = html[:800]
if fp in seen:
break
seen.add(fp)
df = parse_html_table(html)
if df is not None:
frames.append(df)
try:
next_btn = driver.find_element(
By.CSS_SELECTOR, f"#{tab_id} .dataTables_paginate .next"
)
if "disabled" in (next_btn.get_attribute("class") or ""):
break
driver.execute_script("arguments[0].click();", next_btn)
time.sleep(0.5)
except Exception:
break
if not frames:
return None
if len(frames) == 1:
return frames[0]
return pd.concat(frames, ignore_index=True).drop_duplicates()
def extract_tab(
driver: webdriver.Chrome,
wait: WebDriverWait,
tab_id: str,
) -> Optional[pd.DataFrame]:
"""Clicks the tab, waits for it to become active, and extracts its data.
Args:
driver: Active Chrome WebDriver instance.
wait: Configured :class:`WebDriverWait` bound to *driver*.
tab_id: HTML ``id`` of the target tab ``<div>``.
Returns:
DataFrame with the tab's data, or ``None`` if no rows are found.
"""
link = driver.find_element(By.CSS_SELECTOR, f"a[href='#{tab_id}']")
driver.execute_script("arguments[0].click();", link)
try:
wait.until(lambda d: "active" in (
d.find_element(By.ID, tab_id).get_attribute("class") or ""
))
except Exception:
time.sleep(1.5)
show_all_rows(driver, tab_id)
return get_full_table(driver, tab_id)
# ---------------------------------------------------------------------------
# Saída
# ---------------------------------------------------------------------------
def make_filename(tab_name: str, localidade: str, start: date, end: date) -> str:
"""Builds the standard output CSV filename for one tab and period.
Args:
tab_name: Display name of the meteorological tab (e.g. ``"Temperatura"``).
localidade: Aerodrome label returned by the site's selectbox.
start: First day of the scraped period.
end: Last day of the scraped period.
Returns:
Filename string without a directory component.
"""
safe = re.sub(r'[<>:"/\\|?*\r\n]', '', localidade).strip()
s = start.strftime("%d%m%Y")
e = end.strftime("%d%m%Y")
return f"Dados de Superfície {tab_name} - Localidade {safe} - Período {s} - {e}.csv"
def save_csv(df: pd.DataFrame, path: str) -> None:
"""Saves *df* to *path* as a UTF-8-BOM CSV with full quoting.
Args:
df: DataFrame to serialise.
path: Destination file path (created or overwritten).
"""
df.to_csv(path, index=False, encoding="utf-8-sig", quoting=csv.QUOTE_ALL)
# ---------------------------------------------------------------------------
# Lógica principal de scraping
# ---------------------------------------------------------------------------
def _quarter_chunks(year: int) -> list[tuple[date, date]]:
"""Divide o ano em chunks de CHUNK_MONTHS meses (do mais recente para o mais antigo)."""
chunks = []
month = 1
while month <= 12:
end_month = min(month + CHUNK_MONTHS - 1, 12)
last_day = calendar.monthrange(year, end_month)[1]
chunks.append((date(year, month, 1), date(year, end_month, last_day)))
month = end_month + 1
return list(reversed(chunks)) # Q4 → Q3 → Q2 → Q1
def scrape_period(
driver: webdriver.Chrome,
wait: WebDriverWait,
aerodrome: str,
start: date,
end: date,
output_dir: str,
log: Callable[[str], None] = print,
) -> bool:
"""
Faz o scraping de um único período e salva os CSVs das 9 abas.
Retorna True se pelo menos uma aba teve dados, False se tudo veio vazio.
"""
log(f" [{start.strftime('%d/%m/%Y')} -> {end.strftime('%d/%m/%Y')}]")
driver.get(BASE_URL)
wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, "div.bootstrap-select")))
fetch_site_max_date(driver) # atualiza SITE_MAX_DATE com o valor real do site
time.sleep(1.5)
localidade = select_aerodrome(driver, wait, aerodrome)
if not localidade:
log(f" ERR Aerodrome '{aerodrome}' nao encontrado.")
return False
log(f" Localidade: {localidade}")
set_period(driver, start, end)
submit = wait.until(EC.element_to_be_clickable((By.CSS_SELECTOR, "button[type='submit']")))
submit.click()
wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, ".nav-tabs")))
time.sleep(3)
found_any = False
for tab_id, tab_name in TABS:
try:
df = extract_tab(driver, wait, tab_id)
if df is not None and not df.empty:
found_any = True
fname = make_filename(tab_name, localidade, start, end)
fpath = os.path.join(output_dir, fname)
save_csv(df, fpath)
log(f" OK {tab_name:<15s} {len(df):>6} linhas -> {fname}")
else:
log(f" -- {tab_name:<15s} sem dados")
except Exception as exc:
log(f" ERR {tab_name:<15s} {exc}")
return found_any
def scrape_year(
driver: webdriver.Chrome,
wait: WebDriverWait,
aerodrome: str,
year: int,
output_dir: str,
stop_before: Optional[date] = None,
log: Callable[[str], None] = print,
) -> int:
"""
Itera os chunks trimestrais do ano (do mais recente para o mais antigo).
Para se o trimestre for anterior a stop_before (data já coberta).
Retorna o número de trimestres que tiveram dados.
"""
non_empty = 0
for chunk_start, chunk_end in _quarter_chunks(year):
start = max(chunk_start, SITE_MIN_DATE)
end = min(chunk_end, SITE_MAX_DATE)
if start > end:
continue
# Para quando atingir dados já cobertos pelo arquivo existente
if stop_before is not None and end <= stop_before:
log(f" [trimestre {start.strftime('%d/%m/%Y')} -> {end.strftime('%d/%m/%Y')} já coberto, parando]")
break
had_data = scrape_period(driver, wait, aerodrome, start, end, output_dir, log)
if had_data:
non_empty += 1
return non_empty
# ---------------------------------------------------------------------------
# Entry point (CLI standalone)
# ---------------------------------------------------------------------------
def _fix_stdout() -> None:
"""Re-wraps stdout/stderr to UTF-8 on Windows terminals."""
try:
if hasattr(sys.stdout, "buffer"):
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
if hasattr(sys.stderr, "buffer"):
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")
except Exception:
pass
def main() -> None:
_fix_stdout()
parser = argparse.ArgumentParser(
description="Scraper de meteorologia de superfície — ICEA/DECEA",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
exemplos:
python scraper_meteorologia.py --aerodrome SBGR --all-years
python scraper_meteorologia.py --aerodrome SBGR --start-year 2020 --end-year 2025
python scraper_meteorologia.py --aerodrome SBSP --start-year 2023 --end-year 2023 --no-headless
""",
)
parser.add_argument("--aerodrome", default="SBGR", metavar="ICAO",
help="Código ICAO do aeródromo (padrão: SBGR)")
parser.add_argument("--all-years", action="store_true",
help="Coleta todos os anos disponíveis (para automaticamente sem dados)")
parser.add_argument("--start-year", type=int, metavar="ANO",
help="Ano inicial (obrigatório sem --all-years)")
parser.add_argument("--end-year", type=int, metavar="ANO",
help="Ano final (obrigatório sem --all-years)")
parser.add_argument("--output-dir", default=str(DADOS_DIR), metavar="DIR",
help="Output directory for temporary quarterly CSVs")
parser.add_argument("--no-headless", action="store_true",
help="Open Chrome in visible mode (useful for debugging)")
parser.add_argument("--fetch-catalog", action="store_true",
help="Download aerodrome list from the ICEA site and save to SQLite")
parser.add_argument("--db-path", default=str(DB_PATH), metavar="PATH",
help="Path to the SQLite database (used with --fetch-catalog)")
args = parser.parse_args()
# ── fetch-catalog: ação independente ──────────────────────────────────────
if args.fetch_catalog:
from pathlib import Path
import db as _db
print("Acessando site ICEA para obter lista de aeródromos…")
catalog = fetch_aerodrome_catalog(headless=not args.no_headless)
print(f" {len(catalog)} aeródromos encontrados.")
db_path = Path(args.db_path)
db_path.parent.mkdir(parents=True, exist_ok=True)
conn = _db.get_connection(db_path)
_db.ensure_schema(conn)
n = _db.upsert_aerodromes(conn, catalog)
conn.close()
print(f" {n} registros gravados em '{db_path}'.")
return
if not args.all_years and (args.start_year is None or args.end_year is None):
parser.error("Use --all-years ou informe --start-year e --end-year")
end_year = args.end_year if not args.all_years else date.today().year
start_year = args.start_year if not args.all_years else SITE_MIN_DATE.year
Path(args.output_dir).mkdir(parents=True, exist_ok=True)
driver = make_driver(headless=not args.no_headless)
wait = WebDriverWait(driver, 60)
try:
consecutive_empty = 0
for year in range(end_year, start_year - 1, -1): # do mais recente para o mais antigo
print(f"\n{'=' * 60}")
print(f"Ano {year} — {args.aerodrome}")
print(f"{'=' * 60}")
non_empty = scrape_year(driver, wait, args.aerodrome, year, args.output_dir)
if non_empty == 0:
consecutive_empty += 1
print(f" [sem dados] ({consecutive_empty}/{STOP_EMPTY_YEARS} consecutivos)")
if args.all_years and consecutive_empty >= STOP_EMPTY_YEARS:
print(f"\n{STOP_EMPTY_YEARS} anos consecutivos sem dados — parando busca.")
break
else:
consecutive_empty = 0
except KeyboardInterrupt:
print("\nInterrompido pelo usuário.")
finally:
driver.quit()
print("\nFinalizado.")
if __name__ == "__main__":
main()