7cd7458565
- Gerador sintetico cross-base (50 missoes, 10 bases, haversine/C-105) - Fase 3: arcos de ferry em construir_rede(); ferry consome horas de celula - oamrp_v2.py e resultados CSV anteriores adicionados ao repositorio - README, docs/about.md e changelog atualizados ate v0.6 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
1393 lines
49 KiB
Python
1393 lines
49 KiB
Python
"""
|
||
============================================================================
|
||
AIRCRAFT ROUTING - ESQUADRAO ARARA - V3
|
||
============================================================================
|
||
Modelo OAMRP compacto (Al-Thani et al., 2016) em rede espaço-tempo.
|
||
NAO usa Set Partitioning / Column Generation.
|
||
|
||
Arquitetura:
|
||
A rede é composta por Nos (origem, missao, inspecao, sumidouro) e Arcos
|
||
tipados. A variavel de decisao y[k, arco] unifica o que era x/s/c no v2.
|
||
Para adicionar inspeções na Fase 2, basta estender construir_rede() —
|
||
o modelo de otimização e a conservação de fluxo não mudam.
|
||
|
||
Escopo desta versão — Fase 2 (C1–C9, C11):
|
||
C1 Cobertura: cada missão é servida por no máximo 1 aeronave (EVAM = hard)
|
||
C2 Conservação de fluxo por nó e por aeronave
|
||
C3 Continuidade espacial (dest(m) == orig(m') no arco)
|
||
C4 TAT = 1,5 h entre chegada e próxima decolagem
|
||
C5 Janela de decolagem derivada do CSV
|
||
C6 Orçamento de horas por relógio de inspeção com reset (H_{k,i,n} ≤ LRT_i)
|
||
C9 Downtime: próxima missão só começa após s_{k,i} + DT_i + TAT
|
||
C11 Slot único: no máximo 1 aeronave em inspeção em qualquer instante
|
||
|
||
Fases futuras:
|
||
Fase 3 — arcos de ferry, C7 (limite calendárico), objetivo L2 real
|
||
Fase 4 — replanejamento e visualização
|
||
|
||
Inspeções válidas (Q1 confirmado):
|
||
seq {2, 3, 4, 6, 7, 19, 20, 22}
|
||
Removidas: seq 18 (INSP 100FH) e seq 23 (INSP 50FH)
|
||
|
||
Horas iniciais f0_{k,i}:
|
||
--horas-iniciais <arquivo.json ou string JSON>
|
||
--aleatorio [--seed N]
|
||
padrão: 0,0 para todas (aeronave recém-inspecionada — conservador)
|
||
============================================================================
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import csv
|
||
import json
|
||
import random
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Literal, Optional
|
||
|
||
import pulp
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Caminhos padrão
|
||
# ---------------------------------------------------------------------------
|
||
|
||
BASE_DIR = Path(__file__).resolve().parents[1]
|
||
|
||
CSV_REGISTRO_PADRAO = (
|
||
BASE_DIR / "db" / "processed" / "registro_voo_2025_consolidado.csv"
|
||
)
|
||
CSV_RESULTADO_PADRAO = (
|
||
BASE_DIR / "db" / "processed" / "resultado_oamrp_v3.csv"
|
||
)
|
||
JSON_INSPECOES_PADRAO = (
|
||
BASE_DIR
|
||
/ "db"
|
||
/ "pre_process"
|
||
/ "relatorio_ciclo_inspecoes_c105_2805_2026-06-15"
|
||
/ "relatorio_ciclo_inspecoes_c105_2805_2026-06-15_inspecoes.json"
|
||
)
|
||
|
||
DATA_INICIO_PADRAO = "2025-01-01"
|
||
DATA_FIM_PADRAO = "2025-02-01"
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Constantes operacionais
|
||
# ---------------------------------------------------------------------------
|
||
|
||
BASE_MANUTENCAO = "SBMN"
|
||
TAT_H = 1.5
|
||
|
||
AERONAVES_PADRAO = ["2800", "2803", "2809", "2811"]
|
||
BASE_INICIAL_PADRAO = BASE_MANUTENCAO
|
||
|
||
SEQS_VALIDOS = {2, 3, 4, 6, 7, 19, 20, 22} # Q1 confirmado
|
||
NIVEIS_PERMITIDOS = {"Base", "Orgânico"}
|
||
|
||
N_BASES_PADRAO = 10 # Q4: top-N bases por frequência de uso
|
||
HARD_MISSAO_PADRAO = "EVAM" # Q2: padrão para identificar missões obrigatórias
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Domínio
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Missao:
|
||
id: str
|
||
orig: str
|
||
dest: str
|
||
t_dep: float # horas desde o início do horizonte
|
||
t_arr: float
|
||
dur_h: float
|
||
prioridade: int
|
||
obrigatoria: bool # True = M_obr (EVAM/hard); cobertura == 1
|
||
partida_utc: datetime
|
||
chegada_utc: datetime
|
||
periodo: str
|
||
linhas_origem: str
|
||
aeronave_real: str
|
||
om: str
|
||
codigo_missao: str # campo 'missao' do CSV (ex: "69TV01", "EVAM001")
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class InspecaoParam:
|
||
seq: int
|
||
sigla: str
|
||
descricao: str
|
||
nivel: str
|
||
lrt_h: float
|
||
cal_dias: Optional[float]
|
||
dt_h: float
|
||
|
||
|
||
@dataclass
|
||
class Aeronave:
|
||
id: str
|
||
base_inicial: str
|
||
horas_iniciais: dict[int, float] = field(default_factory=dict)
|
||
|
||
def orcamento_h(self, catalogo: dict[int, InspecaoParam]) -> float:
|
||
"""
|
||
Orçamento de horas disponíveis (C6 — Fase 1 simplificado).
|
||
Retorna min(LRT_i − f0_{k,i}) sobre todas as inspeções ativas.
|
||
"""
|
||
if not catalogo:
|
||
return 0.0
|
||
margens = [
|
||
insp.lrt_h - self.horas_iniciais.get(seq, 0.0)
|
||
for seq, insp in catalogo.items()
|
||
]
|
||
return max(0.0, min(margens))
|
||
|
||
def insp_mais_restritiva(
|
||
self, catalogo: dict[int, InspecaoParam]
|
||
) -> Optional[tuple[int, float, InspecaoParam]]:
|
||
if not catalogo:
|
||
return None
|
||
seq_min = min(
|
||
catalogo,
|
||
key=lambda s: catalogo[s].lrt_h - self.horas_iniciais.get(s, 0.0),
|
||
)
|
||
return seq_min, self.horas_iniciais.get(seq_min, 0.0), catalogo[seq_min]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Rede espaço-tempo
|
||
# ---------------------------------------------------------------------------
|
||
|
||
TipoNo = Literal["origem", "missao", "inspecao", "sumidouro"]
|
||
TipoArco = Literal[
|
||
"inicio", # origem_k → missao
|
||
"missao_missao", # missao → missao
|
||
"missao_insp", # missao → inspecao (Fase 2)
|
||
"insp_missao", # inspecao → missao (Fase 2)
|
||
"ferry", # qualquer → qualquer com reposicionamento (Fase 3)
|
||
"fim", # missao → sumidouro_k
|
||
]
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class No:
|
||
"""
|
||
Nó genérico da rede espaço-tempo.
|
||
|
||
base_in — base onde a aeronave deve estar ao *entrar* neste nó.
|
||
base_out — base onde a aeronave estará ao *sair* deste nó.
|
||
t_ini — instante de início (decolagem para missão; início downtime para inspeção).
|
||
t_fim — instante de fim (chegada para missão; fim downtime para inspeção).
|
||
carga_h — horas de célula consumidas ao atravessar este nó.
|
||
|
||
Nós de inspeção (Fase 2) terão aeronave_id preenchido — são nós por aeronave.
|
||
Nós de missão são compartilhados entre aeronaves.
|
||
"""
|
||
|
||
id: str
|
||
tipo: TipoNo
|
||
base_in: str
|
||
base_out: str
|
||
t_ini: float
|
||
t_fim: float
|
||
carga_h: float = 0.0
|
||
missao: Optional[Missao] = None
|
||
insp_seq: Optional[int] = None
|
||
aeronave_id: Optional[str] = None
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Arco:
|
||
"""
|
||
Arco da rede espaço-tempo.
|
||
|
||
aeronave_id — se preenchido, o arco pertence exclusivamente a esta aeronave
|
||
(arcos de início e fim). None indica arco compartilhado
|
||
(missao_missao, ferry), utilizável por qualquer aeronave.
|
||
|
||
Restrições de viabilidade (continuidade espacial e temporal) são
|
||
verificadas em construir_rede() — não são restrições do modelo PuLP.
|
||
"""
|
||
|
||
id: str
|
||
orig_id: str
|
||
dest_id: str
|
||
tipo: TipoArco
|
||
aeronave_id: Optional[str] = None # None = arco compartilhado
|
||
horas_ferry: float = 0.0 # horas de célula de ferry (Fase 3)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Construção da rede (Fase 1 — sem nós de inspeção)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def construir_rede(
|
||
missoes: list[Missao],
|
||
aeronaves: list[Aeronave],
|
||
catalogo: dict[int, InspecaoParam],
|
||
tat_h: float = TAT_H,
|
||
) -> tuple[dict[str, No], list[Arco]]:
|
||
"""
|
||
Constrói os nós e arcos da rede espaço-tempo (Fase 2).
|
||
|
||
Tipos de nós:
|
||
- origem_k : ponto de partida de cada aeronave k
|
||
- missao_m : cada missão candidata
|
||
- inspecao_k_i : nó de inspeção por (aeronave k, tipo i) — base SBMN
|
||
- sumidouro_k : ponto de chegada de cada aeronave k
|
||
|
||
Tipos de arcos:
|
||
- inicio : origem_k → missao_m (com ou sem ferry inicial)
|
||
- missao_missao : missao_m → missao_m' (dest==orig e gap ≥ TAT)
|
||
- ferry : missao_m → missao_m' (dest≠orig, gap ≥ TAT + tempo_ferry)
|
||
- missao_insp : missao_m → inspecao_k_i (m.dest == SBMN, ou com ferry)
|
||
- insp_missao : inspecao_k_i → missao_m' (m'.orig == SBMN ou com ferry)
|
||
- fim : missao_m / inspecao_k_i → sumidouro_k
|
||
"""
|
||
nos: dict[str, No] = {}
|
||
arcos: list[Arco] = []
|
||
|
||
# --- Nós de missão ---
|
||
for m in missoes:
|
||
no_id = f"M_{m.id}"
|
||
nos[no_id] = No(
|
||
id=no_id,
|
||
tipo="missao",
|
||
base_in=m.orig,
|
||
base_out=m.dest,
|
||
t_ini=m.t_dep,
|
||
t_fim=m.t_arr,
|
||
carga_h=m.dur_h,
|
||
missao=m,
|
||
)
|
||
|
||
# --- Nós de origem e sumidouro por aeronave ---
|
||
for a in aeronaves:
|
||
src_id = f"SRC_{a.id}"
|
||
snk_id = f"SNK_{a.id}"
|
||
nos[src_id] = No(
|
||
id=src_id,
|
||
tipo="origem",
|
||
base_in=a.base_inicial,
|
||
base_out=a.base_inicial,
|
||
t_ini=0.0,
|
||
t_fim=0.0,
|
||
aeronave_id=a.id,
|
||
)
|
||
nos[snk_id] = No(
|
||
id=snk_id,
|
||
tipo="sumidouro",
|
||
base_in="", # sumidouro aceita qualquer base
|
||
base_out="",
|
||
t_ini=float("inf"),
|
||
t_fim=float("inf"),
|
||
aeronave_id=a.id,
|
||
)
|
||
|
||
# Helper: tempo de ferry entre bases (0 se mesma base)
|
||
def _ferry_h(b1: str, b2: str) -> float:
|
||
if b1 == b2:
|
||
return 0.0
|
||
if b1 not in BASES_COORDS or b2 not in BASES_COORDS:
|
||
return float("inf")
|
||
return _haversine_km(b1, b2) / _VELOCIDADE_CRUZEIRO_KMH
|
||
|
||
# --- Arcos início: origem_k → missao_m (com ou sem ferry) ---
|
||
for a in aeronaves:
|
||
src_id = f"SRC_{a.id}"
|
||
for m in missoes:
|
||
fh = _ferry_h(a.base_inicial, m.orig)
|
||
if fh == float("inf"):
|
||
continue
|
||
if m.t_dep < fh: # missão começa antes do ferry terminar
|
||
continue
|
||
arc_id = f"INI_{a.id}_{m.id}"
|
||
arcos.append(
|
||
Arco(
|
||
id=arc_id,
|
||
orig_id=src_id,
|
||
dest_id=f"M_{m.id}",
|
||
tipo="inicio" if fh == 0.0 else "ferry",
|
||
aeronave_id=a.id,
|
||
horas_ferry=fh,
|
||
)
|
||
)
|
||
|
||
# --- Arcos missao → missao (direto ou com ferry) ---
|
||
for mi in missoes:
|
||
for mj in missoes:
|
||
if mi.id == mj.id:
|
||
continue
|
||
fh = _ferry_h(mi.dest, mj.orig)
|
||
if fh == float("inf"):
|
||
continue
|
||
if mj.t_dep < mi.t_arr + tat_h + fh:
|
||
continue
|
||
arc_id = f"MM_{mi.id}_{mj.id}"
|
||
arcos.append(
|
||
Arco(
|
||
id=arc_id,
|
||
orig_id=f"M_{mi.id}",
|
||
dest_id=f"M_{mj.id}",
|
||
tipo="missao_missao" if fh == 0.0 else "ferry",
|
||
horas_ferry=fh,
|
||
)
|
||
)
|
||
|
||
# --- Arcos fim: missao_m → sumidouro_k ---
|
||
for a in aeronaves:
|
||
snk_id = f"SNK_{a.id}"
|
||
for m in missoes:
|
||
arc_id = f"FIM_{m.id}_{a.id}"
|
||
arcos.append(
|
||
Arco(
|
||
id=arc_id,
|
||
orig_id=f"M_{m.id}",
|
||
dest_id=snk_id,
|
||
tipo="fim",
|
||
aeronave_id=a.id,
|
||
)
|
||
)
|
||
|
||
# --- Fase 2: Nós e arcos de inspeção ---
|
||
for a in aeronaves:
|
||
for seq, insp in catalogo.items():
|
||
insp_nid = f"INSP_{a.id}_{seq}"
|
||
nos[insp_nid] = No(
|
||
id=insp_nid,
|
||
tipo="inspecao",
|
||
base_in=BASE_MANUTENCAO,
|
||
base_out=BASE_MANUTENCAO,
|
||
t_ini=0.0, # início flexível: determinado por s[k,i] no modelo
|
||
t_fim=insp.dt_h, # duração do downtime (s[k,i] + dt_h = fim)
|
||
carga_h=0.0, # inspeção não acumula horas de célula
|
||
insp_seq=seq,
|
||
aeronave_id=a.id,
|
||
)
|
||
|
||
# SRC_k → INSP_k_i (com ou sem ferry até SBMN)
|
||
fh_src = _ferry_h(a.base_inicial, BASE_MANUTENCAO)
|
||
if fh_src != float("inf"):
|
||
arcos.append(Arco(
|
||
id=f"SRC_INSP_{a.id}_{seq}",
|
||
orig_id=f"SRC_{a.id}",
|
||
dest_id=insp_nid,
|
||
tipo="inicio",
|
||
aeronave_id=a.id,
|
||
horas_ferry=fh_src,
|
||
))
|
||
|
||
# M_m → INSP_k_i: missões de qualquer base (ferry até SBMN)
|
||
for m in missoes:
|
||
fh = _ferry_h(m.dest, BASE_MANUTENCAO)
|
||
if fh == float("inf"):
|
||
continue
|
||
arcos.append(Arco(
|
||
id=f"MINSP_{a.id}_{seq}_{m.id}",
|
||
orig_id=f"M_{m.id}",
|
||
dest_id=insp_nid,
|
||
tipo="missao_insp",
|
||
aeronave_id=a.id,
|
||
horas_ferry=fh,
|
||
))
|
||
|
||
# INSP_k_i → M_m': missões de qualquer base após downtime (ferry de SBMN)
|
||
for m in missoes:
|
||
fh = _ferry_h(BASE_MANUTENCAO, m.orig)
|
||
if fh == float("inf"):
|
||
continue
|
||
if m.t_dep < insp.dt_h + tat_h + fh:
|
||
continue
|
||
arcos.append(Arco(
|
||
id=f"INSPM_{a.id}_{seq}_{m.id}",
|
||
orig_id=insp_nid,
|
||
dest_id=f"M_{m.id}",
|
||
tipo="insp_missao",
|
||
aeronave_id=a.id,
|
||
horas_ferry=fh,
|
||
))
|
||
|
||
# INSP_k_i → SNK_k (aeronave encerra o horizonte após inspeção)
|
||
arcos.append(Arco(
|
||
id=f"INSP_SNK_{a.id}_{seq}",
|
||
orig_id=insp_nid,
|
||
dest_id=f"SNK_{a.id}",
|
||
tipo="fim",
|
||
aeronave_id=a.id,
|
||
))
|
||
|
||
return nos, arcos
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Modelo de otimização
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def resolver(
|
||
nos: dict[str, No],
|
||
arcos: list[Arco],
|
||
aeronaves: list[Aeronave],
|
||
catalogo: dict[int, InspecaoParam],
|
||
) -> dict:
|
||
ks = [a.id for a in aeronaves]
|
||
acft = {a.id: a for a in aeronaves}
|
||
missoes_nos = {nid: n for nid, n in nos.items() if n.tipo == "missao"}
|
||
|
||
# Índices de adjacência para conservação de fluxo
|
||
arcos_saindo: dict[str, list[Arco]] = {nid: [] for nid in nos}
|
||
arcos_entrando: dict[str, list[Arco]] = {nid: [] for nid in nos}
|
||
for a in arcos:
|
||
arcos_saindo[a.orig_id].append(a)
|
||
arcos_entrando[a.dest_id].append(a)
|
||
|
||
# --- Variáveis ---
|
||
# y[k, arco_id] = 1 se aeronave k usa este arco.
|
||
# Arcos exclusivos de outra aeronave (INI/FIM) não geram variável para k.
|
||
def arco_valido_para(k: str, a: Arco) -> bool:
|
||
return a.aeronave_id is None or a.aeronave_id == k
|
||
|
||
y: dict[tuple[str, str], pulp.LpVariable] = {
|
||
(k, a.id): pulp.LpVariable(f"y_{k}_{a.id}", cat="Binary")
|
||
for k in ks
|
||
for a in arcos
|
||
if arco_valido_para(k, a)
|
||
}
|
||
|
||
# z[missao_no_id] = 1 se a missão é coberta (por qualquer aeronave)
|
||
z: dict[str, pulp.LpVariable] = {
|
||
nid: pulp.LpVariable(f"z_{nid}", cat="Binary")
|
||
for nid in missoes_nos
|
||
}
|
||
|
||
mdl = pulp.LpProblem("ARARA_OAMRP_v3", pulp.LpMaximize)
|
||
|
||
# L1: max Σ_m p_m * z_m
|
||
mdl += (
|
||
pulp.lpSum(n.missao.prioridade * z[nid] for nid, n in missoes_nos.items()),
|
||
"obj_L1",
|
||
)
|
||
|
||
def y_sum(k: str, arcs: list[Arco]) -> pulp.LpAffineExpression:
|
||
"""Soma y[(k, a.id)] apenas para arcos válidos para a aeronave k."""
|
||
return pulp.lpSum(
|
||
y[(k, a.id)] for a in arcs if (k, a.id) in y
|
||
)
|
||
|
||
# --- C1: Cobertura e definição de z ---
|
||
# Missões normais: Σ_k y_entrada ≤ 1 (cobertura opcional)
|
||
# Missões obrigatórias (M_obr / EVAM): Σ_k y_entrada == 1
|
||
for nid, n in missoes_nos.items():
|
||
fluxo_entrada = pulp.lpSum(
|
||
y[(k, a.id)]
|
||
for k in ks
|
||
for a in arcos_entrando[nid]
|
||
if (k, a.id) in y
|
||
)
|
||
mdl += fluxo_entrada == z[nid], f"def_z_{nid}"
|
||
if n.missao.obrigatoria:
|
||
mdl += z[nid] == 1, f"hard_{nid}"
|
||
else:
|
||
mdl += z[nid] <= 1, f"cob_{nid}"
|
||
|
||
# --- C2: Conservação de fluxo por (aeronave, nó) ---
|
||
for k in ks:
|
||
src_id = f"SRC_{k}"
|
||
snk_id = f"SNK_{k}"
|
||
for nid, n in nos.items():
|
||
if n.tipo in ("origem", "sumidouro") and n.aeronave_id != k:
|
||
continue # nó de outra aeronave — irrelevante
|
||
|
||
entrada_k = y_sum(k, arcos_entrando[nid])
|
||
saida_k = y_sum(k, arcos_saindo[nid])
|
||
|
||
if nid == src_id:
|
||
# Origem: no máximo 1 arco de saída (aeronave pode não voar)
|
||
mdl += saida_k <= 1, f"src_{k}"
|
||
elif nid == snk_id:
|
||
# Sumidouro: entrada = saída da origem (fluxo fechado)
|
||
src_saida = y_sum(k, arcos_saindo[src_id])
|
||
mdl += entrada_k == src_saida, f"snk_{k}"
|
||
elif n.tipo == "missao":
|
||
# Missão: fluxo de entrada = fluxo de saída para esta aeronave
|
||
mdl += entrada_k == saida_k, f"fluxo_{k}_{nid}"
|
||
|
||
# --- Fase 2: variáveis e restrições C6-reset, C9, C11 ---
|
||
|
||
insp_nos = {nid: n for nid, n in nos.items() if n.tipo == "inspecao"}
|
||
|
||
# Big-M: usa LRT máximo (para horas) e horizonte temporal (para tempo)
|
||
M_h = max((insp.lrt_h for insp in catalogo.values()), default=1.0) + 100.0
|
||
M_t = (
|
||
max((n.t_fim for n in missoes_nos.values()), default=0.0) + 200.0
|
||
)
|
||
|
||
# s[k,i]: instante de início da inspeção i pela aeronave k
|
||
s_insp: dict[tuple[str, int], pulp.LpVariable] = {
|
||
(k, seq): pulp.LpVariable(f"s_{k}_{seq}", lowBound=0, upBound=M_t)
|
||
for k in ks
|
||
for seq in catalogo
|
||
}
|
||
|
||
# H[k, i, nid]: horas acumuladas no relógio i da aeronave k ao sair do nó nid.
|
||
# Definido para todos os nós de missão e de inspeção (exceto o nó de reset
|
||
# de clock i, onde H = 0 por definição).
|
||
def _h_valido(k: str, seq: int, nid: str) -> bool:
|
||
n = nos[nid]
|
||
if n.tipo in ("origem", "sumidouro"):
|
||
return False
|
||
# Nó de reset do próprio clock i: H = 0, não precisa de variável
|
||
if n.tipo == "inspecao" and n.insp_seq == seq and n.aeronave_id == k:
|
||
return False
|
||
# Nó de inspeção de outra aeronave: irrelevante
|
||
if n.tipo == "inspecao" and n.aeronave_id != k:
|
||
return False
|
||
return True
|
||
|
||
H: dict[tuple[str, int, str], pulp.LpVariable] = {
|
||
(k, seq, nid): pulp.LpVariable(
|
||
f"H_{k}_{seq}_{nid}",
|
||
lowBound=0,
|
||
upBound=catalogo[seq].lrt_h,
|
||
)
|
||
for k in ks
|
||
for seq in catalogo
|
||
for nid in nos
|
||
if _h_valido(k, seq, nid)
|
||
}
|
||
|
||
def _H_prev(k: str, seq: int, u_nid: str) -> object:
|
||
"""Retorna o valor/variável de H_{k,seq} ao sair do nó u."""
|
||
n = nos[u_nid]
|
||
if n.tipo == "origem":
|
||
return acft[k].horas_iniciais.get(seq, 0.0)
|
||
# Nó de reset para este clock: saída é 0
|
||
if n.tipo == "inspecao" and n.insp_seq == seq and n.aeronave_id == k:
|
||
return 0.0
|
||
return H.get((k, seq, u_nid))
|
||
|
||
# Propagação de H ao longo dos arcos (big-M)
|
||
for k in ks:
|
||
for seq in catalogo:
|
||
reset_nid = f"INSP_{k}_{seq}"
|
||
for a in arcos:
|
||
if (k, a.id) not in y:
|
||
continue
|
||
v_nid = a.dest_id
|
||
if v_nid == reset_nid:
|
||
continue # reset: H = 0, sem propagação
|
||
H_v = H.get((k, seq, v_nid))
|
||
if H_v is None:
|
||
continue
|
||
h_prev = _H_prev(k, seq, a.orig_id)
|
||
if h_prev is None:
|
||
continue
|
||
# horas acumuladas = missão + ferry (C6: traslado consome horas de célula)
|
||
carga = nos[v_nid].carga_h + a.horas_ferry
|
||
y_var = y[(k, a.id)]
|
||
mdl += H_v >= h_prev + carga - M_h * (1 - y_var), f"Hlo_{k}_{seq}_{a.id}"
|
||
mdl += H_v <= h_prev + carga + M_h * (1 - y_var), f"Hhi_{k}_{seq}_{a.id}"
|
||
|
||
# --- C6 (reset): H[k,i,M_m] ≤ LRT_i quando aeronave k visita M_m ---
|
||
# (já aplicado como upBound na variável; confirma aqui explicitamente)
|
||
# Nota: se a missão não for visitada, H fica ≤ LRT por bound.
|
||
|
||
# Horizonte temporal em horas (última chegada registrada)
|
||
horizonte_h = max((n.t_fim for n in missoes_nos.values()), default=744.0)
|
||
|
||
# Inspeção só pode começar dentro do horizonte de planejamento
|
||
for k in ks:
|
||
for seq in catalogo:
|
||
insp_nid = f"INSP_{k}_{seq}"
|
||
if insp_nid not in nos:
|
||
continue
|
||
w_ki = pulp.lpSum(
|
||
y[(k, a.id)] for a in arcos_entrando[insp_nid] if (k, a.id) in y
|
||
)
|
||
mdl += (
|
||
s_insp[(k, seq)] <= horizonte_h * w_ki,
|
||
f"s_horizonte_{k}_{seq}",
|
||
)
|
||
|
||
# --- C9: downtime — próxima missão só parte após s[k,i] + DT_i + TAT ---
|
||
for k in ks:
|
||
for seq, insp in catalogo.items():
|
||
insp_nid = f"INSP_{k}_{seq}"
|
||
# s[k,i] ≥ t_arr(m_antes) para cada arco M_m → INSP
|
||
for a in arcos_entrando[insp_nid]:
|
||
if (k, a.id) not in y:
|
||
continue
|
||
u = nos[a.orig_id]
|
||
if u.tipo == "missao":
|
||
mdl += (
|
||
s_insp[(k, seq)]
|
||
>= u.t_fim - M_t * (1 - y[(k, a.id)]),
|
||
f"c9a_{k}_{seq}_{a.id}",
|
||
)
|
||
# t_dep(m') ≥ s[k,i] + DT_i + TAT para cada arco INSP → M_m'
|
||
for a in arcos_saindo[insp_nid]:
|
||
if (k, a.id) not in y:
|
||
continue
|
||
v = nos[a.dest_id]
|
||
if v.tipo == "missao":
|
||
mdl += (
|
||
v.t_ini + M_t * (1 - y[(k, a.id)])
|
||
>= s_insp[(k, seq)] + insp.dt_h + TAT_H,
|
||
f"c9b_{k}_{seq}_{a.id}",
|
||
)
|
||
|
||
# --- C11: slot único — no máximo 1 aeronave em inspeção em qualquer instante ---
|
||
# Para cada par distinto de inspeções de aeronaves diferentes,
|
||
# variável oo = 1 se (k1,i1) termina antes de (k2,i2) começar.
|
||
insp_events = [
|
||
(k, seq)
|
||
for k in ks
|
||
for seq in catalogo
|
||
if f"INSP_{k}_{seq}" in nos
|
||
]
|
||
for idx1, (k1, seq1) in enumerate(insp_events):
|
||
for k2, seq2 in insp_events[idx1 + 1:]:
|
||
if k1 == k2:
|
||
continue # mesma aeronave: C10 já garantido pelo fluxo
|
||
insp1, insp2 = catalogo[seq1], catalogo[seq2]
|
||
nid1, nid2 = f"INSP_{k1}_{seq1}", f"INSP_{k2}_{seq2}"
|
||
w1 = pulp.lpSum(y[(k1, a.id)] for a in arcos_entrando[nid1] if (k1, a.id) in y)
|
||
w2 = pulp.lpSum(y[(k2, a.id)] for a in arcos_entrando[nid2] if (k2, a.id) in y)
|
||
oo = pulp.LpVariable(f"oo_{k1}_{seq1}_{k2}_{seq2}", cat="Binary")
|
||
# (k1,i1) termina antes de (k2,i2): s2 ≥ s1 + DT1
|
||
mdl += (
|
||
s_insp[(k2, seq2)]
|
||
>= s_insp[(k1, seq1)] + insp1.dt_h
|
||
- M_t * (1 - oo) - M_t * (1 - w1) - M_t * (1 - w2),
|
||
f"c11a_{k1}_{seq1}_{k2}_{seq2}",
|
||
)
|
||
# (k2,i2) termina antes de (k1,i1): s1 ≥ s2 + DT2
|
||
mdl += (
|
||
s_insp[(k1, seq1)]
|
||
>= s_insp[(k2, seq2)] + insp2.dt_h
|
||
- M_t * oo - M_t * (1 - w1) - M_t * (1 - w2),
|
||
f"c11b_{k1}_{seq1}_{k2}_{seq2}",
|
||
)
|
||
|
||
# --- C6 (Fase 1 — mantido como redundância explícita para verificação) ---
|
||
for k in ks:
|
||
orcamento = acft[k].orcamento_h(catalogo)
|
||
mdl += (
|
||
pulp.lpSum(
|
||
n.carga_h * y[(k, a.id)]
|
||
for nid, n in missoes_nos.items()
|
||
for a in arcos_entrando[nid]
|
||
if (k, a.id) in y
|
||
)
|
||
<= orcamento,
|
||
f"orcamento_{k}",
|
||
)
|
||
|
||
solver = pulp.PULP_CBC_CMD(msg=False)
|
||
mdl.solve(solver)
|
||
prio_otima = int(round(pulp.value(mdl.objective) or 0))
|
||
|
||
# L2: com prioridade fixada, minimiza folga de horas ao fim do horizonte.
|
||
# folga_k = orcamento_k − horas_voadas_k
|
||
# Minimizar Σ_k folga_k ⟺ maximizar Σ_k horas_voadas_k (orcamento é constante).
|
||
mdl += (
|
||
pulp.lpSum(n.missao.prioridade * z[nid] for nid, n in missoes_nos.items())
|
||
== prio_otima,
|
||
"fixa_L1",
|
||
)
|
||
horas_voadas = pulp.lpSum(
|
||
n.carga_h * y[(k, a.id)]
|
||
for nid, n in missoes_nos.items()
|
||
for k in ks
|
||
for a in arcos_entrando[nid]
|
||
if (k, a.id) in y
|
||
)
|
||
mdl.sense = pulp.LpMaximize
|
||
mdl.setObjective(horas_voadas)
|
||
mdl.solve(solver)
|
||
|
||
return {
|
||
"modelo": mdl,
|
||
"nos": nos,
|
||
"arcos": arcos,
|
||
"arcos_entrando": arcos_entrando,
|
||
"arcos_saindo": arcos_saindo,
|
||
"missoes_nos": missoes_nos,
|
||
"insp_nos": insp_nos,
|
||
"acft": acft,
|
||
"catalogo": catalogo,
|
||
"ks": ks,
|
||
"y": y,
|
||
"z": z,
|
||
"s_insp": s_insp,
|
||
"H": H,
|
||
"prio_otima": prio_otima,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Extração de rota
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def rota_da_aeronave(sol: dict, k: str) -> list[No]:
|
||
"""
|
||
Reconstrói a sequência de nós (missão e inspeção) percorrida pela aeronave k.
|
||
"""
|
||
nos = sol["nos"]
|
||
arcos_saindo = sol["arcos_saindo"]
|
||
y = sol["y"]
|
||
|
||
src_id = f"SRC_{k}"
|
||
seq: list[No] = []
|
||
atual_id = src_id
|
||
|
||
while True:
|
||
prox = [
|
||
a
|
||
for a in arcos_saindo[atual_id]
|
||
if y.get((k, a.id)) is not None
|
||
and y[(k, a.id)].value() is not None
|
||
and y[(k, a.id)].value() > 0.5
|
||
]
|
||
if not prox:
|
||
break
|
||
arco = prox[0]
|
||
no_dest = nos[arco.dest_id]
|
||
if no_dest.tipo == "sumidouro":
|
||
break
|
||
seq.append(no_dest)
|
||
atual_id = arco.dest_id
|
||
|
||
return seq
|
||
|
||
|
||
def inspecoes_da_aeronave(sol: dict, k: str) -> list[tuple[int, float, InspecaoParam]]:
|
||
"""Retorna lista de (seq, s_inicio, InspecaoParam) para inspeções executadas por k."""
|
||
s_insp = sol["s_insp"]
|
||
catalogo = sol["catalogo"]
|
||
nos = sol["nos"]
|
||
arcos_entrando = sol["arcos_entrando"]
|
||
y = sol["y"]
|
||
resultado = []
|
||
for seq, insp in catalogo.items():
|
||
insp_nid = f"INSP_{k}_{seq}"
|
||
if insp_nid not in nos:
|
||
continue
|
||
if any(
|
||
y.get((k, a.id)) is not None and y[(k, a.id)].value() is not None
|
||
and y[(k, a.id)].value() > 0.5
|
||
for a in arcos_entrando[insp_nid]
|
||
):
|
||
s_val = s_insp[(k, seq)].value() or 0.0
|
||
resultado.append((seq, s_val, insp))
|
||
return sorted(resultado, key=lambda t: t[1])
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Saída
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def escrever_resultado(sol: dict, caminho: Path) -> None:
|
||
caminho.parent.mkdir(parents=True, exist_ok=True)
|
||
z = sol["z"]
|
||
missoes_nos = sol["missoes_nos"]
|
||
linhas = []
|
||
|
||
cobertas_ids: set[str] = set()
|
||
for k in sol["ks"]:
|
||
missoes_rota = [n for n in rota_da_aeronave(sol, k) if n.tipo == "missao"]
|
||
for ordem, no in enumerate(missoes_rota, start=1):
|
||
m = no.missao
|
||
cobertas_ids.add(no.id)
|
||
linhas.append({
|
||
"status": "cumprida",
|
||
"aeronave": k,
|
||
"ordem": ordem,
|
||
"id": m.id,
|
||
"om": m.om,
|
||
"orig": m.orig,
|
||
"dest": m.dest,
|
||
"partida_utc": m.partida_utc.isoformat().replace("+00:00", "Z"),
|
||
"chegada_utc": m.chegada_utc.isoformat().replace("+00:00", "Z"),
|
||
"dur_h": f"{m.dur_h:.2f}",
|
||
"prioridade": m.prioridade,
|
||
"periodo": m.periodo,
|
||
"aeronave_real_2025": m.aeronave_real,
|
||
"linhas_origem": m.linhas_origem,
|
||
})
|
||
|
||
for nid, n in missoes_nos.items():
|
||
if nid in cobertas_ids:
|
||
continue
|
||
m = n.missao
|
||
linhas.append({
|
||
"status": "nao_cumprida",
|
||
"aeronave": "",
|
||
"ordem": "",
|
||
"id": m.id,
|
||
"om": m.om,
|
||
"orig": m.orig,
|
||
"dest": m.dest,
|
||
"partida_utc": m.partida_utc.isoformat().replace("+00:00", "Z"),
|
||
"chegada_utc": m.chegada_utc.isoformat().replace("+00:00", "Z"),
|
||
"dur_h": f"{m.dur_h:.2f}",
|
||
"prioridade": m.prioridade,
|
||
"periodo": m.periodo,
|
||
"aeronave_real_2025": m.aeronave_real,
|
||
"linhas_origem": m.linhas_origem,
|
||
})
|
||
|
||
colunas = [
|
||
"status", "aeronave", "ordem", "id", "om",
|
||
"orig", "dest", "partida_utc", "chegada_utc",
|
||
"dur_h", "prioridade", "periodo", "aeronave_real_2025", "linhas_origem",
|
||
]
|
||
with caminho.open("w", newline="", encoding="utf-8-sig") as f:
|
||
w = csv.DictWriter(f, fieldnames=colunas)
|
||
w.writeheader()
|
||
w.writerows(linhas)
|
||
|
||
|
||
def imprimir_relatorio(
|
||
sol: dict, bases: list[str], resultado_csv: Path
|
||
) -> None:
|
||
z = sol["z"]
|
||
missoes_nos = sol["missoes_nos"]
|
||
catalogo: dict[int, InspecaoParam] = sol["catalogo"]
|
||
acft: dict[str, Aeronave] = sol["acft"]
|
||
|
||
cobertas = [nid for nid, n in missoes_nos.items() if z[nid].value() and z[nid].value() > 0.5]
|
||
prio_total = sum(n.missao.prioridade for n in missoes_nos.values())
|
||
prio_obtida = sum(missoes_nos[nid].missao.prioridade for nid in cobertas)
|
||
total = len(missoes_nos)
|
||
|
||
print("=" * 72)
|
||
print(" OAMRP V3 — Fase 3 (C1–C11 + ferry) | Esquadrao Arara C-105")
|
||
print("=" * 72)
|
||
print(f" Status solver : {pulp.LpStatus[sol['modelo'].status]}")
|
||
print(f" Missoes : {len(cobertas)}/{total} cumpridas")
|
||
print(f" Prioridade : {prio_obtida}/{prio_total}")
|
||
print(f" Resultado CSV : {resultado_csv}")
|
||
|
||
print(f"\n Bases observadas ({len(bases)}): {', '.join(bases)}")
|
||
|
||
print("\n Catalogo de inspecoes (Fase 1 — orcamento por aeronave):")
|
||
for seq in sorted(catalogo):
|
||
insp = catalogo[seq]
|
||
cal = f"{insp.cal_dias:.0f}d" if insp.cal_dias else "—"
|
||
print(
|
||
f" seq {seq:2d} {insp.sigla:<18} LRT={insp.lrt_h:>6.0f}h "
|
||
f"CAL={cal:>5} DT={insp.dt_h:.1f}h [{insp.nivel}]"
|
||
)
|
||
|
||
print("\n Orcamento por aeronave:")
|
||
for k in sol["ks"]:
|
||
a = acft[k]
|
||
oc = a.orcamento_h(catalogo)
|
||
info = a.insp_mais_restritiva(catalogo)
|
||
if info:
|
||
seq_min, f0, insp = info
|
||
print(
|
||
f" {k}: {oc:.1f}h "
|
||
f"(limitado por seq {seq_min} {insp.sigla}, "
|
||
f"f0={f0:.1f}h, LRT={insp.lrt_h:.0f}h)"
|
||
)
|
||
else:
|
||
print(f" {k}: {oc:.1f}h (sem restricao de inspecao)")
|
||
|
||
print()
|
||
arcos_por_id = {a.id: a for a in sol["arcos"]}
|
||
y = sol["y"]
|
||
|
||
for k in sol["ks"]:
|
||
rota = rota_da_aeronave(sol, k)
|
||
insps = inspecoes_da_aeronave(sol, k)
|
||
missoes_rota = [n for n in rota if n.tipo == "missao"]
|
||
horas_missao = sum(n.carga_h for n in missoes_rota)
|
||
# horas de ferry: soma horas_ferry dos arcos ativos para este k
|
||
horas_ferry_total = sum(
|
||
a.horas_ferry
|
||
for a in sol["arcos"]
|
||
if (k, a.id) in y
|
||
and y[(k, a.id)].value() is not None
|
||
and y[(k, a.id)].value() > 0.5
|
||
and a.horas_ferry > 0.0
|
||
)
|
||
horas_celula = horas_missao + horas_ferry_total
|
||
orcamento = acft[k].orcamento_h(catalogo)
|
||
folga = orcamento - horas_celula
|
||
print(
|
||
f" Aeronave {k}: {len(missoes_rota)} missoes | "
|
||
f"{horas_missao:.1f}h voadas + {horas_ferry_total:.1f}h ferry = {horas_celula:.1f}h celula | "
|
||
f"folga {folga:.1f}h | {len(insps)} inspecao(oes)"
|
||
)
|
||
# Linha do tempo: missões, inspeções e ferries ordenados por t_ini
|
||
eventos: list[tuple[float, str]] = []
|
||
# rastrear nó anterior para identificar ferry
|
||
nos_rota_ids = [n.id for n in rota]
|
||
for idx, no in enumerate(rota):
|
||
if no.tipo == "missao":
|
||
m = no.missao
|
||
# detecta ferry antes desta missão
|
||
if idx > 0:
|
||
prev = rota[idx - 1]
|
||
if prev.tipo in ("missao", "origem") and prev.base_out != no.base_in:
|
||
fh = _haversine_km(prev.base_out, no.base_in) / _VELOCIDADE_CRUZEIRO_KMH
|
||
eventos.append((
|
||
no.t_ini - fh,
|
||
f" [FERRY {prev.base_out}->{no.base_in} {fh:.1f}h]"
|
||
))
|
||
eventos.append((
|
||
no.t_ini,
|
||
f" {m.partida_utc:%d/%m %H:%M}Z "
|
||
f"{m.orig}->{m.dest} OM {m.om} prio {m.prioridade} ({m.id})"
|
||
))
|
||
elif no.tipo == "inspecao":
|
||
seq = no.insp_seq
|
||
s_val = sol["s_insp"][(k, seq)].value() or 0.0
|
||
insp = catalogo[seq]
|
||
if idx > 0:
|
||
prev = rota[idx - 1]
|
||
if prev.tipo == "missao" and prev.base_out != BASE_MANUTENCAO:
|
||
fh = _haversine_km(prev.base_out, BASE_MANUTENCAO) / _VELOCIDADE_CRUZEIRO_KMH
|
||
eventos.append((
|
||
s_val - fh,
|
||
f" [FERRY {prev.base_out}->{BASE_MANUTENCAO} {fh:.1f}h (retorno insp)]"
|
||
))
|
||
eventos.append((
|
||
s_val,
|
||
f" [INSP seq {seq} {insp.sigla}] "
|
||
f"s={s_val:.1f}h DT={insp.dt_h:.1f}h fim={s_val + insp.dt_h:.1f}h"
|
||
))
|
||
for _, linha in sorted(eventos)[:20]:
|
||
print(linha)
|
||
if len(eventos) > 20:
|
||
print(f" ... mais {len(eventos) - 20} eventos")
|
||
print()
|
||
|
||
nao = [nid for nid in missoes_nos if nid not in cobertas]
|
||
if nao:
|
||
print(f" Missoes nao cumpridas ({len(nao)}, primeiras 20):")
|
||
for nid in nao[:20]:
|
||
m = missoes_nos[nid].missao
|
||
print(
|
||
f" {m.partida_utc:%d/%m %H:%M}Z "
|
||
f"{m.orig}->{m.dest} OM {m.om} ({m.id})"
|
||
)
|
||
if len(nao) > 20:
|
||
print(f" ... mais {len(nao) - 20}")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Carregamento de dados
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _parse_utc(valor: str) -> datetime:
|
||
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(
|
||
caminho: Path,
|
||
inicio: datetime,
|
||
fim: datetime,
|
||
bases_validas: Optional[set[str]] = None,
|
||
hard_pattern: str = HARD_MISSAO_PADRAO,
|
||
prioridade_padrao: int = 1,
|
||
) -> tuple[list[Missao], datetime]:
|
||
"""
|
||
Lê o CSV consolidado e retorna missões na janela [inicio, fim).
|
||
|
||
bases_validas — se fornecido, descarta missões fora deste conjunto de bases (Q4).
|
||
hard_pattern — substring no campo 'missao' que marca M_obr (Q2, ex: 'EVAM').
|
||
"""
|
||
import re
|
||
hard_re = re.compile(hard_pattern, re.IGNORECASE)
|
||
|
||
brutos: list[tuple[datetime, datetime, dict]] = []
|
||
with caminho.open(newline="", encoding="utf-8-sig") as f:
|
||
for linha in csv.DictReader(f):
|
||
partida = _parse_utc(linha["partida_utc"])
|
||
chegada = _parse_utc(linha["chegada_utc"])
|
||
if not (inicio <= partida < fim):
|
||
continue
|
||
if chegada <= partida:
|
||
raise ValueError(
|
||
f"Missao {linha['id_registro']}: chegada antes da partida"
|
||
)
|
||
orig = linha["localidade_dep"]
|
||
dest = linha["localidade_arr"]
|
||
if bases_validas and (orig not in bases_validas or dest not in bases_validas):
|
||
continue
|
||
brutos.append((partida, chegada, linha))
|
||
|
||
if not brutos:
|
||
raise ValueError("Nenhuma missao encontrada na janela/bases informadas.")
|
||
|
||
brutos.sort(key=lambda t: (t[0], t[2]["id_registro"]))
|
||
horizonte = brutos[0][0]
|
||
|
||
missoes = []
|
||
for partida, chegada, linha in brutos:
|
||
t_dep = (partida - horizonte).total_seconds() / 3600
|
||
t_arr = (chegada - horizonte).total_seconds() / 3600
|
||
dur_h = t_arr - t_dep
|
||
prio = int(linha.get("prioridade", prioridade_padrao) or prioridade_padrao)
|
||
cod_missao = linha.get("missao", "")
|
||
obrigatoria = bool(hard_re.search(cod_missao))
|
||
missoes.append(
|
||
Missao(
|
||
id=linha["id_registro"],
|
||
orig=orig,
|
||
dest=dest,
|
||
t_dep=t_dep,
|
||
t_arr=t_arr,
|
||
dur_h=dur_h,
|
||
prioridade=prio,
|
||
obrigatoria=obrigatoria,
|
||
partida_utc=partida,
|
||
chegada_utc=chegada,
|
||
periodo=linha.get("periodo", ""),
|
||
linhas_origem=linha.get("linhas_origem", ""),
|
||
aeronave_real=linha.get("aeronave", ""),
|
||
om=linha.get("om", ""),
|
||
codigo_missao=cod_missao,
|
||
)
|
||
)
|
||
return missoes, horizonte
|
||
|
||
|
||
def carregar_inspecoes(caminho: Path) -> dict[int, InspecaoParam]:
|
||
with caminho.open(encoding="utf-8") as f:
|
||
dados = json.load(f)
|
||
|
||
catalogo: dict[int, InspecaoParam] = {}
|
||
for item in dados["inspecoes"]:
|
||
seq = item["seq"]
|
||
if seq not in SEQS_VALIDOS:
|
||
continue
|
||
nivel_desc = item.get("nivel_descricao", "")
|
||
if nivel_desc not in NIVEIS_PERMITIDOS:
|
||
continue
|
||
lrt = item.get("intervalo_horas_voo_valor")
|
||
if not lrt:
|
||
continue # sem intervalo em horas — só calendárica; entra na Fase 2 via CAL
|
||
|
||
cal_meses = item.get("intervalo_meses_continuos_valor")
|
||
cal_dias = float(cal_meses) * 30.0 if cal_meses else None
|
||
|
||
dur_valor = item.get("duracao_valor", 0)
|
||
dur_unidade = item.get("duracao_unidade", "horas")
|
||
dt_h = float(dur_valor) * 24.0 if dur_unidade == "dias" else float(dur_valor)
|
||
|
||
catalogo[seq] = InspecaoParam(
|
||
seq=seq,
|
||
sigla=item["sigla_mnt"],
|
||
descricao=item["descricao_mnt"],
|
||
nivel=nivel_desc,
|
||
lrt_h=float(lrt),
|
||
cal_dias=cal_dias,
|
||
dt_h=dt_h,
|
||
)
|
||
return catalogo
|
||
|
||
|
||
def derivar_bases(
|
||
caminho: Path,
|
||
inicio: datetime,
|
||
fim: datetime,
|
||
top_n: Optional[int] = None,
|
||
) -> list[str]:
|
||
"""
|
||
Deriva bases ICAO observadas na janela.
|
||
top_n — se fornecido, retorna apenas as N bases mais frequentes (Q4).
|
||
A base de manutenção SBMN é sempre incluída.
|
||
"""
|
||
from collections import Counter
|
||
contagem: Counter = Counter()
|
||
with caminho.open(newline="", encoding="utf-8-sig") as f:
|
||
for linha in csv.DictReader(f):
|
||
try:
|
||
partida = _parse_utc(linha["partida_utc"])
|
||
except Exception:
|
||
continue
|
||
if not (inicio <= partida < fim):
|
||
continue
|
||
contagem[linha["localidade_dep"]] += 1
|
||
contagem[linha["localidade_arr"]] += 1
|
||
|
||
if top_n is None:
|
||
return sorted(contagem)
|
||
|
||
# Garante que SBMN entra independente do ranking
|
||
top = {b for b, _ in contagem.most_common(top_n)}
|
||
top.add(BASE_MANUTENCAO)
|
||
if len(top) > top_n:
|
||
# SBMN foi adicionada extra; remove a menos frequente que não seja SBMN
|
||
extras = sorted(top - {BASE_MANUTENCAO}, key=lambda b: contagem[b])
|
||
top.discard(extras[0])
|
||
return sorted(top)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Missões sintéticas
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# Coordenadas (lat, lon) das 10 bases operacionais do C-105
|
||
BASES_COORDS: dict[str, tuple[float, float]] = {
|
||
"SBMN": (-3.15, -59.99), # Manaus — Ponta Pelada (hub)
|
||
"SBBE": (-1.38, -48.48), # Belém
|
||
"SBBV": ( 2.84, -60.69), # Boa Vista
|
||
"SBSN": (-2.42, -54.79), # Santarém
|
||
"SBPV": (-8.71, -63.90), # Porto Velho
|
||
"SBTS": (-4.25, -69.94), # Tabatinga
|
||
"SBTT": (-3.38, -64.72), # Tefé
|
||
"SBUA": (-0.15, -67.05), # São Gabriel da Cachoeira
|
||
"SWBC": (-0.98, -62.92), # Barcelos
|
||
"SBMY": (-5.81, -61.28), # Manicoré
|
||
}
|
||
|
||
_VELOCIDADE_CRUZEIRO_KMH = 430.0 # C-105 Spartan (típico)
|
||
_VARIACAO_DUR = 0.10 # ± 10 % de variação aleatória na duração
|
||
|
||
|
||
def _haversine_km(b1: str, b2: str) -> float:
|
||
import math
|
||
lat1, lon1 = BASES_COORDS[b1]
|
||
lat2, lon2 = BASES_COORDS[b2]
|
||
R = 6371.0
|
||
dlat = math.radians(lat2 - lat1)
|
||
dlon = math.radians(lon2 - lon1)
|
||
a = math.sin(dlat/2)**2 + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(dlon/2)**2
|
||
return R * 2 * math.asin(math.sqrt(a))
|
||
|
||
|
||
def gerar_missoes_sinteticas(
|
||
n: int = 50,
|
||
seed: Optional[int] = 42,
|
||
inicio: Optional[datetime] = None,
|
||
horizonte_h: float = 744.0, # janeiro = 31 × 24
|
||
n_evam: int = 5,
|
||
bases: Optional[list[str]] = None,
|
||
) -> tuple[list[Missao], datetime]:
|
||
"""
|
||
Gera n missões aleatórias cross-base (orig ≠ dest) entre as bases listadas.
|
||
|
||
Duração estimada via distância haversine ÷ velocidade de cruzeiro do C-105,
|
||
com variação aleatória de ± VARIACAO_DUR.
|
||
n_evam missões são marcadas como obrigatórias (hard).
|
||
"""
|
||
import math
|
||
|
||
if inicio is None:
|
||
inicio = datetime(2025, 1, 1, tzinfo=timezone.utc)
|
||
|
||
bases = list(bases or BASES_COORDS.keys())
|
||
rng = random.Random(seed)
|
||
|
||
pares = [(o, d) for o in bases for d in bases if o != d]
|
||
missoes: list[Missao] = []
|
||
|
||
# sorteia pares com reposição para garantir n missões
|
||
escolhidos = [pares[rng.randrange(len(pares))] for _ in range(n)]
|
||
# garante que SBMN aparece em orig ou dest com frequência (hub real)
|
||
for i in range(0, n, 3):
|
||
outro = rng.choice([b for b in bases if b != "SBMN"])
|
||
if rng.random() < 0.5:
|
||
escolhidos[i] = ("SBMN", outro)
|
||
else:
|
||
escolhidos[i] = (outro, "SBMN")
|
||
|
||
evam_indices = set(rng.sample(range(n), min(n_evam, n)))
|
||
|
||
# distribui decolagens uniformemente ao longo do horizonte com gap mínimo de 2h
|
||
t_deps = sorted(rng.uniform(0, horizonte_h - 10) for _ in range(n))
|
||
|
||
for idx, (t_dep_h, (orig, dest)) in enumerate(zip(t_deps, escolhidos)):
|
||
dist_km = _haversine_km(orig, dest)
|
||
dur_base = dist_km / _VELOCIDADE_CRUZEIRO_KMH
|
||
dur_h = round(dur_base * (1 + rng.uniform(-_VARIACAO_DUR, _VARIACAO_DUR)), 2)
|
||
dur_h = max(0.5, dur_h)
|
||
|
||
t_dep_h = round(t_dep_h, 3)
|
||
t_arr_h = round(t_dep_h + dur_h, 3)
|
||
if t_arr_h > horizonte_h:
|
||
t_arr_h = horizonte_h
|
||
t_dep_h = max(0.0, t_arr_h - dur_h)
|
||
|
||
partida = datetime.fromtimestamp(
|
||
inicio.timestamp() + t_dep_h * 3600, tz=timezone.utc
|
||
)
|
||
chegada = datetime.fromtimestamp(
|
||
inicio.timestamp() + t_arr_h * 3600, tz=timezone.utc
|
||
)
|
||
|
||
obrigatoria = idx in evam_indices
|
||
cod = f"EVAM{idx+1:03d}" if obrigatoria else f"SIM{idx+1:03d}"
|
||
om_str = str(idx + 1)
|
||
|
||
missoes.append(Missao(
|
||
id=f"SIM_{idx+1:03d}_{orig}_{dest}",
|
||
orig=orig,
|
||
dest=dest,
|
||
t_dep=t_dep_h,
|
||
t_arr=t_arr_h,
|
||
dur_h=dur_h,
|
||
prioridade=1,
|
||
obrigatoria=obrigatoria,
|
||
partida_utc=partida,
|
||
chegada_utc=chegada,
|
||
periodo="DIURNO",
|
||
linhas_origem="",
|
||
aeronave_real="",
|
||
om=om_str,
|
||
codigo_missao=cod,
|
||
))
|
||
|
||
return missoes, inicio
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Horas iniciais
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def horas_iniciais_aleatorias(
|
||
aeronaves: list[str],
|
||
catalogo: dict[int, InspecaoParam],
|
||
seed: Optional[int] = None,
|
||
) -> dict[str, dict[int, float]]:
|
||
rng = random.Random(seed)
|
||
return {
|
||
k: {
|
||
seq: round(rng.uniform(0.1 * insp.lrt_h, 0.9 * insp.lrt_h), 1)
|
||
for seq, insp in catalogo.items()
|
||
}
|
||
for k in aeronaves
|
||
}
|
||
|
||
|
||
def carregar_horas_iniciais(
|
||
fonte: Optional[str],
|
||
aleatorio: bool,
|
||
aeronaves: list[str],
|
||
catalogo: dict[int, InspecaoParam],
|
||
seed: Optional[int] = None,
|
||
) -> dict[str, dict[int, float]]:
|
||
if fonte:
|
||
p = Path(fonte)
|
||
raw = json.loads(p.read_text(encoding="utf-8")) if p.exists() else json.loads(fonte)
|
||
return {k: {int(s): float(v) for s, v in d.items()} for k, d in raw.items()}
|
||
if aleatorio:
|
||
return horas_iniciais_aleatorias(aeronaves, catalogo, seed=seed)
|
||
return {k: {seq: 0.0 for seq in catalogo} for k in aeronaves}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# CLI
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(
|
||
description="OAMRP v3 — Fase 1: rede espaço-tempo + cobertura + orçamento."
|
||
)
|
||
parser.add_argument("--csv", type=Path, default=CSV_REGISTRO_PADRAO)
|
||
parser.add_argument("--inspecoes", type=Path, default=JSON_INSPECOES_PADRAO)
|
||
parser.add_argument("--inicio", default=DATA_INICIO_PADRAO)
|
||
parser.add_argument("--fim", default=DATA_FIM_PADRAO)
|
||
parser.add_argument("--resultado", type=Path, default=CSV_RESULTADO_PADRAO)
|
||
parser.add_argument(
|
||
"--horas-iniciais", metavar="JSON",
|
||
help='f0_{k,i}: arquivo ou string JSON. Ex: {"2800": {"23": 35.0}}',
|
||
)
|
||
parser.add_argument(
|
||
"--aleatorio", action="store_true",
|
||
help="Gera f0_{k,i} aleatório (10%–90% do LRT).",
|
||
)
|
||
parser.add_argument("--seed", type=int, default=None)
|
||
parser.add_argument(
|
||
"--aeronaves", nargs="+", default=AERONAVES_PADRAO, metavar="CAUDA"
|
||
)
|
||
parser.add_argument("--base-inicial", default=BASE_INICIAL_PADRAO)
|
||
parser.add_argument(
|
||
"--n-bases", type=int, default=N_BASES_PADRAO,
|
||
help="Top-N bases por frequência de uso (Q4). 0 = todas.",
|
||
)
|
||
parser.add_argument(
|
||
"--hard-pattern", default=HARD_MISSAO_PADRAO,
|
||
help="Regex no campo 'missao' do CSV que marca missões obrigatórias (Q2).",
|
||
)
|
||
parser.add_argument(
|
||
"--sintetico", action="store_true",
|
||
help="Usa missões sintéticas cross-base em vez do CSV real.",
|
||
)
|
||
parser.add_argument(
|
||
"--n-missoes", type=int, default=50,
|
||
help="Número de missões sintéticas a gerar (padrão: 50).",
|
||
)
|
||
parser.add_argument(
|
||
"--n-evam", type=int, default=5,
|
||
help="Missões obrigatórias (EVAM/hard) nas sintéticas (padrão: 5).",
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
inicio = datetime.fromisoformat(args.inicio).replace(tzinfo=timezone.utc)
|
||
fim = datetime.fromisoformat(args.fim).replace(tzinfo=timezone.utc)
|
||
|
||
catalogo = carregar_inspecoes(args.inspecoes)
|
||
|
||
if args.sintetico:
|
||
horizonte_h = (fim - inicio).total_seconds() / 3600
|
||
bases = list(BASES_COORDS.keys())
|
||
missoes, horizonte_dt = gerar_missoes_sinteticas(
|
||
n=args.n_missoes,
|
||
seed=args.seed,
|
||
inicio=inicio,
|
||
horizonte_h=horizonte_h,
|
||
n_evam=args.n_evam,
|
||
bases=bases,
|
||
)
|
||
print(f" Modo sintético: {len(missoes)} missões geradas entre {len(bases)} bases.")
|
||
else:
|
||
top_n = args.n_bases if args.n_bases > 0 else None
|
||
bases = derivar_bases(args.csv, inicio, fim, top_n=top_n)
|
||
bases_validas = set(bases) if top_n else None
|
||
missoes, horizonte_dt = carregar_missoes(
|
||
args.csv, inicio, fim,
|
||
bases_validas=bases_validas,
|
||
hard_pattern=args.hard_pattern,
|
||
)
|
||
|
||
f0 = carregar_horas_iniciais(
|
||
fonte=args.horas_iniciais,
|
||
aleatorio=args.aleatorio,
|
||
aeronaves=args.aeronaves,
|
||
catalogo=catalogo,
|
||
seed=args.seed,
|
||
)
|
||
aeronaves = [
|
||
Aeronave(id=k, base_inicial=args.base_inicial, horas_iniciais=f0.get(k, {}))
|
||
for k in args.aeronaves
|
||
]
|
||
|
||
nos, arcos = construir_rede(missoes, aeronaves, catalogo)
|
||
sol = resolver(nos, arcos, aeronaves, catalogo)
|
||
escrever_resultado(sol, args.resultado)
|
||
imprimir_relatorio(sol, bases, args.resultado)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|