178 KiB
178 KiB
🎯 [OPS01] 드론 영상 공중위협 객체 탐지와 오경보 검증
Google Colab 실습 환경 (작전·전투 수행 / Level 3)
📌 1. 프로젝트 개요 및 배경
| 항목 | 내용 |
|---|---|
| 과정 분류 | 작전·전투 수행 (Level 3) |
| 학습 소요 시간 | 10개 미션 (5일 / 총 27.5시간) |
| 환경 사양 | Python 3.10+, numpy, pandas, matplotlib, scipy, scikit-learn |
📖 프로젝트 목표
러시아-우크라이나 전쟁에서는 소형 드론이 정찰과 공격에 대량으로 쓰이면서, 영상으로 공중위협을 빨리 찾아내는 일만큼 조류·민간 항공기·지상 반사체 때문에 생기는 오경보를 줄이는 일이 중요해졌다. 본 프로젝트에서는 교육용 가상 적외선 영상을 코드로 직접 만들어, 데이터 스키마 정의, 기상 왜곡 전처리, 탐지 Baseline, 군집·급강하 패턴 식별, 신뢰도 점수 산출, 전술 상황판, 인간 개입(HITL) 승인/보류, 포화 공격 스트레스 시험, AAR 재보정, 최종 통합까지 대드론 탐지·경보 파이프라인을 단계별로 구현한다. 최종 목표는 드론을 놓치지 않으면서 오경보를 줄이고, 모든 교전 관련 판단이 사람의 승인을 거치도록 설계하는 것이다.
⚠️ [교육용 가상 데이터셋 원칙]
- 본 노트북의 모든 데이터(제원, 로그, 시나리오 등)는 보안 규정을 준수하여 코드 내 난수 기반 가상 데이터로 생성됩니다.
- 실제 부대 위치, 작전 인가, 실제 전장 정보, 군사 기밀 데이터는 일체 포함되지 않습니다.
In [ ]:
# ============================================================================
# [Google Colab 환경 설정] 필수 라이브러리 설치 및 임포트
# ============================================================================
!pip install -q numpy pandas matplotlib scipy scikit-learn
import os
import sys
import time
import math
import json
import re
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import warnings
warnings.filterwarnings('ignore')
# 그래프 영문 폰트 기본 설정 (환경 독립성 보장)
plt.rcParams['font.sans-serif'] = 'DejaVu Sans'
plt.rcParams['axes.unicode_minus'] = False
print(f"실행 환경 준비 완료 (Python {sys.version.split()[0]})")
In [ ]:
"""드론 탑재 EO/IR 영상의 공중위협 객체(정찰·자폭드론, 순항미사일) 메타데이터·라벨 스키마를 정의하고, 가상 샘플 데이터를 검증한다."""
import json
import re
import numpy as np
import pandas as pd
# 1) 스키마 정의: 센서 규격, 전술 파라미터(기상·위치오차·통신지연), 바운딩 박스 라벨
SCHEMA = {
"required": ["sequence_id", "sensor", "tactical", "n_frames", "labels"],
"fields": {
"sequence_id": {"type": str, "pattern": r"^SEQ\d{3}$"},
"n_frames": {"type": int, "min": 1, "max": 3000},
},
"sensor": {
"type": {"type": str, "enum": ["EO", "IR"]},
"resolution": {"type": list, "length": 2},
"fps": {"type": (int, float), "min": 1, "max": 60},
},
"tactical": {
"weather": {"type": str, "enum": ["clear", "night", "fog", "rain", "smoke"]},
"position_error_m": {"type": (int, float), "min": 0, "max": 500},
"comm_delay_ms": {"type": (int, float), "min": 0, "max": 10000},
},
"label": {
"class": {"type": str, "enum": ["recon_drone", "attack_drone", "cruise_missile", "bird", "civil_aircraft", "clutter"]},
"bbox": {"type": list, "length": 4},
"frame": {"type": int, "min": 0},
"thermal": {"type": (int, float), "min": 0, "max": 255},
},
}
def check(name, value, rule):
"""값 하나를 규칙(type, enum, min, max, length, pattern)으로 검사해 오류 목록을 돌려준다."""
# [TODO : 학생 실습 1] 전술 드론 영상 수집 환경 분석 및 공중위협 객체(드론/미사일) 데이터셋 스키마 정의 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def validate(rec):
# [TODO : 학생 실습 1] 전술 드론 영상 수집 환경 분석 및 공중위협 객체(드론/미사일) 데이터셋 스키마 정의 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
rng = np.random.default_rng(1)
classes = SCHEMA["label"]["class"]["enum"]
records = []
for i in range(20):
labels = [{"class": str(rng.choice(classes)), "frame": int(f),
"bbox": [int(rng.integers(0, 60)), int(rng.integers(0, 60)), 3, 3],
"thermal": float(round(rng.uniform(100, 240), 1))} for f in range(3)]
records.append({"sequence_id": f"SEQ{i:03d}", "n_frames": 30,
"sensor": {"type": str(rng.choice(["EO", "IR"])), "resolution": [64, 64], "fps": 10},
"tactical": {"weather": str(rng.choice(["clear", "night", "fog", "rain", "smoke"])),
"position_error_m": float(round(rng.uniform(2, 30), 1)),
"comm_delay_ms": int(rng.integers(20, 400))},
"labels": labels})
# 3) 정상 데이터 검증
ok = sum(1 for r in records if not validate(r))
print(f"정상 샘플 {len(records)}건 중 검증 통과 {ok}건, 오류 {len(records) - ok}건")
# 4) 오류 주입 시험: 검증기가 잘못된 데이터를 잡아내는지 확인
bad = json.loads(json.dumps(records[0]))
bad["sequence_id"] = "SEQ-1"
bad["sensor"]["fps"] = 120
del bad["tactical"]["comm_delay_ms"]
bad["labels"][0]["class"] = "ufo"
bad["labels"][1]["bbox"] = [1, 2, 3]
print("\n[오류 주입 결과]")
for e in validate(bad):
print(" -", e)
# 5) 스키마 필드 정의서(명세서 초안)
rows = [("기본", k, r["type"].__name__, str({x: y for x, y in r.items() if x != "type"})) for k, r in SCHEMA["fields"].items()]
for g in ("sensor", "tactical", "label"):
for k, r in SCHEMA[g].items():
t = r["type"].__name__ if isinstance(r["type"], type) else "number"
rows.append((g, k, t, str({x: y for x, y in r.items() if x != "type"})))
spec = pd.DataFrame(rows, columns=["그룹", "필드", "자료형", "규칙"])
print("\n[데이터 명세 요약]")
print(spec.to_string(index=False))
print("\n클래스별 라벨 수:", pd.Series([l["class"] for r in records for l in r["labels"]]).value_counts().to_dict())
In [ ]:
"""안개·강우·야간 등 기상 왜곡과 센서 잡음이 섞인 EO/IR 영상을 조건별로 전처리(잡음 제거·빗줄기 제거·명암비 강화·정규화)하고, 처리 속도와 탐지 품질 개선 효과를 확인한다."""
import time
import numpy as np
import matplotlib.pyplot as plt
from scipy import ndimage
# ---------------------------------------------------------------------------
# [교육용가상자료] 합성 적외선 영상 생성기 (실제 영상·부대 정보 없음)
# 64x64 프레임에 드론·조류·민간기·지상 반사체(클러터)를 그리고 정답 라벨을 함께 만든다.
# ---------------------------------------------------------------------------
H = W = 64
PROFILE = {"drone": ((2, 3), (170, 225)), "bird": ((2, 3), (110, 150)),
"civil": ((4, 6), (190, 240)), "clutter": ((2, 4), (140, 190))}
def _new_obj(cls, rng, dive=False, v=None, xy=None):
# [TODO : 학생 실습 2] 적외선·광학(EO/IR) 전장 영상 노이즈 제거 및 기상 왜곡 보정 전처리 모듈 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def _noise(img, cond, rng):
# [TODO : 학생 실습 2] 적외선·광학(EO/IR) 전장 영상 노이즈 제거 및 기상 왜곡 보정 전처리 모듈 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def make_sequence(seq_id, cond, rng, kind="mixed", n_frames=30, n_drone=None, n_bird=None):
"""시퀀스 1개 = (프레임 배열, 라벨 목록). kind: mixed / swarm / dive / saturation"""
objs = []
if kind == "swarm":
v, cx, cy = rng.uniform(-1, 1, 2), rng.uniform(15, 35), rng.uniform(15, 35)
objs = [_new_obj("drone", rng, v=v, xy=(cx + i * 5, cy + (i % 2) * 5)) for i in range(4)]
objs.append(_new_obj("bird", rng))
elif kind == "dive":
d = _new_obj("drone", rng, dive=True)
d["y"] = float(rng.uniform(4, 14))
objs = [d, _new_obj("bird", rng), _new_obj("clutter", rng)]
elif kind == "saturation":
objs = [_new_obj("drone", rng) for _ in range(n_drone)] + [_new_obj("bird", rng) for _ in range(n_bird)]
else:
objs = [_new_obj("drone", rng) for _ in range(int(rng.integers(1, 3)))]
objs += [_new_obj("bird", rng) for _ in range(int(rng.integers(0, 3)))]
if rng.random() < 0.5:
objs.append(_new_obj("civil", rng))
if rng.random() < 0.6:
objs.append(_new_obj("clutter", rng))
frames, labels = np.zeros((n_frames, H, W), np.uint8), []
base = rng.normal(40, 6, (H, W))
for t in range(n_frames):
canvas = base.copy()
for i, o in enumerate(objs):
if t > 0:
if o["cls"] == "bird":
o["vx"] = float(np.clip(o["vx"] + rng.normal(0, 0.45), -1.6, 1.6))
o["vy"] = float(np.clip(o["vy"] + rng.normal(0, 0.45), -1.6, 1.6))
if o["dive"] and t > 10:
o["vy"] += 0.18
o["x"] = float(np.clip(o["x"] + o["vx"], 1, W - o["size"] - 1))
o["y"] = float(np.clip(o["y"] + o["vy"], 1, H - o["size"] - 1))
heat = o["heat"]
if o["cls"] == "clutter" and rng.random() < 0.3:
heat *= 0.7
if o["cls"] == "bird":
heat *= float(rng.uniform(0.85, 1.1))
x, y, s = int(round(o["x"])), int(round(o["y"])), o["size"]
canvas[y:y + s, x:x + s] = np.maximum(canvas[y:y + s, x:x + s], heat)
labels.append({"frame": t, "track": f"{seq_id}_T{i:02d}", "class": o["cls"], "bbox": [x, y, s, s],
"pattern": "dive" if o["dive"] else ("swarm" if kind == "swarm" and o["cls"] == "drone" else "normal")})
frames[t] = _noise(canvas, cond, rng)
return frames, labels
def make_dataset(n=24, seed=7):
rng = np.random.default_rng(seed)
conds = ["clear", "night", "fog", "rain"]
out = []
for i in range(n):
kind = "swarm" if i % 8 == 5 else ("dive" if i % 8 == 6 else "mixed")
f, lab = make_sequence(f"SEQ{i:03d}", conds[i % 4], rng, kind)
out.append({"id": f"SEQ{i:03d}", "cond": conds[i % 4], "kind": kind, "frames": f, "labels": lab})
return out
# ---------------------------------------------------------------------------
def denoise(f, size):
return f if size <= 1 else ndimage.median_filter(f, size=size)
def remove_rain(f):
return ndimage.grey_opening(f, size=(1, 2))
def clahe_like(f, tiles=4, clip=0.02):
"""타일별 히스토그램 평활화 후 전체 대비 확장(간이 CLAHE). 안개로 떨어진 국소 대비를 살린다."""
out = np.zeros_like(f, dtype=float)
th, tw = f.shape[0] // tiles, f.shape[1] // tiles
for i in range(tiles):
for j in range(tiles):
t = f[i * th:(i + 1) * th, j * tw:(j + 1) * tw].astype(float)
hist, edges = np.histogram(t, bins=64, range=(0, 256))
limit = max(1, int(clip * t.size))
hist = np.minimum(hist, limit)
cdf = np.cumsum(hist) / hist.sum()
out[i * th:(i + 1) * th, j * tw:(j + 1) * tw] = np.interp(t, edges[:-1], cdf) * 255
return 0.5 * out + 0.5 * stretch(f)
def stretch(f):
lo, hi = np.percentile(f, [1, 99.5])
return np.clip((f.astype(float) - lo) / (hi - lo + 1e-6) * 255, 0, 255)
def normalize(f):
return f.astype(np.float32) / 255.0
PARAMS = {"clear": dict(median=1, rain=False), "night": dict(median=1, rain=False),
"fog": dict(median=1, rain=False), "rain": dict(median=1, rain=True)}
def preprocess(frame, cond):
p = PARAMS[cond]
f = denoise(frame.astype(float), p["median"])
if p["rain"]:
f = remove_rain(f)
return stretch(f)
def quick_f1(seqs, use_pre, k=2.5):
tp = fp = fn = 0
for s in seqs:
for t, fr in enumerate(s["frames"]):
img = preprocess(fr, s["cond"]) if use_pre else fr.astype(float)
lab, n = ndimage.label(img > img.mean() + k * img.std())
sizes = ndimage.sum(np.ones_like(img), lab, range(1, n + 1))
n_det = int(((sizes >= 3) & (sizes <= 60)).sum())
n_gt = sum(1 for l in s["labels"] if l["frame"] == t)
tp += min(n_det, n_gt); fp += max(0, n_det - n_gt); fn += max(0, n_gt - n_det)
p, r = tp / max(tp + fp, 1), tp / max(tp + fn, 1)
return 2 * p * r / max(p + r, 1e-9)
data = make_dataset(24, seed=7)
# 1) 처리 속도 측정 (프레임/초)
frames = [(fr, s["cond"]) for s in data for fr in s["frames"]]
t0 = time.perf_counter()
for fr, c in frames:
normalize(preprocess(fr, c))
dt = time.perf_counter() - t0
print(f"전처리 {len(frames)}프레임, {len(frames) / dt:,.0f} 프레임/초, 오류 0건")
# 2) 조건별 전처리 전후 탐지 F1 비교
print("\n조건 전처리 전 F1 전처리 후 F1")
for c in ["clear", "night", "fog", "rain"]:
sub = [s for s in data if s["cond"] == c]
print(f"{c:6s} {quick_f1(sub, False):10.3f} {quick_f1(sub, True):10.3f}")
# 3) 필터를 과하게 쓰면 소형 표적이 사라지는지 확인 (3x3 중간값 필터)
PARAMS["night"]["median"] = 3
night = [s for s in data if s["cond"] == "night"]
print(f"\n야간 3x3 중간값 필터 적용 시 F1: {quick_f1(night, True):.3f} (2px 소형 드론이 함께 지워짐)")
PARAMS["night"]["median"] = 1
# 4) 결측 프레임 보정(앞 프레임으로 채우기)
seq = list(data[3]["frames"]); seq[5] = None; seq[6] = None
filled = list(seq)
for i in range(len(filled)):
if filled[i] is None:
filled[i] = filled[i - 1]
print("결측 프레임 2장 보정 완료:", all(f is not None for f in filled))
# 5) 전후 비교 그림
fig, ax = plt.subplots(2, 4, figsize=(12, 6))
for j, c in enumerate(["clear", "night", "fog", "rain"]):
s = next(x for x in data if x["cond"] == c)
ax[0, j].imshow(s["frames"][0], cmap="gray"); ax[0, j].set_title(f"{c} raw")
ax[1, j].imshow(clahe_like(s["frames"][0]) if c == "fog" else preprocess(s["frames"][0], c), cmap="gray")
ax[1, j].set_title(f"{c} preprocessed")
for a in ax.ravel():
a.axis("off")
plt.suptitle("Synthetic IR frames: before / after preprocessing (training data)")
plt.tight_layout()
plt.show()
In [ ]:
"""규칙 기반(밝기 임계값 + 연결 영역 분석) 저고도 공중위협 객체 탐지 Baseline을 구현하고, IoU 매칭으로 탐지 정확도(정밀도·재현율)를 측정한다. 경량 YOLO를 쓸 경우와의 차이도 함께 비교한다."""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from matplotlib.patches import Rectangle
from scipy import ndimage
# ---------------------------------------------------------------------------
# [교육용가상자료] 합성 적외선 영상 생성기 (실제 영상·부대 정보 없음)
# 64x64 프레임에 드론·조류·민간기·지상 반사체(클러터)를 그리고 정답 라벨을 함께 만든다.
# ---------------------------------------------------------------------------
H = W = 64
PROFILE = {"drone": ((2, 3), (170, 225)), "bird": ((2, 3), (110, 150)),
"civil": ((4, 6), (190, 240)), "clutter": ((2, 4), (140, 190))}
def _new_obj(cls, rng, dive=False, v=None, xy=None):
# [TODO : 학생 실습 3] YOLO/규칙 기반 저고도 공중위협 객체 탐지 Baseline 프로토타입 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def _noise(img, cond, rng):
# [TODO : 학생 실습 3] YOLO/규칙 기반 저고도 공중위협 객체 탐지 Baseline 프로토타입 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def make_sequence(seq_id, cond, rng, kind="mixed", n_frames=30, n_drone=None, n_bird=None):
"""시퀀스 1개 = (프레임 배열, 라벨 목록). kind: mixed / swarm / dive / saturation"""
objs = []
if kind == "swarm":
v, cx, cy = rng.uniform(-1, 1, 2), rng.uniform(15, 35), rng.uniform(15, 35)
objs = [_new_obj("drone", rng, v=v, xy=(cx + i * 5, cy + (i % 2) * 5)) for i in range(4)]
objs.append(_new_obj("bird", rng))
elif kind == "dive":
d = _new_obj("drone", rng, dive=True)
d["y"] = float(rng.uniform(4, 14))
objs = [d, _new_obj("bird", rng), _new_obj("clutter", rng)]
elif kind == "saturation":
objs = [_new_obj("drone", rng) for _ in range(n_drone)] + [_new_obj("bird", rng) for _ in range(n_bird)]
else:
objs = [_new_obj("drone", rng) for _ in range(int(rng.integers(1, 3)))]
objs += [_new_obj("bird", rng) for _ in range(int(rng.integers(0, 3)))]
if rng.random() < 0.5:
objs.append(_new_obj("civil", rng))
if rng.random() < 0.6:
objs.append(_new_obj("clutter", rng))
frames, labels = np.zeros((n_frames, H, W), np.uint8), []
base = rng.normal(40, 6, (H, W))
for t in range(n_frames):
canvas = base.copy()
for i, o in enumerate(objs):
if t > 0:
if o["cls"] == "bird":
o["vx"] = float(np.clip(o["vx"] + rng.normal(0, 0.45), -1.6, 1.6))
o["vy"] = float(np.clip(o["vy"] + rng.normal(0, 0.45), -1.6, 1.6))
if o["dive"] and t > 10:
o["vy"] += 0.18
o["x"] = float(np.clip(o["x"] + o["vx"], 1, W - o["size"] - 1))
o["y"] = float(np.clip(o["y"] + o["vy"], 1, H - o["size"] - 1))
heat = o["heat"]
if o["cls"] == "clutter" and rng.random() < 0.3:
heat *= 0.7
if o["cls"] == "bird":
heat *= float(rng.uniform(0.85, 1.1))
x, y, s = int(round(o["x"])), int(round(o["y"])), o["size"]
canvas[y:y + s, x:x + s] = np.maximum(canvas[y:y + s, x:x + s], heat)
labels.append({"frame": t, "track": f"{seq_id}_T{i:02d}", "class": o["cls"], "bbox": [x, y, s, s],
"pattern": "dive" if o["dive"] else ("swarm" if kind == "swarm" and o["cls"] == "drone" else "normal")})
frames[t] = _noise(canvas, cond, rng)
return frames, labels
def make_dataset(n=24, seed=7):
rng = np.random.default_rng(seed)
conds = ["clear", "night", "fog", "rain"]
out = []
for i in range(n):
kind = "swarm" if i % 8 == 5 else ("dive" if i % 8 == 6 else "mixed")
f, lab = make_sequence(f"SEQ{i:03d}", conds[i % 4], rng, kind)
out.append({"id": f"SEQ{i:03d}", "cond": conds[i % 4], "kind": kind, "frames": f, "labels": lab})
return out
# ---------------------------------------------------------------------------
# ---- 전처리·탐지 공통 함수 ----
def preprocess(frame, cond):
f = frame.astype(float)
if cond == "rain": # 폭 1px 세로 빗줄기 제거(가로 방향 형태학적 열림)
f = ndimage.grey_opening(f, size=(1, 2))
lo, hi = np.percentile(f, [1, 99.5]) # 대비 확장
return np.clip((f - lo) / (hi - lo + 1e-6) * 255, 0, 255)
def detect(frame, cond, k=2.0, min_area=3, max_area=60):
img = preprocess(frame, cond)
lab, _ = ndimage.label(img > img.mean() + k * img.std())
dets = []
for i, sl in enumerate(ndimage.find_objects(lab), start=1):
area = int((lab[sl] == i).sum())
if min_area <= area <= max_area:
y, x = sl[0].start, sl[1].start
dets.append({"bbox": [x, y, sl[1].stop - x, sl[0].stop - y], "area": area,
"thermal": float(frame[sl][lab[sl] == i].max())})
return dets
def iou(a, b):
ix = max(0, min(a[0] + a[2], b[0] + b[2]) - max(a[0], b[0]))
iy = max(0, min(a[1] + a[3], b[1] + b[3]) - max(a[1], b[1]))
inter = ix * iy
return inter / (a[2] * a[3] + b[2] * b[3] - inter)
def evaluate(data, k=2.0, classes=("drone", "bird", "civil", "clutter")):
tp = fp = fn = 0
for s in data:
for t, fr in enumerate(s["frames"]):
gts = [l["bbox"] for l in s["labels"] if l["frame"] == t and l["class"] in classes]
used = set()
dets = detect(fr, s["cond"], k)
for d in dets:
best = max(((iou(d["bbox"], g), j) for j, g in enumerate(gts) if j not in used), default=(0, -1))
if best[0] >= 0.1:
tp += 1; used.add(best[1])
fp += len(dets) - len(used); fn += len(gts) - len(used)
p, r = tp / max(tp + fp, 1), tp / max(tp + fn, 1)
return {"TP": tp, "FP": fp, "FN": fn, "정밀도": round(p, 3), "재현율": round(r, 3), "F1": round(2 * p * r / max(p + r, 1e-9), 3)}
data = make_dataset(24, seed=7)
# 1) 기상 조건별 객체 탐지 성능 (드론·조류·민간기·클러터 모두 '객체'로 봄)
rows = []
for c in ["clear", "night", "fog", "rain"]:
rows.append({"조건": c, **evaluate([s for s in data if s["cond"] == c])})
rows.append({"조건": "전체", **evaluate(data)})
print(pd.DataFrame(rows).to_string(index=False))
acc = rows[-1]["재현율"]
print(f"\n목표(탐지 정확도 75% 이상) 달성 여부: {'달성' if acc >= 0.75 else '미달'} (재현율 {acc:.1%})")
# 2) 드론만 기준으로 보면? -> 조류·민간기까지 잡혀 정밀도가 크게 떨어진다 (미션 4·5의 동기)
d = evaluate(data, classes=("drone",))
print(f"드론만 정답으로 볼 때 정밀도 {d['정밀도']:.1%}, 재현율 {d['재현율']:.1%}")
# 3) 임계값 k에 따른 민감도 변화
print("\nk 정밀도 재현율")
for k in [1.5, 2.0, 2.5, 3.0, 3.5]:
e = evaluate(data, k)
print(f"{k:<4} {e['정밀도']:.3f} {e['재현율']:.3f}")
# 4) 규칙 기반 vs 경량 YOLO 비교표
print(pd.DataFrame({"항목": ["학습 데이터", "연산 자원", "소형 표적(2~3px)", "판단 근거 설명", "도입 속도"],
"규칙 기반(본 Baseline)": ["불필요", "CPU로 충분", "밝기 차이에 의존", "쉬움(임계값)", "즉시"],
"경량 YOLO": ["라벨 수천 장 이상", "GPU 권장", "학습 데이터 품질에 의존", "어려움", "학습 후"]}).to_string(index=False))
# 5) 탐지 결과 시각화
s = data[5]
fig, ax = plt.subplots(1, 3, figsize=(12, 4))
for a, t in zip(ax, [0, 10, 20]):
a.imshow(s["frames"][t], cmap="gray")
for det in detect(s["frames"][t], s["cond"]):
x, y, w, h = det["bbox"]
a.add_patch(Rectangle((x - .5, y - .5), w, h, fill=False, color="red"))
a.set_title(f"{s['id']} frame {t} (red = detection)")
a.axis("off")
plt.tight_layout()
plt.show()
In [ ]:
"""프레임별 탐지를 궤적으로 연결(추적)하고, 궤적의 움직임 특징으로 군집 드론(동일 속도 동행)·급강하 자폭 궤적·불규칙(조류형 기만) 비행 패턴을 식별해 복합 위협 탐지 재현율과 처리 지연을 측정한다."""
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy import ndimage
# ---------------------------------------------------------------------------
# [교육용가상자료] 합성 적외선 영상 생성기 (실제 영상·부대 정보 없음)
# 64x64 프레임에 드론·조류·민간기·지상 반사체(클러터)를 그리고 정답 라벨을 함께 만든다.
# ---------------------------------------------------------------------------
H = W = 64
PROFILE = {"drone": ((2, 3), (170, 225)), "bird": ((2, 3), (110, 150)),
"civil": ((4, 6), (190, 240)), "clutter": ((2, 4), (140, 190))}
def _new_obj(cls, rng, dive=False, v=None, xy=None):
# [TODO : 학생 실습 4] 군집 드론 및 비행체 기만 비행 패턴 식별 복합 위협 탐지 모델 고도화 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def _noise(img, cond, rng):
# [TODO : 학생 실습 4] 군집 드론 및 비행체 기만 비행 패턴 식별 복합 위협 탐지 모델 고도화 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def make_sequence(seq_id, cond, rng, kind="mixed", n_frames=30, n_drone=None, n_bird=None):
"""시퀀스 1개 = (프레임 배열, 라벨 목록). kind: mixed / swarm / dive / saturation"""
objs = []
if kind == "swarm":
v, cx, cy = rng.uniform(-1, 1, 2), rng.uniform(15, 35), rng.uniform(15, 35)
objs = [_new_obj("drone", rng, v=v, xy=(cx + i * 5, cy + (i % 2) * 5)) for i in range(4)]
objs.append(_new_obj("bird", rng))
elif kind == "dive":
d = _new_obj("drone", rng, dive=True)
d["y"] = float(rng.uniform(4, 14))
objs = [d, _new_obj("bird", rng), _new_obj("clutter", rng)]
elif kind == "saturation":
objs = [_new_obj("drone", rng) for _ in range(n_drone)] + [_new_obj("bird", rng) for _ in range(n_bird)]
else:
objs = [_new_obj("drone", rng) for _ in range(int(rng.integers(1, 3)))]
objs += [_new_obj("bird", rng) for _ in range(int(rng.integers(0, 3)))]
if rng.random() < 0.5:
objs.append(_new_obj("civil", rng))
if rng.random() < 0.6:
objs.append(_new_obj("clutter", rng))
frames, labels = np.zeros((n_frames, H, W), np.uint8), []
base = rng.normal(40, 6, (H, W))
for t in range(n_frames):
canvas = base.copy()
for i, o in enumerate(objs):
if t > 0:
if o["cls"] == "bird":
o["vx"] = float(np.clip(o["vx"] + rng.normal(0, 0.45), -1.6, 1.6))
o["vy"] = float(np.clip(o["vy"] + rng.normal(0, 0.45), -1.6, 1.6))
if o["dive"] and t > 10:
o["vy"] += 0.18
o["x"] = float(np.clip(o["x"] + o["vx"], 1, W - o["size"] - 1))
o["y"] = float(np.clip(o["y"] + o["vy"], 1, H - o["size"] - 1))
heat = o["heat"]
if o["cls"] == "clutter" and rng.random() < 0.3:
heat *= 0.7
if o["cls"] == "bird":
heat *= float(rng.uniform(0.85, 1.1))
x, y, s = int(round(o["x"])), int(round(o["y"])), o["size"]
canvas[y:y + s, x:x + s] = np.maximum(canvas[y:y + s, x:x + s], heat)
labels.append({"frame": t, "track": f"{seq_id}_T{i:02d}", "class": o["cls"], "bbox": [x, y, s, s],
"pattern": "dive" if o["dive"] else ("swarm" if kind == "swarm" and o["cls"] == "drone" else "normal")})
frames[t] = _noise(canvas, cond, rng)
return frames, labels
def make_dataset(n=24, seed=7):
rng = np.random.default_rng(seed)
conds = ["clear", "night", "fog", "rain"]
out = []
for i in range(n):
kind = "swarm" if i % 8 == 5 else ("dive" if i % 8 == 6 else "mixed")
f, lab = make_sequence(f"SEQ{i:03d}", conds[i % 4], rng, kind)
out.append({"id": f"SEQ{i:03d}", "cond": conds[i % 4], "kind": kind, "frames": f, "labels": lab})
return out
# ---------------------------------------------------------------------------
# ---- 전처리·탐지 공통 함수 ----
def preprocess(frame, cond):
f = frame.astype(float)
if cond == "rain": # 폭 1px 세로 빗줄기 제거(가로 방향 형태학적 열림)
f = ndimage.grey_opening(f, size=(1, 2))
lo, hi = np.percentile(f, [1, 99.5]) # 대비 확장
return np.clip((f - lo) / (hi - lo + 1e-6) * 255, 0, 255)
def detect(frame, cond, k=2.0, min_area=3, max_area=60):
img = preprocess(frame, cond)
lab, _ = ndimage.label(img > img.mean() + k * img.std())
dets = []
for i, sl in enumerate(ndimage.find_objects(lab), start=1):
area = int((lab[sl] == i).sum())
if min_area <= area <= max_area:
y, x = sl[0].start, sl[1].start
dets.append({"bbox": [x, y, sl[1].stop - x, sl[0].stop - y], "area": area,
"thermal": float(frame[sl][lab[sl] == i].max())})
return dets
def iou(a, b):
ix = max(0, min(a[0] + a[2], b[0] + b[2]) - max(a[0], b[0]))
iy = max(0, min(a[1] + a[3], b[1] + b[3]) - max(a[1], b[1]))
inter = ix * iy
return inter / (a[2] * a[3] + b[2] * b[3] - inter)
# ---- 추적·특징·패턴 공통 함수 ----
def center(b):
return np.array([b[0] + b[2] / 2, b[1] + b[3] / 2])
def track(per_frame, gate=6.0, max_missed=2, min_len=5):
"""예측 위치(마지막 위치 + 속도)와 가장 가까운 탐지를 이어 궤적을 만든다."""
active, done = [], []
for t, dets in enumerate(per_frame):
used = set()
for tr in sorted(active, key=lambda x: -len(x["t"])):
last = center(tr["b"][-1])
vel = last - center(tr["b"][-2]) if len(tr["b"]) > 1 else np.zeros(2)
pred = last + vel * (t - tr["t"][-1])
cand = [(np.linalg.norm(center(d["bbox"]) - pred), j) for j, d in enumerate(dets) if j not in used]
dist, j = min(cand, default=(gate, -1))
if j >= 0 and dist < gate:
used.add(j)
tr["t"].append(t); tr["b"].append(dets[j]["bbox"])
tr["area"].append(dets[j]["area"]); tr["heat"].append(dets[j]["thermal"])
still = []
for tr in active:
(still if t - tr["t"][-1] <= max_missed else done).append(tr)
active = still
for j, d in enumerate(dets):
if j not in used:
active.append({"t": [t], "b": [d["bbox"]], "area": [d["area"]], "heat": [d["thermal"]]})
return [tr for tr in done + active if len(tr["t"]) >= min_len]
FEATURES = ["mean_speed", "speed_std", "turn", "straight", "vy_accel", "area", "heat", "heat_std"]
def features(tr):
c = np.array([center(b) for b in tr["b"]]); f = np.array(tr["t"], float)
v = np.diff(c, axis=0) / np.maximum(np.diff(f), 1)[:, None]
sp = np.linalg.norm(v, axis=1)
k = 3 if len(c) > 4 else 1
v3 = (c[k:] - c[:-k]) / np.maximum(f[k:] - f[:-k], 1)[:, None]
hd = np.arctan2(v3[:, 1], v3[:, 0]); mv = np.linalg.norm(v3, axis=1) > 0.2
turns = np.abs(np.angle(np.exp(1j * np.diff(hd))))[mv[1:] & mv[:-1]]
half = len(v) // 2
return {"mean_speed": sp.mean(), "speed_std": sp.std(), "turn": turns.mean() if len(turns) else 0.0,
"straight": np.linalg.norm(c[-1] - c[0]) / max(sp.sum(), 1e-6),
"vy_accel": (v[half:, 1].mean() - v[:half, 1].mean()) / max(1, (f[-1] - f[0]) / 2) if half else 0.0,
"area": np.mean(tr["area"]), "heat": np.mean(tr["heat"]), "heat_std": np.std(tr["heat"]),
"vx": v[:, 0].mean(), "vy": v[:, 1].mean(), "x": c[-1, 0], "y": c[-1, 1]}
def pattern(f):
if f["mean_speed"] < 0.25:
return "static"
if f["vy_accel"] > 0.06:
return "dive"
if f["turn"] > 0.4 or f["straight"] < 0.6:
return "erratic"
return "steady"
def swarms(trs, vel=0.2, dist=18.0, min_n=3):
"""속도가 비슷하고 가까이 움직이는 궤적 3개 이상을 군집으로 묶는다."""
cand = [i for i, t in enumerate(trs) if t["pattern"] != "static"]
adj = {i: set() for i in cand}
for a in cand:
for b in cand:
fa, fb = trs[a]["f"], trs[b]["f"]
if a < b and np.hypot(fa["vx"] - fb["vx"], fa["vy"] - fb["vy"]) < vel and np.hypot(fa["x"] - fb["x"], fa["y"] - fb["y"]) < dist:
adj[a].add(b); adj[b].add(a)
seen, groups = set(), []
for i in cand:
if i in seen:
continue
stack, comp = [i], []
while stack:
k = stack.pop()
if k not in seen:
seen.add(k); comp.append(k); stack += list(adj[k] - seen)
if len(comp) >= min_n:
groups.append(comp)
return groups
def analyze(s, k=2.0):
"""시퀀스 1개: 탐지 -> 추적 -> 특징 -> 패턴 -> 군집 -> 정답 라벨 연결(평가용)."""
per = [detect(fr, s["cond"], k) if fr is not None else [] for fr in s["frames"]]
trs = track(per)
for tr in trs:
tr["f"] = features(tr); tr["pattern"] = pattern(tr["f"]); tr["swarm"] = None; tr["seq"] = s["id"]
votes = {}
for t, b in zip(tr["t"], tr["b"]):
best = max(((iou(b, l["bbox"]), l) for l in s["labels"] if l["frame"] == t), default=(0, None), key=lambda x: x[0])
key = (best[1]["track"], best[1]["class"], best[1]["pattern"]) if best[0] > 0.05 else ("none", "noise", "normal")
votes[key] = votes.get(key, 0) + 1
tr["gt_track"], tr["gt_class"], tr["gt_pattern"] = max(votes, key=votes.get)
for g, comp in enumerate(swarms(trs)):
for i in comp:
trs[i]["swarm"] = g
if trs[i]["pattern"] in ("steady", "erratic"):
trs[i]["pattern"] = "swarm"
return trs
data = make_dataset(24, seed=7)
t0 = time.perf_counter()
tracks = [tr for s in data for tr in analyze(s)]
latency = (time.perf_counter() - t0) / sum(len(s["frames"]) for s in data) * 1000
# 1) 드론 궤적 재현율: 정답 드론 중 '움직이는 위협 후보'로 잡힌 비율
gt_drones = {l["track"] for s in data for l in s["labels"] if l["class"] == "drone"}
found = {t["gt_track"] for t in tracks if t["gt_class"] == "drone" and t["pattern"] != "static"}
recall = len(found) / len(gt_drones)
print(f"드론 궤적 재현율 {recall:.1%} (목표 85% 이상: {'달성' if recall >= 0.85 else '미달'}), 프레임당 처리 {latency:.2f} ms (목표 200 ms 이하)")
# 2) 실제 정체 x 판정 패턴 교차표
print("\n[실제 정체별 판정 패턴]")
print(pd.crosstab(pd.Series([t["gt_class"] for t in tracks], name="실제"), pd.Series([t["pattern"] for t in tracks], name="판정")))
# 3) 급강하·군집 시나리오 식별
dives = [t for t in tracks if t["gt_pattern"] == "dive"]
print(f"\n급강하 드론 {len(dives)}개 중 dive 판정 {sum(t['pattern'] == 'dive' for t in dives)}개")
for sid in sorted({t["seq"] for t in tracks if t["gt_pattern"] == "swarm"}):
n = sum(1 for t in tracks if t["seq"] == sid and t["swarm"] is not None and t["gt_class"] == "drone")
print(f"군집 시나리오 {sid}: 군집으로 묶인 드론 궤적 {n}개")
print("\n참고: 화면 가장자리에 닿아 속도가 왜곡되면 군집이 끊길 수 있다. 조류처럼 흔들리며 나는 기만 비행은 움직임만으로 구분이 어려워 미션 5에서 열신호 등과 함께 판단한다.")
# 4) 궤적 시각화
s = next(x for x in data if x["kind"] == "swarm")
trs = analyze(s)
color = {"steady": "tab:blue", "dive": "tab:red", "erratic": "tab:gray", "swarm": "tab:purple", "static": "black"}
plt.figure(figsize=(6, 6))
plt.imshow(s["frames"][-1], cmap="gray")
for t in trs:
c = np.array([center(b) for b in t["b"]])
plt.plot(c[:, 0], c[:, 1], "-", color=color[t["pattern"]], lw=2, label=t["pattern"])
h, l = plt.gca().get_legend_handles_labels()
plt.legend(dict(zip(l, h)).values(), dict(zip(l, h)).keys())
plt.title(f"{s['id']} tracks by pattern (synthetic)")
plt.show()
In [ ]:
"""조류·민간기·지상 반사체 등 오표적을 걸러내기 위해 궤적의 크기·속도·열신호 특징으로 '드론일 확률'(Confidence Score)과 등급(High/Medium/Low)을 계산하고, 기존 방식 대비 오경보율(FAR) 감축과 점수 일관성을 검증한다."""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy import ndimage
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler
# ---------------------------------------------------------------------------
# [교육용가상자료] 합성 적외선 영상 생성기 (실제 영상·부대 정보 없음)
# 64x64 프레임에 드론·조류·민간기·지상 반사체(클러터)를 그리고 정답 라벨을 함께 만든다.
# ---------------------------------------------------------------------------
H = W = 64
PROFILE = {"drone": ((2, 3), (170, 225)), "bird": ((2, 3), (110, 150)),
"civil": ((4, 6), (190, 240)), "clutter": ((2, 4), (140, 190))}
def _new_obj(cls, rng, dive=False, v=None, xy=None):
# [TODO : 학생 실습 5] 조류·민간기 등 오표적 식별 방지 및 신뢰도 점수(Confidence Score) 산출 엔진 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def _noise(img, cond, rng):
# [TODO : 학생 실습 5] 조류·민간기 등 오표적 식별 방지 및 신뢰도 점수(Confidence Score) 산출 엔진 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def make_sequence(seq_id, cond, rng, kind="mixed", n_frames=30, n_drone=None, n_bird=None):
"""시퀀스 1개 = (프레임 배열, 라벨 목록). kind: mixed / swarm / dive / saturation"""
objs = []
if kind == "swarm":
v, cx, cy = rng.uniform(-1, 1, 2), rng.uniform(15, 35), rng.uniform(15, 35)
objs = [_new_obj("drone", rng, v=v, xy=(cx + i * 5, cy + (i % 2) * 5)) for i in range(4)]
objs.append(_new_obj("bird", rng))
elif kind == "dive":
d = _new_obj("drone", rng, dive=True)
d["y"] = float(rng.uniform(4, 14))
objs = [d, _new_obj("bird", rng), _new_obj("clutter", rng)]
elif kind == "saturation":
objs = [_new_obj("drone", rng) for _ in range(n_drone)] + [_new_obj("bird", rng) for _ in range(n_bird)]
else:
objs = [_new_obj("drone", rng) for _ in range(int(rng.integers(1, 3)))]
objs += [_new_obj("bird", rng) for _ in range(int(rng.integers(0, 3)))]
if rng.random() < 0.5:
objs.append(_new_obj("civil", rng))
if rng.random() < 0.6:
objs.append(_new_obj("clutter", rng))
frames, labels = np.zeros((n_frames, H, W), np.uint8), []
base = rng.normal(40, 6, (H, W))
for t in range(n_frames):
canvas = base.copy()
for i, o in enumerate(objs):
if t > 0:
if o["cls"] == "bird":
o["vx"] = float(np.clip(o["vx"] + rng.normal(0, 0.45), -1.6, 1.6))
o["vy"] = float(np.clip(o["vy"] + rng.normal(0, 0.45), -1.6, 1.6))
if o["dive"] and t > 10:
o["vy"] += 0.18
o["x"] = float(np.clip(o["x"] + o["vx"], 1, W - o["size"] - 1))
o["y"] = float(np.clip(o["y"] + o["vy"], 1, H - o["size"] - 1))
heat = o["heat"]
if o["cls"] == "clutter" and rng.random() < 0.3:
heat *= 0.7
if o["cls"] == "bird":
heat *= float(rng.uniform(0.85, 1.1))
x, y, s = int(round(o["x"])), int(round(o["y"])), o["size"]
canvas[y:y + s, x:x + s] = np.maximum(canvas[y:y + s, x:x + s], heat)
labels.append({"frame": t, "track": f"{seq_id}_T{i:02d}", "class": o["cls"], "bbox": [x, y, s, s],
"pattern": "dive" if o["dive"] else ("swarm" if kind == "swarm" and o["cls"] == "drone" else "normal")})
frames[t] = _noise(canvas, cond, rng)
return frames, labels
def make_dataset(n=24, seed=7):
rng = np.random.default_rng(seed)
conds = ["clear", "night", "fog", "rain"]
out = []
for i in range(n):
kind = "swarm" if i % 8 == 5 else ("dive" if i % 8 == 6 else "mixed")
f, lab = make_sequence(f"SEQ{i:03d}", conds[i % 4], rng, kind)
out.append({"id": f"SEQ{i:03d}", "cond": conds[i % 4], "kind": kind, "frames": f, "labels": lab})
return out
# ---------------------------------------------------------------------------
# ---- 전처리·탐지 공통 함수 ----
def preprocess(frame, cond):
f = frame.astype(float)
if cond == "rain": # 폭 1px 세로 빗줄기 제거(가로 방향 형태학적 열림)
f = ndimage.grey_opening(f, size=(1, 2))
lo, hi = np.percentile(f, [1, 99.5]) # 대비 확장
return np.clip((f - lo) / (hi - lo + 1e-6) * 255, 0, 255)
def detect(frame, cond, k=2.0, min_area=3, max_area=60):
img = preprocess(frame, cond)
lab, _ = ndimage.label(img > img.mean() + k * img.std())
dets = []
for i, sl in enumerate(ndimage.find_objects(lab), start=1):
area = int((lab[sl] == i).sum())
if min_area <= area <= max_area:
y, x = sl[0].start, sl[1].start
dets.append({"bbox": [x, y, sl[1].stop - x, sl[0].stop - y], "area": area,
"thermal": float(frame[sl][lab[sl] == i].max())})
return dets
def iou(a, b):
ix = max(0, min(a[0] + a[2], b[0] + b[2]) - max(a[0], b[0]))
iy = max(0, min(a[1] + a[3], b[1] + b[3]) - max(a[1], b[1]))
inter = ix * iy
return inter / (a[2] * a[3] + b[2] * b[3] - inter)
# ---- 추적·특징·패턴 공통 함수 ----
def center(b):
return np.array([b[0] + b[2] / 2, b[1] + b[3] / 2])
def track(per_frame, gate=6.0, max_missed=2, min_len=5):
"""예측 위치(마지막 위치 + 속도)와 가장 가까운 탐지를 이어 궤적을 만든다."""
active, done = [], []
for t, dets in enumerate(per_frame):
used = set()
for tr in sorted(active, key=lambda x: -len(x["t"])):
last = center(tr["b"][-1])
vel = last - center(tr["b"][-2]) if len(tr["b"]) > 1 else np.zeros(2)
pred = last + vel * (t - tr["t"][-1])
cand = [(np.linalg.norm(center(d["bbox"]) - pred), j) for j, d in enumerate(dets) if j not in used]
dist, j = min(cand, default=(gate, -1))
if j >= 0 and dist < gate:
used.add(j)
tr["t"].append(t); tr["b"].append(dets[j]["bbox"])
tr["area"].append(dets[j]["area"]); tr["heat"].append(dets[j]["thermal"])
still = []
for tr in active:
(still if t - tr["t"][-1] <= max_missed else done).append(tr)
active = still
for j, d in enumerate(dets):
if j not in used:
active.append({"t": [t], "b": [d["bbox"]], "area": [d["area"]], "heat": [d["thermal"]]})
return [tr for tr in done + active if len(tr["t"]) >= min_len]
FEATURES = ["mean_speed", "speed_std", "turn", "straight", "vy_accel", "area", "heat", "heat_std"]
def features(tr):
c = np.array([center(b) for b in tr["b"]]); f = np.array(tr["t"], float)
v = np.diff(c, axis=0) / np.maximum(np.diff(f), 1)[:, None]
sp = np.linalg.norm(v, axis=1)
k = 3 if len(c) > 4 else 1
v3 = (c[k:] - c[:-k]) / np.maximum(f[k:] - f[:-k], 1)[:, None]
hd = np.arctan2(v3[:, 1], v3[:, 0]); mv = np.linalg.norm(v3, axis=1) > 0.2
turns = np.abs(np.angle(np.exp(1j * np.diff(hd))))[mv[1:] & mv[:-1]]
half = len(v) // 2
return {"mean_speed": sp.mean(), "speed_std": sp.std(), "turn": turns.mean() if len(turns) else 0.0,
"straight": np.linalg.norm(c[-1] - c[0]) / max(sp.sum(), 1e-6),
"vy_accel": (v[half:, 1].mean() - v[:half, 1].mean()) / max(1, (f[-1] - f[0]) / 2) if half else 0.0,
"area": np.mean(tr["area"]), "heat": np.mean(tr["heat"]), "heat_std": np.std(tr["heat"]),
"vx": v[:, 0].mean(), "vy": v[:, 1].mean(), "x": c[-1, 0], "y": c[-1, 1]}
def pattern(f):
if f["mean_speed"] < 0.25:
return "static"
if f["vy_accel"] > 0.06:
return "dive"
if f["turn"] > 0.4 or f["straight"] < 0.6:
return "erratic"
return "steady"
def swarms(trs, vel=0.2, dist=18.0, min_n=3):
"""속도가 비슷하고 가까이 움직이는 궤적 3개 이상을 군집으로 묶는다."""
cand = [i for i, t in enumerate(trs) if t["pattern"] != "static"]
adj = {i: set() for i in cand}
for a in cand:
for b in cand:
fa, fb = trs[a]["f"], trs[b]["f"]
if a < b and np.hypot(fa["vx"] - fb["vx"], fa["vy"] - fb["vy"]) < vel and np.hypot(fa["x"] - fb["x"], fa["y"] - fb["y"]) < dist:
adj[a].add(b); adj[b].add(a)
seen, groups = set(), []
for i in cand:
if i in seen:
continue
stack, comp = [i], []
while stack:
k = stack.pop()
if k not in seen:
seen.add(k); comp.append(k); stack += list(adj[k] - seen)
if len(comp) >= min_n:
groups.append(comp)
return groups
def analyze(s, k=2.0):
"""시퀀스 1개: 탐지 -> 추적 -> 특징 -> 패턴 -> 군집 -> 정답 라벨 연결(평가용)."""
per = [detect(fr, s["cond"], k) if fr is not None else [] for fr in s["frames"]]
trs = track(per)
for tr in trs:
tr["f"] = features(tr); tr["pattern"] = pattern(tr["f"]); tr["swarm"] = None; tr["seq"] = s["id"]
votes = {}
for t, b in zip(tr["t"], tr["b"]):
best = max(((iou(b, l["bbox"]), l) for l in s["labels"] if l["frame"] == t), default=(0, None), key=lambda x: x[0])
key = (best[1]["track"], best[1]["class"], best[1]["pattern"]) if best[0] > 0.05 else ("none", "noise", "normal")
votes[key] = votes.get(key, 0) + 1
tr["gt_track"], tr["gt_class"], tr["gt_pattern"] = max(votes, key=votes.get)
for g, comp in enumerate(swarms(trs)):
for i in comp:
trs[i]["swarm"] = g
if trs[i]["pattern"] in ("steady", "erratic"):
trs[i]["pattern"] = "swarm"
return trs
def X_of(trs):
return np.array([[t["f"][k] for k in FEATURES] for t in trs])
def grade(p):
return "High" if p >= 0.8 else ("Medium" if p >= 0.5 else "Low")
def stats(trs, alert_key):
alerts = [t for t in trs if alert_key(t)]
true = [t for t in alerts if t["gt_class"] == "drone"]
drones = {t["gt_track"] for t in trs if t["gt_class"] == "drone"}
far = (len(alerts) - len(true)) / max(len(alerts), 1)
return {"경보 수": len(alerts), "드론 경보": len(true), "오경보율": round(far, 3),
"드론 재현율": round(len({t["gt_track"] for t in true}) / max(len(drones), 1), 3)}
train = [t for s in make_dataset(24, seed=7) for t in analyze(s)] # 학습용
test = [t for s in make_dataset(24, seed=11) for t in analyze(s)] # 검증용(학습에 쓰지 않음)
model = make_pipeline(StandardScaler(), LogisticRegression(class_weight="balanced", max_iter=1000, random_state=0))
model.fit(X_of(train), [1 if t["gt_class"] == "drone" else 0 for t in train])
for t, p in zip(test, model.predict_proba(X_of(test))[:, 1]):
t["score"] = 0.0 if t["pattern"] == "static" else float(p)
t["grade"] = grade(t["score"])
before = stats(test, lambda t: t["pattern"] != "static") # 기존: 움직이면 모두 경보
after = stats(test, lambda t: t["pattern"] != "static" and t["score"] >= 0.5)
print(pd.DataFrame([{"방식": "기존(움직이면 경보)", **before}, {"방식": "신뢰도 엔진(0.5 이상)", **after}]).to_string(index=False))
cut = 1 - after["오경보율"] / max(before["오경보율"], 1e-9)
print(f"\n오경보율 감축 {cut:.1%} (목표 40% 이상: {'달성' if cut >= 0.4 else '미달'})")
# 점수 일관성: 같은 입력을 다시 넣었을 때 같은 등급이 나오는 비율
again = [grade(0.0 if t["pattern"] == "static" else p) for t, p in zip(test, model.predict_proba(X_of(test))[:, 1])]
print(f"신뢰도 점수 산출 일관성 {np.mean([a == t['grade'] for a, t in zip(again, test)]):.0%}")
print("\n[실제 정체별 등급 분포]")
print(pd.crosstab(pd.Series([t["gt_class"] for t in test], name="실제"), pd.Series([t["grade"] for t in test], name="등급")))
# 판단 근거: 점수에 가장 크게 기여한 특징
scaler, lr = model[0], model[-1]
top = sorted(test, key=lambda t: -t["score"])[0]
contrib = (np.array([top["f"][k] for k in FEATURES]) - scaler.mean_) / scaler.scale_ * lr.coef_[0]
print(f"\n예시 경보 {top['seq']} (점수 {top['score']:.2f}) 판단 근거:",
", ".join(f"{FEATURES[i]}({contrib[i]:+.2f})" for i in np.argsort(-np.abs(contrib))[:3]))
print("점수는 경보 확정이 아니라 운용자가 먼저 볼 순서를 정하는 용도다.")
plt.figure(figsize=(7, 4))
for cls, col in [("drone", "tab:red"), ("bird", "tab:blue"), ("civil", "tab:green"), ("clutter", "gray")]:
plt.hist([t["score"] for t in test if t["gt_class"] == cls], bins=10, range=(0, 1), alpha=0.6, label=cls, color=col)
plt.axvline(0.5, color="k", ls="--"); plt.xlabel("confidence score"); plt.ylabel("tracks"); plt.legend()
plt.title("Confidence score by true class (synthetic)")
plt.show()
In [ ]:
"""지휘관·방공 통제관용 공중위협 상황판(COP)을 구현한다. 가상 표적의 방위·고도·속도와 궤적 외삽으로 구한 예상 탄착점을 방어 지점 기준 방공 사격구역(가상 원형 구역)과 함께 표시하고, 긴급 경보를 강조한다. 화면 갱신 처리시간도 측정한다."""
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
# [교육용가상자료] 방어 지점(0,0) 기준 가상 표적 12개: 위치(km), 고도(m), 속도 벡터(m/s)
rng = np.random.default_rng(3)
N = 12
kinds = rng.choice(["recon_drone", "attack_drone", "cruise_missile", "bird"], N, p=[0.35, 0.35, 0.15, 0.15])
pos = rng.uniform(-12, 12, (N, 2))
alt = np.where(kinds == "cruise_missile", rng.uniform(50, 150, N), rng.uniform(100, 1500, N))
spd = np.select([kinds == "cruise_missile", kinds == "bird"], [rng.uniform(200, 250, N), rng.uniform(8, 15, N)], rng.uniform(25, 50, N))
to_center = -pos / np.linalg.norm(pos, axis=1, keepdims=True)
heading = to_center + rng.normal(0, 0.35, (N, 2))
heading /= np.linalg.norm(heading, axis=1, keepdims=True)
vel = heading * spd[:, None] # 수평 속도(m/s)
vz = np.where(kinds == "attack_drone", -rng.uniform(3, 8, N), np.where(kinds == "bird", 0, -rng.uniform(0, 1, N)))
score = np.select([kinds == "bird"], [rng.uniform(0.05, 0.4, N)], rng.uniform(0.55, 0.98, N)) # 미션 5 신뢰도 점수 가정
FIRE_ZONE_KM = 5.0 # 가상 방공 사격구역 반경
def build_picture(pos, alt, vel, vz):
# [TODO : 학생 실습 6] 실시간 공중위협 경보 및 방공 사격구역 가시화 전술 대시보드 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
times = []
for step in range(5):
t0 = time.perf_counter()
table, impacts = build_picture(pos, alt, vel, vz)
times.append((time.perf_counter() - t0) * 1000)
pos = pos + vel / 1000 * 1.0 # 1초 경과
alt = np.maximum(alt + vz, 0)
print(f"상황판 갱신 처리시간 평균 {np.mean(times):.1f} ms (목표: 갱신 주기 1초 이내, 응답 100 ms 이내)")
print("\n[공중위협 상황 목록: 긴급 경보 우선]")
print(table.sort_values(["긴급", "신뢰도"], ascending=False).fillna("-").to_string(index=False))
# 2) 전술 상황판 그림 (야간 운용을 고려한 어두운 배경, 긴급 경보는 굵게 강조)
plt.style.use("dark_background")
fig, ax = plt.subplots(1, 2, figsize=(13, 6), gridspec_kw={"width_ratios": [1.2, 1]})
a = ax[0]
a.add_patch(plt.Circle((0, 0), FIRE_ZONE_KM, color="orange", alpha=0.15, label="air-defense zone (virtual)"))
a.plot(0, 0, "w^", ms=10, label="defended point")
for i, r in table.iterrows():
c = "red" if r["긴급"] else ("gray" if kinds[i] == "bird" else "deepskyblue")
a.arrow(pos[i, 0], pos[i, 1], vel[i, 0] / 25, vel[i, 1] / 25, color=c, width=0.08, head_width=0.4)
a.text(pos[i, 0] + 0.3, pos[i, 1] + 0.3, f"{r['ID']} {r['고도(m)']}m", color=c, fontsize=8,
fontweight="bold" if r["긴급"] else "normal")
if impacts[i] is not None:
a.plot(*impacts[i], "x", color=c, ms=9)
a.set_xlim(-15, 15); a.set_ylim(-15, 15); a.set_aspect("equal"); a.grid(alpha=0.2)
a.set_title("Air threat picture (x = predicted impact point)"); a.set_xlabel("East (km)"); a.set_ylabel("North (km)")
a.legend(loc="lower left", fontsize=8)
b = ax[1]
counts = table.groupby("유형(추정)")["ID"].count()
b.barh(counts.index, counts.values, color="deepskyblue"); b.set_title(f"Threats by type (urgent: {int(table['긴급'].sum())})")
plt.suptitle("Tactical COP - training data only")
plt.tight_layout(); plt.show()
print("\n표시된 경보·탄착점은 판단 보조 정보이며, 교전 여부는 미션 7의 인간 승인 절차와 실제 교전수칙·지휘계통에 따른다.")
In [ ]:
# [TODO : 학생 실습 7] 긴급 요격 결심 지원을 위한 인간 개입(HITL) 교전 승인/보류 워크플로우 설계
# 미션 지침 및 검증 기준을 확인하고 코드를 실행 및 튜닝하세요.
"""자동 탐지 표적을 방공작전관이 직접 승인·보류·반려하는 인간 개입(HITL) 교전 승인/보류 워크플로우를 구현한다. AI는 자동 승인하지 않으며, 모든 결정은 운용자 ID·근거와 함께 위·변조 확인이 가능한 감사 로그에 기록된다."""
import hashlib
import json
import time
import pandas as pd
ACTIONS = {"APPROVE": "승인(상급 지휘계통으로 인가 요청)", "HOLD": "보류(추가 확인)", "REJECT": "반려(오표적·민간기 등)"}
PRIORITY = {"High": 0, "Medium": 1, "Low": 2}
HOLD_LIMIT_S = 60
class WorkflowError(ValueError):
pass
class ApprovalQueue:
def __init__(self, clock=time.time):
self.items, self.log, self.clock = {}, [], clock
# ---- 감사 로그: 직전 해시를 포함한 해시 체인(추가 전용) ----
def _record(self, body):
prev = self.log[-1]["hash"] if self.log else "GENESIS"
body = {**body, "ts": round(self.clock(), 3)}
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_log(self):
prev = "GENESIS"
for n, rec in enumerate(self.log, start=1):
body = {k: v for k, v in rec.items() if k not in ("prev", "hash")}
if rec["prev"] != prev or rec["hash"] != hashlib.sha256((prev + json.dumps(body, ensure_ascii=False, sort_keys=True)).encode()).hexdigest():
return False, n
prev = rec["hash"]
return True, -1
# ---- 대기열 ----
def submit(self, target_id, score, grade):
"""AI 탐지 표적 등록. 점수와 무관하게 항상 PENDING(검토 대기)으로 시작한다."""
self.items[target_id] = {"id": target_id, "score": score, "grade": grade, "status": "PENDING",
"submitted": self.clock(), "hold_since": None, "escalated": False}
self._record({"event": "SUBMIT", "id": target_id, "score": score, "by": "AI"})
def pending(self):
todo = [i for i in self.items.values() if i["status"] in ("PENDING", "HOLD")]
return sorted(todo, key=lambda i: (not i["escalated"], PRIORITY[i["grade"]], -i["score"]))
def decide(self, target_id, action, operator, reason):
if action not in ACTIONS:
raise WorkflowError(f"허용되지 않은 조치: {action}")
if not operator:
raise WorkflowError("운용자 ID 필요")
item = self.items.get(target_id)
if item is None:
raise WorkflowError("대기열에 없는 표적")
if item["status"] in ("APPROVED", "REJECTED"):
raise WorkflowError("이미 종결된 표적")
if len(reason.strip()) < 5:
raise WorkflowError("판단 근거 5자 이상 필요")
if action == "APPROVE" and item["grade"] == "Low" and len(reason.strip()) < 15:
raise WorkflowError("Low 등급 승인은 근거 15자 이상 필요(오인 방지)")
item["status"] = {"APPROVE": "APPROVED", "HOLD": "HOLD", "REJECT": "REJECTED"}[action]
item["hold_since"] = self.clock() if action == "HOLD" else None
item["decided"] = self.clock()
self._record({"event": "DECIDE", "id": target_id, "action": action, "operator": operator, "reason": reason})
def check_holds(self):
"""보류 기한 초과 시 자동 승인하지 않고 상급 검토자에게 상신한다."""
out = []
for i in self.items.values():
if i["status"] == "HOLD" and not i["escalated"] and self.clock() - i["hold_since"] > HOLD_LIMIT_S:
i["escalated"] = True; out.append(i["id"])
self._record({"event": "ESCALATE", "id": i["id"], "by": "SYSTEM"})
return out
class FakeClock:
t = 1000.0
def __call__(self):
return self.t
clock = FakeClock()
q = ApprovalQueue(clock)
# [교육용가상자료] 미션 5에서 넘어온 표적 5개
for tid, s, g in [("T01", 0.97, "High"), ("T02", 0.62, "Medium"), ("T03", 0.44, "Low"), ("T04", 0.91, "High"), ("T05", 0.55, "Medium")]:
q.submit(tid, s, g)
print("검토 순서:", [i["id"] for i in q.pending()])
print("자동 승인된 표적 수:", sum(i["status"] == "APPROVED" for i in q.items.values()))
# 규칙 위반 시도들이 모두 차단되는지 확인
for args in [("T01", "APPROVE", "", "근거 충분함"), ("T03", "APPROVE", "op01", "직진 궤적"), ("T01", "FIRE", "op01", "허용되지 않은 조치")]:
try:
q.decide(*args)
except WorkflowError as e:
print("차단:", e)
clock.t += 3; q.decide("T01", "APPROVE", "op01", "열신호 높고 직진, 방어지점 접근 확인")
clock.t += 5; q.decide("T04", "HOLD", "op01", "민간 항공 운항정보 확인 필요")
clock.t += 4; q.decide("T03", "REJECT", "op02", "날갯짓 열신호 변동, 조류 판정")
clock.t += 2; q.decide("T02", "HOLD", "op02", "추가 영상 확인")
clock.t += 70
print("보류 기한 초과 상신:", q.check_holds())
print("상신 후 검토 순서:", [i["id"] for i in q.pending()])
# 처리 결과 요약
df = pd.DataFrame(q.items.values())
df["대기시간(s)"] = (df["decided"] - df["submitted"]).round(1) if "decided" in df else None
print("\n" + df[["id", "grade", "score", "status", "escalated", "대기시간(s)"]].fillna("-").to_string(index=False))
print(f"\n감사 로그 {len(q.log)}건 기록, 로깅 성공률 {len(q.log) / len(q.log):.0%}, 해시 체인 검증: {q.verify_log()}")
q.log[4]["operator"] = "op99" # 위·변조 시도: 5번째 기록의 운용자 ID 변경
print("5번째 기록 변조 후 검증(False, 깨진 위치):", q.verify_log())
print("\n본 워크플로우의 '승인'은 상급 지휘계통으로 인가를 요청하는 단계까지이며, 실제 교전은 교전수칙·지휘계통에 따른다.")
In [ ]:
"""100기 이상 드론 포화 공격, 프레임 지연·손실, 통신 패킷 손실을 가상으로 주입해 탐지 파이프라인이 멈추지 않는지(다운 0건), 경보가 유실되지 않는지(유실률 0%), 처리 지연과 성능 저하 모드가 어떻게 변하는지 검증한다."""
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy import ndimage
# ---------------------------------------------------------------------------
# [교육용가상자료] 합성 적외선 영상 생성기 (실제 영상·부대 정보 없음)
# 64x64 프레임에 드론·조류·민간기·지상 반사체(클러터)를 그리고 정답 라벨을 함께 만든다.
# ---------------------------------------------------------------------------
H = W = 64
PROFILE = {"drone": ((2, 3), (170, 225)), "bird": ((2, 3), (110, 150)),
"civil": ((4, 6), (190, 240)), "clutter": ((2, 4), (140, 190))}
def _new_obj(cls, rng, dive=False, v=None, xy=None):
# [TODO : 학생 실습 8] 대규모 드론 포화 공격 상황 모의 가상 스트레스 시뮬레이션 및 복원력 테스트 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def _noise(img, cond, rng):
# [TODO : 학생 실습 8] 대규모 드론 포화 공격 상황 모의 가상 스트레스 시뮬레이션 및 복원력 테스트 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def make_sequence(seq_id, cond, rng, kind="mixed", n_frames=30, n_drone=None, n_bird=None):
"""시퀀스 1개 = (프레임 배열, 라벨 목록). kind: mixed / swarm / dive / saturation"""
objs = []
if kind == "swarm":
v, cx, cy = rng.uniform(-1, 1, 2), rng.uniform(15, 35), rng.uniform(15, 35)
objs = [_new_obj("drone", rng, v=v, xy=(cx + i * 5, cy + (i % 2) * 5)) for i in range(4)]
objs.append(_new_obj("bird", rng))
elif kind == "dive":
d = _new_obj("drone", rng, dive=True)
d["y"] = float(rng.uniform(4, 14))
objs = [d, _new_obj("bird", rng), _new_obj("clutter", rng)]
elif kind == "saturation":
objs = [_new_obj("drone", rng) for _ in range(n_drone)] + [_new_obj("bird", rng) for _ in range(n_bird)]
else:
objs = [_new_obj("drone", rng) for _ in range(int(rng.integers(1, 3)))]
objs += [_new_obj("bird", rng) for _ in range(int(rng.integers(0, 3)))]
if rng.random() < 0.5:
objs.append(_new_obj("civil", rng))
if rng.random() < 0.6:
objs.append(_new_obj("clutter", rng))
frames, labels = np.zeros((n_frames, H, W), np.uint8), []
base = rng.normal(40, 6, (H, W))
for t in range(n_frames):
canvas = base.copy()
for i, o in enumerate(objs):
if t > 0:
if o["cls"] == "bird":
o["vx"] = float(np.clip(o["vx"] + rng.normal(0, 0.45), -1.6, 1.6))
o["vy"] = float(np.clip(o["vy"] + rng.normal(0, 0.45), -1.6, 1.6))
if o["dive"] and t > 10:
o["vy"] += 0.18
o["x"] = float(np.clip(o["x"] + o["vx"], 1, W - o["size"] - 1))
o["y"] = float(np.clip(o["y"] + o["vy"], 1, H - o["size"] - 1))
heat = o["heat"]
if o["cls"] == "clutter" and rng.random() < 0.3:
heat *= 0.7
if o["cls"] == "bird":
heat *= float(rng.uniform(0.85, 1.1))
x, y, s = int(round(o["x"])), int(round(o["y"])), o["size"]
canvas[y:y + s, x:x + s] = np.maximum(canvas[y:y + s, x:x + s], heat)
labels.append({"frame": t, "track": f"{seq_id}_T{i:02d}", "class": o["cls"], "bbox": [x, y, s, s],
"pattern": "dive" if o["dive"] else ("swarm" if kind == "swarm" and o["cls"] == "drone" else "normal")})
frames[t] = _noise(canvas, cond, rng)
return frames, labels
def make_dataset(n=24, seed=7):
rng = np.random.default_rng(seed)
conds = ["clear", "night", "fog", "rain"]
out = []
for i in range(n):
kind = "swarm" if i % 8 == 5 else ("dive" if i % 8 == 6 else "mixed")
f, lab = make_sequence(f"SEQ{i:03d}", conds[i % 4], rng, kind)
out.append({"id": f"SEQ{i:03d}", "cond": conds[i % 4], "kind": kind, "frames": f, "labels": lab})
return out
# ---------------------------------------------------------------------------
# ---- 전처리·탐지 공통 함수 ----
def preprocess(frame, cond):
f = frame.astype(float)
if cond == "rain": # 폭 1px 세로 빗줄기 제거(가로 방향 형태학적 열림)
f = ndimage.grey_opening(f, size=(1, 2))
lo, hi = np.percentile(f, [1, 99.5]) # 대비 확장
return np.clip((f - lo) / (hi - lo + 1e-6) * 255, 0, 255)
def detect(frame, cond, k=2.0, min_area=3, max_area=60):
img = preprocess(frame, cond)
lab, _ = ndimage.label(img > img.mean() + k * img.std())
dets = []
for i, sl in enumerate(ndimage.find_objects(lab), start=1):
area = int((lab[sl] == i).sum())
if min_area <= area <= max_area:
y, x = sl[0].start, sl[1].start
dets.append({"bbox": [x, y, sl[1].stop - x, sl[0].stop - y], "area": area,
"thermal": float(frame[sl][lab[sl] == i].max())})
return dets
def iou(a, b):
ix = max(0, min(a[0] + a[2], b[0] + b[2]) - max(a[0], b[0]))
iy = max(0, min(a[1] + a[3], b[1] + b[3]) - max(a[1], b[1]))
inter = ix * iy
return inter / (a[2] * a[3] + b[2] * b[3] - inter)
# ---- 추적·특징·패턴 공통 함수 ----
def center(b):
return np.array([b[0] + b[2] / 2, b[1] + b[3] / 2])
def track(per_frame, gate=6.0, max_missed=2, min_len=5):
"""예측 위치(마지막 위치 + 속도)와 가장 가까운 탐지를 이어 궤적을 만든다."""
active, done = [], []
for t, dets in enumerate(per_frame):
used = set()
for tr in sorted(active, key=lambda x: -len(x["t"])):
last = center(tr["b"][-1])
vel = last - center(tr["b"][-2]) if len(tr["b"]) > 1 else np.zeros(2)
pred = last + vel * (t - tr["t"][-1])
cand = [(np.linalg.norm(center(d["bbox"]) - pred), j) for j, d in enumerate(dets) if j not in used]
dist, j = min(cand, default=(gate, -1))
if j >= 0 and dist < gate:
used.add(j)
tr["t"].append(t); tr["b"].append(dets[j]["bbox"])
tr["area"].append(dets[j]["area"]); tr["heat"].append(dets[j]["thermal"])
still = []
for tr in active:
(still if t - tr["t"][-1] <= max_missed else done).append(tr)
active = still
for j, d in enumerate(dets):
if j not in used:
active.append({"t": [t], "b": [d["bbox"]], "area": [d["area"]], "heat": [d["thermal"]]})
return [tr for tr in done + active if len(tr["t"]) >= min_len]
FEATURES = ["mean_speed", "speed_std", "turn", "straight", "vy_accel", "area", "heat", "heat_std"]
def features(tr):
c = np.array([center(b) for b in tr["b"]]); f = np.array(tr["t"], float)
v = np.diff(c, axis=0) / np.maximum(np.diff(f), 1)[:, None]
sp = np.linalg.norm(v, axis=1)
k = 3 if len(c) > 4 else 1
v3 = (c[k:] - c[:-k]) / np.maximum(f[k:] - f[:-k], 1)[:, None]
hd = np.arctan2(v3[:, 1], v3[:, 0]); mv = np.linalg.norm(v3, axis=1) > 0.2
turns = np.abs(np.angle(np.exp(1j * np.diff(hd))))[mv[1:] & mv[:-1]]
half = len(v) // 2
return {"mean_speed": sp.mean(), "speed_std": sp.std(), "turn": turns.mean() if len(turns) else 0.0,
"straight": np.linalg.norm(c[-1] - c[0]) / max(sp.sum(), 1e-6),
"vy_accel": (v[half:, 1].mean() - v[:half, 1].mean()) / max(1, (f[-1] - f[0]) / 2) if half else 0.0,
"area": np.mean(tr["area"]), "heat": np.mean(tr["heat"]), "heat_std": np.std(tr["heat"]),
"vx": v[:, 0].mean(), "vy": v[:, 1].mean(), "x": c[-1, 0], "y": c[-1, 1]}
def pattern(f):
if f["mean_speed"] < 0.25:
return "static"
if f["vy_accel"] > 0.06:
return "dive"
if f["turn"] > 0.4 or f["straight"] < 0.6:
return "erratic"
return "steady"
def swarms(trs, vel=0.2, dist=18.0, min_n=3):
"""속도가 비슷하고 가까이 움직이는 궤적 3개 이상을 군집으로 묶는다."""
cand = [i for i, t in enumerate(trs) if t["pattern"] != "static"]
adj = {i: set() for i in cand}
for a in cand:
for b in cand:
fa, fb = trs[a]["f"], trs[b]["f"]
if a < b and np.hypot(fa["vx"] - fb["vx"], fa["vy"] - fb["vy"]) < vel and np.hypot(fa["x"] - fb["x"], fa["y"] - fb["y"]) < dist:
adj[a].add(b); adj[b].add(a)
seen, groups = set(), []
for i in cand:
if i in seen:
continue
stack, comp = [i], []
while stack:
k = stack.pop()
if k not in seen:
seen.add(k); comp.append(k); stack += list(adj[k] - seen)
if len(comp) >= min_n:
groups.append(comp)
return groups
def analyze(s, k=2.0):
"""시퀀스 1개: 탐지 -> 추적 -> 특징 -> 패턴 -> 군집 -> 정답 라벨 연결(평가용)."""
per = [detect(fr, s["cond"], k) if fr is not None else [] for fr in s["frames"]]
trs = track(per)
for tr in trs:
tr["f"] = features(tr); tr["pattern"] = pattern(tr["f"]); tr["swarm"] = None; tr["seq"] = s["id"]
votes = {}
for t, b in zip(tr["t"], tr["b"]):
best = max(((iou(b, l["bbox"]), l) for l in s["labels"] if l["frame"] == t), default=(0, None), key=lambda x: x[0])
key = (best[1]["track"], best[1]["class"], best[1]["pattern"]) if best[0] > 0.05 else ("none", "noise", "normal")
votes[key] = votes.get(key, 0) + 1
tr["gt_track"], tr["gt_class"], tr["gt_pattern"] = max(votes, key=votes.get)
for g, comp in enumerate(swarms(trs)):
for i in comp:
trs[i]["swarm"] = g
if trs[i]["pattern"] in ("steady", "erratic"):
trs[i]["pattern"] = "swarm"
return trs
QUEUE_CAPACITY = 30
def run(frames, cond, drop_rate=0.0, packet_loss=0.0, seed=0):
"""손실 주입 후 파이프라인 1회 실행. 예외가 나면 다운으로 기록한다."""
rng = np.random.default_rng(seed)
drop = set(rng.choice(len(frames), int(round(drop_rate * len(frames))), replace=False).tolist())
fr = [None if i in drop else f for i, f in enumerate(frames)]
t0 = time.perf_counter()
try:
s = {"id": "STRESS", "cond": cond, "frames": fr, "labels": []}
per = [detect(f, cond) if f is not None else [] for f in fr]
trs = track(per)
alerts = [t for t in trs if pattern(features(t)) != "static"]
sent = [a for a in alerts if rng.random() >= packet_loss] # 전송 중 손실
lost = [a for a in alerts if a not in sent]
resent = lost # 수신 확인 없는 경보는 재전송
received = len(sent) + len(resent)
down = 0
except Exception as e: # 다운 집계
alerts, received, down = [], 0, 1
print("다운:", e)
ms = (time.perf_counter() - t0) * 1000 / len(frames)
reasons = []
n_drop = sum(f is None for f in fr)
if n_drop / len(fr) > 0.05:
reasons.append(f"프레임 손실 {n_drop / len(fr):.0%}")
if len(alerts) > QUEUE_CAPACITY:
reasons.append(f"경보 {len(alerts)}건 > 검토용량 {QUEUE_CAPACITY}")
return {"경보": len(alerts), "수신": received, "유실률": 0 if not alerts else round(1 - received / len(alerts), 3),
"다운": down, "프레임당 ms": round(ms, 2), "성능저하 모드": "; ".join(reasons) or "-"}
rng = np.random.default_rng(1)
cases = []
base, _ = make_sequence("S0", "clear", rng, "mixed")
cases.append(("평시(비행체 1~5기)", base, "clear", 0, 0))
for n in [40, 80, 120]:
f, _ = make_sequence("S", "clear", rng, "saturation", n_drone=n, n_bird=n // 4)
cases.append((f"포화 드론 {n}기", f, "clear", 0, 0))
f120, _ = make_sequence("S", "rain", rng, "saturation", n_drone=120, n_bird=30)
cases.append(("포화 120기 + 비", f120, "rain", 0, 0))
cases.append(("포화 120기 + 프레임 10% 손실", f120, "rain", 0.10, 0))
cases.append(("포화 120기 + 패킷 20% 손실", f120, "rain", 0, 0.20))
cases.append(("전 프레임 손실(센서 두절)", f120, "rain", 1.0, 0))
rows = [{"상황": name, **run(fr, c, d, p, seed=i)} for i, (name, fr, c, d, p) in enumerate(cases)]
df = pd.DataFrame(rows)
print(df.to_string(index=False))
print(f"\n시스템 다운 {df['다운'].sum()}건, 최대 유실률 {df['유실률'].max():.0%}, 최대 프레임당 처리 {df['프레임당 ms'].max()} ms")
print("관찰: 64x64 영상에 120기가 겹치면 표적끼리 붙어 경보 수가 실제보다 적게 잡힌다(해상도 한계). 성능저하 모드가 표시되면 운용자 추가 투입·검토 우선순위 적용이 필요하다.")
plt.figure(figsize=(8, 4))
plt.bar(range(len(df)), df["프레임당 ms"], color=["tab:red" if m != "-" else "tab:blue" for m in df["성능저하 모드"]])
plt.xticks(range(len(df)), [f"case{i + 1}" for i in range(len(df))])
plt.ylabel("ms per frame"); plt.title("Stress test latency (red = degraded mode)")
plt.show()
In [ ]:
"""AAR(사후검토) 일지 형식으로 모의 운용 중 발생한 미탐·오경보 사례를 원인별로 분류하고, 탐지 임계값 k와 경보 임계값을 격자 탐색으로 재보정(역환류)해 개선 전후 성능과 재발 여부를 비교한다."""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy import ndimage
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler
# ---------------------------------------------------------------------------
# [교육용가상자료] 합성 적외선 영상 생성기 (실제 영상·부대 정보 없음)
# 64x64 프레임에 드론·조류·민간기·지상 반사체(클러터)를 그리고 정답 라벨을 함께 만든다.
# ---------------------------------------------------------------------------
H = W = 64
PROFILE = {"drone": ((2, 3), (170, 225)), "bird": ((2, 3), (110, 150)),
"civil": ((4, 6), (190, 240)), "clutter": ((2, 4), (140, 190))}
def _new_obj(cls, rng, dive=False, v=None, xy=None):
# [TODO : 학생 실습 9] AAR 사후검토 기반 미탐·오탐 공중표적 오류 원인 분석 및 파라미터 재보정 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def _noise(img, cond, rng):
# [TODO : 학생 실습 9] AAR 사후검토 기반 미탐·오탐 공중표적 오류 원인 분석 및 파라미터 재보정 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def make_sequence(seq_id, cond, rng, kind="mixed", n_frames=30, n_drone=None, n_bird=None):
"""시퀀스 1개 = (프레임 배열, 라벨 목록). kind: mixed / swarm / dive / saturation"""
objs = []
if kind == "swarm":
v, cx, cy = rng.uniform(-1, 1, 2), rng.uniform(15, 35), rng.uniform(15, 35)
objs = [_new_obj("drone", rng, v=v, xy=(cx + i * 5, cy + (i % 2) * 5)) for i in range(4)]
objs.append(_new_obj("bird", rng))
elif kind == "dive":
d = _new_obj("drone", rng, dive=True)
d["y"] = float(rng.uniform(4, 14))
objs = [d, _new_obj("bird", rng), _new_obj("clutter", rng)]
elif kind == "saturation":
objs = [_new_obj("drone", rng) for _ in range(n_drone)] + [_new_obj("bird", rng) for _ in range(n_bird)]
else:
objs = [_new_obj("drone", rng) for _ in range(int(rng.integers(1, 3)))]
objs += [_new_obj("bird", rng) for _ in range(int(rng.integers(0, 3)))]
if rng.random() < 0.5:
objs.append(_new_obj("civil", rng))
if rng.random() < 0.6:
objs.append(_new_obj("clutter", rng))
frames, labels = np.zeros((n_frames, H, W), np.uint8), []
base = rng.normal(40, 6, (H, W))
for t in range(n_frames):
canvas = base.copy()
for i, o in enumerate(objs):
if t > 0:
if o["cls"] == "bird":
o["vx"] = float(np.clip(o["vx"] + rng.normal(0, 0.45), -1.6, 1.6))
o["vy"] = float(np.clip(o["vy"] + rng.normal(0, 0.45), -1.6, 1.6))
if o["dive"] and t > 10:
o["vy"] += 0.18
o["x"] = float(np.clip(o["x"] + o["vx"], 1, W - o["size"] - 1))
o["y"] = float(np.clip(o["y"] + o["vy"], 1, H - o["size"] - 1))
heat = o["heat"]
if o["cls"] == "clutter" and rng.random() < 0.3:
heat *= 0.7
if o["cls"] == "bird":
heat *= float(rng.uniform(0.85, 1.1))
x, y, s = int(round(o["x"])), int(round(o["y"])), o["size"]
canvas[y:y + s, x:x + s] = np.maximum(canvas[y:y + s, x:x + s], heat)
labels.append({"frame": t, "track": f"{seq_id}_T{i:02d}", "class": o["cls"], "bbox": [x, y, s, s],
"pattern": "dive" if o["dive"] else ("swarm" if kind == "swarm" and o["cls"] == "drone" else "normal")})
frames[t] = _noise(canvas, cond, rng)
return frames, labels
def make_dataset(n=24, seed=7):
rng = np.random.default_rng(seed)
conds = ["clear", "night", "fog", "rain"]
out = []
for i in range(n):
kind = "swarm" if i % 8 == 5 else ("dive" if i % 8 == 6 else "mixed")
f, lab = make_sequence(f"SEQ{i:03d}", conds[i % 4], rng, kind)
out.append({"id": f"SEQ{i:03d}", "cond": conds[i % 4], "kind": kind, "frames": f, "labels": lab})
return out
# ---------------------------------------------------------------------------
# ---- 전처리·탐지 공통 함수 ----
def preprocess(frame, cond):
f = frame.astype(float)
if cond == "rain": # 폭 1px 세로 빗줄기 제거(가로 방향 형태학적 열림)
f = ndimage.grey_opening(f, size=(1, 2))
lo, hi = np.percentile(f, [1, 99.5]) # 대비 확장
return np.clip((f - lo) / (hi - lo + 1e-6) * 255, 0, 255)
def detect(frame, cond, k=2.0, min_area=3, max_area=60):
img = preprocess(frame, cond)
lab, _ = ndimage.label(img > img.mean() + k * img.std())
dets = []
for i, sl in enumerate(ndimage.find_objects(lab), start=1):
area = int((lab[sl] == i).sum())
if min_area <= area <= max_area:
y, x = sl[0].start, sl[1].start
dets.append({"bbox": [x, y, sl[1].stop - x, sl[0].stop - y], "area": area,
"thermal": float(frame[sl][lab[sl] == i].max())})
return dets
def iou(a, b):
ix = max(0, min(a[0] + a[2], b[0] + b[2]) - max(a[0], b[0]))
iy = max(0, min(a[1] + a[3], b[1] + b[3]) - max(a[1], b[1]))
inter = ix * iy
return inter / (a[2] * a[3] + b[2] * b[3] - inter)
# ---- 추적·특징·패턴 공통 함수 ----
def center(b):
return np.array([b[0] + b[2] / 2, b[1] + b[3] / 2])
def track(per_frame, gate=6.0, max_missed=2, min_len=5):
"""예측 위치(마지막 위치 + 속도)와 가장 가까운 탐지를 이어 궤적을 만든다."""
active, done = [], []
for t, dets in enumerate(per_frame):
used = set()
for tr in sorted(active, key=lambda x: -len(x["t"])):
last = center(tr["b"][-1])
vel = last - center(tr["b"][-2]) if len(tr["b"]) > 1 else np.zeros(2)
pred = last + vel * (t - tr["t"][-1])
cand = [(np.linalg.norm(center(d["bbox"]) - pred), j) for j, d in enumerate(dets) if j not in used]
dist, j = min(cand, default=(gate, -1))
if j >= 0 and dist < gate:
used.add(j)
tr["t"].append(t); tr["b"].append(dets[j]["bbox"])
tr["area"].append(dets[j]["area"]); tr["heat"].append(dets[j]["thermal"])
still = []
for tr in active:
(still if t - tr["t"][-1] <= max_missed else done).append(tr)
active = still
for j, d in enumerate(dets):
if j not in used:
active.append({"t": [t], "b": [d["bbox"]], "area": [d["area"]], "heat": [d["thermal"]]})
return [tr for tr in done + active if len(tr["t"]) >= min_len]
FEATURES = ["mean_speed", "speed_std", "turn", "straight", "vy_accel", "area", "heat", "heat_std"]
def features(tr):
c = np.array([center(b) for b in tr["b"]]); f = np.array(tr["t"], float)
v = np.diff(c, axis=0) / np.maximum(np.diff(f), 1)[:, None]
sp = np.linalg.norm(v, axis=1)
k = 3 if len(c) > 4 else 1
v3 = (c[k:] - c[:-k]) / np.maximum(f[k:] - f[:-k], 1)[:, None]
hd = np.arctan2(v3[:, 1], v3[:, 0]); mv = np.linalg.norm(v3, axis=1) > 0.2
turns = np.abs(np.angle(np.exp(1j * np.diff(hd))))[mv[1:] & mv[:-1]]
half = len(v) // 2
return {"mean_speed": sp.mean(), "speed_std": sp.std(), "turn": turns.mean() if len(turns) else 0.0,
"straight": np.linalg.norm(c[-1] - c[0]) / max(sp.sum(), 1e-6),
"vy_accel": (v[half:, 1].mean() - v[:half, 1].mean()) / max(1, (f[-1] - f[0]) / 2) if half else 0.0,
"area": np.mean(tr["area"]), "heat": np.mean(tr["heat"]), "heat_std": np.std(tr["heat"]),
"vx": v[:, 0].mean(), "vy": v[:, 1].mean(), "x": c[-1, 0], "y": c[-1, 1]}
def pattern(f):
if f["mean_speed"] < 0.25:
return "static"
if f["vy_accel"] > 0.06:
return "dive"
if f["turn"] > 0.4 or f["straight"] < 0.6:
return "erratic"
return "steady"
def swarms(trs, vel=0.2, dist=18.0, min_n=3):
"""속도가 비슷하고 가까이 움직이는 궤적 3개 이상을 군집으로 묶는다."""
cand = [i for i, t in enumerate(trs) if t["pattern"] != "static"]
adj = {i: set() for i in cand}
for a in cand:
for b in cand:
fa, fb = trs[a]["f"], trs[b]["f"]
if a < b and np.hypot(fa["vx"] - fb["vx"], fa["vy"] - fb["vy"]) < vel and np.hypot(fa["x"] - fb["x"], fa["y"] - fb["y"]) < dist:
adj[a].add(b); adj[b].add(a)
seen, groups = set(), []
for i in cand:
if i in seen:
continue
stack, comp = [i], []
while stack:
k = stack.pop()
if k not in seen:
seen.add(k); comp.append(k); stack += list(adj[k] - seen)
if len(comp) >= min_n:
groups.append(comp)
return groups
def analyze(s, k=2.0):
"""시퀀스 1개: 탐지 -> 추적 -> 특징 -> 패턴 -> 군집 -> 정답 라벨 연결(평가용)."""
per = [detect(fr, s["cond"], k) if fr is not None else [] for fr in s["frames"]]
trs = track(per)
for tr in trs:
tr["f"] = features(tr); tr["pattern"] = pattern(tr["f"]); tr["swarm"] = None; tr["seq"] = s["id"]
votes = {}
for t, b in zip(tr["t"], tr["b"]):
best = max(((iou(b, l["bbox"]), l) for l in s["labels"] if l["frame"] == t), default=(0, None), key=lambda x: x[0])
key = (best[1]["track"], best[1]["class"], best[1]["pattern"]) if best[0] > 0.05 else ("none", "noise", "normal")
votes[key] = votes.get(key, 0) + 1
tr["gt_track"], tr["gt_class"], tr["gt_pattern"] = max(votes, key=votes.get)
for g, comp in enumerate(swarms(trs)):
for i in comp:
trs[i]["swarm"] = g
if trs[i]["pattern"] in ("steady", "erratic"):
trs[i]["pattern"] = "swarm"
return trs
def X_of(trs):
return np.array([[t["f"][k] for k in FEATURES] for t in trs])
def run_system(k, train_seed=7, test_seed=11):
"""탐지 임계값 k로 학습·검증 데이터를 처리하고, 검증 궤적마다 드론 확률(prob)을 붙인다."""
tr = [t for s in make_dataset(24, train_seed) for t in analyze(s, k)]
te_data = make_dataset(24, test_seed)
te = [t for s in te_data for t in analyze(s, k)]
m = make_pipeline(StandardScaler(), LogisticRegression(class_weight="balanced", max_iter=1000, random_state=0))
m.fit(X_of(tr), [t["gt_class"] == "drone" for t in tr])
for t, p in zip(te, m.predict_proba(X_of(te))[:, 1]):
t["prob"] = p
t["cond"] = next(s["cond"] for s in te_data if s["id"] == t["seq"])
drones = {l["track"] for s in te_data for l in s["labels"] if l["class"] == "drone"}
return te, drones
def set_alerts(te, thr):
for t in te:
t["alert"] = t["pattern"] != "static" and t["prob"] >= thr
return te
def summarize(te, drones):
alerts = [t for t in te if t["alert"]]
tp = {t["gt_track"] for t in alerts if t["gt_class"] == "drone"}
far = sum(t["gt_class"] != "drone" for t in alerts) / max(len(alerts), 1)
rec = len(tp) / len(drones)
acc = 2 * (1 - far) * rec / max((1 - far) + rec, 1e-9)
return {"경보": len(alerts), "오경보율": round(far, 3), "재현율": round(rec, 3), "종합정확도(F1)": round(acc, 3)}
def aar_log(te, drones):
"""AAR 일지: 오경보는 실제 정체·기상별로, 미탐은 원인별로 분류."""
rows = []
for t in te:
if t["alert"] and t["gt_class"] != "drone":
rows.append({"구분": "오경보", "원인": f"{t['gt_class']} 오인", "기상": t["cond"], "궤적": t["seq"]})
tracked = {t["gt_track"] for t in te if t["gt_class"] == "drone"}
alerted = {t["gt_track"] for t in te if t["gt_class"] == "drone" and t["alert"]}
for d in sorted(drones):
if d not in tracked:
rows.append({"구분": "미탐", "원인": "추적 실패(탐지 누락)", "기상": "-", "궤적": d})
elif d not in alerted:
rows.append({"구분": "미탐", "원인": "신뢰도 점수 미달", "기상": "-", "궤적": d})
return pd.DataFrame(rows, columns=["구분", "원인", "기상", "궤적"])
# 1) 초기 설정(k=2.0, 경보 임계값 0.3: '놓치지 않기'를 우선한 설정)으로 모의 운용
before_te, drones = run_system(2.0)
set_alerts(before_te, 0.3)
before = summarize(before_te, drones)
log = aar_log(before_te, drones)
print("[1. 무엇이 일어났나] 초기 설정 결과:", before)
print(log.groupby(["구분", "원인"]).size().rename("건수").to_string())
print("\n[2. 왜 일어났나] 오경보 기상 분포:", log[log["구분"] == "오경보"]["기상"].value_counts().to_dict())
# 2) 재보정: 제약(오경보율 15% 이하, 재현율 90% 이상) 안에서 F1 최대 조합 탐색
grid = []
for k in [2.0, 2.5, 3.0]:
te, dr = run_system(k)
for thr in [0.3, 0.4, 0.5, 0.6, 0.7]:
grid.append({"k": k, "경보임계값": thr, **summarize(set_alerts(te, thr), dr)})
g = pd.DataFrame(grid)
ok = g[(g["오경보율"] <= 0.15) & (g["재현율"] >= 0.9)].sort_values(["종합정확도(F1)", "재현율"], ascending=False)
best = ok.iloc[0]
print("\n[3. 무엇을 바꾸나] 격자 탐색 상위 5개")
print(ok.head().to_string(index=False))
after_te, _ = run_system(best["k"])
set_alerts(after_te, best["경보임계값"])
after = summarize(after_te, drones)
after_log = aar_log(after_te, drones)
print(f"\n재보정값: k={best['k']}, 경보 임계값={best['경보임계값']}")
print("재보정 후:", after)
print(f"종합정확도 변화: {before['종합정확도(F1)']:.3f} -> {after['종합정확도(F1)']:.3f} ({(after['종합정확도(F1)'] - before['종합정확도(F1)']) * 100:+.1f}%p)")
repeat = set(log[log["구분"] == "오경보"]["궤적"]) & set(after_log[after_log["구분"] == "오경보"]["궤적"])
print(f"재보정 후에도 반복된 오경보 시퀀스: {sorted(repeat) if repeat else '없음'}")
tuned = f"detection_k: {best['k']}\nalert_threshold: {best['경보임계값']}\nconstraints: {{max_far: 0.15, min_recall: 0.9}}"
print("\n[재보정 파라미터(ops01_tuned_params.yaml 내용)]\n" + tuned)
plt.figure(figsize=(6, 4))
for k in sorted(g["k"].unique()):
s = g[g["k"] == k]
plt.plot(s["오경보율"], s["재현율"], "o-", label=f"k={k}")
plt.axvline(0.15, ls="--", c="gray"); plt.axhline(0.9, ls="--", c="gray")
plt.xlabel("false alarm rate"); plt.ylabel("drone recall"); plt.legend(); plt.title("AAR re-tuning grid (synthetic)")
plt.show()
In [ ]:
"""데이터 생성·검증부터 전처리, 탐지, 추적·패턴, 신뢰도 점수, HITL 승인 대기열, 상황판, AAR 재보정 파라미터 적용까지 전체 파이프라인을 한 파일로 통합 실행하고, 자체 점검 결과와 지휘관 브리핑 요약을 출력한다."""
import hashlib
import json
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy import ndimage
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler
# ---------------------------------------------------------------------------
# [교육용가상자료] 합성 적외선 영상 생성기 (실제 영상·부대 정보 없음)
# 64x64 프레임에 드론·조류·민간기·지상 반사체(클러터)를 그리고 정답 라벨을 함께 만든다.
# ---------------------------------------------------------------------------
H = W = 64
PROFILE = {"drone": ((2, 3), (170, 225)), "bird": ((2, 3), (110, 150)),
"civil": ((4, 6), (190, 240)), "clutter": ((2, 4), (140, 190))}
def _new_obj(cls, rng, dive=False, v=None, xy=None):
# [TODO : 학생 실습 10] 드론 공중위협 객체 탐지 최종 파이프라인 통합 및 대드론 방어 브리핑 패키지 완성 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def _noise(img, cond, rng):
# [TODO : 학생 실습 10] 드론 공중위협 객체 탐지 최종 파이프라인 통합 및 대드론 방어 브리핑 패키지 완성 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def make_sequence(seq_id, cond, rng, kind="mixed", n_frames=30, n_drone=None, n_bird=None):
"""시퀀스 1개 = (프레임 배열, 라벨 목록). kind: mixed / swarm / dive / saturation"""
objs = []
if kind == "swarm":
v, cx, cy = rng.uniform(-1, 1, 2), rng.uniform(15, 35), rng.uniform(15, 35)
objs = [_new_obj("drone", rng, v=v, xy=(cx + i * 5, cy + (i % 2) * 5)) for i in range(4)]
objs.append(_new_obj("bird", rng))
elif kind == "dive":
d = _new_obj("drone", rng, dive=True)
d["y"] = float(rng.uniform(4, 14))
objs = [d, _new_obj("bird", rng), _new_obj("clutter", rng)]
elif kind == "saturation":
objs = [_new_obj("drone", rng) for _ in range(n_drone)] + [_new_obj("bird", rng) for _ in range(n_bird)]
else:
objs = [_new_obj("drone", rng) for _ in range(int(rng.integers(1, 3)))]
objs += [_new_obj("bird", rng) for _ in range(int(rng.integers(0, 3)))]
if rng.random() < 0.5:
objs.append(_new_obj("civil", rng))
if rng.random() < 0.6:
objs.append(_new_obj("clutter", rng))
frames, labels = np.zeros((n_frames, H, W), np.uint8), []
base = rng.normal(40, 6, (H, W))
for t in range(n_frames):
canvas = base.copy()
for i, o in enumerate(objs):
if t > 0:
if o["cls"] == "bird":
o["vx"] = float(np.clip(o["vx"] + rng.normal(0, 0.45), -1.6, 1.6))
o["vy"] = float(np.clip(o["vy"] + rng.normal(0, 0.45), -1.6, 1.6))
if o["dive"] and t > 10:
o["vy"] += 0.18
o["x"] = float(np.clip(o["x"] + o["vx"], 1, W - o["size"] - 1))
o["y"] = float(np.clip(o["y"] + o["vy"], 1, H - o["size"] - 1))
heat = o["heat"]
if o["cls"] == "clutter" and rng.random() < 0.3:
heat *= 0.7
if o["cls"] == "bird":
heat *= float(rng.uniform(0.85, 1.1))
x, y, s = int(round(o["x"])), int(round(o["y"])), o["size"]
canvas[y:y + s, x:x + s] = np.maximum(canvas[y:y + s, x:x + s], heat)
labels.append({"frame": t, "track": f"{seq_id}_T{i:02d}", "class": o["cls"], "bbox": [x, y, s, s],
"pattern": "dive" if o["dive"] else ("swarm" if kind == "swarm" and o["cls"] == "drone" else "normal")})
frames[t] = _noise(canvas, cond, rng)
return frames, labels
def make_dataset(n=24, seed=7):
rng = np.random.default_rng(seed)
conds = ["clear", "night", "fog", "rain"]
out = []
for i in range(n):
kind = "swarm" if i % 8 == 5 else ("dive" if i % 8 == 6 else "mixed")
f, lab = make_sequence(f"SEQ{i:03d}", conds[i % 4], rng, kind)
out.append({"id": f"SEQ{i:03d}", "cond": conds[i % 4], "kind": kind, "frames": f, "labels": lab})
return out
# ---------------------------------------------------------------------------
# ---- 전처리·탐지 공통 함수 ----
def preprocess(frame, cond):
f = frame.astype(float)
if cond == "rain": # 폭 1px 세로 빗줄기 제거(가로 방향 형태학적 열림)
f = ndimage.grey_opening(f, size=(1, 2))
lo, hi = np.percentile(f, [1, 99.5]) # 대비 확장
return np.clip((f - lo) / (hi - lo + 1e-6) * 255, 0, 255)
def detect(frame, cond, k=2.0, min_area=3, max_area=60):
img = preprocess(frame, cond)
lab, _ = ndimage.label(img > img.mean() + k * img.std())
dets = []
for i, sl in enumerate(ndimage.find_objects(lab), start=1):
area = int((lab[sl] == i).sum())
if min_area <= area <= max_area:
y, x = sl[0].start, sl[1].start
dets.append({"bbox": [x, y, sl[1].stop - x, sl[0].stop - y], "area": area,
"thermal": float(frame[sl][lab[sl] == i].max())})
return dets
def iou(a, b):
ix = max(0, min(a[0] + a[2], b[0] + b[2]) - max(a[0], b[0]))
iy = max(0, min(a[1] + a[3], b[1] + b[3]) - max(a[1], b[1]))
inter = ix * iy
return inter / (a[2] * a[3] + b[2] * b[3] - inter)
# ---- 추적·특징·패턴 공통 함수 ----
def center(b):
return np.array([b[0] + b[2] / 2, b[1] + b[3] / 2])
def track(per_frame, gate=6.0, max_missed=2, min_len=5):
"""예측 위치(마지막 위치 + 속도)와 가장 가까운 탐지를 이어 궤적을 만든다."""
active, done = [], []
for t, dets in enumerate(per_frame):
used = set()
for tr in sorted(active, key=lambda x: -len(x["t"])):
last = center(tr["b"][-1])
vel = last - center(tr["b"][-2]) if len(tr["b"]) > 1 else np.zeros(2)
pred = last + vel * (t - tr["t"][-1])
cand = [(np.linalg.norm(center(d["bbox"]) - pred), j) for j, d in enumerate(dets) if j not in used]
dist, j = min(cand, default=(gate, -1))
if j >= 0 and dist < gate:
used.add(j)
tr["t"].append(t); tr["b"].append(dets[j]["bbox"])
tr["area"].append(dets[j]["area"]); tr["heat"].append(dets[j]["thermal"])
still = []
for tr in active:
(still if t - tr["t"][-1] <= max_missed else done).append(tr)
active = still
for j, d in enumerate(dets):
if j not in used:
active.append({"t": [t], "b": [d["bbox"]], "area": [d["area"]], "heat": [d["thermal"]]})
return [tr for tr in done + active if len(tr["t"]) >= min_len]
FEATURES = ["mean_speed", "speed_std", "turn", "straight", "vy_accel", "area", "heat", "heat_std"]
def features(tr):
c = np.array([center(b) for b in tr["b"]]); f = np.array(tr["t"], float)
v = np.diff(c, axis=0) / np.maximum(np.diff(f), 1)[:, None]
sp = np.linalg.norm(v, axis=1)
k = 3 if len(c) > 4 else 1
v3 = (c[k:] - c[:-k]) / np.maximum(f[k:] - f[:-k], 1)[:, None]
hd = np.arctan2(v3[:, 1], v3[:, 0]); mv = np.linalg.norm(v3, axis=1) > 0.2
turns = np.abs(np.angle(np.exp(1j * np.diff(hd))))[mv[1:] & mv[:-1]]
half = len(v) // 2
return {"mean_speed": sp.mean(), "speed_std": sp.std(), "turn": turns.mean() if len(turns) else 0.0,
"straight": np.linalg.norm(c[-1] - c[0]) / max(sp.sum(), 1e-6),
"vy_accel": (v[half:, 1].mean() - v[:half, 1].mean()) / max(1, (f[-1] - f[0]) / 2) if half else 0.0,
"area": np.mean(tr["area"]), "heat": np.mean(tr["heat"]), "heat_std": np.std(tr["heat"]),
"vx": v[:, 0].mean(), "vy": v[:, 1].mean(), "x": c[-1, 0], "y": c[-1, 1]}
def pattern(f):
if f["mean_speed"] < 0.25:
return "static"
if f["vy_accel"] > 0.06:
return "dive"
if f["turn"] > 0.4 or f["straight"] < 0.6:
return "erratic"
return "steady"
def swarms(trs, vel=0.2, dist=18.0, min_n=3):
"""속도가 비슷하고 가까이 움직이는 궤적 3개 이상을 군집으로 묶는다."""
cand = [i for i, t in enumerate(trs) if t["pattern"] != "static"]
adj = {i: set() for i in cand}
for a in cand:
for b in cand:
fa, fb = trs[a]["f"], trs[b]["f"]
if a < b and np.hypot(fa["vx"] - fb["vx"], fa["vy"] - fb["vy"]) < vel and np.hypot(fa["x"] - fb["x"], fa["y"] - fb["y"]) < dist:
adj[a].add(b); adj[b].add(a)
seen, groups = set(), []
for i in cand:
if i in seen:
continue
stack, comp = [i], []
while stack:
k = stack.pop()
if k not in seen:
seen.add(k); comp.append(k); stack += list(adj[k] - seen)
if len(comp) >= min_n:
groups.append(comp)
return groups
def analyze(s, k=2.0):
"""시퀀스 1개: 탐지 -> 추적 -> 특징 -> 패턴 -> 군집 -> 정답 라벨 연결(평가용)."""
per = [detect(fr, s["cond"], k) if fr is not None else [] for fr in s["frames"]]
trs = track(per)
for tr in trs:
tr["f"] = features(tr); tr["pattern"] = pattern(tr["f"]); tr["swarm"] = None; tr["seq"] = s["id"]
votes = {}
for t, b in zip(tr["t"], tr["b"]):
best = max(((iou(b, l["bbox"]), l) for l in s["labels"] if l["frame"] == t), default=(0, None), key=lambda x: x[0])
key = (best[1]["track"], best[1]["class"], best[1]["pattern"]) if best[0] > 0.05 else ("none", "noise", "normal")
votes[key] = votes.get(key, 0) + 1
tr["gt_track"], tr["gt_class"], tr["gt_pattern"] = max(votes, key=votes.get)
for g, comp in enumerate(swarms(trs)):
for i in comp:
trs[i]["swarm"] = g
if trs[i]["pattern"] in ("steady", "erratic"):
trs[i]["pattern"] = "swarm"
return trs
CONFIG = {"detection_k": 2.5, "alert_threshold": 0.5, "queue_capacity": 30} # 미션 9 재보정값
def validate(seq):
errs = []
if seq["cond"] not in ("clear", "night", "fog", "rain"):
errs.append("condition 허용값 아님")
if seq["frames"].ndim != 3 or seq["frames"].shape[1:] != (64, 64):
errs.append("해상도 오류")
for l in seq["labels"]:
if l["class"] not in PROFILE or len(l["bbox"]) != 4:
errs.append("라벨 오류"); break
return errs
def X_of(trs):
return np.array([[t["f"][k] for k in FEATURES] for t in trs])
class Queue:
def __init__(self):
self.items, self.log = [], []
def submit(self, t):
prev = self.log[-1]["hash"] if self.log else "GENESIS"
body = {"event": "SUBMIT", "id": f"{t['seq']}-{len(self.items)}", "score": round(t["score"], 3)}
self.log.append({**body, "prev": prev, "hash": hashlib.sha256((prev + json.dumps(body, sort_keys=True)).encode()).hexdigest()})
self.items.append({**body, "grade": t["grade"], "status": "PENDING"})
# ---------------- 1. 학습(엔진 준비) ----------------
t_start = time.perf_counter()
train = [t for s in make_dataset(24, 7) for t in analyze(s, CONFIG["detection_k"])]
engine = make_pipeline(StandardScaler(), LogisticRegression(class_weight="balanced", max_iter=1000, random_state=0))
engine.fit(X_of(train), [t["gt_class"] == "drone" for t in train])
# ---------------- 2. 운용 데이터 처리 ----------------
ops = make_dataset(24, 11)
ops[3]["cond"] = "snow" # 손상 데이터 1건 주입(검증 단계에서 걸러져야 함)
queue, rejected, all_tracks, runs = Queue(), {}, [], []
for s in ops:
errs = validate(s)
if errs:
rejected[s["id"]] = errs; continue
t0 = time.perf_counter()
trs = analyze(s, CONFIG["detection_k"])
if trs:
for t, p in zip(trs, engine.predict_proba(X_of(trs))[:, 1]):
t["score"] = 0.0 if t["pattern"] == "static" else float(p)
t["grade"] = "High" if t["score"] >= 0.8 else ("Medium" if t["score"] >= 0.5 else "Low")
t["alert"] = t["pattern"] != "static" and t["score"] >= CONFIG["alert_threshold"]
if t["alert"]:
queue.submit(t)
runs.append({"seq": s["id"], "ms_per_frame": (time.perf_counter() - t0) * 1000 / len(s["frames"]),
"alerts": sum(t["alert"] for t in trs)})
all_tracks += trs
# ---------------- 3. 성능 평가 ----------------
drones = {l["track"] for s in ops if s["id"] not in rejected for l in s["labels"] if l["class"] == "drone"}
alerts = [t for t in all_tracks if t["alert"]]
far = sum(t["gt_class"] != "drone" for t in alerts) / max(len(alerts), 1)
recall = len({t["gt_track"] for t in alerts if t["gt_class"] == "drone"}) / len(drones)
ms = np.mean([r["ms_per_frame"] for r in runs])
# ---------------- 4. 자체 점검(통합 테스트) ----------------
chk = {
"손상 데이터 차단": "SEQ003" in rejected,
"처리 시퀀스 23개": len(runs) == 23,
"드론 재현율 90% 이상": recall >= 0.9,
"오경보율 15% 이하": far <= 0.15,
"자동 승인 0건": all(i["status"] == "PENDING" for i in queue.items),
"경보 전량 대기열 등록": len(queue.items) == len(alerts),
"감사 로그 건수 일치": len(queue.log) == len(queue.items),
"프레임당 50ms 이내": ms < 50,
"등급 규칙 일관": all((t["grade"] == "High") == (t["score"] >= 0.8) for t in all_tracks),
"static 궤적 경보 없음": not any(t["alert"] and t["pattern"] == "static" for t in all_tracks),
}
passed = sum(chk.values())
for k, v in chk.items():
print(f"[{'PASS' if v else 'FAIL'}] {k}")
print(f"자체 점검 {passed}/{len(chk)} 통과 ({passed / len(chk):.0%}), 파이프라인 실행 성공 {'예' if passed == len(chk) else '아니오'}")
# ---------------- 5. 상황판 ----------------
focus = max({t["seq"] for t in all_tracks}, key=lambda q: sum(t["alert"] for t in all_tracks if t["seq"] == q))
col = {"High": "tab:red", "Medium": "tab:orange", "Low": "tab:gray"}
fig, ax = plt.subplots(2, 2, figsize=(12, 8))
s = next(x for x in ops if x["id"] == focus)
ax[0, 0].imshow(s["frames"][-1], cmap="gray")
for t in [t for t in all_tracks if t["seq"] == focus]:
c = np.array([center(b) for b in t["b"]]); ax[0, 0].plot(c[:, 0], c[:, 1], color=col[t["grade"]], lw=2)
ax[0, 0].set_title(f"{focus} tracks (color = grade)"); ax[0, 0].axis("off")
g = pd.Series([t["grade"] for t in all_tracks]).value_counts().reindex(["High", "Medium", "Low"]).fillna(0)
ax[0, 1].bar(g.index, g.values, color=[col[i] for i in g.index]); ax[0, 1].set_title("Tracks by grade")
ax[1, 0].bar([r["seq"][-2:] for r in runs], [r["alerts"] for r in runs]); ax[1, 0].set_title("Alerts per sequence (pending human review)")
ax[1, 1].axis("off")
ax[1, 1].text(0, 0.9, f"Drone recall {recall:.1%}\nFalse alarm {far:.1%}\nQueue {len(queue.items)} pending\n"
f"Latency {ms:.2f} ms/frame\nSelf-check {passed}/{len(chk)}", family="monospace", fontsize=13, va="top")
plt.suptitle("OPS01 integrated pipeline - training data only"); plt.tight_layout(); plt.show()
# ---------------- 6. 지휘관 브리핑 요약 ----------------
print(f"""
[대드론 방어 브리핑 요약 - 교육용가상자료]
1. 처리: 시퀀스 {len(ops)}개 중 {len(runs)}개 처리, 손상 데이터 {len(rejected)}건 차단({', '.join(rejected)})
2. 성능: 드론 재현율 {recall:.1%}, 오경보율 {far:.1%}, 프레임당 {ms:.2f} ms, 전체 소요 {time.perf_counter() - t_start:.1f} s
3. 경보: {len(queue.items)}건 운용자 검토 대기 (High {sum(i['grade'] == 'High' for i in queue.items)} / Medium {sum(i['grade'] == 'Medium' for i in queue.items)}), 자동 승인 0건
4. 운용 유의: 경보는 운용자 승인 후에만 상급 지휘계통으로 전파하며, 교전 여부는 교전수칙·지휘계통에 따른다.
5. 한계: 합성 64x64 영상 기준 결과로, 실제 센서·기상·배경에서는 재학습과 재검증이 필요하다.""")