104 lines
4.2 KiB
Python
104 lines
4.2 KiB
Python
"""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()
|