"""Orquestrador principal do Planejador Missão. Entradas: Arquivos em dados/, argumentos opcionais de linha de comando e configuracoes centralizadas em scripts/01_config.py. Saídas: Planejamento diário em resultados/, histórico operacional atualizado, relatórios de validação quando solicitado e logs em logs/execucao.log. Papel no pipeline: Executa o fluxo completo: leitura de dados, preparacao, geracao de colunas candidatas, montagem/solução do MILP, exportação de resultados e validação. Exemplos: python scripts/00_main.py --data 2026-01-02 --modo diario python scripts/00_main.py --validacao 2025 """ from __future__ import annotations import argparse import importlib import time import pandas as pd config = importlib.import_module("01_config") io_utils = importlib.import_module("02_io_utils") def atualizar_data_planejamento(data: str) -> None: """Atualiza dados/parametros_missao.csv com a data solicitada.""" path = config.ARQUIVO_PARAMETROS df = pd.read_csv(path) if path.exists() else pd.DataFrame(columns=["parametro", "valor"]) if "parametro" not in df.columns or "valor" not in df.columns: raise ValueError("Erro: parametros_missao.csv deve conter as colunas parametro,valor.") if (df["parametro"] == "data_planejamento").any(): df.loc[df["parametro"] == "data_planejamento", "valor"] = data else: df = pd.concat([df, pd.DataFrame([{"parametro": "data_planejamento", "valor": data}])], ignore_index=True) df.to_csv(path, index=False) def executar_diario(args: argparse.Namespace, logger) -> dict: """Executa o planejamento diário oficial em Python.""" io_utils.configurar_ambiente() if args.data: atualizar_data_planejamento(args.data) logger.info("Data de planejamento ajustada para %s", args.data) if args.aeronave: logger.info("Filtro --aeronave recebido: %s. O modelo diario usa a disponibilidade em dados/aeronaves_disponiveis.csv.", args.aeronave) from src.planejador_missao.main import executar_planejamento inicio = time.perf_counter() resultado = executar_planejamento(config.PROJECT_ROOT) duracao = time.perf_counter() - inicio logger.info("Data planejada: %s", resultado["data_planejamento"]) logger.info("Colunas candidatas: %s", resultado["total_candidatas"]) logger.info("Escalas selecionadas: %s", resultado["total_selecionadas"]) logger.info("Arquivo exportado: %s", resultado["arquivo_saida"]) logger.info("Tempo de execução: %.1f s", duracao) return resultado def executar_validacao(ano: str, logger) -> dict: """Executa validação retrospectiva suportada pelo projeto.""" if str(ano) != "2025": raise ValueError("Erro: no momento a validação disponível é --validacao 2025.") validacao = importlib.import_module("07_validacao_2025") return validacao.executar_validacao_2025(importar=True, logger=logger) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Executa o Planejador Missão em Python.") parser.add_argument("--data", help="Data do planejamento diário no formato AAAA-MM-DD.") parser.add_argument("--aeronave", choices=config.AERONAVES, help="Aeronave de referência para diagnóstico.") parser.add_argument("--modo", default="diario", choices=["diario"], help="Modo de execução do planejamento.") parser.add_argument("--validacao", help="Ano da validação retrospectiva. Exemplo: --validacao 2025.") return parser.parse_args() def main() -> None: args = parse_args() logger = io_utils.configurar_logger("main") if args.validacao: saida = executar_validacao(args.validacao, logger) print("\n=== Validação concluída ===") for path in saida["arquivos"]: print(f"- {path}") return resultado = executar_diario(args, logger) print("\n=== Planejamento diário concluído ===") print(f"Data: {resultado['data_planejamento']}") print(f"Arquivo: {resultado['arquivo_saida']}") print(f"Colunas candidatas: {resultado['total_candidatas']}") print(f"Escalas selecionadas: {resultado['total_selecionadas']}") if __name__ == "__main__": main()