""" ============================================================================ 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}: --f0 --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 date, datetime, timedelta, 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 f_max: float # F do Al-Thani: intervalo máximo legal entre manutenções cal_dias: Optional[float] dt_h: float @dataclass class Aeronave: id: str base_inicial: str f0: dict[int, float] = field(default_factory=dict) # f_k do Al-Thani: horas acumuladas no início 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 return max(0.0, min( insp.f_max - self.f0.get(seq, 0.0) for seq, insp in catalogo.items() )) 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].f_max - self.f0.get(s, 0.0), ) return seq_min, self.f0.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], **kwargs, ) -> 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 — Set Partitioning: toda missão coberta por exatamente 1 aeronave --- 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}" mdl += z[nid] == 1, f"part_{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}" 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.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 ) # 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].f_max, ) 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].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 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 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. time_limit = kwargs.get("time_limit", 120) solver = pulp.PULP_CBC_CMD(msg=False, timeLimit=time_limit) mdl.solve(solver) status_l1 = mdl.status prio_otima = int(round(pulp.value(mdl.objective) or 0)) # L2: com prioridade fixada, maximiza horas voadas. 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(pulp.PULP_CBC_CMD(msg=False, timeLimit=time_limit)) # PuLP redefine mdl.status após setObjective; preservar o status real (L1). if mdl.status == 0 and status_l1 == 1: mdl.status = status_l1 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 = [] horizonte_dt = sol.get("horizonte_dt") 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, }) # Inspeções executadas por esta aeronave if horizonte_dt is not None: for seq, s_val, insp in inspecoes_da_aeronave(sol, k): ini_dt = horizonte_dt + timedelta(hours=s_val) fim_dt = horizonte_dt + timedelta(hours=s_val + insp.dt_h) linhas.append({ "status": "inspecao", "aeronave": k, "ordem": "", "id": f"INSP_{k}_{seq}", "om": insp.sigla, "orig": "SBMN", "dest": "SBMN", "partida_utc": ini_dt.isoformat().replace("+00:00", "Z"), "chegada_utc": fim_dt.isoformat().replace("+00:00", "Z"), "dur_h": f"{insp.dt_h:.2f}", "prioridade": "", "periodo": "", "aeronave_real_2025": "", "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} 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] lrt_ini = a.lrt_inicial(catalogo) info = a.insp_mais_restritiva(catalogo) if info: seq_min, f0_val, insp = info print( f" {k}: LRT_ini={lrt_ini:.1f}h " f"(limitado por seq {seq_min} {insp.sigla}, " f"f0={f0_val:.1f}h, F_max={insp.f_max:.0f}h)" ) else: 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"]} 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 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"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]] = [] # 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, f_max=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 bases operacionais do C-105 BASES_COORDS: dict[str, tuple[float, float]] = { # Amazônia (bases primárias) "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é # Bases adicionais (escala de voo real) "SBMQ": ( 0.05, -51.07), # Macapá "SBOI": (-3.77, -38.53), # Fortaleza (Pinto Martins) "SBVH": (-12.69, -60.10), # Vilhena "SBBR": (-15.87, -47.92), # Brasília "SBGL": (-22.81, -43.25), # Rio de Janeiro (Galeão) "SBCY": (-15.65, -56.12), # Cuiabá "SBCC": (-10.86, -51.80), # Canarana "SBYS": (-23.00, -49.71), # Ourinhos "SBCO": (-22.17, -47.88), # Cosmópolis "SBSM": (-29.71, -53.69), # Santa Maria "SBLO": (-23.33, -51.13), # Londrina "SBAN": (-15.59, -56.10), # Anápolis "SWEI": (-7.63, -72.68), # Eirunepé "SWCA": (-8.85, -63.85), # Cacoal "SWKO": (-11.50, -61.45), # Rolim de Moura "SBTF": (-3.38, -64.72), # Tefé (alias SBTT) "SBUY": (-29.78, -57.04), # Uruguaiana } _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 carregar_escala(caminho: Path, inicio: datetime) -> tuple[list[Missao], datetime]: """ Lê a Escala de Voo Modelo 1 (CSV com separador ';') e retorna missões para o solver. Formato esperado (2 linhas de cabeçalho): DATA ; ETAPA ; DEP ; ARR ; HOR_DEP ; HOR_ARR ; DUR ; SEGMTO ; MISSÃO ; OFRAG Cada linha vira uma Missao. Bases sem coordenadas conhecidas são descartadas. A prioridade é derivada do tipo de missão: 69TV → prio 1 | 50TT/05TF → prio 3 | outros → prio 2 """ _MES = { "jan": 1, "fev": 2, "mar": 3, "abr": 4, "mai": 5, "jun": 6, "jul": 7, "ago": 8, "set": 9, "out": 10, "nov": 11, "dez": 12, } def _parse_hora(s: str) -> tuple[int, int]: parts = s.strip().split(":") return int(parts[0]) % 24, int(parts[1]) def _prio(missao_id: str) -> int: m = missao_id.upper() if "69TV" in m: return 1 if any(x in m for x in ("50TT", "05TF", "TT", "TF")): return 3 return 2 missoes: list[Missao] = [] ano = inicio.year data_atual: date | None = None with caminho.open(newline="", encoding="latin-1") as f: reader = csv.reader(f, delimiter=";") next(reader) # cabeçalho 1 next(reader) # cabeçalho 2 for i, row in enumerate(reader): if len(row) < 10: continue data_str, _, dep, arr, h_dep, h_arr, _, _, missao_id, ofrag = ( row[0].strip(), row[1].strip(), row[2].strip(), row[3].strip(), row[4].strip(), row[5].strip(), row[6].strip(), row[7].strip(), row[8].strip(), row[9].strip(), ) if data_str: partes = data_str.split("/") if len(partes) == 2: data_atual = date(ano, _MES.get(partes[1].lower(), 1), int(partes[0])) if data_atual is None or not dep or not arr: continue if dep not in BASES_COORDS or arr not in BASES_COORDS: continue try: hd, md = _parse_hora(h_dep) ha, ma = _parse_hora(h_arr) except Exception: continue t_dep_dt = datetime(data_atual.year, data_atual.month, data_atual.day, hd, md, tzinfo=timezone.utc) t_arr_dt = datetime(data_atual.year, data_atual.month, data_atual.day, ha, ma, tzinfo=timezone.utc) if t_arr_dt <= t_dep_dt: # chegada no dia seguinte t_arr_dt += timedelta(days=1) t_dep_h = (t_dep_dt - inicio).total_seconds() / 3600 t_arr_h = (t_arr_dt - inicio).total_seconds() / 3600 if t_dep_h < 0 or t_arr_h <= t_dep_h: continue missoes.append(Missao( id=f"ESC_{i:03d}_{dep}_{arr}", om=ofrag, orig=dep, dest=arr, t_dep=t_dep_h, t_arr=t_arr_h, dur_h=t_arr_h - t_dep_h, prioridade=_prio(missao_id), obrigatoria=False, partida_utc=t_dep_dt, chegada_utc=t_arr_dt, periodo="", linhas_origem="", aeronave_real="", codigo_missao=missao_id, )) if not missoes: raise ValueError(f"Nenhuma missão válida carregada de {caminho}") horizonte_dt = inicio return missoes, horizonte_dt 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 = 20, seed: Optional[int] = 42, inicio: Optional[datetime] = None, horizonte_h: float = 744.0, # janeiro = 31 × 24 n_evam: int = 4, 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. 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 if inicio is None: inicio = datetime(2025, 1, 1, tzinfo=timezone.utc) 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] = [] # 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) 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=missao_id, orig=orig, dest=dest, t_dep=t_dep_h, t_arr=t_arr_h, dur_h=dur_h, prioridade=prio, 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 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.f_max, 0.9 * insp.f_max), 1) for seq, insp in catalogo.items() } for k in aeronaves } 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 gerar_f0_aleatorio(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( "--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", 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( "--escala", type=Path, default=None, metavar="CSV", help="Escala de Voo Modelo 1 (CSV ';') com missões reais.", ) 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=20, help="Número de missões sintéticas a gerar (padrão: 20).", ) parser.add_argument( "--n-evam", type=int, default=4, help="Missões obrigatórias (EVAM/hard) nas sintéticas (padrão: 4).", ) parser.add_argument( "--ofrag", type=Path, default=None, metavar="CSV", help="CSV de OFRAG com prioridades por missão (missao_id, prioridade).", ) parser.add_argument( "--time-limit", type=int, default=120, dest="time_limit", help="Limite de tempo em segundos para o solver CBC (padrão: 120).", ) parser.add_argument( "--insp-seqs", type=int, nargs="+", default=None, dest="insp_seqs", metavar="SEQ", help="Filtra catálogo para apenas estes seq de inspeção (ex: --insp-seqs 3).", ) 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.insp_seqs: catalogo = {s: v for s, v in catalogo.items() if s in args.insp_seqs} if args.escala: missoes, horizonte_dt = carregar_escala(args.escala, inicio) bases = sorted({m.orig for m in missoes} | {m.dest for m in missoes}) print(f" Escala de voo: {len(missoes)} missões carregadas de {args.escala.name}") elif 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, ofrag_path=args.ofrag, ) 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_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, f0=f0_map.get(k, {})) for k in args.aeronaves ] nos, arcos = construir_rede(missoes, aeronaves, catalogo) sol = resolver(nos, arcos, aeronaves, catalogo, time_limit=args.time_limit) sol["horizonte_dt"] = horizonte_dt escrever_resultado(sol, args.resultado) imprimir_relatorio(sol, bases, args.resultado) if __name__ == "__main__": main()