Corrige modelo OAMRP v3 e alinha nomenclatura com Al-Thani (2016)
Correções ao modelo: - Adiciona conservação de fluxo em nós de inspeção (bug que permitia criação/destruição de unidades de fluxo, inflando cobertura para 50/50 sem atribuição real de rotas) - Remove restrição C6 simplificada que somava horas totais sem respeitar resets de inspeção, tornando 2809/2811 artificialmente ociosos Nomenclatura alinhada ao Al-Thani (2016): - InspecaoParam.lrt_h → f_max (F do artigo: batente máximo legal) - Aeronave.horas_iniciais → f0 (f_k: horas acumuladas no início) - orcamento_h() → lrt_inicial() (LRT = F_max − f0) - horas_iniciais_aleatorias() → gerar_f0_aleatorio() - carregar_horas_iniciais() → carregar_f0() - CLI --horas-iniciais → --f0 Novos arquivos: - software/gerar_ofrag.py: gera tabela de OFRAGs com prioridades 1-5 - software/visualizar_resultado.py: mapa Folium interativo de rotas - db/processed/ofrag.csv: 50 OFRAGs sintéticas (seed 42) - db/processed/mapa_rotas.html: mapa gerado da última rodada Resultado com --sintetico --aleatorio --ofrag: 50/50 missões cumpridas (antes: 30/50 sem ferry, 7/50 com bug de fluxo) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Cesa-V
108
software/gerar_ofrag.py
Normal file
108
software/gerar_ofrag.py
Normal file
@@ -0,0 +1,108 @@
|
||||
"""
|
||||
Gera o arquivo de OFRAG (Ordem de Fragmentação) sintético para o Esquadrão Arara.
|
||||
|
||||
Cada OFRAG autoriza uma missão específica e define:
|
||||
- Prioridade operacional (1=urgente … 5=baixa)
|
||||
- Tipo de missão (Logística, Médico, Pessoal, Reabastecimento, Reconhecimento)
|
||||
- Base de origem e destino
|
||||
|
||||
Uso:
|
||||
python software/gerar_ofrag.py --seed 42 --n 50 --saida db/processed/ofrag.csv
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import csv
|
||||
import random
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Importa constantes e função de geração do modelo principal
|
||||
# ---------------------------------------------------------------------------
|
||||
import sys
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
|
||||
from oamrp_v3 import (
|
||||
BASES_COORDS,
|
||||
TIPOS_MISSAO,
|
||||
_PESOS_PRIO,
|
||||
gerar_missoes_sinteticas,
|
||||
)
|
||||
|
||||
BASE_DIR = Path(__file__).resolve().parents[1]
|
||||
OFRAG_PADRAO = BASE_DIR / "db" / "processed" / "ofrag.csv"
|
||||
|
||||
_TIPOS_EVAM = ["Médico", "Logística"] # EVAM normalmente carrega carga urgente
|
||||
|
||||
|
||||
def gerar_ofrag(
|
||||
n: int = 50,
|
||||
seed: int = 42,
|
||||
saida: Path = OFRAG_PADRAO,
|
||||
) -> None:
|
||||
inicio = datetime(2025, 1, 1, tzinfo=timezone.utc)
|
||||
missoes, _ = gerar_missoes_sinteticas(n=n, seed=seed, inicio=inicio, n_evam=5)
|
||||
|
||||
rng = random.Random(seed + 1) # semente diferente para tipos
|
||||
|
||||
saida.parent.mkdir(parents=True, exist_ok=True)
|
||||
colunas = [
|
||||
"id_ofrag", "missao_id", "om", "orig", "dest",
|
||||
"prioridade", "tipo_missao", "observacao",
|
||||
]
|
||||
|
||||
with saida.open("w", newline="", encoding="utf-8-sig") as f:
|
||||
w = csv.DictWriter(f, fieldnames=colunas)
|
||||
w.writeheader()
|
||||
for idx, m in enumerate(missoes, start=1):
|
||||
if m.obrigatoria:
|
||||
prio = 1
|
||||
tipo = rng.choice(_TIPOS_EVAM)
|
||||
obs = "EVAM — execução obrigatória"
|
||||
else:
|
||||
prio = rng.choices(range(1, 6), weights=_PESOS_PRIO)[0]
|
||||
tipo = rng.choices(TIPOS_MISSAO, weights=[20, 15, 25, 25, 15])[0]
|
||||
obs = ""
|
||||
w.writerow({
|
||||
"id_ofrag": f"OFRAG-2025-{idx:03d}",
|
||||
"missao_id": m.id,
|
||||
"om": m.om,
|
||||
"orig": m.orig,
|
||||
"dest": m.dest,
|
||||
"prioridade": prio,
|
||||
"tipo_missao": tipo,
|
||||
"observacao": obs,
|
||||
})
|
||||
|
||||
print(f" OFRAG gerado: {saida} ({len(missoes)} ordens)")
|
||||
_resumir(saida)
|
||||
|
||||
|
||||
def _resumir(saida: Path) -> None:
|
||||
from collections import Counter
|
||||
prios: Counter = Counter()
|
||||
tipos: Counter = Counter()
|
||||
with saida.open(newline="", encoding="utf-8-sig") as f:
|
||||
for row in csv.DictReader(f):
|
||||
prios[int(row["prioridade"])] += 1
|
||||
tipos[row["tipo_missao"]] += 1
|
||||
print("\n Distribuição de prioridades:")
|
||||
for p in sorted(prios):
|
||||
bar = "#" * prios[p]
|
||||
print(f" Prio {p}: {prios[p]:3d} {bar}")
|
||||
print("\n Distribuição de tipos:")
|
||||
for t, c in tipos.most_common():
|
||||
print(f" {t:<20} {c:3d}")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="Gerador de OFRAG sintético — Arara C-105")
|
||||
parser.add_argument("--seed", type=int, default=42)
|
||||
parser.add_argument("--n", type=int, default=50, help="Número de missões")
|
||||
parser.add_argument("--saida", type=Path, default=OFRAG_PADRAO)
|
||||
args = parser.parse_args()
|
||||
gerar_ofrag(n=args.n, seed=args.seed, saida=args.saida)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -30,7 +30,7 @@
|
||||
Removidas: seq 18 (INSP 100FH) e seq 23 (INSP 50FH)
|
||||
|
||||
Horas iniciais f0_{k,i}:
|
||||
--horas-iniciais <arquivo.json ou string JSON>
|
||||
--f0 <arquivo.json ou string JSON>
|
||||
--aleatorio [--seed N]
|
||||
padrão: 0,0 para todas (aeronave recém-inspecionada — conservador)
|
||||
============================================================================
|
||||
@@ -120,7 +120,7 @@ class InspecaoParam:
|
||||
sigla: str
|
||||
descricao: str
|
||||
nivel: str
|
||||
lrt_h: float
|
||||
f_max: float # F do Al-Thani: intervalo máximo legal entre manutenções
|
||||
cal_dias: Optional[float]
|
||||
dt_h: float
|
||||
|
||||
@@ -129,20 +129,16 @@ class InspecaoParam:
|
||||
class Aeronave:
|
||||
id: str
|
||||
base_inicial: str
|
||||
horas_iniciais: dict[int, float] = field(default_factory=dict)
|
||||
f0: dict[int, float] = field(default_factory=dict) # f_k do Al-Thani: horas acumuladas no início
|
||||
|
||||
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.
|
||||
"""
|
||||
def lrt_inicial(self, catalogo: dict[int, InspecaoParam]) -> float:
|
||||
"""Retorna min(F_max_i − f0_{k,i}): LRT disponível no início do horizonte."""
|
||||
if not catalogo:
|
||||
return 0.0
|
||||
margens = [
|
||||
insp.lrt_h - self.horas_iniciais.get(seq, 0.0)
|
||||
return max(0.0, min(
|
||||
insp.f_max - self.f0.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]
|
||||
@@ -151,9 +147,9 @@ class Aeronave:
|
||||
return None
|
||||
seq_min = min(
|
||||
catalogo,
|
||||
key=lambda s: catalogo[s].lrt_h - self.horas_iniciais.get(s, 0.0),
|
||||
key=lambda s: catalogo[s].f_max - self.f0.get(s, 0.0),
|
||||
)
|
||||
return seq_min, self.horas_iniciais.get(seq_min, 0.0), catalogo[seq_min]
|
||||
return seq_min, self.f0.get(seq_min, 0.0), catalogo[seq_min]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -514,13 +510,16 @@ def resolver(
|
||||
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}"
|
||||
elif n.tipo == "inspecao" and n.aeronave_id == k:
|
||||
# Inspeção exclusiva de k: conservação de fluxo (evita criação/destruição)
|
||||
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_h = max((insp.f_max 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
|
||||
)
|
||||
@@ -551,7 +550,7 @@ def resolver(
|
||||
(k, seq, nid): pulp.LpVariable(
|
||||
f"H_{k}_{seq}_{nid}",
|
||||
lowBound=0,
|
||||
upBound=catalogo[seq].lrt_h,
|
||||
upBound=catalogo[seq].f_max,
|
||||
)
|
||||
for k in ks
|
||||
for seq in catalogo
|
||||
@@ -563,7 +562,7 @@ def resolver(
|
||||
"""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)
|
||||
return acft[k].f0.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
|
||||
@@ -672,19 +671,9 @@ def resolver(
|
||||
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}",
|
||||
)
|
||||
# C6 Fase 1 simplificada foi removida: a propagação de H[k,seq,nid] com reset
|
||||
# já impõe o limite LRT por ciclo corretamente. A versão simplificada somava
|
||||
# horas totais sem respeitar resets de inspeção, super-restringindo o modelo.
|
||||
|
||||
solver = pulp.PULP_CBC_CMD(msg=False)
|
||||
mdl.solve(solver)
|
||||
@@ -881,24 +870,24 @@ def imprimir_relatorio(
|
||||
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" seq {seq:2d} {insp.sigla:<18} F_max={insp.f_max:>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)
|
||||
lrt_ini = a.lrt_inicial(catalogo)
|
||||
info = a.insp_mais_restritiva(catalogo)
|
||||
if info:
|
||||
seq_min, f0, insp = info
|
||||
seq_min, f0_val, insp = info
|
||||
print(
|
||||
f" {k}: {oc:.1f}h "
|
||||
f" {k}: LRT_ini={lrt_ini:.1f}h "
|
||||
f"(limitado por seq {seq_min} {insp.sigla}, "
|
||||
f"f0={f0:.1f}h, LRT={insp.lrt_h:.0f}h)"
|
||||
f"f0={f0_val:.1f}h, F_max={insp.f_max:.0f}h)"
|
||||
)
|
||||
else:
|
||||
print(f" {k}: {oc:.1f}h (sem restricao de inspecao)")
|
||||
print(f" {k}: LRT_ini={lrt_ini:.1f}h (sem restricao de inspecao)")
|
||||
|
||||
print()
|
||||
arcos_por_id = {a.id: a for a in sol["arcos"]}
|
||||
@@ -919,12 +908,12 @@ def imprimir_relatorio(
|
||||
and a.horas_ferry > 0.0
|
||||
)
|
||||
horas_celula = horas_missao + horas_ferry_total
|
||||
orcamento = acft[k].orcamento_h(catalogo)
|
||||
folga = orcamento - horas_celula
|
||||
lrt_ini = acft[k].lrt_inicial(catalogo)
|
||||
folga = lrt_ini - 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)"
|
||||
f"LRT_ini={lrt_ini:.1f}h 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]] = []
|
||||
@@ -1096,7 +1085,7 @@ def carregar_inspecoes(caminho: Path) -> dict[int, InspecaoParam]:
|
||||
sigla=item["sigla_mnt"],
|
||||
descricao=item["descricao_mnt"],
|
||||
nivel=nivel_desc,
|
||||
lrt_h=float(lrt),
|
||||
f_max=float(lrt),
|
||||
cal_dias=cal_dias,
|
||||
dt_h=dt_h,
|
||||
)
|
||||
@@ -1173,6 +1162,12 @@ def _haversine_km(b1: str, b2: str) -> float:
|
||||
return R * 2 * math.asin(math.sqrt(a))
|
||||
|
||||
|
||||
TIPOS_MISSAO = ["Logística", "Médico", "Pessoal", "Reabastecimento", "Reconhecimento"]
|
||||
|
||||
# Pesos de prioridade: prio 1 (urgente) é mais frequente que prio 5 (baixa)
|
||||
_PESOS_PRIO = [35, 30, 20, 10, 5]
|
||||
|
||||
|
||||
def gerar_missoes_sinteticas(
|
||||
n: int = 50,
|
||||
seed: Optional[int] = 42,
|
||||
@@ -1180,6 +1175,7 @@ def gerar_missoes_sinteticas(
|
||||
horizonte_h: float = 744.0, # janeiro = 31 × 24
|
||||
n_evam: int = 5,
|
||||
bases: Optional[list[str]] = None,
|
||||
ofrag_path: Optional[Path] = None,
|
||||
) -> tuple[list[Missao], datetime]:
|
||||
"""
|
||||
Gera n missões aleatórias cross-base (orig ≠ dest) entre as bases listadas.
|
||||
@@ -1187,6 +1183,7 @@ def gerar_missoes_sinteticas(
|
||||
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).
|
||||
Se ofrag_path for fornecido, prioridades são lidas do CSV de OFRAG.
|
||||
"""
|
||||
import math
|
||||
|
||||
@@ -1196,6 +1193,15 @@ def gerar_missoes_sinteticas(
|
||||
bases = list(bases or BASES_COORDS.keys())
|
||||
rng = random.Random(seed)
|
||||
|
||||
# Carrega prioridades do CSV de OFRAG, se fornecido
|
||||
ofrag_prios: dict[str, int] = {}
|
||||
ofrag_tipos: dict[str, str] = {}
|
||||
if ofrag_path and ofrag_path.exists():
|
||||
with ofrag_path.open(newline="", encoding="utf-8-sig") as f:
|
||||
for row in csv.DictReader(f):
|
||||
ofrag_prios[row["missao_id"]] = int(row["prioridade"])
|
||||
ofrag_tipos[row["missao_id"]] = row.get("tipo_missao", "")
|
||||
|
||||
pares = [(o, d) for o in bases for d in bases if o != d]
|
||||
missoes: list[Missao] = []
|
||||
|
||||
@@ -1236,15 +1242,24 @@ def gerar_missoes_sinteticas(
|
||||
obrigatoria = idx in evam_indices
|
||||
cod = f"EVAM{idx+1:03d}" if obrigatoria else f"SIM{idx+1:03d}"
|
||||
om_str = str(idx + 1)
|
||||
missao_id = f"SIM_{idx+1:03d}_{orig}_{dest}"
|
||||
|
||||
# Prioridade: EVAM sempre prio 1 (urgente); demais lidas do OFRAG ou geradas
|
||||
if obrigatoria:
|
||||
prio = 1
|
||||
elif missao_id in ofrag_prios:
|
||||
prio = ofrag_prios[missao_id]
|
||||
else:
|
||||
prio = rng.choices(range(1, 6), weights=_PESOS_PRIO)[0]
|
||||
|
||||
missoes.append(Missao(
|
||||
id=f"SIM_{idx+1:03d}_{orig}_{dest}",
|
||||
id=missao_id,
|
||||
orig=orig,
|
||||
dest=dest,
|
||||
t_dep=t_dep_h,
|
||||
t_arr=t_arr_h,
|
||||
dur_h=dur_h,
|
||||
prioridade=1,
|
||||
prioridade=prio,
|
||||
obrigatoria=obrigatoria,
|
||||
partida_utc=partida,
|
||||
chegada_utc=chegada,
|
||||
@@ -1263,34 +1278,36 @@ def gerar_missoes_sinteticas(
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def horas_iniciais_aleatorias(
|
||||
def gerar_f0_aleatorio(
|
||||
aeronaves: list[str],
|
||||
catalogo: dict[int, InspecaoParam],
|
||||
seed: Optional[int] = None,
|
||||
) -> dict[str, dict[int, float]]:
|
||||
"""Sorteia f0_{k,i} entre 10% e 90% de F_max_i para cada aeronave e inspeção."""
|
||||
rng = random.Random(seed)
|
||||
return {
|
||||
k: {
|
||||
seq: round(rng.uniform(0.1 * insp.lrt_h, 0.9 * insp.lrt_h), 1)
|
||||
seq: round(rng.uniform(0.1 * insp.f_max, 0.9 * insp.f_max), 1)
|
||||
for seq, insp in catalogo.items()
|
||||
}
|
||||
for k in aeronaves
|
||||
}
|
||||
|
||||
|
||||
def carregar_horas_iniciais(
|
||||
def carregar_f0(
|
||||
fonte: Optional[str],
|
||||
aleatorio: bool,
|
||||
aeronaves: list[str],
|
||||
catalogo: dict[int, InspecaoParam],
|
||||
seed: Optional[int] = None,
|
||||
) -> dict[str, dict[int, float]]:
|
||||
"""Retorna dict[aeronave_id -> dict[seq -> f0]] com horas iniciais acumuladas."""
|
||||
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 gerar_f0_aleatorio(aeronaves, catalogo, seed=seed)
|
||||
return {k: {seq: 0.0 for seq in catalogo} for k in aeronaves}
|
||||
|
||||
|
||||
@@ -1309,8 +1326,8 @@ def main() -> None:
|
||||
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}}',
|
||||
"--f0", metavar="JSON", dest="f0",
|
||||
help='f0_{k,i} (horas acumuladas iniciais): arquivo ou string JSON. Ex: {"2800": {"2": 150.0}}',
|
||||
)
|
||||
parser.add_argument(
|
||||
"--aleatorio", action="store_true",
|
||||
@@ -1341,6 +1358,10 @@ def main() -> None:
|
||||
"--n-evam", type=int, default=5,
|
||||
help="Missões obrigatórias (EVAM/hard) nas sintéticas (padrão: 5).",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--ofrag", type=Path, default=None, metavar="CSV",
|
||||
help="CSV de OFRAG com prioridades por missão (missao_id, prioridade).",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
inicio = datetime.fromisoformat(args.inicio).replace(tzinfo=timezone.utc)
|
||||
@@ -1358,6 +1379,7 @@ def main() -> None:
|
||||
horizonte_h=horizonte_h,
|
||||
n_evam=args.n_evam,
|
||||
bases=bases,
|
||||
ofrag_path=args.ofrag,
|
||||
)
|
||||
print(f" Modo sintético: {len(missoes)} missões geradas entre {len(bases)} bases.")
|
||||
else:
|
||||
@@ -1370,15 +1392,15 @@ def main() -> None:
|
||||
hard_pattern=args.hard_pattern,
|
||||
)
|
||||
|
||||
f0 = carregar_horas_iniciais(
|
||||
fonte=args.horas_iniciais,
|
||||
f0_map = carregar_f0(
|
||||
fonte=args.f0,
|
||||
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, {}))
|
||||
Aeronave(id=k, base_inicial=args.base_inicial, f0=f0_map.get(k, {}))
|
||||
for k in args.aeronaves
|
||||
]
|
||||
|
||||
|
||||
232
software/visualizar_resultado.py
Normal file
232
software/visualizar_resultado.py
Normal file
@@ -0,0 +1,232 @@
|
||||
"""
|
||||
Visualização interativa do resultado OAMRP — Esquadrão Arara C-105.
|
||||
|
||||
Gera um mapa HTML (Folium) com:
|
||||
- Marcadores para cada base operacional
|
||||
- Rotas por aeronave (cor distinta por EVAM)
|
||||
- Missões não cumpridas em cinza tracejado
|
||||
- Popup com detalhes de cada missão
|
||||
|
||||
Uso:
|
||||
python software/visualizar_resultado.py \
|
||||
--resultado db/processed/resultado_oamrp_sintetico.csv \
|
||||
--saida db/processed/mapa_rotas.html
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import csv
|
||||
from pathlib import Path
|
||||
|
||||
import folium
|
||||
from folium import plugins
|
||||
|
||||
BASE_DIR = Path(__file__).resolve().parents[1]
|
||||
|
||||
RESULTADO_PADRAO = BASE_DIR / "db" / "processed" / "resultado_oamrp_sintetico.csv"
|
||||
MAPA_PADRAO = BASE_DIR / "db" / "processed" / "mapa_rotas.html"
|
||||
|
||||
# Coordenadas das bases (lat, lon)
|
||||
BASES_COORDS: dict[str, tuple[float, float]] = {
|
||||
"SBMN": (-3.15, -59.99),
|
||||
"SBBE": (-1.38, -48.48),
|
||||
"SBBV": ( 2.84, -60.69),
|
||||
"SBSN": (-2.42, -54.79),
|
||||
"SBPV": (-8.71, -63.90),
|
||||
"SBTS": (-4.25, -69.94),
|
||||
"SBTT": (-3.38, -64.72),
|
||||
"SBUA": (-0.15, -67.05),
|
||||
"SWBC": (-0.98, -62.92),
|
||||
"SBMY": (-5.81, -61.28),
|
||||
}
|
||||
|
||||
# Nomes legíveis das bases
|
||||
BASES_NOMES = {
|
||||
"SBMN": "Manaus / Ponta Pelada (Hub)",
|
||||
"SBBE": "Belém",
|
||||
"SBBV": "Boa Vista",
|
||||
"SBSN": "Santarém",
|
||||
"SBPV": "Porto Velho",
|
||||
"SBTS": "Tabatinga",
|
||||
"SBTT": "Tefé",
|
||||
"SBUA": "São Gabriel da Cachoeira",
|
||||
"SWBC": "Barcelos",
|
||||
"SBMY": "Manicoré",
|
||||
}
|
||||
|
||||
# Paleta de cores por aeronave
|
||||
CORES_AERONAVE = {
|
||||
"2800": "#1f77b4", # azul
|
||||
"2803": "#2ca02c", # verde
|
||||
"2809": "#d62728", # vermelho
|
||||
"2811": "#ff7f0e", # laranja
|
||||
}
|
||||
COR_NAO_CUMPRIDA = "#aaaaaa"
|
||||
COR_HUB = "#d62728"
|
||||
COR_BASE = "#1f77b4"
|
||||
|
||||
|
||||
def _cor_prio(prio: str) -> str:
|
||||
mapa = {"1": "red", "2": "orange", "3": "yellow", "4": "lightblue", "5": "lightgray"}
|
||||
return mapa.get(str(prio), "white")
|
||||
|
||||
|
||||
def carregar_resultado(caminho: Path) -> list[dict]:
|
||||
with caminho.open(newline="", encoding="utf-8-sig") as f:
|
||||
return list(csv.DictReader(f))
|
||||
|
||||
|
||||
def adicionar_bases(mapa: folium.Map) -> None:
|
||||
for icao, (lat, lon) in BASES_COORDS.items():
|
||||
cor = COR_HUB if icao == "SBMN" else COR_BASE
|
||||
icone = folium.Icon(color="red" if icao == "SBMN" else "blue", icon="plane", prefix="fa")
|
||||
folium.Marker(
|
||||
location=[lat, lon],
|
||||
tooltip=f"<b>{icao}</b><br>{BASES_NOMES.get(icao, '')}",
|
||||
popup=folium.Popup(f"<b>{icao}</b><br>{BASES_NOMES.get(icao, '')}", max_width=200),
|
||||
icon=icone,
|
||||
).add_to(mapa)
|
||||
|
||||
|
||||
def gerar_mapa(resultado: Path, saida: Path) -> None:
|
||||
linhas = carregar_resultado(resultado)
|
||||
|
||||
mapa = folium.Map(
|
||||
location=[-3.5, -62.0],
|
||||
zoom_start=5,
|
||||
tiles="CartoDB positron",
|
||||
)
|
||||
|
||||
# Grupos por aeronave + grupo para não cumpridas
|
||||
aeronaves_vistas = sorted({r["aeronave"] for r in linhas if r["aeronave"]})
|
||||
grupos: dict[str, folium.FeatureGroup] = {}
|
||||
for k in aeronaves_vistas:
|
||||
grupos[k] = folium.FeatureGroup(name=f"EVAM {k}", show=True)
|
||||
grupos["nao_cumprida"] = folium.FeatureGroup(name="Não cumpridas", show=True)
|
||||
|
||||
cumpridas = [r for r in linhas if r["status"] == "cumprida"]
|
||||
nao_cumpridas = [r for r in linhas if r["status"] != "cumprida"]
|
||||
|
||||
# Rotas cumpridas
|
||||
for r in cumpridas:
|
||||
orig = r["orig"]
|
||||
dest = r["dest"]
|
||||
if orig not in BASES_COORDS or dest not in BASES_COORDS:
|
||||
continue
|
||||
lat_o, lon_o = BASES_COORDS[orig]
|
||||
lat_d, lon_d = BASES_COORDS[dest]
|
||||
k = r["aeronave"]
|
||||
cor = CORES_AERONAVE.get(k, "#333333")
|
||||
|
||||
# Linha de rota
|
||||
folium.PolyLine(
|
||||
locations=[[lat_o, lon_o], [lat_d, lon_d]],
|
||||
color=cor,
|
||||
weight=2.5,
|
||||
opacity=0.8,
|
||||
tooltip=f"EVAM {k} | {orig}→{dest} | OM {r['om']} | prio {r['prioridade']}",
|
||||
popup=folium.Popup(
|
||||
f"<b>EVAM {k}</b> — Missão #{r['ordem']}<br>"
|
||||
f"{orig} → {dest}<br>"
|
||||
f"OM: {r['om']} | Prio: {r['prioridade']}<br>"
|
||||
f"Partida: {r['partida_utc'][:16].replace('T',' ')}Z<br>"
|
||||
f"Chegada: {r['chegada_utc'][:16].replace('T',' ')}Z<br>"
|
||||
f"Duração: {r['dur_h']}h",
|
||||
max_width=250,
|
||||
),
|
||||
).add_to(grupos[k])
|
||||
|
||||
# Marcador de origem (pequeno círculo)
|
||||
folium.CircleMarker(
|
||||
location=[lat_o, lon_o],
|
||||
radius=4,
|
||||
color=cor,
|
||||
fill=True,
|
||||
fill_color=_cor_prio(r["prioridade"]),
|
||||
fill_opacity=0.9,
|
||||
tooltip=f"{orig} | prio {r['prioridade']}",
|
||||
).add_to(grupos[k])
|
||||
|
||||
# Missões não cumpridas (linha tracejada cinza)
|
||||
for r in nao_cumpridas:
|
||||
orig = r["orig"]
|
||||
dest = r["dest"]
|
||||
if orig not in BASES_COORDS or dest not in BASES_COORDS:
|
||||
continue
|
||||
lat_o, lon_o = BASES_COORDS[orig]
|
||||
lat_d, lon_d = BASES_COORDS[dest]
|
||||
folium.PolyLine(
|
||||
locations=[[lat_o, lon_o], [lat_d, lon_d]],
|
||||
color=COR_NAO_CUMPRIDA,
|
||||
weight=1.5,
|
||||
opacity=0.5,
|
||||
dash_array="6 4",
|
||||
tooltip=f"NÃO CUMPRIDA | {orig}→{dest} | OM {r['om']} | prio {r['prioridade']}",
|
||||
).add_to(grupos["nao_cumprida"])
|
||||
|
||||
# Adiciona grupos ao mapa
|
||||
for g in grupos.values():
|
||||
g.add_to(mapa)
|
||||
|
||||
adicionar_bases(mapa)
|
||||
folium.LayerControl(collapsed=False).add_to(mapa)
|
||||
|
||||
# Legenda HTML
|
||||
legenda_html = """
|
||||
<div style="position:fixed;bottom:30px;left:30px;z-index:1000;background:white;
|
||||
padding:12px 16px;border-radius:8px;box-shadow:2px 2px 6px rgba(0,0,0,0.3);
|
||||
font-family:sans-serif;font-size:12px;">
|
||||
<b>Esquadrão Arara — OAMRP v3</b><br><br>
|
||||
<b>Aeronaves:</b><br>
|
||||
"""
|
||||
for k, cor in CORES_AERONAVE.items():
|
||||
legenda_html += (
|
||||
f' <span style="display:inline-block;width:16px;height:4px;'
|
||||
f'background:{cor};margin-right:6px;vertical-align:middle;"></span>'
|
||||
f'EVAM {k}<br>'
|
||||
)
|
||||
legenda_html += """
|
||||
<br><b>Prioridade (preenchimento):</b><br>
|
||||
<span style="display:inline-block;width:12px;height:12px;border-radius:50%;
|
||||
background:red;margin-right:6px;"></span>Prio 1 (urgente)<br>
|
||||
<span style="display:inline-block;width:12px;height:12px;border-radius:50%;
|
||||
background:orange;margin-right:6px;"></span>Prio 2<br>
|
||||
<span style="display:inline-block;width:12px;height:12px;border-radius:50%;
|
||||
background:yellow;margin-right:6px;"></span>Prio 3<br>
|
||||
<span style="display:inline-block;width:12px;height:12px;border-radius:50%;
|
||||
background:lightblue;margin-right:6px;"></span>Prio 4–5<br>
|
||||
<br>
|
||||
<span style="border-bottom:2px dashed #aaa;display:inline-block;width:20px;
|
||||
margin-right:6px;"></span>Não cumprida
|
||||
</div>
|
||||
"""
|
||||
mapa.get_root().html.add_child(folium.Element(legenda_html))
|
||||
|
||||
saida.parent.mkdir(parents=True, exist_ok=True)
|
||||
mapa.save(str(saida))
|
||||
print(f" Mapa salvo: {saida}")
|
||||
_resumir_resultado(linhas)
|
||||
|
||||
|
||||
def _resumir_resultado(linhas: list[dict]) -> None:
|
||||
total = len(linhas)
|
||||
cumpridas = sum(1 for r in linhas if r["status"] == "cumprida")
|
||||
print(f" Missões: {cumpridas}/{total} cumpridas")
|
||||
aeronaves = {}
|
||||
for r in linhas:
|
||||
if r["aeronave"]:
|
||||
aeronaves.setdefault(r["aeronave"], 0)
|
||||
aeronaves[r["aeronave"]] += 1
|
||||
for k, c in sorted(aeronaves.items()):
|
||||
print(f" EVAM {k}: {c} missões")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="Visualização OAMRP — Arara C-105")
|
||||
parser.add_argument("--resultado", type=Path, default=RESULTADO_PADRAO)
|
||||
parser.add_argument("--saida", type=Path, default=MAPA_PADRAO)
|
||||
args = parser.parse_args()
|
||||
gerar_mapa(resultado=args.resultado, saida=args.saida)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user