Files
arara_oarmp/software/oamrp_v1.py
2026-06-15 20:22:57 -03:00

272 lines
12 KiB
Python

"""
============================================================================
AIRCRAFT ROUTING - ESQUADRAO ARARA (Transporte C-95/C-105)
============================================================================
Modelo MIP compacto que ATRIBUI aeronaves a missoes de transporte,
maximizando o numero (ponderado por prioridade) de missoes cumpridas,
respeitando:
- rota viavel (continuidade espacial e temporal por aeronave)
- turnaround minimo entre missoes
- limite de horas de celula ate a proxima inspecao (LRT)
- base inicial e horas acumuladas reais de cada aeronave
Diferenca chave vs. modelo civil (Bazargan): aqui a cobertura e' "<= 1"
(missao OPCIONAL, selecionada se compensar) e nao "= 1". O objetivo e'
prontidao/cumprimento de missao, nao custo.
Solver: CBC (vem junto com o PuLP, zero instalacao). Para instancias
maiores troque por HiGHS (ver nota no fim do arquivo).
Autor: gerado como ponto de partida. pt-BR nos comentarios, codigo limpo.
============================================================================
"""
import csv
from datetime import datetime, timezone
from pathlib import Path
import pulp
# ===========================================================================
# 1) DADOS DE ENTRADA (substitua pelos dados reais do ARARA)
# ===========================================================================
# Bases (codigos curtos ficticios). 'S' marca bases HABILITADAS a inspecao.
# Na FAB o conjunto de bases habilitadas e' restrito -> isso aperta o routing.
# MN representa Manaus, base operacional do Esquadrao Arara nesta simulacao.
BASES_INSPECAO = {"MN"} # Manaus
TURNAROUND_MIN_H = 1.0 # tempo minimo em solo entre 2 missoes (h)
LIMITE_HORAS_F = 100.0 # horas de celula ate inspecao obrigatoria
# (AJUSTE ao intervalo real da sua frota)
# ----- Aeronaves (tail number, base inicial, horas acumuladas desde a ult. insp.)
# 'horas_acum' = f_k. LRT inicial de cada aeronave = LIMITE_HORAS_F - f_k.
AERONAVES = [
# id base horas_acum
("FAB-2301", "MN", 12.0),
("FAB-2302", "MN", 71.0), # ja gastou muito -> pouco LRT disponivel
("FAB-2303", "MN", 30.0),
("FAB-2304", "MN", 95.0), # quase no limite -> mal consegue voar
]
# ----- Missoes de transporte
# Entrada recomendada: data/hora UTC, como viria de uma planilha.
# Para usar uma planilha, exporte para CSV com as colunas:
# id,orig,dest,partida_utc,chegada_utc,prioridade
# e preencha CSV_MISSOES_PATH com o caminho do arquivo.
CSV_MISSOES_PATH = None
MISSOES_UTC = [
# id orig dest partida UTC chegada UTC prio
("M01", "MN", "BR", "2026-07-01T06:00Z", "2026-07-01T10:00Z", 5),
("M02", "BR", "MN", "2026-07-01T12:00Z", "2026-07-01T16:00Z", 5),
("M03", "MN", "BE", "2026-07-01T06:30Z", "2026-07-01T09:00Z", 4),
("M04", "BE", "MN", "2026-07-01T11:00Z", "2026-07-01T13:30Z", 4),
("M05", "MN", "PV", "2026-07-01T07:00Z", "2026-07-01T09:00Z", 3),
("M06", "PV", "MN", "2026-07-01T10:30Z", "2026-07-01T12:30Z", 3),
("M07", "MN", "BV", "2026-07-01T08:00Z", "2026-07-01T09:30Z", 3),
("M08", "BV", "MN", "2026-07-01T11:00Z", "2026-07-01T12:30Z", 3),
("M09", "MN", "GL", "2026-07-01T18:00Z", "2026-07-01T23:00Z", 4),
("M10", "GL", "MN", "2026-07-02T01:00Z", "2026-07-02T06:00Z", 4),
("M11", "MN", "CG", "2026-07-01T20:00Z", "2026-07-01T23:00Z", 2),
("M12", "CG", "MN", "2026-07-02T01:00Z", "2026-07-02T04:00Z", 2),
("M13", "MN", "RF", "2026-07-02T06:00Z", "2026-07-02T11:00Z", 4),
("M14", "RF", "MN", "2026-07-02T13:00Z", "2026-07-02T18:00Z", 4),
("M15", "MN", "NT", "2026-07-02T08:00Z", "2026-07-02T12:30Z", 3),
("M16", "NT", "MN", "2026-07-02T14:00Z", "2026-07-02T18:30Z", 3),
("M17", "MN", "BR", "2026-07-03T00:00Z", "2026-07-03T04:00Z", 5),
("M18", "BR", "MN", "2026-07-03T06:00Z", "2026-07-03T10:00Z", 5),
("M19", "MN", "BE", "2026-07-03T02:00Z", "2026-07-03T04:30Z", 4),
("M20", "BE", "MN", "2026-07-03T06:00Z", "2026-07-03T08:30Z", 4),
]
def parse_utc(valor):
"""Converte texto ISO UTC da planilha para datetime com timezone."""
texto = str(valor).strip()
if texto.endswith("Z"):
texto = texto[:-1] + "+00:00"
dt = datetime.fromisoformat(texto)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.astimezone(timezone.utc)
def carregar_missoes_csv(caminho):
with Path(caminho).open(newline="", encoding="utf-8-sig") as arq:
leitor = csv.DictReader(arq)
return [
(
linha["id"],
linha["orig"],
linha["dest"],
linha["partida_utc"],
linha["chegada_utc"],
int(linha["prioridade"]),
)
for linha in leitor
]
def converter_missoes_para_horas(missoes_utc):
datas_partida = [parse_utc(m[3]) for m in missoes_utc]
horizonte_inicio = min(datas_partida)
missoes_horas = []
for mid, orig, dest, partida_utc, chegada_utc, prio in missoes_utc:
partida = parse_utc(partida_utc)
chegada = parse_utc(chegada_utc)
if chegada <= partida:
raise ValueError(f"Missao {mid}: chegada_utc deve ser depois de partida_utc")
part_h = (partida - horizonte_inicio).total_seconds() / 3600
cheg_h = (chegada - horizonte_inicio).total_seconds() / 3600
missoes_horas.append((mid, orig, dest, part_h, cheg_h, prio))
return missoes_horas, horizonte_inicio
MISSOES_FONTE = carregar_missoes_csv(CSV_MISSOES_PATH) if CSV_MISSOES_PATH else MISSOES_UTC
MISSOES, HORIZONTE_INICIO_UTC = converter_missoes_para_horas(MISSOES_FONTE)
# ===========================================================================
# 2) PRE-PROCESSAMENTO (gera conexoes viaveis)
# ===========================================================================
miss = {m[0]: dict(orig=m[1], dest=m[2], part=m[3], cheg=m[4], prio=m[5],
dur=m[4] - m[3]) for m in MISSOES}
acft = {a[0]: dict(base=a[1], f=a[2]) for a in AERONAVES}
IDS = list(miss.keys())
KS = list(acft.keys())
# Conexao i->j valida: destino de i == origem de j e ha turnaround suficiente
conex = [(i, j) for i in IDS for j in IDS
if i != j
and miss[i]["dest"] == miss[j]["orig"]
and miss[j]["part"] >= miss[i]["cheg"] + TURNAROUND_MIN_H]
# Aeronave k pode INICIAR pela missao i se i parte da base inicial de k
def pode_iniciar(k, i):
return miss[i]["orig"] == acft[k]["base"]
# ===========================================================================
# 3) MODELO MIP
# ===========================================================================
mdl = pulp.LpProblem("ARARA_Aircraft_Routing", pulp.LpMaximize)
# --- Variaveis
x = {(k, i): pulp.LpVariable(f"x_{k}_{i}", cat="Binary") for k in KS for i in IDS}
s = {(k, i): pulp.LpVariable(f"s_{k}_{i}", cat="Binary") # k inicia por i
for k in KS for i in IDS if pode_iniciar(k, i)}
c = {(k, i, j): pulp.LpVariable(f"c_{k}_{i}_{j}", cat="Binary") # k voa i depois j
for k in KS for (i, j) in conex}
z = {i: pulp.LpVariable(f"z_{i}", cat="Binary") for i in IDS} # missao coberta
# --- Objetivo primario: maximizar prioridade total das missoes cumpridas
mdl += pulp.lpSum(miss[i]["prio"] * z[i] for i in IDS), "Missoes_cumpridas_ponderadas"
# --- (R1) Cobertura: cada missao por NO MAXIMO uma aeronave
for i in IDS:
mdl += z[i] == pulp.lpSum(x[(k, i)] for k in KS), f"def_z_{i}"
# z[i] ja e' binaria, entao automaticamente <= 1
# --- (R2) Fluxo de entrada: se k voa i, ele iniciou por i OU chegou de outra missao
for k in KS:
for i in IDS:
entradas = pulp.lpSum(c[(k, jj, i)] for (jj, ii) in conex if ii == i)
ini = s[(k, i)] if (k, i) in s else 0
mdl += ini + entradas == x[(k, i)], f"in_{k}_{i}"
# --- (R3) Fluxo de saida: depois de i, k segue para no max uma proxima missao
for k in KS:
for i in IDS:
saidas = pulp.lpSum(c[(k, i, jj)] for (ii, jj) in conex if ii == i)
mdl += saidas <= x[(k, i)], f"out_{k}_{i}"
# --- (R4) Cada aeronave inicia no maximo uma rota
for k in KS:
inis = [s[(k, i)] for i in IDS if (k, i) in s]
if inis:
mdl += pulp.lpSum(inis) <= 1, f"start_{k}"
# --- (R5) Limite de horas de celula (LRT): total voado por k <= F - f_k
# Esta e' a restricao CENTRAL de manutencao (versao "orcamento de horas").
# -> A inspecao com RESET de horas em base habilitada e' a extensao da Fase 3.
for k in KS:
mdl += pulp.lpSum(miss[i]["dur"] * x[(k, i)] for i in IDS) <= \
LIMITE_HORAS_F - acft[k]["f"], f"horas_{k}"
# ===========================================================================
# 4) RESOLVER
# ===========================================================================
solver = pulp.PULP_CBC_CMD(msg=False) # troque por HiGHS se instalado
# Etapa 1: maximiza o cumprimento ponderado das missoes.
mdl.solve(solver)
prio_otima = int(pulp.value(mdl.objective))
# Etapa 2: mantendo a prioridade otima, minimiza o LRT restante ponderado.
# Como as horas totais voadas podem ser constantes quando todas as missoes sao
# cumpridas, o peso prioriza consumir LRT das aeronaves mais proximas da
# manutencao, sem sacrificar nenhuma missao prioritaria da primeira etapa.
lrt_restante = {
k: (LIMITE_HORAS_F - acft[k]["f"]) -
pulp.lpSum(miss[i]["dur"] * x[(k, i)] for i in IDS)
for k in KS
}
pesos_lrt = {k: 1 / (LIMITE_HORAS_F - acft[k]["f"]) for k in KS}
mdl += pulp.lpSum(miss[i]["prio"] * z[i] for i in IDS) == prio_otima, "fixa_prioridade_otima"
mdl.sense = pulp.LpMinimize
mdl.setObjective(pulp.lpSum(pesos_lrt[k] * lrt_restante[k] for k in KS))
mdl.solve(solver)
# ===========================================================================
# 5) RELATORIO
# ===========================================================================
print("=" * 70)
print(f" STATUS: {pulp.LpStatus[mdl.status]}")
print(f" Horizonte inicia em UTC: {HORIZONTE_INICIO_UTC.isoformat().replace('+00:00', 'Z')}")
prio_tot = sum(miss[i]["prio"] for i in IDS)
prio_obt = prio_otima
cob = [i for i in IDS if z[i].value() and z[i].value() > 0.5]
print(f" Missoes cumpridas: {len(cob)}/{len(IDS)} "
f"| Prioridade obtida: {prio_obt}/{prio_tot}")
lrt_total = sum((LIMITE_HORAS_F - acft[k]["f"]) -
sum(miss[i]["dur"] * x[(k, i)].value() for i in IDS)
for k in KS)
print(f" LRT restante total: {lrt_total:g}h")
print(" Criterio secundario: menor LRT restante ponderado p/ manutencao")
print("=" * 70)
def rota_da_aeronave(k):
"""Reconstroi a sequencia de missoes da aeronave k seguindo as conexoes."""
inicio = [i for i in IDS if (k, i) in s and s[(k, i)].value() and s[(k, i)].value() > 0.5]
if not inicio:
return []
seq = [inicio[0]]
while True:
atual = seq[-1]
prox = [j for (ii, jj) in conex if ii == atual
for j in [jj] if c[(k, atual, j)].value() and c[(k, atual, j)].value() > 0.5]
if not prox:
break
seq.append(prox[0])
return seq
for k in KS:
seq = rota_da_aeronave(k)
horas = sum(miss[i]["dur"] for i in seq)
lrt_ini = LIMITE_HORAS_F - acft[k]["f"]
if seq:
trechos = " -> ".join(
f"{i}({miss[i]['orig']}-{miss[i]['dest']} {miss[i]['part']:g}h)" for i in seq)
print(f"\n {k} [base {acft[k]['base']}, f={acft[k]['f']:g}h, LRT={lrt_ini:g}h]")
print(f" {trechos}")
print(f" horas voadas: {horas:g}h | LRT restante: {lrt_ini - horas:g}h")
else:
print(f"\n {k} [base {acft[k]['base']}, LRT={lrt_ini:g}h] -> sem missoes")
descobertas = [i for i in IDS if i not in cob]
if descobertas:
print("\n " + "-" * 60)
print(" Missoes NAO cumpridas:",
", ".join(f"{i}(prio {miss[i]['prio']})" for i in descobertas))
# ---------------------------------------------------------------------------
# NOTA - usar HiGHS (recomendado p/ instancias maiores, sem licenca):
# pip install highspy
# solver = pulp.HiGHS(msg=False) # PuLP 3.x expoe HiGHS nativo
# ---------------------------------------------------------------------------