""" Concatena os CSVs de dados de superfície (por aba/trimestre) numa única tabela por aeródromo. Inclui validação amostral e limpeza opcional dos arquivos intermediários. Uso standalone: python concat_meteorologia.py [--dados-dir dados] [--validate] [--n-samples 20] [--cleanup] Importável pelo 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 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: 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): 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]: 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: str = "dados", do_validate: bool = True, n_samples: int = 20, do_cleanup: bool = False, log: Callable[[str], None] = print, ) -> Optional[pd.DataFrame]: 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}") # Verifica se já existe arquivo compilado (para append incremental) compiled_dir = base / aero extra_df: Optional[pd.DataFrame] = None existing = sorted(compiled_dir.glob(f"{aero}_*.csv")) if compiled_dir.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) compiled_dir.mkdir(exist_ok=True) out_name = f"{aero}_{start_str}_{end_str}.csv" out_path = compiled_dir / 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}") # Remove arquivos compilados anteriores (com datas diferentes) 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="Concatena CSVs de superfície por aeródromo") ap.add_argument("--dados-dir", default="dados", metavar="DIR") 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 CSVs trimestrais após compilação validada") ap.add_argument("--db", action="store_true", help="Salva analytics no SQLite (dados/met.db) além do CSV") args = ap.parse_args() result_df = main( dados_dir=args.dados_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 pathlib import Path db_path = Path(args.dados_dir) / "met.db" conn = _db.get_connection(db_path) _db.ensure_schema(conn) # Detecta aeródromo pelo nome da primeira coluna analítica presente # (reutiliza build_analytics do pipeline para converter) from pipeline import build_analytics anl = build_analytics(result_df) # Tenta inferir aerodrome dos arquivos presentes import re as _re _RE = _re.compile(r"Dados de Superfície .+ - Localidade (\w+) -") aeros = set() for f in Path(args.dados_dir).glob("Dados de Superfície*.csv"): m = _RE.match(f.name) if m: aeros.add(m.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}")