120 KiB
120 KiB
OPS04 정답 코드 - 샤헤드·미사일 포화경보 우선순위 대시보드 모의 구현
[교육용가상자료] 모든 데이터는 코드 안에서 난수로 만든 가상 자료이며 실제 부대·장비·인원·작전 정보와 무관합니다.
실행 환경
- Python 3.10 이상, numpy, pandas, scipy, scikit-learn, matplotlib
- 미션별 코드는 각각 한 파일로 독립 실행됩니다. 외부 파일 없이 코드 안에서 가상 데이터를 생성합니다.
- 그래프의 글자는 한글 글꼴이 없는 실행 환경에서도 깨지지 않도록 영문으로 표기했습니다. 표·출력 결과는 한글입니다.
- 난수 시드를 고정해 두어 같은 코드를 실행하면 같은 결과가 나옵니다.
미션 1. 샤헤드 자폭드론 및 순항미사일 복합 비행 특성 분석 및 포화경보 데이터 규격 정의
- 난이도: 초급 / 예상 소요시간: 1.5시간
- 코드 설명: 포화경보 입력 메시지의 데이터 규격(JSON 스키마)을 정의하고, 가상 샘플 20건 검증과 오류 5종 주입 검출로 검증기를 확인한 뒤 데이터 명세서를 출력한다.
"""포화경보 입력 메시지의 데이터 규격(JSON 스키마)을 정의하고, 가상 샘플 20건 검증과 오류 5종 주입 검출로 검증기를 확인한 뒤 데이터 명세서를 출력한다."""
import json
import numpy as np
import pandas as pd
# ---------------- [교육용가상자료] 가상 방어지역·가상 비행체 생성기 ----------------
# 좌표는 가상 지휘소를 원점으로 한 km 단위 평면이다. 비행체 유형은 추상 범주이며
# 속도·고도·반사강도는 학습용 임의 범위로, 실제 무기 제원·사거리·탄도와 무관하다.
ASSETS = pd.DataFrame([["A1", "CP(virtual)", 0.0, 0.0, 1.0],
["A2", "Radar(virtual)", 4.0, -3.0, 0.8],
["A3", "Depot(virtual)", -5.0, 4.0, 0.7],
["A4", "Power(virtual)", 7.0, 6.0, 0.5]], columns=["asset", "name", "x", "y", "value"])
TYPES = { # 유형: (위협 여부, 속도 m/s 범위, 고도 m 범위, 출현 비율)
"uav_type": (True, (30, 60), (100, 800), 0.42), # 저속 무인기형(가상)
"fast_type": (True, (150, 250), (50, 200), 0.16), # 고속 순항형(가상)
"decoy": (False, (30, 60), (100, 800), 0.16), # 기만체(가상)
"civil": (False, (120, 220), (3000, 9000), 0.10), # 민항기(가상)
"bird": (False, (8, 18), (50, 400), 0.16), # 조류 군집
}
TTI_URGENT = 600 # 긴급 판정 기준: 가상 자산 도달 예상시간 600초 이내
CPA_KM = 1.5 # 자산 상공 통과 판정 반경(가상)
def make_raid(seed, n=80, spoof=0):
"""가상 포화 상황 1회분(비행체 n개 + 스푸핑 가짜 항적 spoof개)을 만든다."""
rng = np.random.default_rng(seed)
names = list(TYPES)
rows = []
for i in range(n + spoof):
typ = "spoof" if i >= n else str(rng.choice(names, p=[TYPES[k][3] for k in names]))
threat, sp, al, _ = TYPES["decoy" if typ == "spoof" else typ]
brg = np.radians(rng.uniform(0, 360)); dist = rng.uniform(6, 60)
x, y = dist * np.sin(brg), dist * np.cos(brg)
spd = rng.uniform(*sp); alt = rng.uniform(*al)
aim = int(rng.integers(0, len(ASSETS)))
if typ in ("uav_type", "fast_type", "decoy", "spoof"): # 가상 자산 방향(작은 방위 오차)
ax_, ay_ = ASSETS.loc[aim, ["x", "y"]]
hd = np.arctan2(ax_ - x, ay_ - y) + np.radians(rng.normal(0, 0.6))
else:
hd = rng.uniform(0, 2 * np.pi); aim = -1
vx, vy = spd * np.sin(hd) / 1000, spd * np.cos(hd) / 1000 # km/s
sig = {"uav_type": ((0.2, 0.5), (0.02, 0.12), [0, .1, .5, .4]), "fast_type": ((0.1, 0.4), (0.02, 0.12), [0, .2, .4, .4]),
"decoy": ((0.3, 0.8), (0.08, 0.35), [0, .55, .4, .05]), "spoof": ((0.3, 0.8), (0.05, 0.3), [0, .7, .3, 0]),
"civil": ((0.7, 1.0), (0.01, 0.05), [0, 0, .1, .9]), "bird": ((0.01, 0.12), (0.1, 0.4), [0, .6, .4, 0])}[typ]
jump = spd * rng.uniform(0.95, 1.2) if typ != "spoof" else spd * rng.uniform(4, 15)
true_tti = np.inf
if threat:
d_aim = np.hypot(ASSETS.loc[aim, "x"] - x, ASSETS.loc[aim, "y"] - y)
true_tti = d_aim * 1000 / spd
rows.append({"id": f"T{seed % 100:02d}{i:03d}", "type": typ, "x": x, "y": y, "vx": vx, "vy": vy, "alt": alt, "spd": spd,
"rcs": rng.uniform(*sig[0]), "rcs_var": rng.uniform(*sig[1]), "n_sensors": int(rng.choice(4, p=sig[2])),
"max_jump": jump, "civ_iff": int(typ == "civil" and rng.random() < 0.95),
"aim": aim, "threat": threat, "true_tti": true_tti,
"urgent": bool(threat and true_tti <= TTI_URGENT)})
df = pd.DataFrame(rows)
df["true_prio"] = [ASSETS.loc[a, "value"] * np.exp(-t / 300) if th else 0.0
for a, t, th in zip(df["aim"], df["true_tti"], df["threat"])]
return df
def geometry(df):
"""각 항적과 가상 자산 사이 최근접점(CPA) 거리와 도달 예상시간(TTI, s)을 계산한다."""
px, py = df["x"].to_numpy()[:, None], df["y"].to_numpy()[:, None]
vx, vy = df["vx"].to_numpy()[:, None], df["vy"].to_numpy()[:, None]
rx, ry = ASSETS["x"].to_numpy()[None] - px, ASSETS["y"].to_numpy()[None] - py
v2 = vx ** 2 + vy ** 2 + 1e-12
t = (rx * vx + ry * vy) / v2 # CPA 도달 시각(s)
cpa = np.hypot(rx - vx * t, ry - vy * t)
tti = np.where((t > 0) & (cpa < CPA_KM), t, np.inf)
return cpa, tti
# 포화경보 입력 메시지 스키마. 실제 무기 제원(탄두 위력·사거리 등) 대신 추상 '위협 가중 등급'을 쓴다.
SCHEMA = {
"track_id": {"type": str, "rule": "T+숫자 5자리", "check": lambda v: v[0] == "T" and v[1:].isdigit() and len(v) == 6},
"sensor_id": {"type": str, "allowed": ["RDR-1", "RDR-2", "EO-1", "ACU-1"]},
"timestamp_ms": {"type": int, "range": (0, 10 ** 13)},
"x_km": {"type": float, "range": (-100, 100)},
"y_km": {"type": float, "range": (-100, 100)},
"alt_m": {"type": float, "range": (0, 15000)},
"speed_mps": {"type": float, "range": (0, 400)},
"heading_deg": {"type": float, "range": (0, 360)},
"approach_dir": {"type": str, "allowed": ["N", "NE", "E", "SE", "S", "SW", "W", "NW"]},
"class_hint": {"type": str, "allowed": ["uav_type", "fast_type", "unknown", "civil", "bird"]},
"threat_grade": {"type": int, "range": (1, 3)}, # 위협 가중 등급(가상, 1 낮음 ~ 3 높음)
"rcs": {"type": float, "range": (0, 1)}, # 정규화 반사강도(가상)
"weather": {"type": str, "allowed": ["clear", "rain", "fog", "snow"]},
"pos_error_m": {"type": float, "range": (0, 2000)}, # 보완 전술 파라미터: 위치 오차
"comm_delay_ms": {"type": int, "range": (0, 5000)}, # 보완 전술 파라미터: 통신 지연
}
REQUIRED = list(SCHEMA)
def validate(msg):
errs = []
for f in REQUIRED:
if f not in msg or msg[f] is None:
errs.append(f"{f}: 누락"); continue
spec, v = SCHEMA[f], msg[f]
if spec["type"] is float and isinstance(v, int) and not isinstance(v, bool):
v = float(v)
if not isinstance(v, spec["type"]):
errs.append(f"{f}: 형식 오류({type(v).__name__})"); continue
if "allowed" in spec and v not in spec["allowed"]:
errs.append(f"{f}: 허용값 아님({v})")
if "range" in spec and not (spec["range"][0] <= v <= spec["range"][1]):
errs.append(f"{f}: 범위 초과({v})")
if "check" in spec and not spec["check"](v):
errs.append(f"{f}: 규칙 위반({v})")
extra = set(msg) - set(SCHEMA)
if extra:
errs.append(f"정의되지 않은 필드: {sorted(extra)}")
return errs
def to_message(r, rng):
hd = (np.degrees(np.arctan2(r.vx, r.vy)) + 360) % 360
frm = (hd + 180) % 360
hint = {"uav_type": "uav_type", "fast_type": "fast_type", "civil": "civil", "bird": "bird"}.get(r.type, "unknown")
return {"track_id": r.id, "sensor_id": str(rng.choice(SCHEMA["sensor_id"]["allowed"])), "timestamp_ms": int(1.7e12 + rng.integers(0, 60000)),
"x_km": round(float(r.x), 2), "y_km": round(float(r.y), 2), "alt_m": round(float(r.alt), 0), "speed_mps": round(float(r.spd), 1),
"heading_deg": round(float(hd), 1), "approach_dir": ["N", "NE", "E", "SE", "S", "SW", "W", "NW"][int(((frm + 22.5) % 360) // 45)],
"class_hint": hint, "threat_grade": 3 if r.type == "fast_type" else (2 if r.threat else 1), "rcs": round(float(r.rcs), 2),
"weather": str(rng.choice(["clear", "rain", "fog", "snow"])), "pos_error_m": round(float(rng.uniform(20, 300)), 0),
"comm_delay_ms": int(rng.integers(50, 800))}
rng = np.random.default_rng(1)
raid = make_raid(11, 20)
samples = [to_message(r, rng) for r in raid.itertuples()]
normal_err = sum(len(validate(m)) for m in samples)
print(f"[정상 샘플 검증] {len(samples)}건, 오류 {normal_err}건")
print("샘플 1건:", json.dumps(samples[0], ensure_ascii=False)[:150], "...")
bad = [dict(samples[0]) for _ in range(5)]
bad[0]["speed_mps"] = "fast" # 형식 오류
bad[1]["sensor_id"] = "RDR-9" # 허용값 오류
bad[2]["alt_m"] = -50.0 # 범위 오류
del bad[3]["comm_delay_ms"] # 누락(통신 지연)
bad[4]["warhead_kg"] = 50 # 정의되지 않은 필드(실제 제원 입력 차단)
kinds = ["형식", "허용값", "범위", "누락", "미정의 필드"]
print("\n[오류 주입 검출]")
detected = 0
for k, m in zip(kinds, bad):
e = validate(m); detected += bool(e)
print(f" {k:7s} -> {'검출' if e else '미검출'}: {e[0] if e else ''}")
print("\n[데이터 명세 요약]")
spec = pd.DataFrame([{"필드": f, "형식": s["type"].__name__, "규칙": s.get("allowed", s.get("range", s.get("rule")))} for f, s in SCHEMA.items()])
print(spec.to_string(index=False))
complete = sum(all(k in s for k in ("type",)) and any(k in s for k in ("allowed", "range", "check")) for s in SCHEMA.values()) / len(SCHEMA)
print(f"\n스키마 필드 정의 완성도 {complete:.0%} (목표 100%) -> {'달성' if complete == 1 else '미달'}")
print(f"샘플 검증 오류 {normal_err}건 (목표 0건) -> {'달성' if normal_err == 0 else '미달'}, 주입 오류 {detected}/5 검출")
print("참고: 실제 무기 제원·사거리 필드는 두지 않으며, 미정의 필드는 검증 단계에서 차단한다.")
실행 결과 예시
[정상 샘플 검증] 20건, 오류 0건
샘플 1건: {"track_id": "T11000", "sensor_id": "RDR-2", "timestamp_ms": 1700000030709, "x_km": 0.17, "y_km": -38.48, "alt_m": 204.0, "speed_mps": 30.9, "heading_ ...
[오류 주입 검출]
형식 -> 검출: speed_mps: 형식 오류(str)
허용값 -> 검출: sensor_id: 허용값 아님(RDR-9)
범위 -> 검출: alt_m: 범위 초과(-50.0)
누락 -> 검출: comm_delay_ms: 누락
미정의 필드 -> 검출: 정의되지 않은 필드: ['warhead_kg']
[데이터 명세 요약]
필드 형식 규칙
track_id str T+숫자 5자리
sensor_id str [RDR-1, RDR-2, EO-1, ACU-1]
timestamp_ms int (0, 10000000000000)
x_km float (-100, 100)
y_km float (-100, 100)
alt_m float (0, 15000)
speed_mps float (0, 400)
heading_deg float (0, 360)
approach_dir str [N, NE, E, SE, S, SW, W, NW]
class_hint str [uav_type, fast_type, unknown, civil, bird]
threat_grade int (1, 3)
rcs float (0, 1)
weather str [clear, rain, fog, snow]
pos_error_m float (0, 2000)
comm_delay_ms int (0, 5000)
스키마 필드 정의 완성도 100% (목표 100%) -> 달성
샘플 검증 오류 0건 (목표 0건) -> 달성, 주입 오류 5/5 검출
참고: 실제 무기 제원·사거리 필드는 두지 않으며, 미정의 필드는 검증 단계에서 차단한다.
미션 2. 다방향 동시 침투 항적 데이터 정제 및 센서별 경보 메시지 큐잉 전처리 구현
- 난이도: 초급~중급 / 예상 소요시간: 2.0시간
- 코드 설명: 다방향에서 동시에 들어오는 가상 센서 경보 메시지를 센서별 큐에 받아 타임스탬프 순으로 정렬하고, 중복·반사파(유령 항적)·스푸핑·센서 단절을 정제하는 전처리 파이프라인을 구현해 처리 속도와 오류율을 측정한다.
"""다방향에서 동시에 들어오는 가상 센서 경보 메시지를 센서별 큐에 받아 타임스탬프 순으로 정렬하고, 중복·반사파(유령 항적)·스푸핑·센서 단절을 정제하는 전처리 파이프라인을 구현해 처리 속도와 오류율을 측정한다."""
import heapq
import time
from collections import defaultdict, deque
import numpy as np
import pandas as pd
# ---------------- [교육용가상자료] 가상 방어지역·가상 비행체 생성기 ----------------
# 좌표는 가상 지휘소를 원점으로 한 km 단위 평면이다. 비행체 유형은 추상 범주이며
# 속도·고도·반사강도는 학습용 임의 범위로, 실제 무기 제원·사거리·탄도와 무관하다.
ASSETS = pd.DataFrame([["A1", "CP(virtual)", 0.0, 0.0, 1.0],
["A2", "Radar(virtual)", 4.0, -3.0, 0.8],
["A3", "Depot(virtual)", -5.0, 4.0, 0.7],
["A4", "Power(virtual)", 7.0, 6.0, 0.5]], columns=["asset", "name", "x", "y", "value"])
TYPES = { # 유형: (위협 여부, 속도 m/s 범위, 고도 m 범위, 출현 비율)
"uav_type": (True, (30, 60), (100, 800), 0.42), # 저속 무인기형(가상)
"fast_type": (True, (150, 250), (50, 200), 0.16), # 고속 순항형(가상)
"decoy": (False, (30, 60), (100, 800), 0.16), # 기만체(가상)
"civil": (False, (120, 220), (3000, 9000), 0.10), # 민항기(가상)
"bird": (False, (8, 18), (50, 400), 0.16), # 조류 군집
}
TTI_URGENT = 600 # 긴급 판정 기준: 가상 자산 도달 예상시간 600초 이내
CPA_KM = 1.5 # 자산 상공 통과 판정 반경(가상)
def make_raid(seed, n=80, spoof=0):
"""가상 포화 상황 1회분(비행체 n개 + 스푸핑 가짜 항적 spoof개)을 만든다."""
rng = np.random.default_rng(seed)
names = list(TYPES)
rows = []
for i in range(n + spoof):
typ = "spoof" if i >= n else str(rng.choice(names, p=[TYPES[k][3] for k in names]))
threat, sp, al, _ = TYPES["decoy" if typ == "spoof" else typ]
brg = np.radians(rng.uniform(0, 360)); dist = rng.uniform(6, 60)
x, y = dist * np.sin(brg), dist * np.cos(brg)
spd = rng.uniform(*sp); alt = rng.uniform(*al)
aim = int(rng.integers(0, len(ASSETS)))
if typ in ("uav_type", "fast_type", "decoy", "spoof"): # 가상 자산 방향(작은 방위 오차)
ax_, ay_ = ASSETS.loc[aim, ["x", "y"]]
hd = np.arctan2(ax_ - x, ay_ - y) + np.radians(rng.normal(0, 0.6))
else:
hd = rng.uniform(0, 2 * np.pi); aim = -1
vx, vy = spd * np.sin(hd) / 1000, spd * np.cos(hd) / 1000 # km/s
sig = {"uav_type": ((0.2, 0.5), (0.02, 0.12), [0, .1, .5, .4]), "fast_type": ((0.1, 0.4), (0.02, 0.12), [0, .2, .4, .4]),
"decoy": ((0.3, 0.8), (0.08, 0.35), [0, .55, .4, .05]), "spoof": ((0.3, 0.8), (0.05, 0.3), [0, .7, .3, 0]),
"civil": ((0.7, 1.0), (0.01, 0.05), [0, 0, .1, .9]), "bird": ((0.01, 0.12), (0.1, 0.4), [0, .6, .4, 0])}[typ]
jump = spd * rng.uniform(0.95, 1.2) if typ != "spoof" else spd * rng.uniform(4, 15)
true_tti = np.inf
if threat:
d_aim = np.hypot(ASSETS.loc[aim, "x"] - x, ASSETS.loc[aim, "y"] - y)
true_tti = d_aim * 1000 / spd
rows.append({"id": f"T{seed % 100:02d}{i:03d}", "type": typ, "x": x, "y": y, "vx": vx, "vy": vy, "alt": alt, "spd": spd,
"rcs": rng.uniform(*sig[0]), "rcs_var": rng.uniform(*sig[1]), "n_sensors": int(rng.choice(4, p=sig[2])),
"max_jump": jump, "civ_iff": int(typ == "civil" and rng.random() < 0.95),
"aim": aim, "threat": threat, "true_tti": true_tti,
"urgent": bool(threat and true_tti <= TTI_URGENT)})
df = pd.DataFrame(rows)
df["true_prio"] = [ASSETS.loc[a, "value"] * np.exp(-t / 300) if th else 0.0
for a, t, th in zip(df["aim"], df["true_tti"], df["threat"])]
return df
def geometry(df):
"""각 항적과 가상 자산 사이 최근접점(CPA) 거리와 도달 예상시간(TTI, s)을 계산한다."""
px, py = df["x"].to_numpy()[:, None], df["y"].to_numpy()[:, None]
vx, vy = df["vx"].to_numpy()[:, None], df["vy"].to_numpy()[:, None]
rx, ry = ASSETS["x"].to_numpy()[None] - px, ASSETS["y"].to_numpy()[None] - py
v2 = vx ** 2 + vy ** 2 + 1e-12
t = (rx * vx + ry * vy) / v2 # CPA 도달 시각(s)
cpa = np.hypot(rx - vx * t, ry - vy * t)
tti = np.where((t > 0) & (cpa < CPA_KM), t, np.inf)
return cpa, tti
SENSORS = ["RDR-1", "RDR-2", "EO-1", "ACU-1"]
def make_stream(seed=21, n=120, steps=30):
"""[교육용가상자료] 센서별 메시지 스트림. label: valid/dup/ghost/spoof (채점용 정답)"""
rng = np.random.default_rng(seed)
raid = make_raid(seed, n, spoof=6)
queues = {s: [] for s in SENSORS}
mid = 0
for r in raid.itertuples():
sens = list(rng.choice(SENSORS, size=max(1, r.n_sensors), replace=False))
for k in range(steps):
if r.Index % 9 == 0 and 10 <= k < 15: # 일부 항적은 센서 단절 구간
continue
t = k * 1000 + int(rng.integers(0, 300))
x, y = r.x + r.vx * k, r.y + r.vy * k
if r.type == "spoof": # 스푸핑: 위치가 비정상적으로 도약
x, y = x + rng.normal(0, 3), y + rng.normal(0, 3)
for s in sens:
m = {"mid": mid, "sensor": s, "track": r.id, "ts": t + int(rng.integers(0, 80)), "x": x, "y": y,
"spd": r.spd, "snr": rng.uniform(0.4, 1.0), "label": "spoof" if r.type == "spoof" else "valid"}
queues[s].append(m); mid += 1
if rng.random() < 0.08: # 중복 재전송
queues[s].append({**m, "mid": mid, "label": "dup"}); mid += 1
if s.startswith("RDR") and rng.random() < 0.03: # 반사파 유령 항적(약한 신호, 대칭 위치)
queues[s].append({**m, "mid": mid, "track": "G" + r.id[1:], "x": -x, "snr": rng.uniform(0.05, 0.2),
"label": "ghost"}); mid += 1
for s in SENSORS: # 전송 지연으로 순서가 섞여 도착
q = queues[s]
idx = rng.permutation(len(q))[: len(q) // 10]
for a, b in zip(idx[::2], idx[1::2]):
q[a], q[b] = q[b], q[a]
return queues
def preprocess(queues, snr_min=0.25, jump_ratio=3.0, gap_ms=5000):
stats = defaultdict(int)
# 1) 센서별 큐 -> 타임스탬프 순 병합(각 큐를 먼저 정렬한 뒤 k-way merge)
sorted_q = [sorted(q, key=lambda m: (m["ts"], m["mid"])) for q in queues.values()]
merged = list(heapq.merge(*sorted_q, key=lambda m: (m["ts"], m["mid"])))
# 2) 동일 센서 중복 제거 + 반사파(약신호) 제거
seen, clean = set(), []
for m in merged:
key = (m["sensor"], m["track"], m["ts"] // 1000)
if key in seen:
stats["중복 제거"] += 1; m["drop"] = True; continue
if m["snr"] < snr_min:
stats["반사파 제거"] += 1; m["drop"] = True; continue
seen.add(key); clean.append(m)
# 3) 센서 간 융합(같은 항적, 같은 1초 창) 및 스푸핑 판정(속도 대비 위치 도약)
fused = {}
for m in clean:
k = (m["track"], m["ts"] // 1000)
if k in fused:
fused[k]["src"].append(m["mid"]); fused[k]["sensors"].add(m["sensor"])
else:
fused[k] = {"track": m["track"], "sec": m["ts"] // 1000, "x": m["x"], "y": m["y"], "spd": m["spd"],
"src": [m["mid"]], "sensors": {m["sensor"]}}
by_track = defaultdict(list)
for f in fused.values():
by_track[f["track"]].append(f)
spoofed, gaps = set(), 0
for tr, fs in by_track.items():
fs.sort(key=lambda f: f["sec"])
jumps = [np.hypot(b["x"] - a["x"], b["y"] - a["y"]) * 1000 / max(b["sec"] - a["sec"], 1) for a, b in zip(fs, fs[1:])]
if jumps and np.median(jumps) > jump_ratio * fs[0]["spd"]:
spoofed.add(tr)
gaps += sum((b["sec"] - a["sec"]) * 1000 > gap_ms for a, b in zip(fs, fs[1:]))
out = [f for f in fused.values() if f["track"] not in spoofed]
stats["스푸핑 항적 격리"] = len(spoofed); stats["센서 단절 구간 표시"] = gaps
return out, stats, merged
queues = make_stream()
n_in = sum(len(q) for q in queues.values())
t0 = time.perf_counter()
out, stats, merged = preprocess(queues)
dt = time.perf_counter() - t0
rate = n_in / dt
kept = {mid for f in out for mid in f["src"]}
labels = {m["mid"]: m["label"] for m in merged}
wrong_drop = sum(1 for k, l in labels.items() if l == "valid" and k not in kept)
wrong_keep = sum(1 for k, l in labels.items() if l != "valid" and k in kept)
err = (wrong_drop + wrong_keep) / n_in
print(f"[전처리 전] 메시지 {n_in}건 (센서 {len(queues)}개), 유형별:", pd.Series(labels).value_counts().to_dict())
print(f"[전처리 후] 융합 보고 {len(out)}건, 항적 {len({f['track'] for f in out})}개")
for k, v in stats.items():
print(f" - {k}: {v}")
ts = [m["ts"] for m in merged]
print(f" - 타임스탬프 정렬 위반: {sum(b < a for a, b in zip(ts, ts[1:]))}건")
multi = sum(len(f["sensors"]) > 1 for f in out) / len(out)
print(f" - 다중 센서 융합 비율: {multi:.1%}")
print(f"\n처리 시간 {dt * 1000:.1f} ms, 처리 속도 {rate:,.0f}건/초 (목표 1,000건/초 이상) -> {'달성' if rate >= 1000 else '미달'}")
print(f"정상 메시지 오삭제 {wrong_drop}건, 잡음 미제거 {wrong_keep}건, 전처리 오류율 {err:.2%} (목표 0%) -> {'달성' if err == 0 else '미달'}")
print("\n[전처리 후 데이터 앞부분]")
print(pd.DataFrame([{"track": f["track"], "sec": f["sec"], "x_km": round(f["x"], 2), "y_km": round(f["y"], 2), "센서수": len(f["sensors"])}
for f in sorted(out, key=lambda f: (f["sec"], f["track"]))[:6]]).to_string(index=False))
실행 결과 예시
[전처리 전] 메시지 8100건 (센서 4개), 유형별: {'valid': 7145, 'dup': 612, 'spoof': 240, 'ghost': 103}
[전처리 후] 융합 보고 3530건, 항적 120개
- 반사파 제거: 103
- 중복 제거: 612
- 스푸핑 항적 격리: 6
- 센서 단절 구간 표시: 14
- 타임스탬프 정렬 위반: 0건
- 다중 센서 융합 비율: 71.8%
처리 시간 49.3 ms, 처리 속도 164,466건/초 (목표 1,000건/초 이상) -> 달성
정상 메시지 오삭제 0건, 잡음 미제거 0건, 전처리 오류율 0.00% (목표 0%) -> 달성
[전처리 후 데이터 앞부분]
track sec x_km y_km 센서수
T21000 0 -27.36 -34.88 3
T21001 0 50.43 13.12 2
T21002 0 0.03 20.05 3
T21003 0 -8.61 -25.31 3
T21004 0 -13.38 -24.24 1
T21005 0 50.83 -22.68 2
미션 3. 속도·고도·접근 방위각 기반 1차 공중위협 우선순위 산정 알고리즘 구현
- 난이도: 중급 / 예상 소요시간: 2.5시간
- 코드 설명: 가상 방어자산까지의 최근접 거리와 잔여 도달 예상시간(TTI), 접근 속도만으로 1차 위협 순위를 정렬하는 Baseline 랭킹 알고리즘을 구현하고 긴급 판정 정확도를 측정한다.
"""가상 방어자산까지의 최근접 거리와 잔여 도달 예상시간(TTI), 접근 속도만으로 1차 위협 순위를 정렬하는 Baseline 랭킹 알고리즘을 구현하고 긴급 판정 정확도를 측정한다."""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler
# ---------------- [교육용가상자료] 가상 방어지역·가상 비행체 생성기 ----------------
# 좌표는 가상 지휘소를 원점으로 한 km 단위 평면이다. 비행체 유형은 추상 범주이며
# 속도·고도·반사강도는 학습용 임의 범위로, 실제 무기 제원·사거리·탄도와 무관하다.
ASSETS = pd.DataFrame([["A1", "CP(virtual)", 0.0, 0.0, 1.0],
["A2", "Radar(virtual)", 4.0, -3.0, 0.8],
["A3", "Depot(virtual)", -5.0, 4.0, 0.7],
["A4", "Power(virtual)", 7.0, 6.0, 0.5]], columns=["asset", "name", "x", "y", "value"])
TYPES = { # 유형: (위협 여부, 속도 m/s 범위, 고도 m 범위, 출현 비율)
"uav_type": (True, (30, 60), (100, 800), 0.42), # 저속 무인기형(가상)
"fast_type": (True, (150, 250), (50, 200), 0.16), # 고속 순항형(가상)
"decoy": (False, (30, 60), (100, 800), 0.16), # 기만체(가상)
"civil": (False, (120, 220), (3000, 9000), 0.10), # 민항기(가상)
"bird": (False, (8, 18), (50, 400), 0.16), # 조류 군집
}
TTI_URGENT = 600 # 긴급 판정 기준: 가상 자산 도달 예상시간 600초 이내
CPA_KM = 1.5 # 자산 상공 통과 판정 반경(가상)
def make_raid(seed, n=80, spoof=0):
"""가상 포화 상황 1회분(비행체 n개 + 스푸핑 가짜 항적 spoof개)을 만든다."""
rng = np.random.default_rng(seed)
names = list(TYPES)
rows = []
for i in range(n + spoof):
typ = "spoof" if i >= n else str(rng.choice(names, p=[TYPES[k][3] for k in names]))
threat, sp, al, _ = TYPES["decoy" if typ == "spoof" else typ]
brg = np.radians(rng.uniform(0, 360)); dist = rng.uniform(6, 60)
x, y = dist * np.sin(brg), dist * np.cos(brg)
spd = rng.uniform(*sp); alt = rng.uniform(*al)
aim = int(rng.integers(0, len(ASSETS)))
if typ in ("uav_type", "fast_type", "decoy", "spoof"): # 가상 자산 방향(작은 방위 오차)
ax_, ay_ = ASSETS.loc[aim, ["x", "y"]]
hd = np.arctan2(ax_ - x, ay_ - y) + np.radians(rng.normal(0, 0.6))
else:
hd = rng.uniform(0, 2 * np.pi); aim = -1
vx, vy = spd * np.sin(hd) / 1000, spd * np.cos(hd) / 1000 # km/s
sig = {"uav_type": ((0.2, 0.5), (0.02, 0.12), [0, .1, .5, .4]), "fast_type": ((0.1, 0.4), (0.02, 0.12), [0, .2, .4, .4]),
"decoy": ((0.3, 0.8), (0.08, 0.35), [0, .55, .4, .05]), "spoof": ((0.3, 0.8), (0.05, 0.3), [0, .7, .3, 0]),
"civil": ((0.7, 1.0), (0.01, 0.05), [0, 0, .1, .9]), "bird": ((0.01, 0.12), (0.1, 0.4), [0, .6, .4, 0])}[typ]
jump = spd * rng.uniform(0.95, 1.2) if typ != "spoof" else spd * rng.uniform(4, 15)
true_tti = np.inf
if threat:
d_aim = np.hypot(ASSETS.loc[aim, "x"] - x, ASSETS.loc[aim, "y"] - y)
true_tti = d_aim * 1000 / spd
rows.append({"id": f"T{seed % 100:02d}{i:03d}", "type": typ, "x": x, "y": y, "vx": vx, "vy": vy, "alt": alt, "spd": spd,
"rcs": rng.uniform(*sig[0]), "rcs_var": rng.uniform(*sig[1]), "n_sensors": int(rng.choice(4, p=sig[2])),
"max_jump": jump, "civ_iff": int(typ == "civil" and rng.random() < 0.95),
"aim": aim, "threat": threat, "true_tti": true_tti,
"urgent": bool(threat and true_tti <= TTI_URGENT)})
df = pd.DataFrame(rows)
df["true_prio"] = [ASSETS.loc[a, "value"] * np.exp(-t / 300) if th else 0.0
for a, t, th in zip(df["aim"], df["true_tti"], df["threat"])]
return df
def geometry(df):
"""각 항적과 가상 자산 사이 최근접점(CPA) 거리와 도달 예상시간(TTI, s)을 계산한다."""
px, py = df["x"].to_numpy()[:, None], df["y"].to_numpy()[:, None]
vx, vy = df["vx"].to_numpy()[:, None], df["vy"].to_numpy()[:, None]
rx, ry = ASSETS["x"].to_numpy()[None] - px, ASSETS["y"].to_numpy()[None] - py
v2 = vx ** 2 + vy ** 2 + 1e-12
t = (rx * vx + ry * vy) / v2 # CPA 도달 시각(s)
cpa = np.hypot(rx - vx * t, ry - vy * t)
tti = np.where((t > 0) & (cpa < CPA_KM), t, np.inf)
return cpa, tti
# ---------------- 위협 점수 모델(추상 모델: 거리·도달시간·자산 가치·위협 유사도) ----------------
FEATS = ["log_spd", "alt_km", "rcs", "rcs_var", "n_sensors", "civ_iff", "jump_ratio"]
def features(df):
return pd.DataFrame({"log_spd": np.log(df["spd"]), "alt_km": df["alt"] / 1000, "rcs": df["rcs"], "rcs_var": df["rcs_var"],
"n_sensors": df["n_sensors"], "civ_iff": df["civ_iff"], "jump_ratio": df["max_jump"] / df["spd"]})
def train_classifier(seeds=(101, 102, 103, 104)):
"""학습용 가상 상황(시험 상황과 다른 시드)으로 '위협 유사도' 분류기를 학습한다."""
tr = pd.concat([make_raid(s, 120, spoof=10) for s in seeds], ignore_index=True)
clf = make_pipeline(StandardScaler(), LogisticRegression(max_iter=500, C=1.0))
clf.fit(features(tr)[FEATS], tr["threat"].astype(int))
return clf
def baseline_rank(df):
"""Baseline: 거리·도달시간만으로 1차 위협 점수를 매긴다(자산 가치·유형 판별 없음)."""
_, tti = geometry(df)
t_min = tti.min(axis=1)
out = df[["id"]].copy()
out["tti_s"] = t_min
out["score"] = np.exp(-t_min / 300)
out["urgent_pred"] = t_min <= TTI_URGENT
return out.sort_values("score", ascending=False)
def advanced_rank(df, clf, w=None):
"""고도화: 자산 가치 x 도달시간 x 위협 유사도, 운동학 불일치(스푸핑 의심) 항적은 별도 표시한다."""
w = w or {"tau": 300, "p_min": 0.5, "value_pow": 1.0, "jump_max": 2.0}
_, tti = geometry(df)
val = ASSETS["value"].to_numpy()[None] ** w["value_pow"]
asset_score = val * np.exp(-tti / w["tau"])
best = asset_score.argmax(axis=1)
p = clf.predict_proba(features(df)[FEATS])[:, 1]
spoof = (df["max_jump"] / df["spd"]).to_numpy() > w["jump_max"]
out = df[["id"]].copy()
out["asset"] = ASSETS["asset"].to_numpy()[best]
out["tti_s"] = tti[np.arange(len(df)), best]
out["p_threat"] = p
out["spoof_flag"] = spoof
out["score"] = np.where(spoof, 0, p * asset_score.max(axis=1))
out["urgent_pred"] = (p >= w["p_min"]) & (out["tti_s"] <= TTI_URGENT) & ~spoof
return out.sort_values("score", ascending=False)
# 판단 기준(투명성): ① 비행 경로가 가상 자산 반경 1.5 km 안을 지나는가(CPA)
# ② 그 자산까지 도달 예상시간(TTI)이 짧은가 ③ TTI 600초 이내면 '긴급'
raid = make_raid(7, 80)
rank = baseline_rank(raid).merge(raid[["id", "type", "spd", "urgent"]], on="id")
_, tti = geometry(raid)
dist = np.hypot(raid["x"], raid["y"])
y, p = rank["urgent"].to_numpy(), rank["urgent_pred"].to_numpy()
acc = (y == p).mean()
print(f"[Baseline 1차 위협 순위] 가상 비행체 {len(raid)}개, 실제 긴급 {y.sum()}개, 긴급 판정 {p.sum()}개")
top = rank.head(10).copy()
top["순위"] = range(1, 11)
top["tti_s"] = top["tti_s"].round(0); top["spd"] = top["spd"].round(0)
print(top[["순위", "id", "tti_s", "spd", "type", "urgent"]].rename(columns={"tti_s": "도달예상(s)", "spd": "속도(m/s)", "type": "실제유형", "urgent": "실제긴급"}).to_string(index=False))
print("\n[유형별 판정 결과] (긴급 판정 수 / 전체)")
for t, g in rank.groupby("type"):
print(f" {t:10s}: {int(g['urgent_pred'].sum())}/{len(g)} (실제 긴급 {int(g['urgent'].sum())})")
top10_true = set(raid.sort_values("true_prio", ascending=False).head(10)["id"])
overlap = len(top10_true & set(rank.head(10)["id"])) / 10
print(f"\n긴급 판정 정확도 {acc:.1%} (목표 75% 이상) -> {'달성' if acc >= 0.75 else '미달'}")
print(f"상위 10개 순위 일치율 {overlap:.0%}, 오경보 {int((p & ~y).sum())}건(기만체·조류 등 유형 판별 없음이 원인)")
print("한계: 자산 가치와 비행체 유형을 고려하지 않으므로 미션 4에서 고도화한다. 순위는 '권고'이며 판단은 지휘관이 한다.")
fig, ax = plt.subplots(1, 2, figsize=(12, 5))
ax[0].scatter(raid["x"], raid["y"], c=np.where(rank.set_index("id").loc[raid["id"], "urgent_pred"], "red", "gray"), s=18)
ax[0].scatter(ASSETS["x"], ASSETS["y"], marker="^", s=120, c="blue", label="virtual assets")
ax[0].set_title("Baseline: red = urgent (TTI <= 600 s)"); ax[0].set_xlabel("East (km)"); ax[0].set_ylabel("North (km)"); ax[0].legend()
ax[0].set_aspect("equal")
t15 = rank.head(15)
ax[1].barh(t15["id"][::-1], t15["tti_s"][::-1], color=["red" if u else "gray" for u in t15["urgent"][::-1]])
ax[1].axvline(TTI_URGENT, ls="--", c="k"); ax[1].set_title("Top-15 by TTI (red = truly urgent)"); ax[1].set_xlabel("TTI (s)")
plt.tight_layout(); plt.show()
실행 결과 예시
[Baseline 1차 위협 순위] 가상 비행체 80개, 실제 긴급 31개, 긴급 판정 39개
순위 id 도달예상(s) 속도(m/s) 실제유형 실제긴급
1 T07061 42.0 59.0 decoy False
2 T07055 53.0 55.0 uav_type True
3 T07048 68.0 221.0 fast_type True
4 T07014 103.0 231.0 fast_type True
5 T07004 152.0 224.0 fast_type True
6 T07038 158.0 59.0 uav_type True
7 T07019 162.0 54.0 uav_type True
8 T07033 173.0 45.0 uav_type True
9 T07065 176.0 154.0 fast_type True
10 T07062 180.0 158.0 fast_type True
[유형별 판정 결과] (긴급 판정 수 / 전체)
bird : 0/6 (실제 긴급 0)
civil : 1/4 (실제 긴급 0)
decoy : 7/16 (실제 긴급 0)
fast_type : 13/13 (실제 긴급 13)
uav_type : 18/41 (실제 긴급 18)
긴급 판정 정확도 90.0% (목표 75% 이상) -> 달성
상위 10개 순위 일치율 80%, 오경보 8건(기만체·조류 등 유형 판별 없음이 원인)
한계: 자산 가치와 비행체 유형을 고려하지 않으므로 미션 4에서 고도화한다. 순위는 '권고'이며 판단은 지휘관이 한다.
미션 4. 핵심 방어자산(지휘소·인프라) 거리 및 도달 예상 시간(TTI) 연계 다차원 위협 평가 고도화
- 난이도: 중급 / 예상 소요시간: 3.0시간
- 코드 설명: 가상 방어자산의 방호 가치와 자산별 도달 예상시간(TTI), 위협 유사도 분류기를 결합한 다차원 위협 평가 모델로 우선순위를 동적으로 재계산하고, 기만체·스푸핑 항적에 대한 방어 로직과 재현율·처리 지연을 검증한다.
"""가상 방어자산의 방호 가치와 자산별 도달 예상시간(TTI), 위협 유사도 분류기를 결합한 다차원 위협 평가 모델로 우선순위를 동적으로 재계산하고, 기만체·스푸핑 항적에 대한 방어 로직과 재현율·처리 지연을 검증한다."""
import time
import numpy as np
import pandas as pd
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler
# ---------------- [교육용가상자료] 가상 방어지역·가상 비행체 생성기 ----------------
# 좌표는 가상 지휘소를 원점으로 한 km 단위 평면이다. 비행체 유형은 추상 범주이며
# 속도·고도·반사강도는 학습용 임의 범위로, 실제 무기 제원·사거리·탄도와 무관하다.
ASSETS = pd.DataFrame([["A1", "CP(virtual)", 0.0, 0.0, 1.0],
["A2", "Radar(virtual)", 4.0, -3.0, 0.8],
["A3", "Depot(virtual)", -5.0, 4.0, 0.7],
["A4", "Power(virtual)", 7.0, 6.0, 0.5]], columns=["asset", "name", "x", "y", "value"])
TYPES = { # 유형: (위협 여부, 속도 m/s 범위, 고도 m 범위, 출현 비율)
"uav_type": (True, (30, 60), (100, 800), 0.42), # 저속 무인기형(가상)
"fast_type": (True, (150, 250), (50, 200), 0.16), # 고속 순항형(가상)
"decoy": (False, (30, 60), (100, 800), 0.16), # 기만체(가상)
"civil": (False, (120, 220), (3000, 9000), 0.10), # 민항기(가상)
"bird": (False, (8, 18), (50, 400), 0.16), # 조류 군집
}
TTI_URGENT = 600 # 긴급 판정 기준: 가상 자산 도달 예상시간 600초 이내
CPA_KM = 1.5 # 자산 상공 통과 판정 반경(가상)
def make_raid(seed, n=80, spoof=0):
"""가상 포화 상황 1회분(비행체 n개 + 스푸핑 가짜 항적 spoof개)을 만든다."""
rng = np.random.default_rng(seed)
names = list(TYPES)
rows = []
for i in range(n + spoof):
typ = "spoof" if i >= n else str(rng.choice(names, p=[TYPES[k][3] for k in names]))
threat, sp, al, _ = TYPES["decoy" if typ == "spoof" else typ]
brg = np.radians(rng.uniform(0, 360)); dist = rng.uniform(6, 60)
x, y = dist * np.sin(brg), dist * np.cos(brg)
spd = rng.uniform(*sp); alt = rng.uniform(*al)
aim = int(rng.integers(0, len(ASSETS)))
if typ in ("uav_type", "fast_type", "decoy", "spoof"): # 가상 자산 방향(작은 방위 오차)
ax_, ay_ = ASSETS.loc[aim, ["x", "y"]]
hd = np.arctan2(ax_ - x, ay_ - y) + np.radians(rng.normal(0, 0.6))
else:
hd = rng.uniform(0, 2 * np.pi); aim = -1
vx, vy = spd * np.sin(hd) / 1000, spd * np.cos(hd) / 1000 # km/s
sig = {"uav_type": ((0.2, 0.5), (0.02, 0.12), [0, .1, .5, .4]), "fast_type": ((0.1, 0.4), (0.02, 0.12), [0, .2, .4, .4]),
"decoy": ((0.3, 0.8), (0.08, 0.35), [0, .55, .4, .05]), "spoof": ((0.3, 0.8), (0.05, 0.3), [0, .7, .3, 0]),
"civil": ((0.7, 1.0), (0.01, 0.05), [0, 0, .1, .9]), "bird": ((0.01, 0.12), (0.1, 0.4), [0, .6, .4, 0])}[typ]
jump = spd * rng.uniform(0.95, 1.2) if typ != "spoof" else spd * rng.uniform(4, 15)
true_tti = np.inf
if threat:
d_aim = np.hypot(ASSETS.loc[aim, "x"] - x, ASSETS.loc[aim, "y"] - y)
true_tti = d_aim * 1000 / spd
rows.append({"id": f"T{seed % 100:02d}{i:03d}", "type": typ, "x": x, "y": y, "vx": vx, "vy": vy, "alt": alt, "spd": spd,
"rcs": rng.uniform(*sig[0]), "rcs_var": rng.uniform(*sig[1]), "n_sensors": int(rng.choice(4, p=sig[2])),
"max_jump": jump, "civ_iff": int(typ == "civil" and rng.random() < 0.95),
"aim": aim, "threat": threat, "true_tti": true_tti,
"urgent": bool(threat and true_tti <= TTI_URGENT)})
df = pd.DataFrame(rows)
df["true_prio"] = [ASSETS.loc[a, "value"] * np.exp(-t / 300) if th else 0.0
for a, t, th in zip(df["aim"], df["true_tti"], df["threat"])]
return df
def geometry(df):
"""각 항적과 가상 자산 사이 최근접점(CPA) 거리와 도달 예상시간(TTI, s)을 계산한다."""
px, py = df["x"].to_numpy()[:, None], df["y"].to_numpy()[:, None]
vx, vy = df["vx"].to_numpy()[:, None], df["vy"].to_numpy()[:, None]
rx, ry = ASSETS["x"].to_numpy()[None] - px, ASSETS["y"].to_numpy()[None] - py
v2 = vx ** 2 + vy ** 2 + 1e-12
t = (rx * vx + ry * vy) / v2 # CPA 도달 시각(s)
cpa = np.hypot(rx - vx * t, ry - vy * t)
tti = np.where((t > 0) & (cpa < CPA_KM), t, np.inf)
return cpa, tti
# ---------------- 위협 점수 모델(추상 모델: 거리·도달시간·자산 가치·위협 유사도) ----------------
FEATS = ["log_spd", "alt_km", "rcs", "rcs_var", "n_sensors", "civ_iff", "jump_ratio"]
def features(df):
return pd.DataFrame({"log_spd": np.log(df["spd"]), "alt_km": df["alt"] / 1000, "rcs": df["rcs"], "rcs_var": df["rcs_var"],
"n_sensors": df["n_sensors"], "civ_iff": df["civ_iff"], "jump_ratio": df["max_jump"] / df["spd"]})
def train_classifier(seeds=(101, 102, 103, 104)):
"""학습용 가상 상황(시험 상황과 다른 시드)으로 '위협 유사도' 분류기를 학습한다."""
tr = pd.concat([make_raid(s, 120, spoof=10) for s in seeds], ignore_index=True)
clf = make_pipeline(StandardScaler(), LogisticRegression(max_iter=500, C=1.0))
clf.fit(features(tr)[FEATS], tr["threat"].astype(int))
return clf
def baseline_rank(df):
"""Baseline: 거리·도달시간만으로 1차 위협 점수를 매긴다(자산 가치·유형 판별 없음)."""
_, tti = geometry(df)
t_min = tti.min(axis=1)
out = df[["id"]].copy()
out["tti_s"] = t_min
out["score"] = np.exp(-t_min / 300)
out["urgent_pred"] = t_min <= TTI_URGENT
return out.sort_values("score", ascending=False)
def advanced_rank(df, clf, w=None):
"""고도화: 자산 가치 x 도달시간 x 위협 유사도, 운동학 불일치(스푸핑 의심) 항적은 별도 표시한다."""
w = w or {"tau": 300, "p_min": 0.5, "value_pow": 1.0, "jump_max": 2.0}
_, tti = geometry(df)
val = ASSETS["value"].to_numpy()[None] ** w["value_pow"]
asset_score = val * np.exp(-tti / w["tau"])
best = asset_score.argmax(axis=1)
p = clf.predict_proba(features(df)[FEATS])[:, 1]
spoof = (df["max_jump"] / df["spd"]).to_numpy() > w["jump_max"]
out = df[["id"]].copy()
out["asset"] = ASSETS["asset"].to_numpy()[best]
out["tti_s"] = tti[np.arange(len(df)), best]
out["p_threat"] = p
out["spoof_flag"] = spoof
out["score"] = np.where(spoof, 0, p * asset_score.max(axis=1))
out["urgent_pred"] = (p >= w["p_min"]) & (out["tti_s"] <= TTI_URGENT) & ~spoof
return out.sort_values("score", ascending=False)
clf = train_classifier()
rows = []
for seed in (7, 8, 9): # 시험용 가상 상황 3개(스푸핑 항적 포함)
raid = make_raid(seed, 80, spoof=6)
t0 = time.perf_counter()
adv = advanced_rank(raid, clf)
ms = (time.perf_counter() - t0) * 1000
base = baseline_rank(raid)
for name, r in (("Baseline", base), ("고도화", adv)):
m = raid.merge(r[["id", "urgent_pred"]], on="id")
y, p = m["urgent"].to_numpy(), m["urgent_pred"].to_numpy()
rows.append({"상황": seed, "모델": name, "재현율": (p & y).sum() / y.sum(), "정밀도": (p & y).sum() / max(p.sum(), 1),
"기만·스푸핑 오판": int((p & m["type"].isin(["decoy", "spoof"]).to_numpy()).sum()),
"지연(ms)": ms if name == "고도화" else np.nan})
res = pd.DataFrame(rows)
print("[모델 비교] 실제 긴급 표적 기준")
print(res.round(3).fillna("-").to_string(index=False))
# 동적 재계산: 10초 경과 후 위치를 갱신해 순위 변화를 확인
raid = make_raid(7, 80, spoof=6)
r0 = advanced_rank(raid, clf).head(5)
moved = raid.copy(); moved["x"] += moved["vx"] * 10; moved["y"] += moved["vy"] * 10
r1 = advanced_rank(moved, clf).head(5)
print("\n[동적 재계산] 상위 5개 (현재 -> 10초 후)")
for a, b in zip(r0.itertuples(), r1.itertuples()):
print(f" {a.id} {a.asset} TTI {a.tti_s:5.0f}s 점수 {a.score:.2f} -> {b.id} {b.asset} TTI {b.tti_s:5.0f}s 점수 {b.score:.2f}")
adv_res = res[res["모델"] == "고도화"]
rec, lat = adv_res["재현율"].mean(), adv_res["지연(ms)"].max()
print(f"\n복합 위협 재현율 평균 {rec:.1%} (목표 85% 이상) -> {'달성' if rec >= 0.85 else '미달'}")
print(f"처리 지연 최대 {lat:.1f} ms (목표 200 ms 이하) -> {'달성' if lat <= 200 else '미달'}")
print(f"스푸핑 의심 항적 격리 {int(advanced_rank(raid, clf)['spoof_flag'].sum())}/6건 (운동학 불일치 = 보고 속도 대비 위치 도약 2배 초과)")
print("유의: 위협 유사도는 가상 반사강도·센서 합의·민간 응답 신호 등 추상 특징 기반이며, 최종 식별은 운용자 확인이 필요하다.")
실행 결과 예시
[모델 비교] 실제 긴급 표적 기준
상황 모델 재현율 정밀도 기만·스푸핑 오판 지연(ms)
7 Baseline 1.000 0.721 11 -
7 고도화 0.968 1.000 0 7.677
8 Baseline 1.000 0.861 3 -
8 고도화 1.000 1.000 0 6.501
9 Baseline 1.000 0.703 9 -
9 고도화 1.000 0.963 0 6.603
[동적 재계산] 상위 5개 (현재 -> 10초 후)
T07055 A2 TTI 53s 점수 0.60 -> T07055 A2 TTI 43s 점수 0.62
T07014 A2 TTI 103s 점수 0.57 -> T07014 A2 TTI 93s 점수 0.59
T07033 A1 TTI 173s 점수 0.56 -> T07033 A1 TTI 163s 점수 0.58
T07004 A1 TTI 181s 점수 0.55 -> T07004 A1 TTI 171s 점수 0.57
T07053 A1 TTI 185s 점수 0.54 -> T07053 A1 TTI 175s 점수 0.56
복합 위협 재현율 평균 98.9% (목표 85% 이상) -> 달성
처리 지연 최대 7.7 ms (목표 200 ms 이하) -> 달성
스푸핑 의심 항적 격리 6/6건 (운동학 불일치 = 보고 속도 대비 위치 도약 2배 초과)
유의: 위협 유사도는 가상 반사강도·센서 합의·민간 응답 신호 등 추상 특징 기반이며, 최종 식별은 운용자 확인이 필요하다.
미션 5. 방공 유도무기 교전 우선순위 랭킹 및 복합 신뢰도 지수 산출 모듈 구현
- 난이도: 중급~고급 / 예상 소요시간: 3.0시간
- 코드 설명: 위협 유사도·센서 합의·신호 안정성·운동학 일관성을 결합한 복합 신뢰도 지수(High/Med/Low)를 산출해 오경보를 줄이고, 가상 대응 자원의 잔여 수량과 가정 대응 성공 점수를 반영한 '권고' 대응 순위를 만든다(모든 권고는 지휘관 승인 대기).
"""위협 유사도·센서 합의·신호 안정성·운동학 일관성을 결합한 복합 신뢰도 지수(High/Med/Low)를 산출해 오경보를 줄이고, 가상 대응 자원의 잔여 수량과 가정 대응 성공 점수를 반영한 '권고' 대응 순위를 만든다(모든 권고는 지휘관 승인 대기)."""
import numpy as np
import pandas as pd
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler
# ---------------- [교육용가상자료] 가상 방어지역·가상 비행체 생성기 ----------------
# 좌표는 가상 지휘소를 원점으로 한 km 단위 평면이다. 비행체 유형은 추상 범주이며
# 속도·고도·반사강도는 학습용 임의 범위로, 실제 무기 제원·사거리·탄도와 무관하다.
ASSETS = pd.DataFrame([["A1", "CP(virtual)", 0.0, 0.0, 1.0],
["A2", "Radar(virtual)", 4.0, -3.0, 0.8],
["A3", "Depot(virtual)", -5.0, 4.0, 0.7],
["A4", "Power(virtual)", 7.0, 6.0, 0.5]], columns=["asset", "name", "x", "y", "value"])
TYPES = { # 유형: (위협 여부, 속도 m/s 범위, 고도 m 범위, 출현 비율)
"uav_type": (True, (30, 60), (100, 800), 0.42), # 저속 무인기형(가상)
"fast_type": (True, (150, 250), (50, 200), 0.16), # 고속 순항형(가상)
"decoy": (False, (30, 60), (100, 800), 0.16), # 기만체(가상)
"civil": (False, (120, 220), (3000, 9000), 0.10), # 민항기(가상)
"bird": (False, (8, 18), (50, 400), 0.16), # 조류 군집
}
TTI_URGENT = 600 # 긴급 판정 기준: 가상 자산 도달 예상시간 600초 이내
CPA_KM = 1.5 # 자산 상공 통과 판정 반경(가상)
def make_raid(seed, n=80, spoof=0):
"""가상 포화 상황 1회분(비행체 n개 + 스푸핑 가짜 항적 spoof개)을 만든다."""
rng = np.random.default_rng(seed)
names = list(TYPES)
rows = []
for i in range(n + spoof):
typ = "spoof" if i >= n else str(rng.choice(names, p=[TYPES[k][3] for k in names]))
threat, sp, al, _ = TYPES["decoy" if typ == "spoof" else typ]
brg = np.radians(rng.uniform(0, 360)); dist = rng.uniform(6, 60)
x, y = dist * np.sin(brg), dist * np.cos(brg)
spd = rng.uniform(*sp); alt = rng.uniform(*al)
aim = int(rng.integers(0, len(ASSETS)))
if typ in ("uav_type", "fast_type", "decoy", "spoof"): # 가상 자산 방향(작은 방위 오차)
ax_, ay_ = ASSETS.loc[aim, ["x", "y"]]
hd = np.arctan2(ax_ - x, ay_ - y) + np.radians(rng.normal(0, 0.6))
else:
hd = rng.uniform(0, 2 * np.pi); aim = -1
vx, vy = spd * np.sin(hd) / 1000, spd * np.cos(hd) / 1000 # km/s
sig = {"uav_type": ((0.2, 0.5), (0.02, 0.12), [0, .1, .5, .4]), "fast_type": ((0.1, 0.4), (0.02, 0.12), [0, .2, .4, .4]),
"decoy": ((0.3, 0.8), (0.08, 0.35), [0, .55, .4, .05]), "spoof": ((0.3, 0.8), (0.05, 0.3), [0, .7, .3, 0]),
"civil": ((0.7, 1.0), (0.01, 0.05), [0, 0, .1, .9]), "bird": ((0.01, 0.12), (0.1, 0.4), [0, .6, .4, 0])}[typ]
jump = spd * rng.uniform(0.95, 1.2) if typ != "spoof" else spd * rng.uniform(4, 15)
true_tti = np.inf
if threat:
d_aim = np.hypot(ASSETS.loc[aim, "x"] - x, ASSETS.loc[aim, "y"] - y)
true_tti = d_aim * 1000 / spd
rows.append({"id": f"T{seed % 100:02d}{i:03d}", "type": typ, "x": x, "y": y, "vx": vx, "vy": vy, "alt": alt, "spd": spd,
"rcs": rng.uniform(*sig[0]), "rcs_var": rng.uniform(*sig[1]), "n_sensors": int(rng.choice(4, p=sig[2])),
"max_jump": jump, "civ_iff": int(typ == "civil" and rng.random() < 0.95),
"aim": aim, "threat": threat, "true_tti": true_tti,
"urgent": bool(threat and true_tti <= TTI_URGENT)})
df = pd.DataFrame(rows)
df["true_prio"] = [ASSETS.loc[a, "value"] * np.exp(-t / 300) if th else 0.0
for a, t, th in zip(df["aim"], df["true_tti"], df["threat"])]
return df
def geometry(df):
"""각 항적과 가상 자산 사이 최근접점(CPA) 거리와 도달 예상시간(TTI, s)을 계산한다."""
px, py = df["x"].to_numpy()[:, None], df["y"].to_numpy()[:, None]
vx, vy = df["vx"].to_numpy()[:, None], df["vy"].to_numpy()[:, None]
rx, ry = ASSETS["x"].to_numpy()[None] - px, ASSETS["y"].to_numpy()[None] - py
v2 = vx ** 2 + vy ** 2 + 1e-12
t = (rx * vx + ry * vy) / v2 # CPA 도달 시각(s)
cpa = np.hypot(rx - vx * t, ry - vy * t)
tti = np.where((t > 0) & (cpa < CPA_KM), t, np.inf)
return cpa, tti
# ---------------- 위협 점수 모델(추상 모델: 거리·도달시간·자산 가치·위협 유사도) ----------------
FEATS = ["log_spd", "alt_km", "rcs", "rcs_var", "n_sensors", "civ_iff", "jump_ratio"]
def features(df):
return pd.DataFrame({"log_spd": np.log(df["spd"]), "alt_km": df["alt"] / 1000, "rcs": df["rcs"], "rcs_var": df["rcs_var"],
"n_sensors": df["n_sensors"], "civ_iff": df["civ_iff"], "jump_ratio": df["max_jump"] / df["spd"]})
def train_classifier(seeds=(101, 102, 103, 104)):
"""학습용 가상 상황(시험 상황과 다른 시드)으로 '위협 유사도' 분류기를 학습한다."""
tr = pd.concat([make_raid(s, 120, spoof=10) for s in seeds], ignore_index=True)
clf = make_pipeline(StandardScaler(), LogisticRegression(max_iter=500, C=1.0))
clf.fit(features(tr)[FEATS], tr["threat"].astype(int))
return clf
def baseline_rank(df):
"""Baseline: 거리·도달시간만으로 1차 위협 점수를 매긴다(자산 가치·유형 판별 없음)."""
_, tti = geometry(df)
t_min = tti.min(axis=1)
out = df[["id"]].copy()
out["tti_s"] = t_min
out["score"] = np.exp(-t_min / 300)
out["urgent_pred"] = t_min <= TTI_URGENT
return out.sort_values("score", ascending=False)
def advanced_rank(df, clf, w=None):
"""고도화: 자산 가치 x 도달시간 x 위협 유사도, 운동학 불일치(스푸핑 의심) 항적은 별도 표시한다."""
w = w or {"tau": 300, "p_min": 0.5, "value_pow": 1.0, "jump_max": 2.0}
_, tti = geometry(df)
val = ASSETS["value"].to_numpy()[None] ** w["value_pow"]
asset_score = val * np.exp(-tti / w["tau"])
best = asset_score.argmax(axis=1)
p = clf.predict_proba(features(df)[FEATS])[:, 1]
spoof = (df["max_jump"] / df["spd"]).to_numpy() > w["jump_max"]
out = df[["id"]].copy()
out["asset"] = ASSETS["asset"].to_numpy()[best]
out["tti_s"] = tti[np.arange(len(df)), best]
out["p_threat"] = p
out["spoof_flag"] = spoof
out["score"] = np.where(spoof, 0, p * asset_score.max(axis=1))
out["urgent_pred"] = (p >= w["p_min"]) & (out["tti_s"] <= TTI_URGENT) & ~spoof
return out.sort_values("score", ascending=False)
GRADE = [(0.70, "High"), (0.45, "Med"), (0.0, "Low")]
# [교육용가상자료] 가상 대응 자원: 잔여 수량, 담당 자산, 유형별 '가정 대응 성공 점수'(추상값, 실제 무기 성능 아님)
RESOURCES = pd.DataFrame([["R1", 6, ["A1", "A2"], {"uav_type": 0.8, "fast_type": 0.5}],
["R2", 4, ["A1", "A3"], {"uav_type": 0.7, "fast_type": 0.7}],
["R3", 8, ["A2", "A3", "A4"], {"uav_type": 0.6, "fast_type": 0.3}]],
columns=["res", "remain", "covers", "score_by_type"])
def confidence(df, clf):
p = clf.predict_proba(features(df)[FEATS])[:, 1]
agree = df["n_sensors"].to_numpy() / 3
stable = 1 - np.clip(df["rcs_var"].to_numpy() / 0.4, 0, 1)
kin = np.clip(2 - df["max_jump"].to_numpy() / df["spd"].to_numpy(), 0, 1)
conf = 0.5 * p + 0.2 * agree + 0.15 * stable + 0.15 * kin
grade = np.select([conf >= GRADE[0][0], conf >= GRADE[1][0]], ["High", "Med"], "Low")
return conf, grade
clf = train_classifier()
raid = make_raid(8, 80, spoof=6)
base = raid.merge(baseline_rank(raid)[["id", "urgent_pred"]], on="id")
adv = raid.merge(advanced_rank(raid, clf), on="id")
conf, grade = confidence(raid, clf)
adv["conf"] = conf[raid.set_index("id").index.get_indexer(adv["id"])]
adv["grade"] = grade[raid.set_index("id").index.get_indexer(adv["id"])]
alert = adv["urgent_pred"] & (adv["grade"] != "Low")
far_b = (base["urgent_pred"] & ~base["urgent"]).sum() / base["urgent_pred"].sum()
far_n = (alert & ~adv["urgent"]).sum() / max(alert.sum(), 1)
cut = (far_b - far_n) / far_b if far_b > 0 else 1.0
print(f"[오경보율] Baseline {far_b:.1%} -> 신뢰도 엔진 {far_n:.1%}, 감축률 {cut:.0%} (목표 40% 이상) -> {'달성' if cut >= 0.4 else '미달'}")
print("[등급 기준] High: 지수 0.70 이상(복수 센서·안정 신호), Med: 0.45~0.70(추가 확인 필요), Low: 0.45 미만(경보 보류)")
print(pd.crosstab(adv["type"], adv["grade"]).to_string())
# 일관성: 측정 잡음(반사강도·속도 ±5%)을 넣어 20회 반복할 때 등급이 유지되는 비율
rng = np.random.default_rng(0)
same = []
for _ in range(20):
noisy = raid.copy()
noisy["rcs"] *= rng.normal(1, 0.05, len(raid)); noisy["spd"] *= rng.normal(1, 0.05, len(raid))
same.append((confidence(noisy, clf)[1] == grade).mean())
cons = float(np.mean(same))
print(f"\n신뢰도 등급 일관성 {cons:.1%} (목표 90% 이상) -> {'달성' if cons >= 0.9 else '미달'}")
# 권고 대응 순위: 점수 순으로 담당 자원 중 (가정 대응 점수 x 잔여 비율)이 가장 큰 자원을 '권고'
remain = dict(zip(RESOURCES["res"], RESOURCES["remain"]))
recs = []
for r in adv[alert].sort_values("score", ascending=False).itertuples():
cand = [(row.score_by_type.get(r.type, 0.4) * remain[row.res] / row.remain, row.res) for row in RESOURCES.itertuples()
if r.asset in row.covers and remain[row.res] > 0]
res = max(cand)[1] if cand else "자원부족-상급보고"
if cand:
remain[res] -= 1
recs.append({"권고순위": len(recs) + 1, "표적": r.id, "지향자산": r.asset, "도달예상(s)": int(r.tti_s), "신뢰도": round(r.conf, 2),
"등급": r.grade, "권고자원": res, "상태": "승인 대기"})
rec = pd.DataFrame(recs)
print(f"\n[권고 대응 순위] 경보 {len(rec)}건 중 상위 10건 (자동 실행 없음, 지휘관 승인 필요)")
print(rec.head(10).to_string(index=False))
print("잔여 수량(가상):", remain, f"| 자원 부족으로 상급 보고 권고 {int((rec['권고자원'] == '자원부족-상급보고').sum())}건")
실행 결과 예시
[오경보율] Baseline 13.9% -> 신뢰도 엔진 0.0%, 감축률 100% (목표 40% 이상) -> 달성
[등급 기준] High: 지수 0.70 이상(복수 센서·안정 신호), Med: 0.45~0.70(추가 확인 필요), Low: 0.45 미만(경보 보류)
grade High Low Med
type
bird 0 8 1
civil 0 0 10
decoy 1 7 2
fast_type 15 0 0
spoof 0 6 0
uav_type 35 0 1
신뢰도 등급 일관성 99.4% (목표 90% 이상) -> 달성
[권고 대응 순위] 경보 31건 중 상위 10건 (자동 실행 없음, 지휘관 승인 필요)
권고순위 표적 지향자산 도달예상(s) 신뢰도 등급 권고자원 상태
1 T08023 A1 163 0.96 High R2 승인 대기
2 T08044 A2 99 0.90 High R1 승인 대기
3 T08034 A1 170 0.94 High R2 승인 대기
4 T08015 A3 69 0.94 High R2 승인 대기
5 T08036 A1 192 0.96 High R1 승인 대기
6 T08027 A2 107 0.89 High R3 승인 대기
7 T08058 A2 143 0.97 High R1 승인 대기
8 T08006 A2 145 0.87 High R3 승인 대기
9 T08068 A3 118 0.98 High R3 승인 대기
10 T08003 A2 153 0.90 High R1 승인 대기
잔여 수량(가상): {'R1': 0, 'R2': 0, 'R3': 0} | 자원 부족으로 상급 보고 권고 13건
미션 6. 방공 지휘관 결심 지원을 위한 실시간 포화경보 우선순위 대시보드 구현
- 난이도: 중급 / 예상 소요시간: 2.5시간
- 코드 설명: 방공 지휘관 결심 지원용 포화경보 우선순위 대시보드를 구현한다. 가상 방어지역 지도에 표적 항적과 우선순위를, 옆에 권고 우선순위 목록·도달시간 분포·가상 대응 자원 현황을 표시하고 1초 주기 갱신 처리시간과 조회 응답시간을 측정한다.
"""방공 지휘관 결심 지원용 포화경보 우선순위 대시보드를 구현한다. 가상 방어지역 지도에 표적 항적과 우선순위를, 옆에 권고 우선순위 목록·도달시간 분포·가상 대응 자원 현황을 표시하고 1초 주기 갱신 처리시간과 조회 응답시간을 측정한다."""
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler
# ---------------- [교육용가상자료] 가상 방어지역·가상 비행체 생성기 ----------------
# 좌표는 가상 지휘소를 원점으로 한 km 단위 평면이다. 비행체 유형은 추상 범주이며
# 속도·고도·반사강도는 학습용 임의 범위로, 실제 무기 제원·사거리·탄도와 무관하다.
ASSETS = pd.DataFrame([["A1", "CP(virtual)", 0.0, 0.0, 1.0],
["A2", "Radar(virtual)", 4.0, -3.0, 0.8],
["A3", "Depot(virtual)", -5.0, 4.0, 0.7],
["A4", "Power(virtual)", 7.0, 6.0, 0.5]], columns=["asset", "name", "x", "y", "value"])
TYPES = { # 유형: (위협 여부, 속도 m/s 범위, 고도 m 범위, 출현 비율)
"uav_type": (True, (30, 60), (100, 800), 0.42), # 저속 무인기형(가상)
"fast_type": (True, (150, 250), (50, 200), 0.16), # 고속 순항형(가상)
"decoy": (False, (30, 60), (100, 800), 0.16), # 기만체(가상)
"civil": (False, (120, 220), (3000, 9000), 0.10), # 민항기(가상)
"bird": (False, (8, 18), (50, 400), 0.16), # 조류 군집
}
TTI_URGENT = 600 # 긴급 판정 기준: 가상 자산 도달 예상시간 600초 이내
CPA_KM = 1.5 # 자산 상공 통과 판정 반경(가상)
def make_raid(seed, n=80, spoof=0):
"""가상 포화 상황 1회분(비행체 n개 + 스푸핑 가짜 항적 spoof개)을 만든다."""
rng = np.random.default_rng(seed)
names = list(TYPES)
rows = []
for i in range(n + spoof):
typ = "spoof" if i >= n else str(rng.choice(names, p=[TYPES[k][3] for k in names]))
threat, sp, al, _ = TYPES["decoy" if typ == "spoof" else typ]
brg = np.radians(rng.uniform(0, 360)); dist = rng.uniform(6, 60)
x, y = dist * np.sin(brg), dist * np.cos(brg)
spd = rng.uniform(*sp); alt = rng.uniform(*al)
aim = int(rng.integers(0, len(ASSETS)))
if typ in ("uav_type", "fast_type", "decoy", "spoof"): # 가상 자산 방향(작은 방위 오차)
ax_, ay_ = ASSETS.loc[aim, ["x", "y"]]
hd = np.arctan2(ax_ - x, ay_ - y) + np.radians(rng.normal(0, 0.6))
else:
hd = rng.uniform(0, 2 * np.pi); aim = -1
vx, vy = spd * np.sin(hd) / 1000, spd * np.cos(hd) / 1000 # km/s
sig = {"uav_type": ((0.2, 0.5), (0.02, 0.12), [0, .1, .5, .4]), "fast_type": ((0.1, 0.4), (0.02, 0.12), [0, .2, .4, .4]),
"decoy": ((0.3, 0.8), (0.08, 0.35), [0, .55, .4, .05]), "spoof": ((0.3, 0.8), (0.05, 0.3), [0, .7, .3, 0]),
"civil": ((0.7, 1.0), (0.01, 0.05), [0, 0, .1, .9]), "bird": ((0.01, 0.12), (0.1, 0.4), [0, .6, .4, 0])}[typ]
jump = spd * rng.uniform(0.95, 1.2) if typ != "spoof" else spd * rng.uniform(4, 15)
true_tti = np.inf
if threat:
d_aim = np.hypot(ASSETS.loc[aim, "x"] - x, ASSETS.loc[aim, "y"] - y)
true_tti = d_aim * 1000 / spd
rows.append({"id": f"T{seed % 100:02d}{i:03d}", "type": typ, "x": x, "y": y, "vx": vx, "vy": vy, "alt": alt, "spd": spd,
"rcs": rng.uniform(*sig[0]), "rcs_var": rng.uniform(*sig[1]), "n_sensors": int(rng.choice(4, p=sig[2])),
"max_jump": jump, "civ_iff": int(typ == "civil" and rng.random() < 0.95),
"aim": aim, "threat": threat, "true_tti": true_tti,
"urgent": bool(threat and true_tti <= TTI_URGENT)})
df = pd.DataFrame(rows)
df["true_prio"] = [ASSETS.loc[a, "value"] * np.exp(-t / 300) if th else 0.0
for a, t, th in zip(df["aim"], df["true_tti"], df["threat"])]
return df
def geometry(df):
"""각 항적과 가상 자산 사이 최근접점(CPA) 거리와 도달 예상시간(TTI, s)을 계산한다."""
px, py = df["x"].to_numpy()[:, None], df["y"].to_numpy()[:, None]
vx, vy = df["vx"].to_numpy()[:, None], df["vy"].to_numpy()[:, None]
rx, ry = ASSETS["x"].to_numpy()[None] - px, ASSETS["y"].to_numpy()[None] - py
v2 = vx ** 2 + vy ** 2 + 1e-12
t = (rx * vx + ry * vy) / v2 # CPA 도달 시각(s)
cpa = np.hypot(rx - vx * t, ry - vy * t)
tti = np.where((t > 0) & (cpa < CPA_KM), t, np.inf)
return cpa, tti
# ---------------- 위협 점수 모델(추상 모델: 거리·도달시간·자산 가치·위협 유사도) ----------------
FEATS = ["log_spd", "alt_km", "rcs", "rcs_var", "n_sensors", "civ_iff", "jump_ratio"]
def features(df):
return pd.DataFrame({"log_spd": np.log(df["spd"]), "alt_km": df["alt"] / 1000, "rcs": df["rcs"], "rcs_var": df["rcs_var"],
"n_sensors": df["n_sensors"], "civ_iff": df["civ_iff"], "jump_ratio": df["max_jump"] / df["spd"]})
def train_classifier(seeds=(101, 102, 103, 104)):
"""학습용 가상 상황(시험 상황과 다른 시드)으로 '위협 유사도' 분류기를 학습한다."""
tr = pd.concat([make_raid(s, 120, spoof=10) for s in seeds], ignore_index=True)
clf = make_pipeline(StandardScaler(), LogisticRegression(max_iter=500, C=1.0))
clf.fit(features(tr)[FEATS], tr["threat"].astype(int))
return clf
def baseline_rank(df):
"""Baseline: 거리·도달시간만으로 1차 위협 점수를 매긴다(자산 가치·유형 판별 없음)."""
_, tti = geometry(df)
t_min = tti.min(axis=1)
out = df[["id"]].copy()
out["tti_s"] = t_min
out["score"] = np.exp(-t_min / 300)
out["urgent_pred"] = t_min <= TTI_URGENT
return out.sort_values("score", ascending=False)
def advanced_rank(df, clf, w=None):
"""고도화: 자산 가치 x 도달시간 x 위협 유사도, 운동학 불일치(스푸핑 의심) 항적은 별도 표시한다."""
w = w or {"tau": 300, "p_min": 0.5, "value_pow": 1.0, "jump_max": 2.0}
_, tti = geometry(df)
val = ASSETS["value"].to_numpy()[None] ** w["value_pow"]
asset_score = val * np.exp(-tti / w["tau"])
best = asset_score.argmax(axis=1)
p = clf.predict_proba(features(df)[FEATS])[:, 1]
spoof = (df["max_jump"] / df["spd"]).to_numpy() > w["jump_max"]
out = df[["id"]].copy()
out["asset"] = ASSETS["asset"].to_numpy()[best]
out["tti_s"] = tti[np.arange(len(df)), best]
out["p_threat"] = p
out["spoof_flag"] = spoof
out["score"] = np.where(spoof, 0, p * asset_score.max(axis=1))
out["urgent_pred"] = (p >= w["p_min"]) & (out["tti_s"] <= TTI_URGENT) & ~spoof
return out.sort_values("score", ascending=False)
clf = train_classifier()
raid = make_raid(9, 100, spoof=6)
RES = {"R1": [6, 6], "R2": [4, 4], "R3": [8, 8]} # [잔여, 최초] 가상 대응 자원
def refresh(df):
r = advanced_rank(df, clf).merge(df[["id", "x", "y", "vx", "vy"]], on="id")
r["prio"] = np.where(r["urgent_pred"] & (r["tti_s"] <= 180), "P1", np.where(r["urgent_pred"], "P2", np.where(r["spoof_flag"], "SPOOF", "P3")))
return r
# 1) 1초 주기 갱신 10회 모의: 갱신 처리시간과 조회(필터·정렬) 응답시간 측정
upd, qry = [], []
for k in range(10):
t0 = time.perf_counter(); board = refresh(raid); upd.append((time.perf_counter() - t0) * 1000)
t0 = time.perf_counter(); view = board[board["prio"].isin(["P1", "P2"])].sort_values("tti_s").head(12); qry.append((time.perf_counter() - t0) * 1000)
raid = raid.assign(x=raid["x"] + raid["vx"], y=raid["y"] + raid["vy"])
raid = raid[np.hypot(raid["x"], raid["y"]) > 0.5].reset_index(drop=True) # 도달 처리된 항적 제거(가상)
print(f"갱신 처리시간 평균 {np.mean(upd):.1f} ms / 최대 {np.max(upd):.1f} ms (목표: 갱신 주기 1초 이내) -> {'달성' if max(upd) < 1000 else '미달'}")
print(f"조회 응답시간 최대 {np.max(qry):.2f} ms (목표 100 ms 이내) -> {'달성' if max(qry) < 100 else '미달'}")
cnt = board["prio"].value_counts()
print(f"현재 표시: P1(180초 이내) {cnt.get('P1', 0)}건, P2(600초 이내) {cnt.get('P2', 0)}건, 스푸핑 의심 {cnt.get('SPOOF', 0)}건, 기타 {cnt.get('P3', 0)}건")
print("\n[권고 우선순위 목록] (P1·P2, 도달 예상 순)")
print(view[["id", "prio", "asset", "tti_s", "p_threat"]].round({"tti_s": 0, "p_threat": 2}).rename(
columns={"id": "표적", "prio": "등급", "asset": "지향자산", "tti_s": "도달예상(s)", "p_threat": "위협유사도"}).to_string(index=False))
# 2) 대시보드(어두운 배경, P1은 빨강 굵게, 정보는 4개 영역으로 고정 배치해 인지 부하를 줄인다)
plt.style.use("dark_background")
fig = plt.figure(figsize=(14, 7))
gs = fig.add_gridspec(2, 3)
a = fig.add_subplot(gs[:, :2])
col = {"P1": "red", "P2": "orange", "P3": "gray", "SPOOF": "magenta"}
for p, g in board.groupby("prio"):
a.quiver(g["x"], g["y"], g["vx"], g["vy"], color=col[p], scale=3, width=0.003, label=f"{p} ({len(g)})")
a.scatter(ASSETS["x"], ASSETS["y"], marker="^", s=150, c="cyan")
for r in ASSETS.itertuples():
a.text(r.x + 0.8, r.y + 0.8, f"{r.asset} v={r.value}", color="cyan", fontsize=9)
for r in board[board["prio"] == "P1"].itertuples():
a.text(r.x, r.y, r.id[-3:], color="red", fontsize=8, fontweight="bold")
a.set_xlim(-60, 60); a.set_ylim(-60, 60); a.set_aspect("equal"); a.grid(alpha=0.2); a.legend(loc="lower left")
a.set_title("Saturation alert picture (virtual area)"); a.set_xlabel("East (km)"); a.set_ylabel("North (km)")
b = fig.add_subplot(gs[0, 2])
urg = board[board["urgent_pred"]]
b.hist(urg["tti_s"], bins=np.arange(0, 660, 60), color="orange"); b.axvline(180, c="red", ls="--")
b.set_title("Urgent tracks by TTI (s)")
c = fig.add_subplot(gs[1, 2])
c.barh(list(RES), [v[1] for v in RES.values()], color="dimgray", label="initial")
c.barh(list(RES), [v[0] for v in RES.values()], color="lime", label="remaining")
c.set_title("Virtual response resources"); c.legend(fontsize=8)
plt.suptitle("OPS04 dashboard - recommendations only, commander approval required (training data)")
plt.tight_layout(); plt.show()
print("\n대시보드의 순위는 권고 정보이며, 대응 여부는 미션 7의 지휘관 승인 절차와 교전수칙·지휘계통에 따른다.")
실행 결과 예시
갱신 처리시간 평균 9.4 ms / 최대 10.9 ms (목표: 갱신 주기 1초 이내) -> 달성
조회 응답시간 최대 1.64 ms (목표 100 ms 이내) -> 달성
현재 표시: P1(180초 이내) 13건, P2(600초 이내) 20건, 스푸핑 의심 6건, 기타 67건
[권고 우선순위 목록] (P1·P2, 도달 예상 순)
표적 등급 지향자산 도달예상(s) 위협유사도
T09027 P1 A3 0.0 1.00
T09017 P1 A4 70.0 1.00
T09072 P1 A1 76.0 1.00
T09047 P1 A2 78.0 0.99
T09038 P1 A3 85.0 0.96
T09069 P1 A4 97.0 1.00
T09049 P1 A3 136.0 0.99
T09023 P1 A4 147.0 1.00
T09092 P1 A3 149.0 0.97
T09006 P1 A1 155.0 0.93
T09089 P1 A2 160.0 1.00
T09026 P1 A2 167.0 1.00
대시보드의 순위는 권고 정보이며, 대응 여부는 미션 7의 지휘관 승인 절차와 교전수칙·지휘계통에 따른다.
미션 7. 자동 할당된 교전 순위 재조정 및 긴급 수동 전환을 위한 HITL 승인 절차 설계
- 난이도: 고급 / 예상 소요시간: 3.0시간
- 코드 설명: 자동 계산된 대응 우선순위를 방공 지휘관이 승인·보류·반려하거나 수동으로 순위를 바꾸고, 긴급 시 수동 통제 모드로 전환하는 인간 개입(HITL) 승인 워크플로우를 구현한다. 자동 승인은 없으며, 모든 시도(거부된 시도 포함)는 역할·근거와 함께 해시 체인 감사 로그에 기록된다.
"""자동 계산된 대응 우선순위를 방공 지휘관이 승인·보류·반려하거나 수동으로 순위를 바꾸고, 긴급 시 수동 통제 모드로 전환하는 인간 개입(HITL) 승인 워크플로우를 구현한다. 자동 승인은 없으며, 모든 시도(거부된 시도 포함)는 역할·근거와 함께 해시 체인 감사 로그에 기록된다."""
import hashlib
import json
import numpy as np
import pandas as pd
ROLES = {"CDR-01": "commander", "OPR-07": "operator", "OPR-09": "operator"} # 가상 운용자
PERM = {"APPROVE": {"commander"}, "REORDER": {"commander"}, "MODE": {"commander"},
"HOLD": {"commander", "operator"}, "REJECT": {"commander", "operator"}}
HOLD_LIMIT_S = 30
class FakeClock:
t = 0.0
def __call__(self):
return self.t
class HitlBoard:
def __init__(self, clock):
self.clock, self.items, self.log, self.mode = clock, {}, [], "RECOMMEND"
self.attempts = 0
def _record(self, body):
prev = self.log[-1]["hash"] if self.log else "GENESIS"
body = {**body, "ts": self.clock(), "mode": self.mode}
h = hashlib.sha256((prev + json.dumps(body, ensure_ascii=False, sort_keys=True)).encode()).hexdigest()
self.log.append({**body, "prev": prev, "hash": h})
def verify(self):
prev = "GENESIS"
for n, r in enumerate(self.log, 1):
body = {k: v for k, v in r.items() if k not in ("prev", "hash")}
if r["prev"] != prev or r["hash"] != hashlib.sha256((prev + json.dumps(body, ensure_ascii=False, sort_keys=True)).encode()).hexdigest():
return False, n
prev = r["hash"]
return True, -1
def submit(self, tid, ai_rank, tti, grade):
self.items[tid] = {"id": tid, "ai_rank": ai_rank, "rank": ai_rank, "tti": tti, "grade": grade, "status": "PENDING",
"t_in": self.clock(), "t_out": None, "hold_at": None, "escalated": False}
self._record({"event": "SUBMIT", "id": tid, "ai_rank": ai_rank, "by": "AI"})
def act(self, op, action, tid=None, reason="", new_rank=None):
"""모든 시도는 성공·거부와 무관하게 감사 로그에 남긴다."""
self.attempts += 1
err = None
role = ROLES.get(op)
it = self.items.get(tid)
if role is None or role not in PERM.get(action, set()):
err = f"권한 없음({op}:{action})"
elif len(reason.strip()) < 6:
err = "판단 근거 6자 이상 필요"
elif action != "MODE" and (it is None or it["status"] in ("APPROVED", "REJECTED")):
err = "대상 없음 또는 이미 종결"
elif action == "APPROVE" and it["grade"] == "Low":
err = "Low 등급은 승인 전 추가 확인(HOLD) 필요"
if err:
self._record({"event": "REFUSED", "action": action, "id": tid, "operator": op, "reason": reason, "error": err})
return err
if action == "MODE":
self.mode = "MANUAL" if self.mode == "RECOMMEND" else "RECOMMEND"
elif action == "REORDER":
it["rank"] = new_rank
else:
it["status"] = {"APPROVE": "APPROVED", "HOLD": "HOLD", "REJECT": "REJECTED"}[action]
it["hold_at"] = self.clock() if action == "HOLD" else None
if action != "HOLD":
it["t_out"] = self.clock()
self._record({"event": action, "id": tid, "operator": op, "role": role, "reason": reason, "new_rank": new_rank})
return "OK"
def tick(self):
"""보류 기한 초과 시 자동 승인하지 않고 지휘관에게 상신(ESCALATE)만 한다."""
for it in self.items.values():
if it["status"] == "HOLD" and not it["escalated"] and self.clock() - it["hold_at"] > HOLD_LIMIT_S:
it["escalated"] = True
self._record({"event": "ESCALATE", "id": it["id"], "by": "SYSTEM"})
def queue(self):
todo = [i for i in self.items.values() if i["status"] in ("PENDING", "HOLD")]
return sorted(todo, key=lambda i: (not i["escalated"], i["rank"]))
clk = FakeClock()
hb = HitlBoard(clk)
rng = np.random.default_rng(5)
# [교육용가상자료] 미션 5 권고 결과를 가정한 12건
for k in range(12):
hb.submit(f"T09{k:03d}", k + 1, int(rng.integers(60, 600)), ["High", "High", "Med", "Low"][k % 4])
scenario = [ # (경과 s, 운용자, 조치, 표적, 근거, 새 순위)
(2, "OPR-07", "APPROVE", "T09000", "도달 임박 표적", None), # 권한 없음 -> 거부
(3, "CDR-01", "APPROVE", "T09000", "복수 센서 확인, 지휘소 지향", None),
(5, "CDR-01", "REORDER", "T09005", "탄약고 지향 확인, 우선 대응 필요", 2),
(6, "OPR-09", "HOLD", "T09003", "단일 센서, 육안 확인 요청", None),
(7, "CDR-01", "APPROVE", "T09003", "급함", None), # 근거 부족 -> 거부
(8, "OPR-07", "REJECT", "T09007", "민간 응답 신호 확인, 오경보", None),
(10, "CDR-01", "MODE", None, "통신 지연으로 긴급 수동 통제 전환", None),
(12, "CDR-01", "APPROVE", "T09001", "수동 통제 중 지휘관 직접 판단", None),
(15, "CDR-01", "APPROVE", "T09011", "Low 등급이나 도달 임박 판단", None), # Low 승인 -> 거부
(20, "CDR-01", "APPROVE", "T09001", "중복 승인 시도 확인용", None), # 이미 종결 -> 거부
(25, "CDR-01", "MODE", None, "통신 복구, 권고 모드 복귀", None),
]
for t, op, a, tid, rsn, nr in scenario:
clk.t = t; hb.tick()
res = hb.act(op, a, tid, rsn, nr)
print(f"[t={t:>3}s] {op} {a:8s} {tid or '-':7s} -> {res}")
clk.t = 45; hb.tick()
print("\n[검토 대기열] (상신 우선, 그다음 순위)")
print(pd.DataFrame(hb.queue())[["id", "ai_rank", "rank", "grade", "status", "escalated"]].head(6).to_string(index=False))
auto = sum(1 for r in hb.log if r["event"] == "APPROVE" and r.get("operator") in (None, "AI", "SYSTEM"))
logged = sum(1 for r in hb.log if r["event"] not in ("SUBMIT", "ESCALATE"))
ok, _ = hb.verify()
done = [i for i in hb.items.values() if i["t_out"] is not None]
waits = [i["t_out"] - i["t_in"] for i in done]
print(f"\n조치 시도 {hb.attempts}건 중 로그 기록 {logged}건 -> 로깅 성공률 {logged / hb.attempts:.0%} (목표 100%) -> {'달성' if logged == hb.attempts else '미달'}")
print(f"종결 {len(done)}건 대기시간 평균 {np.mean(waits):.1f}s / 최대 {max(waits):.0f}s, 미결 {len(hb.queue())}건, 자동 승인 {auto}건")
print(f"감사 로그 무결성: {'정상' if ok else '훼손'} ({len(hb.log)}건)")
hb.log[3]["reason"] = "사후 수정"
ok2, n2 = hb.verify()
print(f"위·변조 시험(4번째 기록 근거 수정) -> {'검출(' + str(n2) + '번 기록)' if not ok2 else '미검출'}")
print("\n[지휘 책임 분계선] AI: 순위 '권고'만 / 운용자: 보류·반려 / 지휘관: 승인·순위 변경·수동 전환")
print("승인은 지휘계통에 대응 인가를 요청하는 단계이며, 실제 교전은 교전수칙과 지휘계통에 따른다.")
실행 결과 예시
[t= 2s] OPR-07 APPROVE T09000 -> 권한 없음(OPR-07:APPROVE)
[t= 3s] CDR-01 APPROVE T09000 -> OK
[t= 5s] CDR-01 REORDER T09005 -> OK
[t= 6s] OPR-09 HOLD T09003 -> OK
[t= 7s] CDR-01 APPROVE T09003 -> 판단 근거 6자 이상 필요
[t= 8s] OPR-07 REJECT T09007 -> OK
[t= 10s] CDR-01 MODE - -> OK
[t= 12s] CDR-01 APPROVE T09001 -> OK
[t= 15s] CDR-01 APPROVE T09011 -> Low 등급은 승인 전 추가 확인(HOLD) 필요
[t= 20s] CDR-01 APPROVE T09001 -> 대상 없음 또는 이미 종결
[t= 25s] CDR-01 MODE - -> OK
[검토 대기열] (상신 우선, 그다음 순위)
id ai_rank rank grade status escalated
T09003 4 4 Low HOLD True
T09005 6 2 High PENDING False
T09002 3 3 Med PENDING False
T09004 5 5 High PENDING False
T09006 7 7 Med PENDING False
T09008 9 9 High PENDING False
조치 시도 11건 중 로그 기록 11건 -> 로깅 성공률 100% (목표 100%) -> 달성
종결 3건 대기시간 평균 7.7s / 최대 12s, 미결 9건, 자동 승인 0건
감사 로그 무결성: 정상 (24건)
위·변조 시험(4번째 기록 근거 수정) -> 검출(4번 기록)
[지휘 책임 분계선] AI: 순위 '권고'만 / 운용자: 보류·반려 / 지휘관: 승인·순위 변경·수동 전환
승인은 지휘계통에 대응 인가를 요청하는 단계이며, 실제 교전은 교전수칙과 지휘계통에 따른다.
미션 8. 100기 이상 동시 포화 공격 가상 전장 모의 부하 테스트 및 처리 지연 검증
- 난이도: 고급 / 예상 소요시간: 3.5시간
- 코드 설명: 평시 100기 기준에서 최대 600기(500% 증가)까지 가상 비행체가 동시에 유입되는 포화 공격 상황에 손상 메시지·중복·순서 뒤섞임을 주입해 부하 시험을 수행하고, 단계별 처리 지연·시스템 다운·데이터 유실과 시간 예산 초과 시 기능 축소(Fail-safe) 동작을 검증한다.
"""평시 100기 기준에서 최대 600기(500% 증가)까지 가상 비행체가 동시에 유입되는 포화 공격 상황에 손상 메시지·중복·순서 뒤섞임을 주입해 부하 시험을 수행하고, 단계별 처리 지연·시스템 다운·데이터 유실과 시간 예산 초과 시 기능 축소(Fail-safe) 동작을 검증한다."""
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler
# ---------------- [교육용가상자료] 가상 방어지역·가상 비행체 생성기 ----------------
# 좌표는 가상 지휘소를 원점으로 한 km 단위 평면이다. 비행체 유형은 추상 범주이며
# 속도·고도·반사강도는 학습용 임의 범위로, 실제 무기 제원·사거리·탄도와 무관하다.
ASSETS = pd.DataFrame([["A1", "CP(virtual)", 0.0, 0.0, 1.0],
["A2", "Radar(virtual)", 4.0, -3.0, 0.8],
["A3", "Depot(virtual)", -5.0, 4.0, 0.7],
["A4", "Power(virtual)", 7.0, 6.0, 0.5]], columns=["asset", "name", "x", "y", "value"])
TYPES = { # 유형: (위협 여부, 속도 m/s 범위, 고도 m 범위, 출현 비율)
"uav_type": (True, (30, 60), (100, 800), 0.42), # 저속 무인기형(가상)
"fast_type": (True, (150, 250), (50, 200), 0.16), # 고속 순항형(가상)
"decoy": (False, (30, 60), (100, 800), 0.16), # 기만체(가상)
"civil": (False, (120, 220), (3000, 9000), 0.10), # 민항기(가상)
"bird": (False, (8, 18), (50, 400), 0.16), # 조류 군집
}
TTI_URGENT = 600 # 긴급 판정 기준: 가상 자산 도달 예상시간 600초 이내
CPA_KM = 1.5 # 자산 상공 통과 판정 반경(가상)
def make_raid(seed, n=80, spoof=0):
"""가상 포화 상황 1회분(비행체 n개 + 스푸핑 가짜 항적 spoof개)을 만든다."""
rng = np.random.default_rng(seed)
names = list(TYPES)
rows = []
for i in range(n + spoof):
typ = "spoof" if i >= n else str(rng.choice(names, p=[TYPES[k][3] for k in names]))
threat, sp, al, _ = TYPES["decoy" if typ == "spoof" else typ]
brg = np.radians(rng.uniform(0, 360)); dist = rng.uniform(6, 60)
x, y = dist * np.sin(brg), dist * np.cos(brg)
spd = rng.uniform(*sp); alt = rng.uniform(*al)
aim = int(rng.integers(0, len(ASSETS)))
if typ in ("uav_type", "fast_type", "decoy", "spoof"): # 가상 자산 방향(작은 방위 오차)
ax_, ay_ = ASSETS.loc[aim, ["x", "y"]]
hd = np.arctan2(ax_ - x, ay_ - y) + np.radians(rng.normal(0, 0.6))
else:
hd = rng.uniform(0, 2 * np.pi); aim = -1
vx, vy = spd * np.sin(hd) / 1000, spd * np.cos(hd) / 1000 # km/s
sig = {"uav_type": ((0.2, 0.5), (0.02, 0.12), [0, .1, .5, .4]), "fast_type": ((0.1, 0.4), (0.02, 0.12), [0, .2, .4, .4]),
"decoy": ((0.3, 0.8), (0.08, 0.35), [0, .55, .4, .05]), "spoof": ((0.3, 0.8), (0.05, 0.3), [0, .7, .3, 0]),
"civil": ((0.7, 1.0), (0.01, 0.05), [0, 0, .1, .9]), "bird": ((0.01, 0.12), (0.1, 0.4), [0, .6, .4, 0])}[typ]
jump = spd * rng.uniform(0.95, 1.2) if typ != "spoof" else spd * rng.uniform(4, 15)
true_tti = np.inf
if threat:
d_aim = np.hypot(ASSETS.loc[aim, "x"] - x, ASSETS.loc[aim, "y"] - y)
true_tti = d_aim * 1000 / spd
rows.append({"id": f"T{seed % 100:02d}{i:03d}", "type": typ, "x": x, "y": y, "vx": vx, "vy": vy, "alt": alt, "spd": spd,
"rcs": rng.uniform(*sig[0]), "rcs_var": rng.uniform(*sig[1]), "n_sensors": int(rng.choice(4, p=sig[2])),
"max_jump": jump, "civ_iff": int(typ == "civil" and rng.random() < 0.95),
"aim": aim, "threat": threat, "true_tti": true_tti,
"urgent": bool(threat and true_tti <= TTI_URGENT)})
df = pd.DataFrame(rows)
df["true_prio"] = [ASSETS.loc[a, "value"] * np.exp(-t / 300) if th else 0.0
for a, t, th in zip(df["aim"], df["true_tti"], df["threat"])]
return df
def geometry(df):
"""각 항적과 가상 자산 사이 최근접점(CPA) 거리와 도달 예상시간(TTI, s)을 계산한다."""
px, py = df["x"].to_numpy()[:, None], df["y"].to_numpy()[:, None]
vx, vy = df["vx"].to_numpy()[:, None], df["vy"].to_numpy()[:, None]
rx, ry = ASSETS["x"].to_numpy()[None] - px, ASSETS["y"].to_numpy()[None] - py
v2 = vx ** 2 + vy ** 2 + 1e-12
t = (rx * vx + ry * vy) / v2 # CPA 도달 시각(s)
cpa = np.hypot(rx - vx * t, ry - vy * t)
tti = np.where((t > 0) & (cpa < CPA_KM), t, np.inf)
return cpa, tti
# ---------------- 위협 점수 모델(추상 모델: 거리·도달시간·자산 가치·위협 유사도) ----------------
FEATS = ["log_spd", "alt_km", "rcs", "rcs_var", "n_sensors", "civ_iff", "jump_ratio"]
def features(df):
return pd.DataFrame({"log_spd": np.log(df["spd"]), "alt_km": df["alt"] / 1000, "rcs": df["rcs"], "rcs_var": df["rcs_var"],
"n_sensors": df["n_sensors"], "civ_iff": df["civ_iff"], "jump_ratio": df["max_jump"] / df["spd"]})
def train_classifier(seeds=(101, 102, 103, 104)):
"""학습용 가상 상황(시험 상황과 다른 시드)으로 '위협 유사도' 분류기를 학습한다."""
tr = pd.concat([make_raid(s, 120, spoof=10) for s in seeds], ignore_index=True)
clf = make_pipeline(StandardScaler(), LogisticRegression(max_iter=500, C=1.0))
clf.fit(features(tr)[FEATS], tr["threat"].astype(int))
return clf
def baseline_rank(df):
"""Baseline: 거리·도달시간만으로 1차 위협 점수를 매긴다(자산 가치·유형 판별 없음)."""
_, tti = geometry(df)
t_min = tti.min(axis=1)
out = df[["id"]].copy()
out["tti_s"] = t_min
out["score"] = np.exp(-t_min / 300)
out["urgent_pred"] = t_min <= TTI_URGENT
return out.sort_values("score", ascending=False)
def advanced_rank(df, clf, w=None):
"""고도화: 자산 가치 x 도달시간 x 위협 유사도, 운동학 불일치(스푸핑 의심) 항적은 별도 표시한다."""
w = w or {"tau": 300, "p_min": 0.5, "value_pow": 1.0, "jump_max": 2.0}
_, tti = geometry(df)
val = ASSETS["value"].to_numpy()[None] ** w["value_pow"]
asset_score = val * np.exp(-tti / w["tau"])
best = asset_score.argmax(axis=1)
p = clf.predict_proba(features(df)[FEATS])[:, 1]
spoof = (df["max_jump"] / df["spd"]).to_numpy() > w["jump_max"]
out = df[["id"]].copy()
out["asset"] = ASSETS["asset"].to_numpy()[best]
out["tti_s"] = tti[np.arange(len(df)), best]
out["p_threat"] = p
out["spoof_flag"] = spoof
out["score"] = np.where(spoof, 0, p * asset_score.max(axis=1))
out["urgent_pred"] = (p >= w["p_min"]) & (out["tti_s"] <= TTI_URGENT) & ~spoof
return out.sort_values("score", ascending=False)
BUDGET_MS = 200 # 1회 처리 시간 예산
clf = train_classifier()
def inject(df, rng):
"""손상(NaN·음수 속도) 5%, 중복 10%, 순서 뒤섞기"""
d = df.copy()
bad = rng.random(len(d)) < 0.05
d.loc[bad, "spd"] = rng.choice([np.nan, -1.0], bad.sum())
dup = d.sample(frac=0.10, random_state=int(rng.integers(1e6)))
return pd.concat([d, dup]).sample(frac=1, random_state=1).reset_index(drop=True)
def process(msgs, degrade=False):
"""검증 -> 격리 -> 중복 제거 -> 순위 산정. 어떤 메시지도 조용히 버리지 않고 결과/격리/중복 중 하나로 계수한다."""
ok = msgs["spd"].notna() & (msgs["spd"] > 0) & msgs[["x", "y"]].notna().all(axis=1)
quarantine = msgs[~ok]
good = msgs[ok]
uniq = good.drop_duplicates("id")
if degrade: # 기능 축소: 분류기 생략, 도달시간 순 정렬만 수행(경보 누락 방지 우선)
ranked = baseline_rank(uniq)
else:
ranked = advanced_rank(uniq, clf)
return ranked, len(quarantine), len(good) - len(uniq)
def run_stage(msgs, budget_ms):
"""시간 예산을 넘으면 기능 축소 모드로 다시 처리한다(Fail-safe)."""
t0 = time.perf_counter()
out = process(msgs)
ms = (time.perf_counter() - t0) * 1000
if ms <= budget_ms:
return out, ms, False
t0 = time.perf_counter()
out = process(msgs, degrade=True)
return out, (time.perf_counter() - t0) * 1000, True
rows = []
rng = np.random.default_rng(8)
for n in (100, 200, 300, 450, 600):
raid = make_raid(30 + n, n, spoof=n // 20)
msgs = inject(raid, rng)
crash = 0
try:
(ranked, nq, ndup), ms, degraded = run_stage(msgs, BUDGET_MS)
except Exception: # 다운 계수(발생 시에도 다음 단계 계속)
crash, ranked, nq, ndup, ms, degraded = 1, pd.DataFrame(columns=["id"]), 0, 0, np.nan, False
accounted = len(ranked) + nq + ndup
expect_ids = set(msgs.loc[msgs["spd"].notna() & (msgs["spd"] > 0), "id"])
lost = len(expect_ids - set(ranked["id"]))
rows.append({"비행체": n, "증가율": f"{(n / 100 - 1):.0%}", "메시지": len(msgs), "처리(ms)": round(ms, 1), "격리": nq, "중복": ndup,
"계수누락": len(msgs) - accounted, "유실": lost, "다운": crash, "기능축소": degraded})
res = pd.DataFrame(rows)
print("[부하 단계별 결과]")
print(res.to_string(index=False))
# Fail-safe 확인: 예산을 0.1 ms로 강제해 기능 축소 경로가 실제로 동작하는지 시험
raid = make_raid(99, 600, spoof=30); msgs = inject(raid, rng)
(r_deg, nq, _), ms, degraded = run_stage(msgs, 0.1)
valid = set(msgs.loc[msgs["spd"].notna() & (msgs["spd"] > 0), "id"])
urgent_ids = set(raid.loc[raid["urgent"], "id"]) & valid
miss = len(urgent_ids - set(r_deg.loc[r_deg["urgent_pred"], "id"]))
print(f"\n[Fail-safe 시험] 예산 0.1 ms 강제 -> 기능 축소 {'전환' if degraded else '미전환'}, 긴급 표적 {len(urgent_ids)}건 중 누락 {miss}건")
print(f" 기능 축소 모드는 도달시간 기준 경보를 유지하고(누락 방지 우선), 늘어난 오경보와 격리 메시지 {nq}건은 운용자 확인으로 처리한다.")
print("병목 구간: 분류기 추론·자산별 CPA 행렬 계산(항적 수 x 자산 수)에 비례, 600기에서도 예산 내")
down, loss = int(res["다운"].sum()), int(res["유실"].sum() + res["계수누락"].sum())
print(f"\n트래픽 500% 증가 시 시스템 다운 {down}건 (목표 0건) -> {'달성' if down == 0 else '미달'}")
print(f"데이터 유실 {loss}건, 유실률 {loss / res['메시지'].sum():.2%} (목표 0%) -> {'달성' if loss == 0 else '미달'}")
plt.figure(figsize=(7, 4))
plt.plot(res["비행체"], res["처리(ms)"], "o-", label="processing time")
plt.axhline(200, ls="--", c="r", label="budget 200 ms")
plt.xlabel("Simultaneous tracks"); plt.ylabel("ms"); plt.title("Stress test: latency vs load (virtual)"); plt.legend(); plt.grid(alpha=.3)
plt.tight_layout(); plt.show()
실행 결과 예시
[부하 단계별 결과]
비행체 증가율 메시지 처리(ms) 격리 중복 계수누락 유실 다운 기능축소
100 0% 115 9.8 2 10 0 0 0 False
200 100% 231 9.8 9 20 0 0 0 False
300 200% 347 11.8 9 31 0 0 0 False
450 350% 519 14.8 27 46 0 0 0 False
600 500% 693 10.9 28 60 0 0 0 False
[Fail-safe 시험] 예산 0.1 ms 강제 -> 기능 축소 전환, 긴급 표적 197건 중 누락 0건
기능 축소 모드는 도달시간 기준 경보를 유지하고(누락 방지 우선), 늘어난 오경보와 격리 메시지 23건은 운용자 확인으로 처리한다.
병목 구간: 분류기 추론·자산별 CPA 행렬 계산(항적 수 x 자산 수)에 비례, 600기에서도 예산 내
트래픽 500% 증가 시 시스템 다운 0건 (목표 0건) -> 달성
데이터 유실 0건, 유실률 0.00% (목표 0%) -> 달성
미션 9. AAR 기반 방공 요격 순위 판정 오류 분석 및 위협 가중치 매트릭스 재보정
- 난이도: 고급 / 예상 소요시간: 3.0시간
- 코드 설명: AAR(사후검토)로 초기 위협 가중치 매트릭스의 우선순위 판정 오류를 유형별로 분석하고, 학습용 가상 상황에서 가중치를 재보정한 뒤 별도 시험 상황에서 정확도 향상과 같은 오류의 재발 여부를 확인해 조정 파라미터를 YAML 문자열로 출력한다.
"""AAR(사후검토)로 초기 위협 가중치 매트릭스의 우선순위 판정 오류를 유형별로 분석하고, 학습용 가상 상황에서 가중치를 재보정한 뒤 별도 시험 상황에서 정확도 향상과 같은 오류의 재발 여부를 확인해 조정 파라미터를 YAML 문자열로 출력한다."""
import itertools
import numpy as np
import pandas as pd
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler
# ---------------- [교육용가상자료] 가상 방어지역·가상 비행체 생성기 ----------------
# 좌표는 가상 지휘소를 원점으로 한 km 단위 평면이다. 비행체 유형은 추상 범주이며
# 속도·고도·반사강도는 학습용 임의 범위로, 실제 무기 제원·사거리·탄도와 무관하다.
ASSETS = pd.DataFrame([["A1", "CP(virtual)", 0.0, 0.0, 1.0],
["A2", "Radar(virtual)", 4.0, -3.0, 0.8],
["A3", "Depot(virtual)", -5.0, 4.0, 0.7],
["A4", "Power(virtual)", 7.0, 6.0, 0.5]], columns=["asset", "name", "x", "y", "value"])
TYPES = { # 유형: (위협 여부, 속도 m/s 범위, 고도 m 범위, 출현 비율)
"uav_type": (True, (30, 60), (100, 800), 0.42), # 저속 무인기형(가상)
"fast_type": (True, (150, 250), (50, 200), 0.16), # 고속 순항형(가상)
"decoy": (False, (30, 60), (100, 800), 0.16), # 기만체(가상)
"civil": (False, (120, 220), (3000, 9000), 0.10), # 민항기(가상)
"bird": (False, (8, 18), (50, 400), 0.16), # 조류 군집
}
TTI_URGENT = 600 # 긴급 판정 기준: 가상 자산 도달 예상시간 600초 이내
CPA_KM = 1.5 # 자산 상공 통과 판정 반경(가상)
def make_raid(seed, n=80, spoof=0):
"""가상 포화 상황 1회분(비행체 n개 + 스푸핑 가짜 항적 spoof개)을 만든다."""
rng = np.random.default_rng(seed)
names = list(TYPES)
rows = []
for i in range(n + spoof):
typ = "spoof" if i >= n else str(rng.choice(names, p=[TYPES[k][3] for k in names]))
threat, sp, al, _ = TYPES["decoy" if typ == "spoof" else typ]
brg = np.radians(rng.uniform(0, 360)); dist = rng.uniform(6, 60)
x, y = dist * np.sin(brg), dist * np.cos(brg)
spd = rng.uniform(*sp); alt = rng.uniform(*al)
aim = int(rng.integers(0, len(ASSETS)))
if typ in ("uav_type", "fast_type", "decoy", "spoof"): # 가상 자산 방향(작은 방위 오차)
ax_, ay_ = ASSETS.loc[aim, ["x", "y"]]
hd = np.arctan2(ax_ - x, ay_ - y) + np.radians(rng.normal(0, 0.6))
else:
hd = rng.uniform(0, 2 * np.pi); aim = -1
vx, vy = spd * np.sin(hd) / 1000, spd * np.cos(hd) / 1000 # km/s
sig = {"uav_type": ((0.2, 0.5), (0.02, 0.12), [0, .1, .5, .4]), "fast_type": ((0.1, 0.4), (0.02, 0.12), [0, .2, .4, .4]),
"decoy": ((0.3, 0.8), (0.08, 0.35), [0, .55, .4, .05]), "spoof": ((0.3, 0.8), (0.05, 0.3), [0, .7, .3, 0]),
"civil": ((0.7, 1.0), (0.01, 0.05), [0, 0, .1, .9]), "bird": ((0.01, 0.12), (0.1, 0.4), [0, .6, .4, 0])}[typ]
jump = spd * rng.uniform(0.95, 1.2) if typ != "spoof" else spd * rng.uniform(4, 15)
true_tti = np.inf
if threat:
d_aim = np.hypot(ASSETS.loc[aim, "x"] - x, ASSETS.loc[aim, "y"] - y)
true_tti = d_aim * 1000 / spd
rows.append({"id": f"T{seed % 100:02d}{i:03d}", "type": typ, "x": x, "y": y, "vx": vx, "vy": vy, "alt": alt, "spd": spd,
"rcs": rng.uniform(*sig[0]), "rcs_var": rng.uniform(*sig[1]), "n_sensors": int(rng.choice(4, p=sig[2])),
"max_jump": jump, "civ_iff": int(typ == "civil" and rng.random() < 0.95),
"aim": aim, "threat": threat, "true_tti": true_tti,
"urgent": bool(threat and true_tti <= TTI_URGENT)})
df = pd.DataFrame(rows)
df["true_prio"] = [ASSETS.loc[a, "value"] * np.exp(-t / 300) if th else 0.0
for a, t, th in zip(df["aim"], df["true_tti"], df["threat"])]
return df
def geometry(df):
"""각 항적과 가상 자산 사이 최근접점(CPA) 거리와 도달 예상시간(TTI, s)을 계산한다."""
px, py = df["x"].to_numpy()[:, None], df["y"].to_numpy()[:, None]
vx, vy = df["vx"].to_numpy()[:, None], df["vy"].to_numpy()[:, None]
rx, ry = ASSETS["x"].to_numpy()[None] - px, ASSETS["y"].to_numpy()[None] - py
v2 = vx ** 2 + vy ** 2 + 1e-12
t = (rx * vx + ry * vy) / v2 # CPA 도달 시각(s)
cpa = np.hypot(rx - vx * t, ry - vy * t)
tti = np.where((t > 0) & (cpa < CPA_KM), t, np.inf)
return cpa, tti
# ---------------- 위협 점수 모델(추상 모델: 거리·도달시간·자산 가치·위협 유사도) ----------------
FEATS = ["log_spd", "alt_km", "rcs", "rcs_var", "n_sensors", "civ_iff", "jump_ratio"]
def features(df):
return pd.DataFrame({"log_spd": np.log(df["spd"]), "alt_km": df["alt"] / 1000, "rcs": df["rcs"], "rcs_var": df["rcs_var"],
"n_sensors": df["n_sensors"], "civ_iff": df["civ_iff"], "jump_ratio": df["max_jump"] / df["spd"]})
def train_classifier(seeds=(101, 102, 103, 104)):
"""학습용 가상 상황(시험 상황과 다른 시드)으로 '위협 유사도' 분류기를 학습한다."""
tr = pd.concat([make_raid(s, 120, spoof=10) for s in seeds], ignore_index=True)
clf = make_pipeline(StandardScaler(), LogisticRegression(max_iter=500, C=1.0))
clf.fit(features(tr)[FEATS], tr["threat"].astype(int))
return clf
def baseline_rank(df):
"""Baseline: 거리·도달시간만으로 1차 위협 점수를 매긴다(자산 가치·유형 판별 없음)."""
_, tti = geometry(df)
t_min = tti.min(axis=1)
out = df[["id"]].copy()
out["tti_s"] = t_min
out["score"] = np.exp(-t_min / 300)
out["urgent_pred"] = t_min <= TTI_URGENT
return out.sort_values("score", ascending=False)
def advanced_rank(df, clf, w=None):
"""고도화: 자산 가치 x 도달시간 x 위협 유사도, 운동학 불일치(스푸핑 의심) 항적은 별도 표시한다."""
w = w or {"tau": 300, "p_min": 0.5, "value_pow": 1.0, "jump_max": 2.0}
_, tti = geometry(df)
val = ASSETS["value"].to_numpy()[None] ** w["value_pow"]
asset_score = val * np.exp(-tti / w["tau"])
best = asset_score.argmax(axis=1)
p = clf.predict_proba(features(df)[FEATS])[:, 1]
spoof = (df["max_jump"] / df["spd"]).to_numpy() > w["jump_max"]
out = df[["id"]].copy()
out["asset"] = ASSETS["asset"].to_numpy()[best]
out["tti_s"] = tti[np.arange(len(df)), best]
out["p_threat"] = p
out["spoof_flag"] = spoof
out["score"] = np.where(spoof, 0, p * asset_score.max(axis=1))
out["urgent_pred"] = (p >= w["p_min"]) & (out["tti_s"] <= TTI_URGENT) & ~spoof
return out.sort_values("score", ascending=False)
LEVELS = ["긴급", "주의", "일반"]
def truth(df):
return np.select([df["threat"] & (df["true_tti"] <= 300), df["threat"] & (df["true_tti"] <= 900)], ["긴급", "주의"], "일반")
def judge(df, clf, W):
"""위협 지수 = exp(-TTI/300) x 위협유사도^w_cls x (1 - w_spoof x 스푸핑의심) -> 임계값으로 3단계 판정"""
_, tti = geometry(df)
t = tti.min(axis=1)
p = clf.predict_proba(features(df)[FEATS])[:, 1]
spoof = (df["max_jump"] / df["spd"]).to_numpy() > 2.0
idx = np.exp(-t / 300) * p ** W["w_cls"] * (1 - W["w_spoof"] * spoof)
return np.select([idx >= W["th_urgent"], idx >= W["th_watch"]], ["긴급", "주의"], "일반")
def acc(df, clf, W):
return (judge(df, clf, W) == truth(df)).mean()
clf = train_classifier()
W0 = {"w_cls": 0.0, "w_spoof": 0.0, "th_urgent": 0.37, "th_watch": 0.05} # 초기: 도달시간만 반영
aar = make_raid(7, 80, spoof=6) # AAR 대상 상황
pred0, y = judge(aar, clf, W0), truth(aar)
err = aar[pred0 != y].assign(pred=pred0[pred0 != y], true=y[pred0 != y])
def cause(r):
if r.type == "spoof":
return "스푸핑 항적 미격리"
if r.type in ("decoy", "bird", "civil"):
return "비위협(기만체·조류·민항기) 과대 판정"
return "등급 경계 오차(도달시간 추정)"
err["원인"] = [cause(r) for r in err.itertuples()]
print(f"[AAR] 초기 가중치 정확도 {acc(aar, clf, W0):.1%}, 오류 {len(err)}건")
print(err["원인"].value_counts().to_string())
print(err[["id", "type", "true", "pred", "원인"]].head(5).to_string(index=False))
# 재보정: 학습용 가상 상황 3개(시험과 다른 시드)에서 격자 탐색
train = [make_raid(s, 80, spoof=6) for s in (201, 202, 203)]
grid = itertools.product([0.0, 0.5, 1.0, 2.0, 4.0], [1.0, 0.0], [0.30, 0.37, 0.45], [0.03, 0.05, 0.08])
cands = [{"w_cls": a, "w_spoof": b, "th_urgent": c, "th_watch": d} for a, b, c, d in grid]
aar_idx = [i for i in np.where(pred0 != y)[0] if cause(aar.iloc[i]) != "등급 경계 오차(도달시간 추정)"]
def recurrence(W):
p = judge(aar, clf, W)
return int(sum(p[i] != y[i] for i in aar_idx))
score = {i: np.mean([acc(r, clf, W) for r in train]) for i, W in enumerate(cands)}
top = max(score.values())
# 역환류: 학습 정확도 최고치와 0.5%p 이내인 후보 중 AAR 지적 오류 재발이 가장 적은 가중치를 채택
best = min((W for i, W in enumerate(cands) if score[i] >= top - 0.005), key=lambda W: (recurrence(W), -np.mean([acc(r, clf, W) for r in train])))
tests = [make_raid(s, 80, spoof=6) for s in (8, 9, 10)]
a0 = np.mean([acc(r, clf, W0) for r in tests]); a1 = np.mean([acc(r, clf, best) for r in tests])
pred1 = judge(aar, clf, best)
recur = recurrence(best)
boundary_left = int(sum((pred1[i] != y[i]) for i in np.where(pred0 != y)[0]))
print("\n[재보정 결과] (시험 상황 3개 평균, 학습에 쓰지 않은 상황)")
print(f" 종합 정확도 {a0:.1%} -> {a1:.1%} ({(a1 - a0) * 100:+.1f}%p, 목표 +10%p 이상) -> {'달성' if a1 - a0 >= 0.10 else '미달'}")
print(f" AAR 대상 상황 정확도 {acc(aar, clf, W0):.1%} -> {acc(aar, clf, best):.1%}")
print(f" AAR 지적 오류(기만·스푸핑·비위협) 재발 {recur}건 (목표 0건) -> {'달성' if recur == 0 else '미달'}, 등급 경계 오류 포함 잔여 {boundary_left}건")
for i in aar_idx:
if pred1[i] != y[i]:
r = aar.iloc[i]
print(f" 재발 사례 {r.id}({r.type}): 자산 근접(도달 {geometry(aar.iloc[[i]])[1].min():.0f}s)으로 지수가 '주의' 임계를 넘음 -> 근거리 미확인 항적은 운용자 육안·추가 센서 확인으로 처리")
print(pd.crosstab(pd.Series(truth(tests[0]), name="true"), pd.Series(judge(tests[0], clf, best), name="pred")).to_string())
print("\n# configs/ops04_tuned_params.yaml")
print(f"""threat_matrix:
w_cls: {best['w_cls']} # 위협 유사도 반영 지수 (초기 0.0)
w_spoof: {best['w_spoof']} # 스푸핑 의심 감점 (초기 0.0)
th_urgent: {best['th_urgent']} # 긴급 임계 (exp(-TTI/300) 기준)
th_watch: {best['th_watch']} # 주의 임계
evaluation:
baseline_acc: {a0:.3f}
tuned_acc: {a1:.3f}
note: "[교육용가상자료] 추상 위협 지수 매개변수이며 실제 무기 제원과 무관"
""")
실행 결과 예시
[AAR] 초기 가중치 정확도 75.6%, 오류 21건
원인
비위협(기만체·조류·민항기) 과대 판정 13
스푸핑 항적 미격리 5
등급 경계 오차(도달시간 추정) 3
id type true pred 원인
T07002 uav_type 주의 긴급 등급 경계 오차(도달시간 추정)
T07005 uav_type 주의 긴급 등급 경계 오차(도달시간 추정)
T07010 decoy 일반 주의 비위협(기만체·조류·민항기) 과대 판정
T07020 decoy 일반 주의 비위협(기만체·조류·민항기) 과대 판정
T07021 decoy 일반 주의 비위협(기만체·조류·민항기) 과대 판정
[재보정 결과] (시험 상황 3개 평균, 학습에 쓰지 않은 상황)
종합 정확도 84.9% -> 95.3% (+10.5%p, 목표 +10%p 이상) -> 달성
AAR 대상 상황 정확도 75.6% -> 95.3%
AAR 지적 오류(기만·스푸핑·비위협) 재발 1건 (목표 0건) -> 미달, 등급 경계 오류 포함 잔여 3건
재발 사례 T07061(decoy): 자산 근접(도달 42s)으로 지수가 '주의' 임계를 넘음 -> 근거리 미확인 항적은 운용자 육안·추가 센서 확인으로 처리
pred 긴급 일반 주의
true
긴급 18 0 2
일반 0 46 1
주의 2 0 17
# configs/ops04_tuned_params.yaml
threat_matrix:
w_cls: 1.0 # 위협 유사도 반영 지수 (초기 0.0)
w_spoof: 1.0 # 스푸핑 의심 감점 (초기 0.0)
th_urgent: 0.37 # 긴급 임계 (exp(-TTI/300) 기준)
th_watch: 0.05 # 주의 임계
evaluation:
baseline_acc: 0.849
tuned_acc: 0.953
note: "[교육용가상자료] 추상 위협 지수 매개변수이며 실제 무기 제원과 무관"
미션 10. 샤헤드·미사일 포화경보 우선순위 시스템 최종 통합 및 방공작전 브리핑 패키지 완성
- 난이도: 최고급 / 예상 소요시간: 3.5시간
- 코드 설명: 메시지 검증·전처리, 다차원 위협 평가(재보정 가중치), 신뢰도 등급, 권고 대응 순위, 지휘관 승인 대기열, 대시보드까지 전체 파이프라인을 한 파일로 통합 실행하고, 자체 점검(함수 커버리지 포함) 결과와 방공작전 브리핑 요약을 출력한다.
"""메시지 검증·전처리, 다차원 위협 평가(재보정 가중치), 신뢰도 등급, 권고 대응 순위, 지휘관 승인 대기열, 대시보드까지 전체 파이프라인을 한 파일로 통합 실행하고, 자체 점검(함수 커버리지 포함) 결과와 방공작전 브리핑 요약을 출력한다."""
import hashlib
import json
import sys
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler
# ---------------- [교육용가상자료] 가상 방어지역·가상 비행체 생성기 ----------------
# 좌표는 가상 지휘소를 원점으로 한 km 단위 평면이다. 비행체 유형은 추상 범주이며
# 속도·고도·반사강도는 학습용 임의 범위로, 실제 무기 제원·사거리·탄도와 무관하다.
ASSETS = pd.DataFrame([["A1", "CP(virtual)", 0.0, 0.0, 1.0],
["A2", "Radar(virtual)", 4.0, -3.0, 0.8],
["A3", "Depot(virtual)", -5.0, 4.0, 0.7],
["A4", "Power(virtual)", 7.0, 6.0, 0.5]], columns=["asset", "name", "x", "y", "value"])
TYPES = { # 유형: (위협 여부, 속도 m/s 범위, 고도 m 범위, 출현 비율)
"uav_type": (True, (30, 60), (100, 800), 0.42), # 저속 무인기형(가상)
"fast_type": (True, (150, 250), (50, 200), 0.16), # 고속 순항형(가상)
"decoy": (False, (30, 60), (100, 800), 0.16), # 기만체(가상)
"civil": (False, (120, 220), (3000, 9000), 0.10), # 민항기(가상)
"bird": (False, (8, 18), (50, 400), 0.16), # 조류 군집
}
TTI_URGENT = 600 # 긴급 판정 기준: 가상 자산 도달 예상시간 600초 이내
CPA_KM = 1.5 # 자산 상공 통과 판정 반경(가상)
def make_raid(seed, n=80, spoof=0):
"""가상 포화 상황 1회분(비행체 n개 + 스푸핑 가짜 항적 spoof개)을 만든다."""
rng = np.random.default_rng(seed)
names = list(TYPES)
rows = []
for i in range(n + spoof):
typ = "spoof" if i >= n else str(rng.choice(names, p=[TYPES[k][3] for k in names]))
threat, sp, al, _ = TYPES["decoy" if typ == "spoof" else typ]
brg = np.radians(rng.uniform(0, 360)); dist = rng.uniform(6, 60)
x, y = dist * np.sin(brg), dist * np.cos(brg)
spd = rng.uniform(*sp); alt = rng.uniform(*al)
aim = int(rng.integers(0, len(ASSETS)))
if typ in ("uav_type", "fast_type", "decoy", "spoof"): # 가상 자산 방향(작은 방위 오차)
ax_, ay_ = ASSETS.loc[aim, ["x", "y"]]
hd = np.arctan2(ax_ - x, ay_ - y) + np.radians(rng.normal(0, 0.6))
else:
hd = rng.uniform(0, 2 * np.pi); aim = -1
vx, vy = spd * np.sin(hd) / 1000, spd * np.cos(hd) / 1000 # km/s
sig = {"uav_type": ((0.2, 0.5), (0.02, 0.12), [0, .1, .5, .4]), "fast_type": ((0.1, 0.4), (0.02, 0.12), [0, .2, .4, .4]),
"decoy": ((0.3, 0.8), (0.08, 0.35), [0, .55, .4, .05]), "spoof": ((0.3, 0.8), (0.05, 0.3), [0, .7, .3, 0]),
"civil": ((0.7, 1.0), (0.01, 0.05), [0, 0, .1, .9]), "bird": ((0.01, 0.12), (0.1, 0.4), [0, .6, .4, 0])}[typ]
jump = spd * rng.uniform(0.95, 1.2) if typ != "spoof" else spd * rng.uniform(4, 15)
true_tti = np.inf
if threat:
d_aim = np.hypot(ASSETS.loc[aim, "x"] - x, ASSETS.loc[aim, "y"] - y)
true_tti = d_aim * 1000 / spd
rows.append({"id": f"T{seed % 100:02d}{i:03d}", "type": typ, "x": x, "y": y, "vx": vx, "vy": vy, "alt": alt, "spd": spd,
"rcs": rng.uniform(*sig[0]), "rcs_var": rng.uniform(*sig[1]), "n_sensors": int(rng.choice(4, p=sig[2])),
"max_jump": jump, "civ_iff": int(typ == "civil" and rng.random() < 0.95),
"aim": aim, "threat": threat, "true_tti": true_tti,
"urgent": bool(threat and true_tti <= TTI_URGENT)})
df = pd.DataFrame(rows)
df["true_prio"] = [ASSETS.loc[a, "value"] * np.exp(-t / 300) if th else 0.0
for a, t, th in zip(df["aim"], df["true_tti"], df["threat"])]
return df
def geometry(df):
"""각 항적과 가상 자산 사이 최근접점(CPA) 거리와 도달 예상시간(TTI, s)을 계산한다."""
px, py = df["x"].to_numpy()[:, None], df["y"].to_numpy()[:, None]
vx, vy = df["vx"].to_numpy()[:, None], df["vy"].to_numpy()[:, None]
rx, ry = ASSETS["x"].to_numpy()[None] - px, ASSETS["y"].to_numpy()[None] - py
v2 = vx ** 2 + vy ** 2 + 1e-12
t = (rx * vx + ry * vy) / v2 # CPA 도달 시각(s)
cpa = np.hypot(rx - vx * t, ry - vy * t)
tti = np.where((t > 0) & (cpa < CPA_KM), t, np.inf)
return cpa, tti
# ---------------- 위협 점수 모델(추상 모델: 거리·도달시간·자산 가치·위협 유사도) ----------------
FEATS = ["log_spd", "alt_km", "rcs", "rcs_var", "n_sensors", "civ_iff", "jump_ratio"]
def features(df):
return pd.DataFrame({"log_spd": np.log(df["spd"]), "alt_km": df["alt"] / 1000, "rcs": df["rcs"], "rcs_var": df["rcs_var"],
"n_sensors": df["n_sensors"], "civ_iff": df["civ_iff"], "jump_ratio": df["max_jump"] / df["spd"]})
def train_classifier(seeds=(101, 102, 103, 104)):
"""학습용 가상 상황(시험 상황과 다른 시드)으로 '위협 유사도' 분류기를 학습한다."""
tr = pd.concat([make_raid(s, 120, spoof=10) for s in seeds], ignore_index=True)
clf = make_pipeline(StandardScaler(), LogisticRegression(max_iter=500, C=1.0))
clf.fit(features(tr)[FEATS], tr["threat"].astype(int))
return clf
def baseline_rank(df):
"""Baseline: 거리·도달시간만으로 1차 위협 점수를 매긴다(자산 가치·유형 판별 없음)."""
_, tti = geometry(df)
t_min = tti.min(axis=1)
out = df[["id"]].copy()
out["tti_s"] = t_min
out["score"] = np.exp(-t_min / 300)
out["urgent_pred"] = t_min <= TTI_URGENT
return out.sort_values("score", ascending=False)
def advanced_rank(df, clf, w=None):
"""고도화: 자산 가치 x 도달시간 x 위협 유사도, 운동학 불일치(스푸핑 의심) 항적은 별도 표시한다."""
w = w or {"tau": 300, "p_min": 0.5, "value_pow": 1.0, "jump_max": 2.0}
_, tti = geometry(df)
val = ASSETS["value"].to_numpy()[None] ** w["value_pow"]
asset_score = val * np.exp(-tti / w["tau"])
best = asset_score.argmax(axis=1)
p = clf.predict_proba(features(df)[FEATS])[:, 1]
spoof = (df["max_jump"] / df["spd"]).to_numpy() > w["jump_max"]
out = df[["id"]].copy()
out["asset"] = ASSETS["asset"].to_numpy()[best]
out["tti_s"] = tti[np.arange(len(df)), best]
out["p_threat"] = p
out["spoof_flag"] = spoof
out["score"] = np.where(spoof, 0, p * asset_score.max(axis=1))
out["urgent_pred"] = (p >= w["p_min"]) & (out["tti_s"] <= TTI_URGENT) & ~spoof
return out.sort_values("score", ascending=False)
CONFIG = {"w_cls": 1.0, "w_spoof": 1.0, "th_urgent": 0.37, "th_watch": 0.05, "budget_ms": 200} # 미션 9 재보정값
def validate(df):
ok = df["spd"].between(0.1, 400) & df["alt"].between(0, 15000) & df[["x", "y"]].abs().le(100).all(axis=1)
return df[ok], df[~ok]
def preprocess(df):
good, bad = validate(df)
return good.drop_duplicates("id").reset_index(drop=True), bad
def assess(df, clf):
_, tti = geometry(df)
best = (ASSETS["value"].to_numpy()[None] * np.exp(-tti / 300)).argmax(axis=1)
t = tti[np.arange(len(df)), best]
p = clf.predict_proba(features(df)[FEATS])[:, 1]
spoof = (df["max_jump"] / df["spd"]).to_numpy() > 2.0
idx = np.exp(-t / 300) * p ** CONFIG["w_cls"] * (1 - CONFIG["w_spoof"] * spoof)
lvl = np.select([idx >= CONFIG["th_urgent"], idx >= CONFIG["th_watch"]], ["긴급", "주의"], "일반")
conf = 0.5 * p + 0.2 * df["n_sensors"].to_numpy() / 3 + 0.15 * (1 - np.clip(df["rcs_var"].to_numpy() / 0.4, 0, 1)) + 0.15 * (~spoof)
grade = np.select([conf >= 0.7, conf >= 0.45], ["High", "Med"], "Low")
return df.assign(asset=ASSETS["asset"].to_numpy()[best], tti_s=t, p=p, spoof_flag=spoof, index_=idx, level=lvl, conf=conf, grade=grade)
def recommend(a):
rec = a[a["level"] != "일반"].sort_values("index_", ascending=False).reset_index(drop=True)
rec["권고순위"] = np.arange(1, len(rec) + 1)
rec["상태"] = "승인 대기" # 자동 실행·자동 승인 없음
return rec
class Queue:
def __init__(self):
self.items, self.log = {}, []
def submit(self, r):
self.items[r.id] = "PENDING"
prev = self.log[-1]["hash"] if self.log else "GENESIS"
body = json.dumps({"id": r.id, "rank": int(r.권고순위), "by": "AI"}, sort_keys=True)
self.log.append({"body": body, "hash": hashlib.sha256((prev + body).encode()).hexdigest()})
def run(seed, clf):
raw = make_raid(seed, 120, spoof=8)
rng = np.random.default_rng(seed)
raw.loc[rng.choice(len(raw), 4, replace=False), "spd"] = np.nan # 손상 메시지 주입
raw = pd.concat([raw, raw.sample(10, random_state=seed)]) # 중복 주입
t0 = time.perf_counter()
df, bad = preprocess(raw)
a = assess(df, clf)
rec = recommend(a)
q = Queue()
for r in rec.itertuples():
q.submit(r)
return {"raw": raw, "df": df, "bad": bad, "a": a, "rec": rec, "q": q, "ms": (time.perf_counter() - t0) * 1000}
# ---------------- 1. 통합 실행 + 함수 커버리지 측정 ----------------
called = set()
mine = {"make_raid", "geometry", "features", "train_classifier", "baseline_rank", "advanced_rank", "validate", "preprocess",
"assess", "recommend", "submit", "run"}
sys.setprofile(lambda f, e, a: called.add(f.f_code.co_name) if e == "call" and f.f_code.co_name in mine else None)
t_start = time.perf_counter()
clf = train_classifier()
results, fails = [], 0
for seed in (8, 9, 10, 11, 12):
try:
results.append(run(seed, clf))
except Exception as e:
fails += 1; print("실행 실패:", seed, e)
base_chk = baseline_rank(results[0]["df"]) # 비교 기준(Baseline) 경로도 점검에 포함
adv_chk = advanced_rank(results[0]["df"], clf)
sys.setprofile(None)
cov = len(called & mine) / len(mine)
# ---------------- 2. 성능 집계 ----------------
acc, rec_u, far = [], [], []
for R in results:
a = R["a"]
y = np.select([a["threat"] & (a["true_tti"] <= 300), a["threat"] & (a["true_tti"] <= 900)], ["긴급", "주의"], "일반")
acc.append((a["level"] == y).mean())
rec_u.append(((a["level"] != "일반") & (y == "긴급")).sum() / max((y == "긴급").sum(), 1))
far.append((~a["threat"] & (a["level"] != "일반")).sum() / max((a["level"] != "일반").sum(), 1))
R = results[0]
print(f"[통합 실행] 상황 {len(results) + fails}개 중 성공 {len(results)}개, 평균 처리 {np.mean([r['ms'] for r in results]):.1f} ms")
print(f" 상황 1: 입력 {len(R['raw'])}건 -> 손상 격리 {len(R['bad'])}건, 중복 제거 {len(R['raw']) - len(R['bad']) - len(R['df'])}건, 평가 {len(R['df'])}건")
print(f" 우선순위 판정 정확도 {np.mean(acc):.1%}, 긴급 표적 경보 재현율 {np.mean(rec_u):.1%}, 비위협 경보 비율 {np.mean(far):.1%}")
print("\n[상황 1 권고 대응 순위 상위 5] (모두 승인 대기)")
print(R["rec"].head(5)[["권고순위", "id", "asset", "tti_s", "level", "grade", "상태"]].round({"tti_s": 0}).to_string(index=False))
# ---------------- 3. 자체 점검 ----------------
chk = {
"E2E 파이프라인 실행 성공률 100%": fails == 0,
"손상 메시지 전량 격리": all(r["bad"]["spd"].isna().all() and r["bad"]["id"].nunique() == 4 for r in results),
"스푸핑 의심 항적 경보 0건": all(((r["a"]["spoof_flag"]) & (r["a"]["level"] != "일반")).sum() == 0 for r in results),
"자동 승인 0건(전부 승인 대기)": all(set(r["q"].items.values()) <= {"PENDING"} for r in results),
"긴급 표적 경보 재현율 90% 이상": np.mean(rec_u) >= 0.9,
"처리 시간 예산 이내": max(r["ms"] for r in results) <= CONFIG["budget_ms"],
"함수 커버리지 85% 이상": cov >= 0.85,
}
passed = sum(chk.values())
for k, v in chk.items():
print(f"[{'PASS' if v else 'FAIL'}] {k}")
print(f"자체 점검 {passed}/{len(chk)} 통과, 함수 커버리지 {cov:.0%} (목표 85% 이상) -> {'달성' if cov >= 0.85 else '미달'}")
# ---------------- 4. 브리핑 상황판 ----------------
a = R["a"]
fig, ax = plt.subplots(1, 2, figsize=(12, 5))
col = {"긴급": "red", "주의": "orange", "일반": "lightgray"}
for lv, g in a.groupby("level"):
ax[0].scatter(g["x"], g["y"], s=14, c=col[lv], label={"긴급": "urgent", "주의": "watch", "일반": "normal"}[lv])
ax[0].scatter(ASSETS["x"], ASSETS["y"], marker="^", s=120, c="blue", label="virtual assets")
ax[0].set_aspect("equal"); ax[0].legend(fontsize=8); ax[0].set_title("Integrated picture (scenario 1)")
ax[1].axis("off")
ax[1].text(0, 0.9, f"Accuracy {np.mean(acc):.1%}\nUrgent recall {np.mean(rec_u):.1%}\nPending recs {len(R['rec'])}\n"
f"Auto-approved 0\nSelf-check {passed}/{len(chk)}\nCoverage {cov:.0%}", family="monospace", fontsize=13, va="top")
plt.suptitle("OPS04 saturation alert system - training data only"); plt.tight_layout(); plt.show()
# ---------------- 5. 방공작전 브리핑 요약 ----------------
lv = R["a"]["level"].value_counts()
print(f"""
[포화경보 우선순위 브리핑 요약 - 교육용가상자료]
1. 상황: 가상 비행체 {len(R['df'])}건 평가, 긴급 {lv.get('긴급', 0)} / 주의 {lv.get('주의', 0)} / 일반 {lv.get('일반', 0)}
2. 성능: 판정 정확도 {np.mean(acc):.1%}, 긴급 재현율 {np.mean(rec_u):.1%}, 처리 {R['ms']:.1f} ms, 전체 소요 {time.perf_counter() - t_start:.1f} s
3. 권고: 대응 권고 {len(R['rec'])}건 모두 지휘관 승인 대기, 자동 승인 0건
4. 운용 유의: 순위는 권고이며 대응 여부는 지휘관 승인과 교전수칙·지휘계통에 따른다.
5. 한계: 추상 위협 지수와 가상 데이터 기반 결과로, 실제 운용 전에는 실제 센서 자료로 재검증해야 한다.""")
실행 결과 예시
[통합 실행] 상황 5개 중 성공 5개, 평균 처리 31.0 ms
상황 1: 입력 138건 -> 손상 격리 4건, 중복 제거 10건, 평가 124건
우선순위 판정 정확도 94.5%, 긴급 표적 경보 재현율 100.0%, 비위협 경보 비율 3.3%
[상황 1 권고 대응 순위 상위 5] (모두 승인 대기)
권고순위 id asset tti_s level grade 상태
1 T08115 A3 11.0 긴급 High 승인 대기
2 T08017 A4 54.0 긴급 High 승인 대기
3 T08094 A3 60.0 긴급 High 승인 대기
4 T08119 A4 64.0 긴급 High 승인 대기
5 T08015 A3 69.0 긴급 High 승인 대기
[PASS] E2E 파이프라인 실행 성공률 100%
[PASS] 손상 메시지 전량 격리
[PASS] 스푸핑 의심 항적 경보 0건
[PASS] 자동 승인 0건(전부 승인 대기)
[PASS] 긴급 표적 경보 재현율 90% 이상
[PASS] 처리 시간 예산 이내
[PASS] 함수 커버리지 85% 이상
자체 점검 7/7 통과, 함수 커버리지 100% (목표 85% 이상) -> 달성
[포화경보 우선순위 브리핑 요약 - 교육용가상자료]
1. 상황: 가상 비행체 124건 평가, 긴급 30 / 주의 30 / 일반 64
2. 성능: 판정 정확도 94.5%, 긴급 재현율 100.0%, 처리 29.1 ms, 전체 소요 1.7 s
3. 권고: 대응 권고 60건 모두 지휘관 승인 대기, 자동 승인 0건
4. 운용 유의: 순위는 권고이며 대응 여부는 지휘관 승인과 교전수칙·지휘계통에 따른다.
5. 한계: 추상 위협 지수와 가상 데이터 기반 결과로, 실제 운용 전에는 실제 센서 자료로 재검증해야 한다.