217 KiB
217 KiB
🎯 [SEC01] 기지 CCTV 드론·비인가 접근 객체 탐지 프로토타입
Google Colab 실습 환경 (시설·방호·경계 / Level 3)
📌 1. 프로젝트 개요 및 배경
| 항목 | 내용 |
|---|---|
| 과정 분류 | 시설·방호·경계 (Level 3) |
| 학습 소요 시간 | 10개 미션 (5일 / 총 27.5시간) |
| 환경 사양 | Python 3.10+, numpy, pandas, matplotlib, scipy, scikit-learn |
📖 프로젝트 목표
미국-이란 전쟁에서는 소형 드론과 차량·인원의 접근 시도가 기지 외곽 경계의 주요 위협으로 부각되었고, 많은 CCTV 화면에서 실제 침입과 수목 흔들림·야생동물을 빠르게 구분하는 일이 시설·방호 업무의 핵심이 되었다. 본 프로젝트에서는 교육용 가상 CCTV 영상을 코드로 직접 그려, 영상 데이터 스키마 정의, 조도·기상 전처리와 경계구역(ROI) 설정, 울타리 접근 객체 탐지 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 [ ]:
"""기지 외곽 CCTV 영상 메타데이터와 비인가 침입 객체(드론·차량·인원) 라벨의 스키마(카메라 규격, 전술 파라미터, 경계구역, 바운딩 박스)를 정의하고, 코드로 그린 합성 영상에서 만든 샘플 20건을 검증한다."""
import json
import re
import numpy as np
import pandas as pd
# ---------------- [교육용가상자료] 합성 CCTV 영상 생성기 ----------------
# 실제 영상 대신 코드로 그린 객체(사각형·점)로 기지 외곽 고정 카메라 화면을 흉내 낸다.
H, W, T = 64, 96, 10 # 화면 크기(px), 시퀀스 길이(프레임)
HORIZON, FENCE = 20, 46 # 하늘/지면 경계, 울타리 행
ROI_ROWS = (32, 48) # 울타리 접근 경계구역(ROI): 지면 32~47행
CLASSES = ["person", "vehicle", "drone", "animal", "bird"]
INTRUDER = {"person", "vehicle", "drone"} # 비인가 접근(경보 대상)
CONDS = ["day", "night", "rain", "fog"]
def background(rng):
bg = np.zeros((H, W))
bg[:HORIZON] = np.linspace(200, 170, HORIZON)[:, None] # 하늘
bg[HORIZON:] = 95 + rng.normal(0, 3, (H - HORIZON, W)) # 지면 질감
bg[26:30] = 115 # 외곽 도로
bg[FENCE] = 60; bg[FENCE - 3:FENCE + 1, ::8] = 60 # 울타리·기둥
return bg
def tree(img, t, wind):
"""좌측 수목 수관: 바람 세기에 따라 좌우로 흔들려 오경보를 만든다."""
s = int(round(wind * np.sin(t * 1.3)))
img[8:30, max(0, 2 + s):14 + s] = 55
def make_seq(seed, cond=None, objs=None, wind=None, camo=False):
rng = np.random.default_rng(seed)
cond = cond or str(rng.choice(CONDS))
wind = rng.uniform(0, 2) if wind is None else wind
bg = background(rng)
n = int(rng.integers(1, 4)) if objs is None else len(objs)
kinds = objs or list(rng.choice(CLASSES, n, p=[0.3, 0.15, 0.25, 0.15, 0.15]))
obs = []
for k in kinds:
if k == "person":
o = dict(y=rng.uniform(30, 36), x=rng.uniform(15, 85), vy=rng.uniform(0.8, 1.2), vx=rng.uniform(-0.6, 0.6), h=7, w=3,
val=72 if camo else rng.uniform(35, 50))
elif k == "vehicle":
o = dict(y=rng.uniform(33, 38), x=rng.uniform(10, 30), vy=0.2, vx=rng.uniform(2, 4), h=5, w=10, val=rng.uniform(150, 170))
elif k == "drone":
o = dict(y=rng.uniform(3, 15), x=rng.uniform(10, 80), vy=rng.uniform(0, 0.6), vx=rng.uniform(-2, 2), h=2, w=2 if camo else 3,
val=rng.uniform(80, 120))
elif k == "animal":
o = dict(y=rng.uniform(34, 42), x=rng.uniform(15, 80), vy=0, vx=rng.uniform(-1.5, 1.5), h=2, w=4, val=rng.uniform(60, 75))
else:
o = dict(y=rng.uniform(3, 15), x=rng.uniform(10, 80), vy=0, vx=rng.uniform(-3, 3), h=1, w=2, val=rng.uniform(100, 130))
o["cls"] = str(k); obs.append(o)
frames, labels = [], []
for t in range(-8, T): # 음수 시각 = 객체 없는 배경 학습 구간(사전 관측)
f = bg.copy(); tree(f, t, wind); lab = []
for oi, o in enumerate(obs if t >= 0 else []):
jit = rng.normal(0, 1.2, 2) if o["cls"] in ("animal", "bird") else (0, 0) # 불규칙 움직임
y, x = int(o["y"] + o["vy"] * t + jit[0]), int(o["x"] + o["vx"] * t + jit[1])
y0, x0, y1, x1 = max(y, 0), max(x, 0), min(y + o["h"], H), min(x + o["w"], W)
if y1 > y0 and x1 > x0:
f[y0:y1, x0:x1] = o["val"]
if o["cls"] == "person":
f[max(y0 - 2, 0):y0, x0 + 1:x0 + 2] = o["val"] # 머리
lab.append({"cls": o["cls"], "oid": oi, "bbox": (x0, y0, x1, y1), "intruder": o["cls"] in INTRUDER})
if cond == "night":
f = f * 0.5 + 10 + rng.normal(0, 3.5, f.shape)
else:
f = f + rng.normal(0, 2.5, f.shape)
if cond == "rain":
m = rng.random(f.shape) < 0.006
for d in range(3):
f[d:][m[:H - d]] += 45
if cond == "fog":
f = 0.45 * f + 0.55 * 150
frames.append(np.clip(f, 0, 255)); labels.append(lab)
fr = np.array(frames)
return {"id": f"CAM-{seed:04d}", "cond": cond, "wind": wind, "ref": fr[:8], "frames": fr[8:], "labels": labels[8:]}
# 1) 스키마 정의: 카메라 규격, 전술 파라미터(기상·위치오차·통신지연), 경계구역(ROI), 객체 라벨
SCHEMA = {
"required": ["camera_id", "n_frames", "camera", "tactical", "roi", "labels"],
"fields": {
"camera_id": {"type": str, "pattern": r"^CAM-\d{4}$"},
"n_frames": {"type": int, "min": 1, "max": 3000},
},
"camera": {
"mode": {"type": str, "enum": ["EO", "IR"]},
"resolution": {"type": list, "length": 2},
"fps": {"type": (int, float), "min": 1, "max": 60},
},
"tactical": {
"condition": {"type": str, "enum": ["day", "night", "rain", "fog"]},
"wind_level": {"type": (int, float), "min": 0, "max": 10},
"position_error_m": {"type": (int, float), "min": 0, "max": 100},
"comm_delay_ms": {"type": (int, float), "min": 0, "max": 10000},
},
"roi": {
"row_start": {"type": int, "min": 0, "max": 200},
"row_end": {"type": int, "min": 1, "max": 200},
},
"label": {
"class": {"type": str, "enum": CLASSES},
"bbox": {"type": list, "length": 4},
"frame": {"type": int, "min": 0},
"intruder": {"type": bool},
},
}
def check(name, value, rule):
"""값 하나를 규칙(type, enum, min, max, length, pattern)으로 검사해 오류 목록을 돌려준다."""
if not isinstance(value, rule["type"]) or (isinstance(value, bool) and rule["type"] is not bool):
return [f"{name}: 자료형 오류({type(value).__name__})"]
errs = []
if "enum" in rule and value not in rule["enum"]:
errs.append(f"{name}: 허용되지 않은 값 '{value}'")
if "min" in rule and value < rule["min"]:
errs.append(f"{name}: 최소값 {rule['min']} 미만")
if "max" in rule and value > rule["max"]:
errs.append(f"{name}: 최대값 {rule['max']} 초과")
if "length" in rule and len(value) != rule["length"]:
errs.append(f"{name}: 길이 {rule['length']} 필요")
if "pattern" in rule and not re.match(rule["pattern"], value):
errs.append(f"{name}: 형식 불일치 '{value}'")
return errs
def validate(rec):
errs = [f"{k}: 필수 필드 누락" for k in SCHEMA["required"] if k not in rec]
for k, rule in SCHEMA["fields"].items():
if k in rec:
errs += check(k, rec[k], rule)
for group in ("camera", "tactical", "roi"):
for k, rule in SCHEMA[group].items():
if k not in rec.get(group, {}):
errs.append(f"{group}.{k}: 필수 필드 누락")
else:
errs += check(f"{group}.{k}", rec[group][k], rule)
roi = rec.get("roi", {})
if "row_start" in roi and "row_end" in roi and roi["row_start"] >= roi["row_end"]:
errs.append("roi: row_start는 row_end보다 작아야 함")
for i, lab in enumerate(rec.get("labels", [])):
for k, rule in SCHEMA["label"].items():
if k not in lab:
errs.append(f"labels[{i}].{k}: 필수 필드 누락")
else:
errs += check(f"labels[{i}].{k}", lab[k], rule)
return errs
# 2) [교육용가상자료] 합성 CCTV 시퀀스 20개에서 메타데이터 샘플 생성
rng = np.random.default_rng(1)
records = []
for i in range(20):
seq = make_seq(100 + i)
labels = [{"class": l["cls"], "frame": t, "bbox": [int(v) for v in l["bbox"]], "intruder": bool(l["intruder"])}
for t in (0, 5, 9) for l in seq["labels"][t]]
records.append({"camera_id": seq["id"], "n_frames": T,
"camera": {"mode": "IR" if seq["cond"] == "night" else "EO", "resolution": [W, H], "fps": 10},
"tactical": {"condition": seq["cond"], "wind_level": round(float(seq["wind"]) * 2.5, 1),
"position_error_m": round(float(rng.uniform(1, 15)), 1), "comm_delay_ms": int(rng.integers(20, 400))},
"roi": {"row_start": ROI_ROWS[0], "row_end": ROI_ROWS[1]}, "labels": labels})
# 3) 정상 데이터 검증
ok = sum(1 for r in records if not validate(r))
print(f"정상 샘플 {len(records)}건 중 검증 통과 {ok}건, 오류 {len(records) - ok}건 -> 목표(검증 오류 0건): {'달성' if ok == len(records) else '미달'}")
# 4) 오류 주입 시험: 검증기가 잘못된 데이터를 잡아내는지 확인
bad = json.loads(json.dumps(records[0]))
bad["camera_id"] = "CAM-1" # 형식
bad["camera"]["fps"] = 120 # 범위
del bad["tactical"]["comm_delay_ms"] # 누락
bad["labels"][0]["class"] = "ufo" # 허용값
bad["roi"]["row_start"] = 50 # 논리(시작 > 끝)
errs = validate(bad)
print(f"\n[오류 주입 결과] 주입 5종 -> 검출 {len(errs)}건")
for e in errs:
print(" -", e)
print(f"주입 오류 전부 검출: {'예' if len(errs) >= 5 else '아니오'}")
# 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 ("camera", "tactical", "roi", "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"})))
print("\n[데이터 명세 요약] 필드 수", len(rows))
print(pd.DataFrame(rows, columns=["그룹", "필드", "자료형", "규칙"]).to_string(index=False))
print("\n침입 대상 클래스:", sorted(INTRUDER), "/ 오경보 유발 클래스: animal, bird, 수목 흔들림(clutter)")
print("클래스별 라벨 수:", pd.Series([l["class"] for r in records for l in r["labels"]]).value_counts().to_dict())
In [ ]:
"""주/야간 조도 보정, 비·잡음 필터, 센서 단절 프레임 보정, 울타리 경계구역(ROI) 마스크를 묶은 CCTV 전처리 모듈을 구현하고, 처리 속도·단위 테스트 5종·조건별 전처리 전후 탐지 성능을 측정한다."""
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy import ndimage
# ---------------- [교육용가상자료] 합성 CCTV 영상 생성기 ----------------
# 실제 영상 대신 코드로 그린 객체(사각형·점)로 기지 외곽 고정 카메라 화면을 흉내 낸다.
H, W, T = 64, 96, 10 # 화면 크기(px), 시퀀스 길이(프레임)
HORIZON, FENCE = 20, 46 # 하늘/지면 경계, 울타리 행
ROI_ROWS = (32, 48) # 울타리 접근 경계구역(ROI): 지면 32~47행
CLASSES = ["person", "vehicle", "drone", "animal", "bird"]
INTRUDER = {"person", "vehicle", "drone"} # 비인가 접근(경보 대상)
CONDS = ["day", "night", "rain", "fog"]
def background(rng):
bg = np.zeros((H, W))
bg[:HORIZON] = np.linspace(200, 170, HORIZON)[:, None] # 하늘
bg[HORIZON:] = 95 + rng.normal(0, 3, (H - HORIZON, W)) # 지면 질감
bg[26:30] = 115 # 외곽 도로
bg[FENCE] = 60; bg[FENCE - 3:FENCE + 1, ::8] = 60 # 울타리·기둥
return bg
def tree(img, t, wind):
"""좌측 수목 수관: 바람 세기에 따라 좌우로 흔들려 오경보를 만든다."""
s = int(round(wind * np.sin(t * 1.3)))
img[8:30, max(0, 2 + s):14 + s] = 55
def make_seq(seed, cond=None, objs=None, wind=None, camo=False):
rng = np.random.default_rng(seed)
cond = cond or str(rng.choice(CONDS))
wind = rng.uniform(0, 2) if wind is None else wind
bg = background(rng)
n = int(rng.integers(1, 4)) if objs is None else len(objs)
kinds = objs or list(rng.choice(CLASSES, n, p=[0.3, 0.15, 0.25, 0.15, 0.15]))
obs = []
for k in kinds:
if k == "person":
o = dict(y=rng.uniform(30, 36), x=rng.uniform(15, 85), vy=rng.uniform(0.8, 1.2), vx=rng.uniform(-0.6, 0.6), h=7, w=3,
val=72 if camo else rng.uniform(35, 50))
elif k == "vehicle":
o = dict(y=rng.uniform(33, 38), x=rng.uniform(10, 30), vy=0.2, vx=rng.uniform(2, 4), h=5, w=10, val=rng.uniform(150, 170))
elif k == "drone":
o = dict(y=rng.uniform(3, 15), x=rng.uniform(10, 80), vy=rng.uniform(0, 0.6), vx=rng.uniform(-2, 2), h=2, w=2 if camo else 3,
val=rng.uniform(80, 120))
elif k == "animal":
o = dict(y=rng.uniform(34, 42), x=rng.uniform(15, 80), vy=0, vx=rng.uniform(-1.5, 1.5), h=2, w=4, val=rng.uniform(60, 75))
else:
o = dict(y=rng.uniform(3, 15), x=rng.uniform(10, 80), vy=0, vx=rng.uniform(-3, 3), h=1, w=2, val=rng.uniform(100, 130))
o["cls"] = str(k); obs.append(o)
frames, labels = [], []
for t in range(-8, T): # 음수 시각 = 객체 없는 배경 학습 구간(사전 관측)
f = bg.copy(); tree(f, t, wind); lab = []
for oi, o in enumerate(obs if t >= 0 else []):
jit = rng.normal(0, 1.2, 2) if o["cls"] in ("animal", "bird") else (0, 0) # 불규칙 움직임
y, x = int(o["y"] + o["vy"] * t + jit[0]), int(o["x"] + o["vx"] * t + jit[1])
y0, x0, y1, x1 = max(y, 0), max(x, 0), min(y + o["h"], H), min(x + o["w"], W)
if y1 > y0 and x1 > x0:
f[y0:y1, x0:x1] = o["val"]
if o["cls"] == "person":
f[max(y0 - 2, 0):y0, x0 + 1:x0 + 2] = o["val"] # 머리
lab.append({"cls": o["cls"], "oid": oi, "bbox": (x0, y0, x1, y1), "intruder": o["cls"] in INTRUDER})
if cond == "night":
f = f * 0.5 + 10 + rng.normal(0, 3.5, f.shape)
else:
f = f + rng.normal(0, 2.5, f.shape)
if cond == "rain":
m = rng.random(f.shape) < 0.006
for d in range(3):
f[d:][m[:H - d]] += 45
if cond == "fog":
f = 0.45 * f + 0.55 * 150
frames.append(np.clip(f, 0, 255)); labels.append(lab)
fr = np.array(frames)
return {"id": f"CAM-{seed:04d}", "cond": cond, "wind": wind, "ref": fr[:8], "frames": fr[8:], "labels": labels[8:]}
# ---------------- 전처리·탐지 공통 함수(배경 차분 기반) ----------------
def enhance(seq):
"""조도 보정(백분위 신장) + 비 조건의 빗줄기 제거(3x3 중앙값). 배경 학습 구간(ref)과 같은 기준으로 보정한다."""
allf = np.concatenate([seq["ref"], seq["frames"]])
f = ndimage.median_filter(allf, size=(1, 3, 3)) if seq["cond"] == "rain" else allf
lo, hi = np.percentile(f, [1, 99.5])
f = np.clip((f - lo) / (hi - lo + 1e-6) * 255, 0, 255)
return f[:len(seq["ref"])], f[len(seq["ref"]):]
def detect(ref, frames, k=4.0, min_area=3, sky_k=None, smooth=False):
"""배경 학습 구간의 중앙값 배경과 차이를 강건 임계값(k x MAD)으로 이진화해 연결 성분을 객체 후보로 만든다.
smooth=True 이면 연속 3프레임 평균으로 시간 잡음을 줄여 저대비 객체를 살린다."""
bg = np.median(ref, axis=0)
if smooth:
frames = ndimage.uniform_filter1d(frames, 3, axis=0, mode="nearest")
out = []
for f in frames:
d = np.abs(f - bg)
mad = np.median(d) * 1.4826 + 1e-6
thr = np.full(d.shape, k * mad)
if sky_k is not None:
thr[:HORIZON] = sky_k * mad
lab, n = ndimage.label(d > thr)
dets = []
for i, sl in enumerate(ndimage.find_objects(lab), start=1):
area = int((lab[sl] == i).sum())
if area >= min_area:
dets.append({"bbox": (sl[1].start, sl[0].start, sl[1].stop, sl[0].stop), "area": area,
"contrast": float(d[sl][lab[sl] == i].mean() / mad)})
out.append(dets)
return out
def center(b):
return ((b[0] + b[2]) / 2, (b[1] + b[3]) / 2)
def match(dets, labels, dist=4.0):
"""탐지와 정답을 중심 거리로 1:1 대응시킨다. 반환: [(det, label or None)], 놓친 label 목록"""
used, pairs = set(), []
for d in dets:
cd = center(d["bbox"]); best, bj = dist, None
for j, l in enumerate(labels):
cl = center(l["bbox"]); e = np.hypot(cd[0] - cl[0], cd[1] - cl[1])
if j not in used and e < best:
best, bj = e, j
if bj is not None:
used.add(bj)
pairs.append((d, labels[bj] if bj is not None else None))
return pairs, [l for j, l in enumerate(labels) if j not in used]
def in_roi(b):
cy = (b[1] + b[3]) / 2
return cy < HORIZON or ROI_ROWS[0] <= cy < ROI_ROWS[1]
def rule_class(b, area):
"""Baseline 규칙: 위치(하늘/지면)·크기·종횡비로 1차 분류(학습 없음, 판단 근거가 그대로 드러난다)"""
w, h = b[2] - b[0], b[3] - b[1]
cy = (b[1] + b[3]) / 2
if cy < HORIZON:
return "drone" if 3 <= area <= 12 else "clutter"
if area >= 30 and w > h:
return "vehicle"
if h >= w and 6 <= area <= 30:
return "person"
return "clutter"
def repair_frames(f):
"""센서 단절(전 화면 0 또는 NaN) 프레임을 직전 정상 프레임으로 대체하고 보정 건수를 돌려준다."""
f = f.copy(); fixed = 0
for t in range(len(f)):
if np.isnan(f[t]).any() or f[t].max() < 1:
f[t] = f[t - 1] if t > 0 else np.nanmedian(f[1:], axis=0); fixed += 1
return f, fixed
def roi_mask():
"""경계구역: 하늘(드론 접근) + 울타리 접근 지면 띠. 수목 영역(좌측 x<16)은 감시 제외 후보로 표시만 한다."""
m = np.zeros((H, W), bool)
m[:HORIZON] = True; m[ROI_ROWS[0]:ROI_ROWS[1]] = True
return m
def preprocess(seq):
"""반환: (보정된 배경 구간, 보정된 감시 구간, 보정 건수). 조건별 처리를 자동 선택한다."""
allf, fixed = repair_frames(np.concatenate([seq["ref"], seq["frames"]]))
s2 = {**seq, "ref": allf[:len(seq["ref"])], "frames": allf[len(seq["ref"]):]}
ref, fr = enhance(s2)
return ref, fr, fixed
def corrupt(seq, rate, rng):
"""센서 단절 모의: 감시 구간 프레임 일부를 0으로 만든다(스푸핑·단절 잡음의 단순화)."""
fr = seq["frames"].copy()
for t in rng.choice(len(fr), int(round(rate * len(fr))), replace=False):
fr[t] = 0
return {**seq, "frames": fr}
def f1_of(seqs, use_pre, mask=None):
tp = fp = fn = 0
for q in seqs:
ref, fr = preprocess(q)[:2] if use_pre else (q["ref"], q["frames"])
for ds, labs in zip(detect(ref, fr), q["labels"]):
labs = [l for l in labs if l["intruder"]]
if mask is not None:
ds = [d for d in ds if in_roi(d["bbox"])]
pairs, missed = match(ds, labs)
tp += sum(l is not None for _, l in pairs); fp += sum(l is None for _, l in pairs); fn += len(missed)
p, r = tp / max(tp + fp, 1), tp / max(tp + fn, 1)
return round(p, 3), round(r, 3), round(2 * p * r / max(p + r, 1e-9), 3)
rng = np.random.default_rng(5)
data = [make_seq(s) for s in range(300, 340)]
noisy = [corrupt(q, 0.2, rng) if i % 2 == 0 else q for i, q in enumerate(data)] # 절반에 센서 단절 20% 주입
# 1) 처리 속도와 오류율: 프레임 단위 전처리(조도 보정 + 필터)를 시퀀스 40개에 적용
t0 = time.perf_counter(); n_frames = 0; errors = 0
for q in noisy:
try:
ref, fr, _ = preprocess(q); n_frames += len(ref) + len(fr)
errors += int(not np.isfinite(fr).all() or fr.min() < 0 or fr.max() > 255)
except Exception:
errors += 1
sec = time.perf_counter() - t0
print(f"전처리 {n_frames}프레임 / {sec:.2f}초 = {n_frames / sec:,.0f}프레임/초 (목표 1,000 이상: {'달성' if n_frames / sec >= 1000 else '미달'})")
print(f"전처리 오류(NaN·범위 이탈) {errors}건 -> 오류율 {errors / len(noisy):.0%} (목표 0%)")
# 2) 단위 테스트 5종
def t1():
f = np.full((4, H, W), 100.0); f[2] = 0
out, n = repair_frames(f); assert n == 1 and out[2].mean() == 100
def t2():
f = np.full((3, H, W), 100.0); f[1, 5, 5] = np.nan
out, n = repair_frames(f); assert n == 1 and np.isfinite(out).all()
def t3():
ref, fr, _ = preprocess(make_seq(1, cond="night")); assert 0 <= fr.min() and fr.max() <= 255
def t4():
m = roi_mask(); assert m[10, 10] and m[40, 40] and not m[25, 40] and not m[55, 40]
def t5():
q = make_seq(2, cond="rain"); ref, fr, _ = preprocess(q); assert fr.shape == q["frames"].shape
tests = {"단절 프레임 보정": t1, "NaN 프레임 보정": t2, "야간 밝기 범위": t3, "ROI 마스크 경계": t4, "출력 형상 보존": t5}
passed = 0
for name, fn in tests.items():
try:
fn(); passed += 1; print(f"[PASS] {name}")
except AssertionError:
print(f"[FAIL] {name}")
print(f"단위 테스트 {passed}/5 통과")
# 3) 조건별 전처리 전후 탐지 성능(센서 단절 주입 포함)
rows = []
for c in CONDS + ["전체"]:
sub = [q for q in noisy if c == "전체" or q["cond"] == c]
b, a = f1_of(sub, False), f1_of(sub, True)
ar = f1_of(sub, True, mask=True)
rows.append({"조건": c, "시퀀스": len(sub), "전 F1": b[2], "후 F1": a[2], "후 재현율": a[1], "후+ROI 정밀도": ar[0], "후+ROI F1": ar[2]})
df = pd.DataFrame(rows)
print("\n[조건별 전처리 전후 탐지 F1] (전: 원본 그대로 / 후: 단절 보정+조도 보정+필터 / ROI: 경계구역 밖 탐지 제외)")
print(df.to_string(index=False))
print("전술 노이즈 대응: 센서 단절 프레임은 직전 프레임으로 대체, 비는 3x3 중앙값, 야간 조도는 백분위 신장으로 보정하며 스푸핑 영상 판별은 본 과제 범위 밖이다.")
q = noisy[2]; ref, fr, _ = preprocess(q)
fig, ax = plt.subplots(1, 3, figsize=(12, 3.5))
for a, im, ti in zip(ax, [q["frames"][4], fr[4], fr[4] * roi_mask()], ["raw", "preprocessed", "preprocessed + ROI mask"]):
a.imshow(im, cmap="gray", vmin=0, vmax=255); a.set_title(f"{q['id']} {q['cond']}: {ti}"); a.axis("off")
plt.tight_layout(); plt.show()
In [ ]:
"""기지 외곽 고정 CCTV 합성 영상에서 배경 차분과 위치·크기·종횡비 규칙으로 울타리 접근 객체(인원·차량·드론)를 1차 감지하는 Baseline 비전 모델을 구현하고, 프레임 단위 탐지 정확도와 오경보를 측정한다."""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy import ndimage
# ---------------- [교육용가상자료] 합성 CCTV 영상 생성기 ----------------
# 실제 영상 대신 코드로 그린 객체(사각형·점)로 기지 외곽 고정 카메라 화면을 흉내 낸다.
H, W, T = 64, 96, 10 # 화면 크기(px), 시퀀스 길이(프레임)
HORIZON, FENCE = 20, 46 # 하늘/지면 경계, 울타리 행
ROI_ROWS = (32, 48) # 울타리 접근 경계구역(ROI): 지면 32~47행
CLASSES = ["person", "vehicle", "drone", "animal", "bird"]
INTRUDER = {"person", "vehicle", "drone"} # 비인가 접근(경보 대상)
CONDS = ["day", "night", "rain", "fog"]
def background(rng):
bg = np.zeros((H, W))
bg[:HORIZON] = np.linspace(200, 170, HORIZON)[:, None] # 하늘
bg[HORIZON:] = 95 + rng.normal(0, 3, (H - HORIZON, W)) # 지면 질감
bg[26:30] = 115 # 외곽 도로
bg[FENCE] = 60; bg[FENCE - 3:FENCE + 1, ::8] = 60 # 울타리·기둥
return bg
def tree(img, t, wind):
"""좌측 수목 수관: 바람 세기에 따라 좌우로 흔들려 오경보를 만든다."""
s = int(round(wind * np.sin(t * 1.3)))
img[8:30, max(0, 2 + s):14 + s] = 55
def make_seq(seed, cond=None, objs=None, wind=None, camo=False):
rng = np.random.default_rng(seed)
cond = cond or str(rng.choice(CONDS))
wind = rng.uniform(0, 2) if wind is None else wind
bg = background(rng)
n = int(rng.integers(1, 4)) if objs is None else len(objs)
kinds = objs or list(rng.choice(CLASSES, n, p=[0.3, 0.15, 0.25, 0.15, 0.15]))
obs = []
for k in kinds:
if k == "person":
o = dict(y=rng.uniform(30, 36), x=rng.uniform(15, 85), vy=rng.uniform(0.8, 1.2), vx=rng.uniform(-0.6, 0.6), h=7, w=3,
val=72 if camo else rng.uniform(35, 50))
elif k == "vehicle":
o = dict(y=rng.uniform(33, 38), x=rng.uniform(10, 30), vy=0.2, vx=rng.uniform(2, 4), h=5, w=10, val=rng.uniform(150, 170))
elif k == "drone":
o = dict(y=rng.uniform(3, 15), x=rng.uniform(10, 80), vy=rng.uniform(0, 0.6), vx=rng.uniform(-2, 2), h=2, w=2 if camo else 3,
val=rng.uniform(80, 120))
elif k == "animal":
o = dict(y=rng.uniform(34, 42), x=rng.uniform(15, 80), vy=0, vx=rng.uniform(-1.5, 1.5), h=2, w=4, val=rng.uniform(60, 75))
else:
o = dict(y=rng.uniform(3, 15), x=rng.uniform(10, 80), vy=0, vx=rng.uniform(-3, 3), h=1, w=2, val=rng.uniform(100, 130))
o["cls"] = str(k); obs.append(o)
frames, labels = [], []
for t in range(-8, T): # 음수 시각 = 객체 없는 배경 학습 구간(사전 관측)
f = bg.copy(); tree(f, t, wind); lab = []
for oi, o in enumerate(obs if t >= 0 else []):
jit = rng.normal(0, 1.2, 2) if o["cls"] in ("animal", "bird") else (0, 0) # 불규칙 움직임
y, x = int(o["y"] + o["vy"] * t + jit[0]), int(o["x"] + o["vx"] * t + jit[1])
y0, x0, y1, x1 = max(y, 0), max(x, 0), min(y + o["h"], H), min(x + o["w"], W)
if y1 > y0 and x1 > x0:
f[y0:y1, x0:x1] = o["val"]
if o["cls"] == "person":
f[max(y0 - 2, 0):y0, x0 + 1:x0 + 2] = o["val"] # 머리
lab.append({"cls": o["cls"], "oid": oi, "bbox": (x0, y0, x1, y1), "intruder": o["cls"] in INTRUDER})
if cond == "night":
f = f * 0.5 + 10 + rng.normal(0, 3.5, f.shape)
else:
f = f + rng.normal(0, 2.5, f.shape)
if cond == "rain":
m = rng.random(f.shape) < 0.006
for d in range(3):
f[d:][m[:H - d]] += 45
if cond == "fog":
f = 0.45 * f + 0.55 * 150
frames.append(np.clip(f, 0, 255)); labels.append(lab)
fr = np.array(frames)
return {"id": f"CAM-{seed:04d}", "cond": cond, "wind": wind, "ref": fr[:8], "frames": fr[8:], "labels": labels[8:]}
# ---------------- 전처리·탐지 공통 함수(배경 차분 기반) ----------------
def enhance(seq):
"""조도 보정(백분위 신장) + 비 조건의 빗줄기 제거(3x3 중앙값). 배경 학습 구간(ref)과 같은 기준으로 보정한다."""
allf = np.concatenate([seq["ref"], seq["frames"]])
f = ndimage.median_filter(allf, size=(1, 3, 3)) if seq["cond"] == "rain" else allf
lo, hi = np.percentile(f, [1, 99.5])
f = np.clip((f - lo) / (hi - lo + 1e-6) * 255, 0, 255)
return f[:len(seq["ref"])], f[len(seq["ref"]):]
def detect(ref, frames, k=4.0, min_area=3, sky_k=None, smooth=False):
"""배경 학습 구간의 중앙값 배경과 차이를 강건 임계값(k x MAD)으로 이진화해 연결 성분을 객체 후보로 만든다.
smooth=True 이면 연속 3프레임 평균으로 시간 잡음을 줄여 저대비 객체를 살린다."""
bg = np.median(ref, axis=0)
if smooth:
frames = ndimage.uniform_filter1d(frames, 3, axis=0, mode="nearest")
out = []
for f in frames:
d = np.abs(f - bg)
mad = np.median(d) * 1.4826 + 1e-6
thr = np.full(d.shape, k * mad)
if sky_k is not None:
thr[:HORIZON] = sky_k * mad
lab, n = ndimage.label(d > thr)
dets = []
for i, sl in enumerate(ndimage.find_objects(lab), start=1):
area = int((lab[sl] == i).sum())
if area >= min_area:
dets.append({"bbox": (sl[1].start, sl[0].start, sl[1].stop, sl[0].stop), "area": area,
"contrast": float(d[sl][lab[sl] == i].mean() / mad)})
out.append(dets)
return out
def center(b):
return ((b[0] + b[2]) / 2, (b[1] + b[3]) / 2)
def match(dets, labels, dist=4.0):
"""탐지와 정답을 중심 거리로 1:1 대응시킨다. 반환: [(det, label or None)], 놓친 label 목록"""
used, pairs = set(), []
for d in dets:
cd = center(d["bbox"]); best, bj = dist, None
for j, l in enumerate(labels):
cl = center(l["bbox"]); e = np.hypot(cd[0] - cl[0], cd[1] - cl[1])
if j not in used and e < best:
best, bj = e, j
if bj is not None:
used.add(bj)
pairs.append((d, labels[bj] if bj is not None else None))
return pairs, [l for j, l in enumerate(labels) if j not in used]
def in_roi(b):
cy = (b[1] + b[3]) / 2
return cy < HORIZON or ROI_ROWS[0] <= cy < ROI_ROWS[1]
def rule_class(b, area):
"""Baseline 규칙: 위치(하늘/지면)·크기·종횡비로 1차 분류(학습 없음, 판단 근거가 그대로 드러난다)"""
w, h = b[2] - b[0], b[3] - b[1]
cy = (b[1] + b[3]) / 2
if cy < HORIZON:
return "drone" if 3 <= area <= 12 else "clutter"
if area >= 30 and w > h:
return "vehicle"
if h >= w and 6 <= area <= 30:
return "person"
return "clutter"
# 판단 기준(투명성): ① 학습된 배경과 밝기 차이가 잡음의 4배(k=4 x MAD) 이상인 픽셀 덩어리 ② 3픽셀 이상
# ③ 하늘 영역 3~12픽셀 = 드론, 지면의 가로로 긴 30픽셀 이상 = 차량, 세로로 긴 6~30픽셀 = 인원
# ④ 경계구역(ROI: 하늘 + 울타리 접근 구역) 안의 인원·차량·드론만 경보
rows, fa_src = [], {}
for s in range(500, 540):
seq = make_seq(s)
ref, fr = enhance(seq)
for t, (ds, labs) in enumerate(zip(detect(ref, fr), seq["labels"])):
pairs, missed = match(ds, labs)
for d, l in pairs:
pred = rule_class(d["bbox"], d["area"])
alarm = pred in INTRUDER and in_roi(d["bbox"])
truth = l["cls"] if l else "clutter"
rows.append({"cond": seq["cond"], "truth": truth, "pred": pred, "alarm": alarm})
for l in missed:
rows.append({"cond": seq["cond"], "truth": l["cls"], "pred": "missed", "alarm": False})
df = pd.DataFrame(rows)
intr = df[df["truth"].isin(INTRUDER)]
acc = (intr["pred"] == intr["truth"]).mean()
det_rate = intr["alarm"].mean()
fa = df[df["alarm"] & ~df["truth"].isin(INTRUDER)]
print(f"[Baseline] 합성 CCTV 시퀀스 40개(프레임 {40 * T}장), 침입 객체 출현 {len(intr)}회")
print(f"침입 객체 탐지·분류 정확도 {acc:.1%} (목표 75% 이상) -> {'달성' if acc >= 0.75 else '미달'}")
print(f"침입 경보 발생률 {det_rate:.1%}, 오경보 {len(fa)}건 (원인: {fa['truth'].value_counts().to_dict()})")
print("\n[정체별 Baseline 판정] (행: 실제, 열: 판정)")
print(pd.crosstab(df["truth"], df["pred"]).to_string())
print("\n[조건별 침입 객체 정확도]")
print(intr.groupby("cond").apply(lambda g: f"{(g['pred'] == g['truth']).mean():.1%} ({len(g)}회)").to_string())
print("한계: 규칙은 위장 인원·소형 드론(저대비·소면적)과 수목·동물 오경보에 약하다 -> 미션 4·5에서 보완한다.")
seq = make_seq(503, cond="day", objs=["person", "vehicle", "drone", "animal"])
ref, fr = enhance(seq)
ds = detect(ref, fr)[6]
fig, ax = plt.subplots(1, 2, figsize=(11, 4))
ax[0].imshow(seq["frames"][6], cmap="gray", vmin=0, vmax=255); ax[0].set_title("Synthetic CCTV frame (training data)")
ax[1].imshow(fr[6], cmap="gray")
ax[1].axhspan(ROI_ROWS[0], ROI_ROWS[1], color="yellow", alpha=0.15)
for d in ds:
b, p = d["bbox"], rule_class(d["bbox"], d["area"])
ax[1].add_patch(plt.Rectangle((b[0] - .5, b[1] - .5), b[2] - b[0], b[3] - b[1], fill=False, ec="red" if p in INTRUDER else "gray"))
ax[1].text(b[0], b[1] - 1, p, color="red" if p in INTRUDER else "gray", fontsize=8)
ax[1].set_title("Baseline detections (yellow = fence ROI)")
plt.tight_layout(); plt.show()
In [ ]:
"""저고도 소형 드론과 위장(저대비) 침입자를 식별하도록, 하늘 영역 민감도 상향·프레임 간 추적·궤적 특징 기반 랜덤포레스트 분류를 결합한 복합 탐지 모델로 고도화하고, 객체 단위 재현율과 프레임당 처리 지연을 Baseline과 비교한다."""
import time
import numpy as np
import pandas as pd
from scipy import ndimage
from sklearn.ensemble import RandomForestClassifier
# ---------------- [교육용가상자료] 합성 CCTV 영상 생성기 ----------------
# 실제 영상 대신 코드로 그린 객체(사각형·점)로 기지 외곽 고정 카메라 화면을 흉내 낸다.
H, W, T = 64, 96, 10 # 화면 크기(px), 시퀀스 길이(프레임)
HORIZON, FENCE = 20, 46 # 하늘/지면 경계, 울타리 행
ROI_ROWS = (32, 48) # 울타리 접근 경계구역(ROI): 지면 32~47행
CLASSES = ["person", "vehicle", "drone", "animal", "bird"]
INTRUDER = {"person", "vehicle", "drone"} # 비인가 접근(경보 대상)
CONDS = ["day", "night", "rain", "fog"]
def background(rng):
bg = np.zeros((H, W))
bg[:HORIZON] = np.linspace(200, 170, HORIZON)[:, None] # 하늘
bg[HORIZON:] = 95 + rng.normal(0, 3, (H - HORIZON, W)) # 지면 질감
bg[26:30] = 115 # 외곽 도로
bg[FENCE] = 60; bg[FENCE - 3:FENCE + 1, ::8] = 60 # 울타리·기둥
return bg
def tree(img, t, wind):
"""좌측 수목 수관: 바람 세기에 따라 좌우로 흔들려 오경보를 만든다."""
s = int(round(wind * np.sin(t * 1.3)))
img[8:30, max(0, 2 + s):14 + s] = 55
def make_seq(seed, cond=None, objs=None, wind=None, camo=False):
rng = np.random.default_rng(seed)
cond = cond or str(rng.choice(CONDS))
wind = rng.uniform(0, 2) if wind is None else wind
bg = background(rng)
n = int(rng.integers(1, 4)) if objs is None else len(objs)
kinds = objs or list(rng.choice(CLASSES, n, p=[0.3, 0.15, 0.25, 0.15, 0.15]))
obs = []
for k in kinds:
if k == "person":
o = dict(y=rng.uniform(30, 36), x=rng.uniform(15, 85), vy=rng.uniform(0.8, 1.2), vx=rng.uniform(-0.6, 0.6), h=7, w=3,
val=72 if camo else rng.uniform(35, 50))
elif k == "vehicle":
o = dict(y=rng.uniform(33, 38), x=rng.uniform(10, 30), vy=0.2, vx=rng.uniform(2, 4), h=5, w=10, val=rng.uniform(150, 170))
elif k == "drone":
o = dict(y=rng.uniform(3, 15), x=rng.uniform(10, 80), vy=rng.uniform(0, 0.6), vx=rng.uniform(-2, 2), h=2, w=2 if camo else 3,
val=rng.uniform(80, 120))
elif k == "animal":
o = dict(y=rng.uniform(34, 42), x=rng.uniform(15, 80), vy=0, vx=rng.uniform(-1.5, 1.5), h=2, w=4, val=rng.uniform(60, 75))
else:
o = dict(y=rng.uniform(3, 15), x=rng.uniform(10, 80), vy=0, vx=rng.uniform(-3, 3), h=1, w=2, val=rng.uniform(100, 130))
o["cls"] = str(k); obs.append(o)
frames, labels = [], []
for t in range(-8, T): # 음수 시각 = 객체 없는 배경 학습 구간(사전 관측)
f = bg.copy(); tree(f, t, wind); lab = []
for oi, o in enumerate(obs if t >= 0 else []):
jit = rng.normal(0, 1.2, 2) if o["cls"] in ("animal", "bird") else (0, 0) # 불규칙 움직임
y, x = int(o["y"] + o["vy"] * t + jit[0]), int(o["x"] + o["vx"] * t + jit[1])
y0, x0, y1, x1 = max(y, 0), max(x, 0), min(y + o["h"], H), min(x + o["w"], W)
if y1 > y0 and x1 > x0:
f[y0:y1, x0:x1] = o["val"]
if o["cls"] == "person":
f[max(y0 - 2, 0):y0, x0 + 1:x0 + 2] = o["val"] # 머리
lab.append({"cls": o["cls"], "oid": oi, "bbox": (x0, y0, x1, y1), "intruder": o["cls"] in INTRUDER})
if cond == "night":
f = f * 0.5 + 10 + rng.normal(0, 3.5, f.shape)
else:
f = f + rng.normal(0, 2.5, f.shape)
if cond == "rain":
m = rng.random(f.shape) < 0.006
for d in range(3):
f[d:][m[:H - d]] += 45
if cond == "fog":
f = 0.45 * f + 0.55 * 150
frames.append(np.clip(f, 0, 255)); labels.append(lab)
fr = np.array(frames)
return {"id": f"CAM-{seed:04d}", "cond": cond, "wind": wind, "ref": fr[:8], "frames": fr[8:], "labels": labels[8:]}
# ---------------- 전처리·탐지 공통 함수(배경 차분 기반) ----------------
def enhance(seq):
"""조도 보정(백분위 신장) + 비 조건의 빗줄기 제거(3x3 중앙값). 배경 학습 구간(ref)과 같은 기준으로 보정한다."""
allf = np.concatenate([seq["ref"], seq["frames"]])
f = ndimage.median_filter(allf, size=(1, 3, 3)) if seq["cond"] == "rain" else allf
lo, hi = np.percentile(f, [1, 99.5])
f = np.clip((f - lo) / (hi - lo + 1e-6) * 255, 0, 255)
return f[:len(seq["ref"])], f[len(seq["ref"]):]
def detect(ref, frames, k=4.0, min_area=3, sky_k=None, smooth=False):
"""배경 학습 구간의 중앙값 배경과 차이를 강건 임계값(k x MAD)으로 이진화해 연결 성분을 객체 후보로 만든다.
smooth=True 이면 연속 3프레임 평균으로 시간 잡음을 줄여 저대비 객체를 살린다."""
bg = np.median(ref, axis=0)
if smooth:
frames = ndimage.uniform_filter1d(frames, 3, axis=0, mode="nearest")
out = []
for f in frames:
d = np.abs(f - bg)
mad = np.median(d) * 1.4826 + 1e-6
thr = np.full(d.shape, k * mad)
if sky_k is not None:
thr[:HORIZON] = sky_k * mad
lab, n = ndimage.label(d > thr)
dets = []
for i, sl in enumerate(ndimage.find_objects(lab), start=1):
area = int((lab[sl] == i).sum())
if area >= min_area:
dets.append({"bbox": (sl[1].start, sl[0].start, sl[1].stop, sl[0].stop), "area": area,
"contrast": float(d[sl][lab[sl] == i].mean() / mad)})
out.append(dets)
return out
def center(b):
return ((b[0] + b[2]) / 2, (b[1] + b[3]) / 2)
def match(dets, labels, dist=4.0):
"""탐지와 정답을 중심 거리로 1:1 대응시킨다. 반환: [(det, label or None)], 놓친 label 목록"""
used, pairs = set(), []
for d in dets:
cd = center(d["bbox"]); best, bj = dist, None
for j, l in enumerate(labels):
cl = center(l["bbox"]); e = np.hypot(cd[0] - cl[0], cd[1] - cl[1])
if j not in used and e < best:
best, bj = e, j
if bj is not None:
used.add(bj)
pairs.append((d, labels[bj] if bj is not None else None))
return pairs, [l for j, l in enumerate(labels) if j not in used]
def in_roi(b):
cy = (b[1] + b[3]) / 2
return cy < HORIZON or ROI_ROWS[0] <= cy < ROI_ROWS[1]
def rule_class(b, area):
"""Baseline 규칙: 위치(하늘/지면)·크기·종횡비로 1차 분류(학습 없음, 판단 근거가 그대로 드러난다)"""
w, h = b[2] - b[0], b[3] - b[1]
cy = (b[1] + b[3]) / 2
if cy < HORIZON:
return "drone" if 3 <= area <= 12 else "clutter"
if area >= 30 and w > h:
return "vehicle"
if h >= w and 6 <= area <= 30:
return "person"
return "clutter"
# ---------------- 프레임 간 추적과 궤적 특징 ----------------
def track(dets, gate=6.0):
"""최근접 이웃으로 프레임별 탐지를 궤적으로 잇는다."""
tracks = []
for t, ds in enumerate(dets):
free = list(range(len(ds)))
for tr in tracks:
if tr["last"] != t - 1 or not free:
continue
c0 = center(tr["b"][-1])
dd = [np.hypot(center(ds[i]["bbox"])[0] - c0[0], center(ds[i]["bbox"])[1] - c0[1]) for i in free]
j = int(np.argmin(dd))
if dd[j] < gate:
i = free.pop(j); tr["b"].append(ds[i]["bbox"]); tr["a"].append(ds[i]["area"]); tr["c"].append(ds[i]["contrast"])
tr["t"].append(t); tr["last"] = t
for i in free:
tracks.append({"b": [ds[i]["bbox"]], "a": [ds[i]["area"]], "c": [ds[i]["contrast"]], "t": [t], "last": t})
return tracks
TFEATS = ["n", "area", "aspect", "row", "sky", "speed", "straight", "disp", "contrast", "area_cv", "col"]
def track_features(tr):
c = np.array([center(b) for b in tr["b"]])
wh = np.array([(b[2] - b[0], b[3] - b[1]) for b in tr["b"]])
steps = np.hypot(*np.diff(c, axis=0).T) if len(c) > 1 else np.array([0.0])
disp = float(np.hypot(*(c[-1] - c[0])))
return {"n": len(c), "area": float(np.mean(tr["a"])), "aspect": float(np.mean(wh[:, 1] / wh[:, 0])), "row": float(c[:, 1].mean()),
"sky": float(c[:, 1].mean() < HORIZON), "speed": float(steps.mean()), "straight": disp / (steps.sum() + 1e-6),
"disp": disp, "contrast": float(np.mean(tr["c"])), "area_cv": float(np.std(tr["a"]) / (np.mean(tr["a"]) + 1e-6)),
"col": float(c[:, 0].mean())}
def track_truth(tr, labels):
"""궤적의 각 프레임을 정답과 대응시켜 다수결로 정체를 정한다(없으면 clutter)."""
votes = []
for b, t in zip(tr["b"], tr["t"]):
pairs, _ = match([{"bbox": b}], labels[t])
votes.append((pairs[0][1]["cls"], pairs[0][1]["oid"]) if pairs[0][1] else ("clutter", -1))
return max(set(votes), key=votes.count)
def seq_tracks(seq, k=4.0, min_area=3, sky_k=None, smooth=False):
ref, fr = enhance(seq)
tracks = track(detect(ref, fr, k, min_area, sky_k, smooth))
rows = []
for tr in tracks:
f = track_features(tr); f["truth"], f["oid"] = track_truth(tr, seq["labels"]); f["seq"] = seq["id"]; f["cond"] = seq["cond"]
f["tr"] = tr; rows.append(f)
return rows
ADV = {"k": 3.0, "min_area": 2, "sky_k": 2.5} # 고도화 탐지 파라미터(저대비 객체 대응)
def make_set(seeds, camo_ratio):
return [make_seq(s, camo=(i % 10) < camo_ratio * 10) for i, s in enumerate(seeds)]
train = make_set(range(0, 80), 0.5)
tr = pd.DataFrame(sum([seq_tracks(q, **ADV) for q in train], []))
clf = RandomForestClassifier(n_estimators=150, min_samples_leaf=2, random_state=0).fit(tr[TFEATS], tr["truth"])
def objects(seqs):
return {(q["id"], l["oid"]): l["cls"] for q in seqs for labs in q["labels"] for l in labs if l["intruder"]}
def baseline_found(seqs):
found = set()
for q in seqs:
ref, fr = enhance(q)
hits = {}
for ds, labs in zip(detect(ref, fr), q["labels"]):
for d, l in match(ds, labs)[0]:
if l and l["intruder"] and rule_class(d["bbox"], d["area"]) == l["cls"]:
hits[l["oid"]] = hits.get(l["oid"], 0) + 1
found |= {(q["id"], o) for o, c in hits.items() if c >= 2}
return found
def advanced_found(seqs):
found, ms = set(), []
for q in seqs:
t0 = time.perf_counter()
rows = seq_tracks(q, **ADV)
if rows:
pred = clf.predict(pd.DataFrame(rows)[TFEATS])
ms.append((time.perf_counter() - t0) * 1000 / T)
for r, p in zip(rows, pred if rows else []):
if p in INTRUDER and r["n"] >= 2 and r["oid"] >= 0 and r["truth"] in INTRUDER:
found.add((q["id"], r["oid"]))
return found, ms
res = []
for name, seqs in (("일반", make_set(range(600, 630), 0.0)), ("위장·소형", make_set(range(700, 720), 1.0))):
obj = objects(seqs)
fb = baseline_found(seqs) & set(obj)
fa, ms = advanced_found(seqs)
for cls in ("person", "vehicle", "drone", "전체"):
keys = [k for k, c in obj.items() if cls in (c, "전체")]
if keys:
res.append({"시험셋": name, "정체": cls, "객체 수": len(keys), "Baseline 재현율": np.mean([k in fb for k in keys]),
"고도화 재현율": np.mean([k in fa for k in keys])})
res[-1]["지연(ms/프레임)"] = max(ms)
out = pd.DataFrame(res)
print("[객체 단위 재현율 비교] (2프레임 이상 올바르게 식별되면 탐지로 인정)")
print(out.round(3).fillna("-").to_string(index=False))
rec = out[out["정체"] == "전체"]["고도화 재현율"].min()
lat = out["지연(ms/프레임)"].dropna().max()
imp = pd.Series(clf.feature_importances_, index=TFEATS).sort_values(ascending=False)
print("\n궤적 특징 중요도 상위:", ", ".join(f"{k} {v:.2f}" for k, v in imp.head(5).items()))
print(f"\n복합 위협 탐지 재현율(최저 시험셋) {rec:.1%} (목표 85% 이상) -> {'달성' if rec >= 0.85 else '미달'}")
print(f"처리 지연 최대 {lat:.1f} ms/프레임 (목표 200 ms 이하) -> {'달성' if lat <= 200 else '미달'}")
print("기만 대응: 한 프레임만 나타나는 섬광·빗줄기는 2프레임 이상 지속 조건으로 걸러지며, 최종 확인은 경계병 육안 확인(미션 7)으로 한다.")
In [ ]:
"""수목 흔들림·야생동물·조류로 인한 오경보를 걸러내기 위해 궤적 특징(크기·종횡비·속도·직진성·대비 등)으로 침입 위험 신뢰도 점수와 등급(High/Medium/Low)을 계산하고, 기존 방식 대비 오경보율(FAR) 감축·점수 일관성·ROC-AUC를 검증한다."""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy import ndimage
from sklearn.ensemble import RandomForestClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import roc_auc_score, roc_curve
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler
# ---------------- [교육용가상자료] 합성 CCTV 영상 생성기 ----------------
# 실제 영상 대신 코드로 그린 객체(사각형·점)로 기지 외곽 고정 카메라 화면을 흉내 낸다.
H, W, T = 64, 96, 10 # 화면 크기(px), 시퀀스 길이(프레임)
HORIZON, FENCE = 20, 46 # 하늘/지면 경계, 울타리 행
ROI_ROWS = (32, 48) # 울타리 접근 경계구역(ROI): 지면 32~47행
CLASSES = ["person", "vehicle", "drone", "animal", "bird"]
INTRUDER = {"person", "vehicle", "drone"} # 비인가 접근(경보 대상)
CONDS = ["day", "night", "rain", "fog"]
def background(rng):
bg = np.zeros((H, W))
bg[:HORIZON] = np.linspace(200, 170, HORIZON)[:, None] # 하늘
bg[HORIZON:] = 95 + rng.normal(0, 3, (H - HORIZON, W)) # 지면 질감
bg[26:30] = 115 # 외곽 도로
bg[FENCE] = 60; bg[FENCE - 3:FENCE + 1, ::8] = 60 # 울타리·기둥
return bg
def tree(img, t, wind):
"""좌측 수목 수관: 바람 세기에 따라 좌우로 흔들려 오경보를 만든다."""
s = int(round(wind * np.sin(t * 1.3)))
img[8:30, max(0, 2 + s):14 + s] = 55
def make_seq(seed, cond=None, objs=None, wind=None, camo=False):
rng = np.random.default_rng(seed)
cond = cond or str(rng.choice(CONDS))
wind = rng.uniform(0, 2) if wind is None else wind
bg = background(rng)
n = int(rng.integers(1, 4)) if objs is None else len(objs)
kinds = objs or list(rng.choice(CLASSES, n, p=[0.3, 0.15, 0.25, 0.15, 0.15]))
obs = []
for k in kinds:
if k == "person":
o = dict(y=rng.uniform(30, 36), x=rng.uniform(15, 85), vy=rng.uniform(0.8, 1.2), vx=rng.uniform(-0.6, 0.6), h=7, w=3,
val=72 if camo else rng.uniform(35, 50))
elif k == "vehicle":
o = dict(y=rng.uniform(33, 38), x=rng.uniform(10, 30), vy=0.2, vx=rng.uniform(2, 4), h=5, w=10, val=rng.uniform(150, 170))
elif k == "drone":
o = dict(y=rng.uniform(3, 15), x=rng.uniform(10, 80), vy=rng.uniform(0, 0.6), vx=rng.uniform(-2, 2), h=2, w=2 if camo else 3,
val=rng.uniform(80, 120))
elif k == "animal":
o = dict(y=rng.uniform(34, 42), x=rng.uniform(15, 80), vy=0, vx=rng.uniform(-1.5, 1.5), h=2, w=4, val=rng.uniform(60, 75))
else:
o = dict(y=rng.uniform(3, 15), x=rng.uniform(10, 80), vy=0, vx=rng.uniform(-3, 3), h=1, w=2, val=rng.uniform(100, 130))
o["cls"] = str(k); obs.append(o)
frames, labels = [], []
for t in range(-8, T): # 음수 시각 = 객체 없는 배경 학습 구간(사전 관측)
f = bg.copy(); tree(f, t, wind); lab = []
for oi, o in enumerate(obs if t >= 0 else []):
jit = rng.normal(0, 1.2, 2) if o["cls"] in ("animal", "bird") else (0, 0) # 불규칙 움직임
y, x = int(o["y"] + o["vy"] * t + jit[0]), int(o["x"] + o["vx"] * t + jit[1])
y0, x0, y1, x1 = max(y, 0), max(x, 0), min(y + o["h"], H), min(x + o["w"], W)
if y1 > y0 and x1 > x0:
f[y0:y1, x0:x1] = o["val"]
if o["cls"] == "person":
f[max(y0 - 2, 0):y0, x0 + 1:x0 + 2] = o["val"] # 머리
lab.append({"cls": o["cls"], "oid": oi, "bbox": (x0, y0, x1, y1), "intruder": o["cls"] in INTRUDER})
if cond == "night":
f = f * 0.5 + 10 + rng.normal(0, 3.5, f.shape)
else:
f = f + rng.normal(0, 2.5, f.shape)
if cond == "rain":
m = rng.random(f.shape) < 0.006
for d in range(3):
f[d:][m[:H - d]] += 45
if cond == "fog":
f = 0.45 * f + 0.55 * 150
frames.append(np.clip(f, 0, 255)); labels.append(lab)
fr = np.array(frames)
return {"id": f"CAM-{seed:04d}", "cond": cond, "wind": wind, "ref": fr[:8], "frames": fr[8:], "labels": labels[8:]}
# ---------------- 전처리·탐지 공통 함수(배경 차분 기반) ----------------
def enhance(seq):
"""조도 보정(백분위 신장) + 비 조건의 빗줄기 제거(3x3 중앙값). 배경 학습 구간(ref)과 같은 기준으로 보정한다."""
allf = np.concatenate([seq["ref"], seq["frames"]])
f = ndimage.median_filter(allf, size=(1, 3, 3)) if seq["cond"] == "rain" else allf
lo, hi = np.percentile(f, [1, 99.5])
f = np.clip((f - lo) / (hi - lo + 1e-6) * 255, 0, 255)
return f[:len(seq["ref"])], f[len(seq["ref"]):]
def detect(ref, frames, k=4.0, min_area=3, sky_k=None, smooth=False):
"""배경 학습 구간의 중앙값 배경과 차이를 강건 임계값(k x MAD)으로 이진화해 연결 성분을 객체 후보로 만든다.
smooth=True 이면 연속 3프레임 평균으로 시간 잡음을 줄여 저대비 객체를 살린다."""
bg = np.median(ref, axis=0)
if smooth:
frames = ndimage.uniform_filter1d(frames, 3, axis=0, mode="nearest")
out = []
for f in frames:
d = np.abs(f - bg)
mad = np.median(d) * 1.4826 + 1e-6
thr = np.full(d.shape, k * mad)
if sky_k is not None:
thr[:HORIZON] = sky_k * mad
lab, n = ndimage.label(d > thr)
dets = []
for i, sl in enumerate(ndimage.find_objects(lab), start=1):
area = int((lab[sl] == i).sum())
if area >= min_area:
dets.append({"bbox": (sl[1].start, sl[0].start, sl[1].stop, sl[0].stop), "area": area,
"contrast": float(d[sl][lab[sl] == i].mean() / mad)})
out.append(dets)
return out
def center(b):
return ((b[0] + b[2]) / 2, (b[1] + b[3]) / 2)
def match(dets, labels, dist=4.0):
"""탐지와 정답을 중심 거리로 1:1 대응시킨다. 반환: [(det, label or None)], 놓친 label 목록"""
used, pairs = set(), []
for d in dets:
cd = center(d["bbox"]); best, bj = dist, None
for j, l in enumerate(labels):
cl = center(l["bbox"]); e = np.hypot(cd[0] - cl[0], cd[1] - cl[1])
if j not in used and e < best:
best, bj = e, j
if bj is not None:
used.add(bj)
pairs.append((d, labels[bj] if bj is not None else None))
return pairs, [l for j, l in enumerate(labels) if j not in used]
def in_roi(b):
cy = (b[1] + b[3]) / 2
return cy < HORIZON or ROI_ROWS[0] <= cy < ROI_ROWS[1]
def rule_class(b, area):
"""Baseline 규칙: 위치(하늘/지면)·크기·종횡비로 1차 분류(학습 없음, 판단 근거가 그대로 드러난다)"""
w, h = b[2] - b[0], b[3] - b[1]
cy = (b[1] + b[3]) / 2
if cy < HORIZON:
return "drone" if 3 <= area <= 12 else "clutter"
if area >= 30 and w > h:
return "vehicle"
if h >= w and 6 <= area <= 30:
return "person"
return "clutter"
# ---------------- 프레임 간 추적과 궤적 특징 ----------------
def track(dets, gate=6.0):
"""최근접 이웃으로 프레임별 탐지를 궤적으로 잇는다."""
tracks = []
for t, ds in enumerate(dets):
free = list(range(len(ds)))
for tr in tracks:
if tr["last"] != t - 1 or not free:
continue
c0 = center(tr["b"][-1])
dd = [np.hypot(center(ds[i]["bbox"])[0] - c0[0], center(ds[i]["bbox"])[1] - c0[1]) for i in free]
j = int(np.argmin(dd))
if dd[j] < gate:
i = free.pop(j); tr["b"].append(ds[i]["bbox"]); tr["a"].append(ds[i]["area"]); tr["c"].append(ds[i]["contrast"])
tr["t"].append(t); tr["last"] = t
for i in free:
tracks.append({"b": [ds[i]["bbox"]], "a": [ds[i]["area"]], "c": [ds[i]["contrast"]], "t": [t], "last": t})
return tracks
TFEATS = ["n", "area", "aspect", "row", "sky", "speed", "straight", "disp", "contrast", "area_cv", "col"]
def track_features(tr):
c = np.array([center(b) for b in tr["b"]])
wh = np.array([(b[2] - b[0], b[3] - b[1]) for b in tr["b"]])
steps = np.hypot(*np.diff(c, axis=0).T) if len(c) > 1 else np.array([0.0])
disp = float(np.hypot(*(c[-1] - c[0])))
return {"n": len(c), "area": float(np.mean(tr["a"])), "aspect": float(np.mean(wh[:, 1] / wh[:, 0])), "row": float(c[:, 1].mean()),
"sky": float(c[:, 1].mean() < HORIZON), "speed": float(steps.mean()), "straight": disp / (steps.sum() + 1e-6),
"disp": disp, "contrast": float(np.mean(tr["c"])), "area_cv": float(np.std(tr["a"]) / (np.mean(tr["a"]) + 1e-6)),
"col": float(c[:, 0].mean())}
def track_truth(tr, labels):
"""궤적의 각 프레임을 정답과 대응시켜 다수결로 정체를 정한다(없으면 clutter)."""
votes = []
for b, t in zip(tr["b"], tr["t"]):
pairs, _ = match([{"bbox": b}], labels[t])
votes.append((pairs[0][1]["cls"], pairs[0][1]["oid"]) if pairs[0][1] else ("clutter", -1))
return max(set(votes), key=votes.count)
def seq_tracks(seq, k=4.0, min_area=3, sky_k=None, smooth=False):
ref, fr = enhance(seq)
tracks = track(detect(ref, fr, k, min_area, sky_k, smooth))
rows = []
for tr in tracks:
f = track_features(tr); f["truth"], f["oid"] = track_truth(tr, seq["labels"]); f["seq"] = seq["id"]; f["cond"] = seq["cond"]
f["tr"] = tr; rows.append(f)
return rows
ADV = {"k": 3.0, "min_area": 2, "sky_k": 2.5}
def collect(seeds, wind=None):
rows = []
for s in seeds:
q = make_seq(s, wind=wind)
for r in seq_tracks(q, **ADV):
if r["n"] >= 2 and in_roi(r["tr"]["b"][-1]): # 2프레임 이상 지속 + 경계구역 안 = 경보 후보
r["intr"] = r["truth"] in INTRUDER
rows.append(r)
return rows
def X_of(rows):
return np.array([[r[k] for k in TFEATS] for r in rows])
def grade(p):
return "High" if p >= 0.8 else ("Medium" if p >= 0.5 else "Low")
train = collect(range(0, 100))
test = collect(range(1000, 1060))
windy = collect(range(2000, 2030), wind=2.0) # 강풍 시험셋: 수목 흔들림이 가장 심한 조건
model = make_pipeline(StandardScaler(), LogisticRegression(class_weight="balanced", C=2.0, max_iter=2000, random_state=0))
model.fit(X_of(train), [r["intr"] for r in train])
def evaluate(rows, thr=0.5):
p = model.predict_proba(X_of(rows))[:, 1]
y = np.array([r["intr"] for r in rows])
far0 = (~y).mean() # 기존: 후보 전부 경보 -> 오경보 비율
sel = p >= thr
far1 = (~y[sel]).sum() / max(sel.sum(), 1)
return p, y, far0, far1, y[sel].sum() / max(y.sum(), 1)
out = []
for name, rows in (("일반 시험셋", test), ("강풍 시험셋", windy)):
p, y, far0, far1, rec = evaluate(rows)
out.append({"시험셋": name, "후보 수": len(rows), "기존 FAR": round(far0, 3), "엔진 FAR": round(far1, 3),
"FAR 감축": f"{1 - far1 / far0:.1%}", "침입 재현율": round(rec, 3), "ROC-AUC": round(roc_auc_score(y, p), 3)})
print(pd.DataFrame(out).to_string(index=False))
p, y, far0, far1, rec = evaluate(test)
cut = 1 - far1 / far0
print(f"\n오경보율 감축 {cut:.1%} (목표 40% 이상: {'달성' if cut >= 0.4 else '미달'})")
# 점수 일관성: 특징에 +-3% 측정 흔들림을 주어도 같은 등급이 유지되는 비율
rng = np.random.default_rng(0)
X = X_of(test)
g0 = [grade(v) for v in p]
same = [np.mean([grade(v) == a for v, a in zip(model.predict_proba(X * (1 + rng.normal(0, 0.03, X.shape)))[:, 1], g0)]) for _ in range(10)]
print(f"신뢰도 등급 일관성 {np.mean(same):.1%} (같은 입력 재실행 100%, 목표 90% 이상: {'달성' if np.mean(same) >= 0.9 else '미달'})")
print("\n[실제 정체별 등급 분포] (행: 실제 정체)")
print(pd.crosstab(pd.Series([r["truth"] for r in test], name="실제"), pd.Series(g0, name="등급")).to_string())
print("\n[임계값 민감도]")
for thr in (0.3, 0.5, 0.7, 0.9):
_, _, _, f1_, r1_ = evaluate(test, thr)
print(f"임계값 {thr}: 오경보율 {f1_:.3f}, 침입 재현율 {r1_:.3f}")
# 판단 근거: 점수에 크게 기여한 특징
sc, lr = model[0], model[-1]
i = int(np.argmax(p)); c = (X[i] - sc.mean_) / sc.scale_ * lr.coef_[0]
print(f"\n최고 점수 후보 {test[i]['seq']}({test[i]['truth']}, 점수 {p[i]:.2f}) 근거:", ", ".join(f"{TFEATS[j]}({c[j]:+.2f})" for j in np.argsort(-np.abs(c))[:3]))
print("등급 기준: High(0.8 이상)는 경계병 즉시 확인, Medium(0.5~0.8)은 영상 재확인, Low(0.5 미만)는 기록 후 감시 유지. 점수는 확인 순서를 정하는 보조 정보이며 침입 확정이 아니다.")
fig, ax = plt.subplots(1, 2, figsize=(11, 4))
for cls, col in [("person", "tab:red"), ("vehicle", "tab:orange"), ("drone", "tab:purple"), ("animal", "tab:green"), ("bird", "tab:blue"), ("clutter", "gray")]:
v = [s for s, r in zip(p, test) if r["truth"] == cls]
if v:
ax[0].hist(v, bins=10, range=(0, 1), alpha=0.6, label=cls, color=col)
ax[0].axvline(0.5, color="k", ls="--"); ax[0].set_xlabel("confidence score"); ax[0].set_ylabel("tracks"); ax[0].legend(); ax[0].set_title("Score by true class")
fpr, tpr, _ = roc_curve(y, p); ax[1].plot(fpr, tpr); ax[1].plot([0, 1], [0, 1], "k:"); ax[1].set_title(f"ROC (AUC {roc_auc_score(y, p):.3f})")
ax[1].set_xlabel("false positive rate"); ax[1].set_ylabel("true positive rate")
plt.tight_layout(); plt.show()
In [ ]:
"""6채널 CCTV 합성 영상을 1초 주기로 갱신하는 관제 상황판을 구현한다. 채널별 상태(경보/주시/정상), 침입 객체 궤적(침투 경로), 경보 목록을 한 화면에 모으고 갱신 처리시간을 측정한다."""
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy import ndimage
# ---------------- [교육용가상자료] 합성 CCTV 영상 생성기 ----------------
# 실제 영상 대신 코드로 그린 객체(사각형·점)로 기지 외곽 고정 카메라 화면을 흉내 낸다.
H, W, T = 64, 96, 10 # 화면 크기(px), 시퀀스 길이(프레임)
HORIZON, FENCE = 20, 46 # 하늘/지면 경계, 울타리 행
ROI_ROWS = (32, 48) # 울타리 접근 경계구역(ROI): 지면 32~47행
CLASSES = ["person", "vehicle", "drone", "animal", "bird"]
INTRUDER = {"person", "vehicle", "drone"} # 비인가 접근(경보 대상)
CONDS = ["day", "night", "rain", "fog"]
def background(rng):
bg = np.zeros((H, W))
bg[:HORIZON] = np.linspace(200, 170, HORIZON)[:, None] # 하늘
bg[HORIZON:] = 95 + rng.normal(0, 3, (H - HORIZON, W)) # 지면 질감
bg[26:30] = 115 # 외곽 도로
bg[FENCE] = 60; bg[FENCE - 3:FENCE + 1, ::8] = 60 # 울타리·기둥
return bg
def tree(img, t, wind):
"""좌측 수목 수관: 바람 세기에 따라 좌우로 흔들려 오경보를 만든다."""
s = int(round(wind * np.sin(t * 1.3)))
img[8:30, max(0, 2 + s):14 + s] = 55
def make_seq(seed, cond=None, objs=None, wind=None, camo=False):
rng = np.random.default_rng(seed)
cond = cond or str(rng.choice(CONDS))
wind = rng.uniform(0, 2) if wind is None else wind
bg = background(rng)
n = int(rng.integers(1, 4)) if objs is None else len(objs)
kinds = objs or list(rng.choice(CLASSES, n, p=[0.3, 0.15, 0.25, 0.15, 0.15]))
obs = []
for k in kinds:
if k == "person":
o = dict(y=rng.uniform(30, 36), x=rng.uniform(15, 85), vy=rng.uniform(0.8, 1.2), vx=rng.uniform(-0.6, 0.6), h=7, w=3,
val=72 if camo else rng.uniform(35, 50))
elif k == "vehicle":
o = dict(y=rng.uniform(33, 38), x=rng.uniform(10, 30), vy=0.2, vx=rng.uniform(2, 4), h=5, w=10, val=rng.uniform(150, 170))
elif k == "drone":
o = dict(y=rng.uniform(3, 15), x=rng.uniform(10, 80), vy=rng.uniform(0, 0.6), vx=rng.uniform(-2, 2), h=2, w=2 if camo else 3,
val=rng.uniform(80, 120))
elif k == "animal":
o = dict(y=rng.uniform(34, 42), x=rng.uniform(15, 80), vy=0, vx=rng.uniform(-1.5, 1.5), h=2, w=4, val=rng.uniform(60, 75))
else:
o = dict(y=rng.uniform(3, 15), x=rng.uniform(10, 80), vy=0, vx=rng.uniform(-3, 3), h=1, w=2, val=rng.uniform(100, 130))
o["cls"] = str(k); obs.append(o)
frames, labels = [], []
for t in range(-8, T): # 음수 시각 = 객체 없는 배경 학습 구간(사전 관측)
f = bg.copy(); tree(f, t, wind); lab = []
for oi, o in enumerate(obs if t >= 0 else []):
jit = rng.normal(0, 1.2, 2) if o["cls"] in ("animal", "bird") else (0, 0) # 불규칙 움직임
y, x = int(o["y"] + o["vy"] * t + jit[0]), int(o["x"] + o["vx"] * t + jit[1])
y0, x0, y1, x1 = max(y, 0), max(x, 0), min(y + o["h"], H), min(x + o["w"], W)
if y1 > y0 and x1 > x0:
f[y0:y1, x0:x1] = o["val"]
if o["cls"] == "person":
f[max(y0 - 2, 0):y0, x0 + 1:x0 + 2] = o["val"] # 머리
lab.append({"cls": o["cls"], "oid": oi, "bbox": (x0, y0, x1, y1), "intruder": o["cls"] in INTRUDER})
if cond == "night":
f = f * 0.5 + 10 + rng.normal(0, 3.5, f.shape)
else:
f = f + rng.normal(0, 2.5, f.shape)
if cond == "rain":
m = rng.random(f.shape) < 0.006
for d in range(3):
f[d:][m[:H - d]] += 45
if cond == "fog":
f = 0.45 * f + 0.55 * 150
frames.append(np.clip(f, 0, 255)); labels.append(lab)
fr = np.array(frames)
return {"id": f"CAM-{seed:04d}", "cond": cond, "wind": wind, "ref": fr[:8], "frames": fr[8:], "labels": labels[8:]}
# ---------------- 전처리·탐지 공통 함수(배경 차분 기반) ----------------
def enhance(seq):
"""조도 보정(백분위 신장) + 비 조건의 빗줄기 제거(3x3 중앙값). 배경 학습 구간(ref)과 같은 기준으로 보정한다."""
allf = np.concatenate([seq["ref"], seq["frames"]])
f = ndimage.median_filter(allf, size=(1, 3, 3)) if seq["cond"] == "rain" else allf
lo, hi = np.percentile(f, [1, 99.5])
f = np.clip((f - lo) / (hi - lo + 1e-6) * 255, 0, 255)
return f[:len(seq["ref"])], f[len(seq["ref"]):]
def detect(ref, frames, k=4.0, min_area=3, sky_k=None, smooth=False):
"""배경 학습 구간의 중앙값 배경과 차이를 강건 임계값(k x MAD)으로 이진화해 연결 성분을 객체 후보로 만든다.
smooth=True 이면 연속 3프레임 평균으로 시간 잡음을 줄여 저대비 객체를 살린다."""
bg = np.median(ref, axis=0)
if smooth:
frames = ndimage.uniform_filter1d(frames, 3, axis=0, mode="nearest")
out = []
for f in frames:
d = np.abs(f - bg)
mad = np.median(d) * 1.4826 + 1e-6
thr = np.full(d.shape, k * mad)
if sky_k is not None:
thr[:HORIZON] = sky_k * mad
lab, n = ndimage.label(d > thr)
dets = []
for i, sl in enumerate(ndimage.find_objects(lab), start=1):
area = int((lab[sl] == i).sum())
if area >= min_area:
dets.append({"bbox": (sl[1].start, sl[0].start, sl[1].stop, sl[0].stop), "area": area,
"contrast": float(d[sl][lab[sl] == i].mean() / mad)})
out.append(dets)
return out
def center(b):
return ((b[0] + b[2]) / 2, (b[1] + b[3]) / 2)
def match(dets, labels, dist=4.0):
"""탐지와 정답을 중심 거리로 1:1 대응시킨다. 반환: [(det, label or None)], 놓친 label 목록"""
used, pairs = set(), []
for d in dets:
cd = center(d["bbox"]); best, bj = dist, None
for j, l in enumerate(labels):
cl = center(l["bbox"]); e = np.hypot(cd[0] - cl[0], cd[1] - cl[1])
if j not in used and e < best:
best, bj = e, j
if bj is not None:
used.add(bj)
pairs.append((d, labels[bj] if bj is not None else None))
return pairs, [l for j, l in enumerate(labels) if j not in used]
def in_roi(b):
cy = (b[1] + b[3]) / 2
return cy < HORIZON or ROI_ROWS[0] <= cy < ROI_ROWS[1]
def rule_class(b, area):
"""Baseline 규칙: 위치(하늘/지면)·크기·종횡비로 1차 분류(학습 없음, 판단 근거가 그대로 드러난다)"""
w, h = b[2] - b[0], b[3] - b[1]
cy = (b[1] + b[3]) / 2
if cy < HORIZON:
return "drone" if 3 <= area <= 12 else "clutter"
if area >= 30 and w > h:
return "vehicle"
if h >= w and 6 <= area <= 30:
return "person"
return "clutter"
# ---------------- 프레임 간 추적과 궤적 특징 ----------------
def track(dets, gate=6.0):
"""최근접 이웃으로 프레임별 탐지를 궤적으로 잇는다."""
tracks = []
for t, ds in enumerate(dets):
free = list(range(len(ds)))
for tr in tracks:
if tr["last"] != t - 1 or not free:
continue
c0 = center(tr["b"][-1])
dd = [np.hypot(center(ds[i]["bbox"])[0] - c0[0], center(ds[i]["bbox"])[1] - c0[1]) for i in free]
j = int(np.argmin(dd))
if dd[j] < gate:
i = free.pop(j); tr["b"].append(ds[i]["bbox"]); tr["a"].append(ds[i]["area"]); tr["c"].append(ds[i]["contrast"])
tr["t"].append(t); tr["last"] = t
for i in free:
tracks.append({"b": [ds[i]["bbox"]], "a": [ds[i]["area"]], "c": [ds[i]["contrast"]], "t": [t], "last": t})
return tracks
TFEATS = ["n", "area", "aspect", "row", "sky", "speed", "straight", "disp", "contrast", "area_cv", "col"]
def track_features(tr):
c = np.array([center(b) for b in tr["b"]])
wh = np.array([(b[2] - b[0], b[3] - b[1]) for b in tr["b"]])
steps = np.hypot(*np.diff(c, axis=0).T) if len(c) > 1 else np.array([0.0])
disp = float(np.hypot(*(c[-1] - c[0])))
return {"n": len(c), "area": float(np.mean(tr["a"])), "aspect": float(np.mean(wh[:, 1] / wh[:, 0])), "row": float(c[:, 1].mean()),
"sky": float(c[:, 1].mean() < HORIZON), "speed": float(steps.mean()), "straight": disp / (steps.sum() + 1e-6),
"disp": disp, "contrast": float(np.mean(tr["c"])), "area_cv": float(np.std(tr["a"]) / (np.mean(tr["a"]) + 1e-6)),
"col": float(c[:, 0].mean())}
def track_truth(tr, labels):
"""궤적의 각 프레임을 정답과 대응시켜 다수결로 정체를 정한다(없으면 clutter)."""
votes = []
for b, t in zip(tr["b"], tr["t"]):
pairs, _ = match([{"bbox": b}], labels[t])
votes.append((pairs[0][1]["cls"], pairs[0][1]["oid"]) if pairs[0][1] else ("clutter", -1))
return max(set(votes), key=votes.count)
def seq_tracks(seq, k=4.0, min_area=3, sky_k=None, smooth=False):
ref, fr = enhance(seq)
tracks = track(detect(ref, fr, k, min_area, sky_k, smooth))
rows = []
for tr in tracks:
f = track_features(tr); f["truth"], f["oid"] = track_truth(tr, seq["labels"]); f["seq"] = seq["id"]; f["cond"] = seq["cond"]
f["tr"] = tr; rows.append(f)
return rows
# [교육용가상자료] 6개 채널: 서로 다른 조건·객체 구성의 합성 CCTV
SPECS = [(11, "day", ["person", "drone"]), (12, "night", ["vehicle"]), (13, "rain", ["animal", "person"]),
(14, "fog", ["bird", "drone", "person"]), (15, "day", ["animal"]), (16, "night", ["drone", "vehicle"])]
CH = {f"CH{i + 1}": make_seq(s, cond=c, objs=o, wind=1.5) for i, (s, c, o) in enumerate(SPECS)}
PRE = {name: enhance(q) for name, q in CH.items()}
BG = {name: np.median(ref, axis=0) for name, (ref, fr) in PRE.items()}
STATE = {name: {"dets": []} for name in CH}
def update(t):
"""tick t: 채널마다 새 프레임을 탐지해 궤적을 갱신하고 경보 목록을 만든다."""
alerts, raw = [], 0
for name in CH:
ref, fr = PRE[name]
STATE[name]["dets"] += detect(ref, fr[t:t + 1])
tracks = track(STATE[name]["dets"])
raw += len(STATE[name]["dets"][-1])
STATE[name]["tracks"] = tracks
for i, tr in enumerate(tracks):
if len(tr["b"]) >= 3 and in_roi(tr["b"][-1]) and tr["last"] == t:
f = track_features(tr)
cls = rule_class(tr["b"][-1], tr["a"][-1])
moving = f["disp"] > 4
if cls in INTRUDER and moving:
alerts.append({"채널": name, "분류": cls, "연속 프레임": len(tr["b"]), "위치(x,y)": tuple(int(v) for v in center(tr["b"][-1])), "tr": tr})
return alerts, raw
# 1) 10틱(10초) 갱신 모의: 처리시간·표시 건수 기록
log = []
for t in range(T):
t0 = time.perf_counter()
alerts, raw = update(t)
ms = (time.perf_counter() - t0) * 1000
log.append({"tick": t + 1, "원시 탐지": raw, "경보 항목": len(alerts), "갱신 ms": round(ms, 1)})
df = pd.DataFrame(log)
print(df.to_string(index=False))
print(f"\n6채널 갱신 평균 {df['갱신 ms'].mean():.1f} ms, 최대 {df['갱신 ms'].max():.1f} ms (목표: 갱신 주기 1초 이내 -> {'달성' if df['갱신 ms'].max() < 1000 else '미달'})")
# 2) 인지 부하 감소: 운용자가 봐야 할 항목 수
print(f"인지 부하: 마지막 틱 원시 탐지 {log[-1]['원시 탐지']}건 -> 경보 항목 {log[-1]['경보 항목']}건으로 요약")
# 3) 채널 상태 표(경보 우선 정렬)
board = []
for name in CH:
mine = [a for a in alerts if a["채널"] == name]
watch = sum(1 for tr in STATE[name]["tracks"] if len(tr["b"]) >= 3 and in_roi(tr["b"][-1]) and tr["last"] == T - 1) - len(mine)
board.append({"채널": name, "조건": CH[name]["cond"], "상태": "경보" if mine else ("주시" if watch > 0 else "정상"),
"경보 분류": ",".join(sorted({a["분류"] for a in mine})) or "-", "주시 항목": max(watch, 0)})
board = pd.DataFrame(board)
print("\n[채널 상태판]")
print(board.sort_values("상태", key=lambda s: s.map({"경보": 0, "주시": 1, "정상": 2})).to_string(index=False))
# 4) 렌더링 시간(UI 응답성): 한 화면을 그리는 시간
t0 = time.perf_counter()
col = {"경보": "red", "주시": "gold", "정상": "limegreen"}
EN = {"경보": "ALERT", "주시": "WATCH", "정상": "NORMAL"}
fig, ax = plt.subplots(2, 3, figsize=(13, 6))
for a, name in zip(ax.ravel(), CH):
st = board[board["채널"] == name].iloc[0]
a.imshow(CH[name]["frames"][-1], cmap="gray", vmin=0, vmax=255)
for al in [x for x in alerts if x["채널"] == name]:
c = np.array([center(b) for b in al["tr"]["b"]])
a.plot(c[:, 0], c[:, 1], "-o", color="red", ms=3, lw=2)
a.text(c[-1, 0] + 2, c[-1, 1], al["분류"], color="red", fontsize=8)
for s in a.spines.values():
s.set_edgecolor(col[st["상태"]]); s.set_linewidth(4)
a.set_title(f"{name} [{EN[st['상태']]}] {CH[name]['cond']}", fontsize=10); a.set_xticks([]); a.set_yticks([])
plt.suptitle("Multi-channel CCTV board (synthetic) - red path = intrusion track")
plt.tight_layout(); fig.canvas.draw()
render = (time.perf_counter() - t0) * 1000
print(f"\n한 화면 렌더링 {render:.0f} ms (UI 응답 목표 100 ms: 데이터 갱신 {df['갱신 ms'].mean():.1f} ms는 달성, 렌더링은 실행 환경에 따라 달라짐)")
print("상황판 경보는 판단 보조 정보이며, 출동·경보 해제 결심은 경계병 육안 확인과 지휘계통(미션 7)에 따른다.")
plt.show()
In [ ]:
"""CCTV 침입 경보를 경계병이 육안으로 확인한 뒤 경보 해제·출동 결심 요청·보류를 결정하는 인간 개입(HITL) 승인 워크플로우를 구현한다. AI는 자동으로 해제하거나 출동을 결정하지 않으며, 모든 결정은 담당자 ID·역할·근거와 함께 위·변조 확인이 가능한 감사 로그에 기록된다."""
import hashlib
import json
import pandas as pd
ACTIONS = {"RELEASE": "경보 해제(육안 확인 결과 오경보)", "DISPATCH": "출동 결심 요청(당직 지휘관 승인 대상)", "HOLD": "보류(영상 재확인)"}
ROLES = {"sentry": "경계병", "duty_officer": "당직 지휘관"}
PRIORITY = {"High": 0, "Medium": 1, "Low": 2}
HOLD_LIMIT_S = 60
class WorkflowError(ValueError):
pass
class ReviewQueue:
def __init__(self, clock):
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")}
ok_hash = hashlib.sha256((prev + json.dumps(body, ensure_ascii=False, sort_keys=True)).encode()).hexdigest()
if rec["prev"] != prev or rec["hash"] != ok_hash:
return False, n
prev = rec["hash"]
return True, -1
def submit(self, alert_id, score, grade, camera):
"""AI 경보 등록. 점수와 무관하게 항상 PENDING(검토 대기)으로 시작한다."""
self.items[alert_id] = {"id": alert_id, "camera": camera, "score": score, "grade": grade, "status": "PENDING",
"submitted": self.clock(), "hold_since": None, "escalated": False, "decided": None}
self._record({"event": "SUBMIT", "id": alert_id, "score": score, "by": "AI"})
def pending(self):
todo = [i for i in self.items.values() if i["status"] in ("PENDING", "HOLD", "DISPATCH_REQUESTED")]
return sorted(todo, key=lambda i: (not i["escalated"], PRIORITY[i["grade"]], -i["score"]))
def decide(self, alert_id, action, operator, role, reason):
if action not in ACTIONS:
raise WorkflowError(f"허용되지 않은 조치: {action}")
if role not in ROLES or not operator:
raise WorkflowError("담당자 ID와 역할(sentry/duty_officer) 필요")
item = self.items.get(alert_id)
if item is None:
raise WorkflowError("대기열에 없는 경보")
if item["status"] in ("RELEASED", "DISPATCHED"):
raise WorkflowError("이미 종결된 경보")
if len(reason.strip()) < 5:
raise WorkflowError("판단 근거 5자 이상 필요")
if action == "RELEASE" and item["grade"] == "High" and role != "duty_officer":
raise WorkflowError("High 등급 경보 해제는 당직 지휘관만 가능")
if action == "RELEASE" and item["grade"] != "Low" and len(reason.strip()) < 15:
raise WorkflowError("Medium 이상 해제는 근거 15자 이상 필요(성급한 해제 방지)")
if action == "DISPATCH" and item["status"] == "DISPATCH_REQUESTED" and role != "duty_officer":
raise WorkflowError("출동 최종 승인은 당직 지휘관만 가능")
if action == "DISPATCH" and item["status"] != "DISPATCH_REQUESTED":
item["status"] = "DISPATCH_REQUESTED" # 경계병의 요청 단계: 아직 확정 아님
elif action == "DISPATCH":
item["status"] = "DISPATCHED" # 당직 지휘관 승인 = 상급 지휘계통 보고·출동 지시 요청까지
else:
item["status"] = {"RELEASE": "RELEASED", "HOLD": "HOLD"}[action]
item["hold_since"] = self.clock() if action == "HOLD" else None
item["decided"] = self.clock() if item["status"] in ("RELEASED", "DISPATCHED") else None
self._record({"event": "DECIDE", "id": alert_id, "action": action, "operator": operator, "role": role, "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 = ReviewQueue(clock)
# [교육용가상자료] 미션 5 신뢰도 엔진에서 넘어온 경보 6건
for aid, cam, s, g in [("A01", "CH1", 0.97, "High"), ("A02", "CH4", 0.66, "Medium"), ("A03", "CH3", 0.42, "Low"),
("A04", "CH6", 0.91, "High"), ("A05", "CH2", 0.58, "Medium"), ("A06", "CH5", 0.35, "Low")]:
q.submit(aid, s, g, cam)
print("검토 순서:", [i["id"] for i in q.pending()])
print("자동 해제·자동 출동된 경보 수:", sum(i["status"] in ("RELEASED", "DISPATCHED") for i in q.items.values()))
# 규칙 위반 시도가 모두 차단되는지 확인
tries = [("A01", "RELEASE", "gd01", "sentry", "수목 흔들림으로 보임 확인 완료"), ("A02", "RELEASE", "gd01", "sentry", "동물"),
("A01", "DISPATCH", "", "sentry", "인원 접근 확인"), ("A04", "SHOOT", "gd01", "sentry", "허용되지 않은 조치"),
("A99", "HOLD", "gd01", "sentry", "없는 경보 확인")]
for args in tries:
try:
q.decide(*args)
except WorkflowError as e:
print("차단:", e)
clock.t += 3; q.decide("A01", "DISPATCH", "gd01", "sentry", "울타리 접근 인원 육안 확인, 출동 요청")
clock.t += 4; q.decide("A01", "DISPATCH", "do01", "duty_officer", "영상 재확인 후 출동 승인, 상급 보고")
clock.t += 5; q.decide("A04", "HOLD", "gd02", "sentry", "야간 영상 선명도 낮아 재확인")
clock.t += 4; q.decide("A03", "RELEASE", "gd01", "sentry", "고라니 확인")
clock.t += 3; q.decide("A05", "RELEASE", "gd02", "sentry", "수목 흔들림과 동일 위치에서 반복, 사람 아님을 확대 영상으로 확인")
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)
print("\n" + df[["id", "camera", "grade", "score", "status", "escalated", "대기시간(s)"]].fillna("-").to_string(index=False))
ev = [r for r in q.log if r["event"] == "DECIDE"]
print(f"\n처리 결정 {len(ev)}건 중 로그 기록 {len(ev)}건(성공률 {len(ev) / len(ev):.0%}), 전체 감사 로그 {len(q.log)}건, 해시 체인 검증: {q.verify_log()}")
q.log[7]["operator"] = "gd99" # 위·변조 시도: 8번째 기록의 담당자 ID 변경
print("8번째 기록 변조 후 검증(False, 깨진 위치):", q.verify_log())
print("\n책임 분계: AI는 경보 제안만, 해제는 경계병(High는 당직 지휘관), 출동 승인은 당직 지휘관이다. 실제 출동·대응은 경계 규정과 지휘계통에 따른다.")
In [ ]:
"""야간·비·안개의 악천후에서 CCTV 채널 수를 평시 대비 최대 6배(트래픽 500% 증가)로 늘리고 센서 단절·프레임 손실·전송 손실·처리 오류를 주입해, 시스템 다운 0건·경보 유실 0건과 안전 실패(Fail-safe)·기능 축소 동작을 검증한다."""
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy import ndimage
# ---------------- [교육용가상자료] 합성 CCTV 영상 생성기 ----------------
# 실제 영상 대신 코드로 그린 객체(사각형·점)로 기지 외곽 고정 카메라 화면을 흉내 낸다.
H, W, T = 64, 96, 10 # 화면 크기(px), 시퀀스 길이(프레임)
HORIZON, FENCE = 20, 46 # 하늘/지면 경계, 울타리 행
ROI_ROWS = (32, 48) # 울타리 접근 경계구역(ROI): 지면 32~47행
CLASSES = ["person", "vehicle", "drone", "animal", "bird"]
INTRUDER = {"person", "vehicle", "drone"} # 비인가 접근(경보 대상)
CONDS = ["day", "night", "rain", "fog"]
def background(rng):
bg = np.zeros((H, W))
bg[:HORIZON] = np.linspace(200, 170, HORIZON)[:, None] # 하늘
bg[HORIZON:] = 95 + rng.normal(0, 3, (H - HORIZON, W)) # 지면 질감
bg[26:30] = 115 # 외곽 도로
bg[FENCE] = 60; bg[FENCE - 3:FENCE + 1, ::8] = 60 # 울타리·기둥
return bg
def tree(img, t, wind):
"""좌측 수목 수관: 바람 세기에 따라 좌우로 흔들려 오경보를 만든다."""
s = int(round(wind * np.sin(t * 1.3)))
img[8:30, max(0, 2 + s):14 + s] = 55
def make_seq(seed, cond=None, objs=None, wind=None, camo=False):
rng = np.random.default_rng(seed)
cond = cond or str(rng.choice(CONDS))
wind = rng.uniform(0, 2) if wind is None else wind
bg = background(rng)
n = int(rng.integers(1, 4)) if objs is None else len(objs)
kinds = objs or list(rng.choice(CLASSES, n, p=[0.3, 0.15, 0.25, 0.15, 0.15]))
obs = []
for k in kinds:
if k == "person":
o = dict(y=rng.uniform(30, 36), x=rng.uniform(15, 85), vy=rng.uniform(0.8, 1.2), vx=rng.uniform(-0.6, 0.6), h=7, w=3,
val=72 if camo else rng.uniform(35, 50))
elif k == "vehicle":
o = dict(y=rng.uniform(33, 38), x=rng.uniform(10, 30), vy=0.2, vx=rng.uniform(2, 4), h=5, w=10, val=rng.uniform(150, 170))
elif k == "drone":
o = dict(y=rng.uniform(3, 15), x=rng.uniform(10, 80), vy=rng.uniform(0, 0.6), vx=rng.uniform(-2, 2), h=2, w=2 if camo else 3,
val=rng.uniform(80, 120))
elif k == "animal":
o = dict(y=rng.uniform(34, 42), x=rng.uniform(15, 80), vy=0, vx=rng.uniform(-1.5, 1.5), h=2, w=4, val=rng.uniform(60, 75))
else:
o = dict(y=rng.uniform(3, 15), x=rng.uniform(10, 80), vy=0, vx=rng.uniform(-3, 3), h=1, w=2, val=rng.uniform(100, 130))
o["cls"] = str(k); obs.append(o)
frames, labels = [], []
for t in range(-8, T): # 음수 시각 = 객체 없는 배경 학습 구간(사전 관측)
f = bg.copy(); tree(f, t, wind); lab = []
for oi, o in enumerate(obs if t >= 0 else []):
jit = rng.normal(0, 1.2, 2) if o["cls"] in ("animal", "bird") else (0, 0) # 불규칙 움직임
y, x = int(o["y"] + o["vy"] * t + jit[0]), int(o["x"] + o["vx"] * t + jit[1])
y0, x0, y1, x1 = max(y, 0), max(x, 0), min(y + o["h"], H), min(x + o["w"], W)
if y1 > y0 and x1 > x0:
f[y0:y1, x0:x1] = o["val"]
if o["cls"] == "person":
f[max(y0 - 2, 0):y0, x0 + 1:x0 + 2] = o["val"] # 머리
lab.append({"cls": o["cls"], "oid": oi, "bbox": (x0, y0, x1, y1), "intruder": o["cls"] in INTRUDER})
if cond == "night":
f = f * 0.5 + 10 + rng.normal(0, 3.5, f.shape)
else:
f = f + rng.normal(0, 2.5, f.shape)
if cond == "rain":
m = rng.random(f.shape) < 0.006
for d in range(3):
f[d:][m[:H - d]] += 45
if cond == "fog":
f = 0.45 * f + 0.55 * 150
frames.append(np.clip(f, 0, 255)); labels.append(lab)
fr = np.array(frames)
return {"id": f"CAM-{seed:04d}", "cond": cond, "wind": wind, "ref": fr[:8], "frames": fr[8:], "labels": labels[8:]}
# ---------------- 전처리·탐지 공통 함수(배경 차분 기반) ----------------
def enhance(seq):
"""조도 보정(백분위 신장) + 비 조건의 빗줄기 제거(3x3 중앙값). 배경 학습 구간(ref)과 같은 기준으로 보정한다."""
allf = np.concatenate([seq["ref"], seq["frames"]])
f = ndimage.median_filter(allf, size=(1, 3, 3)) if seq["cond"] == "rain" else allf
lo, hi = np.percentile(f, [1, 99.5])
f = np.clip((f - lo) / (hi - lo + 1e-6) * 255, 0, 255)
return f[:len(seq["ref"])], f[len(seq["ref"]):]
def detect(ref, frames, k=4.0, min_area=3, sky_k=None, smooth=False):
"""배경 학습 구간의 중앙값 배경과 차이를 강건 임계값(k x MAD)으로 이진화해 연결 성분을 객체 후보로 만든다.
smooth=True 이면 연속 3프레임 평균으로 시간 잡음을 줄여 저대비 객체를 살린다."""
bg = np.median(ref, axis=0)
if smooth:
frames = ndimage.uniform_filter1d(frames, 3, axis=0, mode="nearest")
out = []
for f in frames:
d = np.abs(f - bg)
mad = np.median(d) * 1.4826 + 1e-6
thr = np.full(d.shape, k * mad)
if sky_k is not None:
thr[:HORIZON] = sky_k * mad
lab, n = ndimage.label(d > thr)
dets = []
for i, sl in enumerate(ndimage.find_objects(lab), start=1):
area = int((lab[sl] == i).sum())
if area >= min_area:
dets.append({"bbox": (sl[1].start, sl[0].start, sl[1].stop, sl[0].stop), "area": area,
"contrast": float(d[sl][lab[sl] == i].mean() / mad)})
out.append(dets)
return out
def center(b):
return ((b[0] + b[2]) / 2, (b[1] + b[3]) / 2)
def match(dets, labels, dist=4.0):
"""탐지와 정답을 중심 거리로 1:1 대응시킨다. 반환: [(det, label or None)], 놓친 label 목록"""
used, pairs = set(), []
for d in dets:
cd = center(d["bbox"]); best, bj = dist, None
for j, l in enumerate(labels):
cl = center(l["bbox"]); e = np.hypot(cd[0] - cl[0], cd[1] - cl[1])
if j not in used and e < best:
best, bj = e, j
if bj is not None:
used.add(bj)
pairs.append((d, labels[bj] if bj is not None else None))
return pairs, [l for j, l in enumerate(labels) if j not in used]
def in_roi(b):
cy = (b[1] + b[3]) / 2
return cy < HORIZON or ROI_ROWS[0] <= cy < ROI_ROWS[1]
def rule_class(b, area):
"""Baseline 규칙: 위치(하늘/지면)·크기·종횡비로 1차 분류(학습 없음, 판단 근거가 그대로 드러난다)"""
w, h = b[2] - b[0], b[3] - b[1]
cy = (b[1] + b[3]) / 2
if cy < HORIZON:
return "drone" if 3 <= area <= 12 else "clutter"
if area >= 30 and w > h:
return "vehicle"
if h >= w and 6 <= area <= 30:
return "person"
return "clutter"
# ---------------- 프레임 간 추적과 궤적 특징 ----------------
def track(dets, gate=6.0):
"""최근접 이웃으로 프레임별 탐지를 궤적으로 잇는다."""
tracks = []
for t, ds in enumerate(dets):
free = list(range(len(ds)))
for tr in tracks:
if tr["last"] != t - 1 or not free:
continue
c0 = center(tr["b"][-1])
dd = [np.hypot(center(ds[i]["bbox"])[0] - c0[0], center(ds[i]["bbox"])[1] - c0[1]) for i in free]
j = int(np.argmin(dd))
if dd[j] < gate:
i = free.pop(j); tr["b"].append(ds[i]["bbox"]); tr["a"].append(ds[i]["area"]); tr["c"].append(ds[i]["contrast"])
tr["t"].append(t); tr["last"] = t
for i in free:
tracks.append({"b": [ds[i]["bbox"]], "a": [ds[i]["area"]], "c": [ds[i]["contrast"]], "t": [t], "last": t})
return tracks
TFEATS = ["n", "area", "aspect", "row", "sky", "speed", "straight", "disp", "contrast", "area_cv", "col"]
def track_features(tr):
c = np.array([center(b) for b in tr["b"]])
wh = np.array([(b[2] - b[0], b[3] - b[1]) for b in tr["b"]])
steps = np.hypot(*np.diff(c, axis=0).T) if len(c) > 1 else np.array([0.0])
disp = float(np.hypot(*(c[-1] - c[0])))
return {"n": len(c), "area": float(np.mean(tr["a"])), "aspect": float(np.mean(wh[:, 1] / wh[:, 0])), "row": float(c[:, 1].mean()),
"sky": float(c[:, 1].mean() < HORIZON), "speed": float(steps.mean()), "straight": disp / (steps.sum() + 1e-6),
"disp": disp, "contrast": float(np.mean(tr["c"])), "area_cv": float(np.std(tr["a"]) / (np.mean(tr["a"]) + 1e-6)),
"col": float(c[:, 0].mean())}
def track_truth(tr, labels):
"""궤적의 각 프레임을 정답과 대응시켜 다수결로 정체를 정한다(없으면 clutter)."""
votes = []
for b, t in zip(tr["b"], tr["t"]):
pairs, _ = match([{"bbox": b}], labels[t])
votes.append((pairs[0][1]["cls"], pairs[0][1]["oid"]) if pairs[0][1] else ("clutter", -1))
return max(set(votes), key=votes.count)
def seq_tracks(seq, k=4.0, min_area=3, sky_k=None, smooth=False):
ref, fr = enhance(seq)
tracks = track(detect(ref, fr, k, min_area, sky_k, smooth))
rows = []
for tr in tracks:
f = track_features(tr); f["truth"], f["oid"] = track_truth(tr, seq["labels"]); f["seq"] = seq["id"]; f["cond"] = seq["cond"]
f["tr"] = tr; rows.append(f)
return rows
BASE_CH = 6 # 평시 채널 수
FULL_CAPACITY = 20 # 한 틱(1초)에 정밀 처리하는 채널 수(초과분은 기능 축소 모드)
rng_all = np.random.default_rng(8)
ADVERSE = ["night", "rain", "fog"]
POOL = [make_seq(900 + i, cond=ADVERSE[i % 3], objs=list(rng_all.choice(["person", "vehicle", "drone", "animal", "bird"], 3)), wind=1.8)
for i in range(36)] # 다지점 동시 침입: 채널마다 객체 3개
PRE = {i: enhance(q) for i, q in enumerate(POOL)}
def full_mode(ch, t, state):
"""정밀 모드: 배경 차분 + 연속 3프레임 추적 후 경계구역 안 침입 분류만 경보."""
ref, fr = PRE[ch]
state.setdefault("dets", []).extend(detect(ref, fr[t:t + 1]))
out = []
for tr in track(state["dets"]):
if len(tr["b"]) >= 3 and tr["last"] == t and in_roi(tr["b"][-1]) and rule_class(tr["b"][-1], tr["a"][-1]) in INTRUDER and track_features(tr)["disp"] > 4:
out.append(("FULL", rule_class(tr["b"][-1], tr["a"][-1]), id(tr)))
return out
def light_mode(ch, t):
"""기능 축소 모드: 추적 없이 한 프레임 탐지만으로 경보(오경보는 늘어도 미탐을 피하는 안전 실패 방향)."""
ref, fr = PRE[ch]
return [("LIGHT", rule_class(d["bbox"], d["area"]), (ch, t, d["bbox"])) for d in detect(ref, fr[t:t + 1])[0]
if in_roi(d["bbox"]) and rule_class(d["bbox"], d["area"]) in INTRUDER]
def scenario(n_ch, frame_loss=0.0, sensor_out=0, packet_loss=0.0, fault_rate=0.0, seed=0):
rng = np.random.default_rng(seed)
states = [{} for _ in range(n_ch)]
generated, ledger, buffer, down = 0, [], [], 0
tick_ms, degraded, blind, caught = [], 0, 0, 0 # 다운 집계는 호출부에서 처리되지 않은 예외로 센다
dead = set(rng.choice(n_ch, sensor_out, replace=False).tolist()) if sensor_out else set()
for t in range(T):
t0 = time.perf_counter()
for ch in range(n_ch):
alerts = []
try:
if ch in dead or rng.random() < frame_loss: # 프레임 손실·센서 단절
alerts = [("BLIND", "sensor_gap", (ch, t))]; blind += 1 # 모르면 '정상'이 아니라 '감시 공백 경보'
elif rng.random() < fault_rate:
raise RuntimeError("주입된 처리 오류")
elif ch < FULL_CAPACITY:
alerts = full_mode(ch % 36, t, states[ch])
else:
alerts = light_mode(ch % 36, t); degraded += 1
except Exception:
caught += 1; alerts = [("FAULT", "process_error", (ch, t))] # 오류는 격리하고 감시 공백으로 기록
generated += len(alerts)
buffer += alerts
for a in list(buffer): # 전송: 손실 시 버퍼에 남겨 다음 틱 재전송
if rng.random() >= packet_loss:
ledger.append(a); buffer.remove(a)
tick_ms.append((time.perf_counter() - t0) * 1000)
for a in buffer: # 마지막에 남은 건 재전송(수신 확인 후 삭제)
ledger.append(a)
kinds = pd.Series([a[0] for a in ledger]).value_counts().to_dict()
return {"채널": n_ch, "배율": f"x{n_ch / BASE_CH:.0f}", "다운": down, "생성 경보": generated, "기록 경보": len(ledger),
"유실률": f"{(1 - len(ledger) / max(generated, 1)):.0%}", "처리오류 격리": caught, "축소모드 호출": degraded,
"감시공백": blind, "틱 평균 ms": round(float(np.mean(tick_ms)), 1)}, kinds
cases = [("평시 6채널", dict(n_ch=6)), ("x2 12채널", dict(n_ch=12)), ("x4 24채널", dict(n_ch=24)),
("x6 36채널(트래픽 500% 증가)", dict(n_ch=36)),
("x6 + 프레임손실 15% + 전송손실 30%", dict(n_ch=36, frame_loss=0.15, packet_loss=0.3)),
("x6 + 센서두절 4개 + 처리오류 5%", dict(n_ch=36, sensor_out=4, fault_rate=0.05, packet_loss=0.3))]
rows, kinds_log = [], []
for i, (name, kw) in enumerate(cases):
try:
r, kinds = scenario(seed=i, **kw)
except Exception as e: # 처리되지 않은 예외 = 시스템 다운
r, kinds = {"채널": kw["n_ch"], "배율": "-", "다운": 1, "생성 경보": 0, "기록 경보": 0, "유실률": "100%", "처리오류 격리": 0,
"축소모드 호출": 0, "감시공백": 0, "틱 평균 ms": 0}, {}
print("다운:", e)
rows.append({"상황": name, **r}); kinds_log.append(kinds)
df = pd.DataFrame(rows)
print(df.to_string(index=False))
print(f"\n시스템 다운 {df['다운'].sum()}건(목표 0건), 최대 유실률 {df['유실률'].max()}(목표 0%) -> {'달성' if df['다운'].sum() == 0 and df['유실률'].max() == '0%' else '미달'}")
print("기록 경보 유형(최악 상황):", kinds_log[-1])
# 기능 축소 모드의 대가: 채널·틱당 경보 수(운용자 부담)와 침입 객체 프레임 단위 탐지율
def load_and_hit():
f_cnt = l_cnt = hit = tot = 0
for ch in range(36):
st = {}
for t in range(T):
f_cnt += len(full_mode(ch, t, st)); l_cnt += len(light_mode(ch, t))
ref, fr = PRE[ch]
labs = [l for l in POOL[ch]["labels"][t] if l["intruder"]]
ds = [d for d in detect(ref, fr[t:t + 1])[0] if in_roi(d["bbox"]) and rule_class(d["bbox"], d["area"]) in INTRUDER]
pairs, missed = match(ds, labs)
hit += sum(l is not None for _, l in pairs); tot += len(labs)
return f_cnt / (36 * T), l_cnt / (36 * T), hit / max(tot, 1)
fc, lc, hr = load_and_hit()
print(f"\n채널·틱당 경보: 정밀 모드 {fc:.2f}건, 축소 모드 {lc:.2f}건(운용자 부담 {lc / max(fc, 1e-9):.1f}배), 축소 모드 침입 객체 프레임 탐지율 {hr:.1%}")
print("안전 실패 원칙: 처리 용량을 넘으면 정밀도를 낮춰 경보를 더 내는 쪽(축소 모드)으로, 센서 두절·오류는 '정상'이 아니라 '감시 공백 경보'로 처리한다. 경보는 어느 경우에도 버리지 않고 전송 확인 후 삭제한다.")
plt.figure(figsize=(9, 4))
plt.bar(range(len(df)), df["틱 평균 ms"], color=["tab:blue"] * 3 + ["tab:orange"] * 3)
plt.xticks(range(len(df)), [f"case{i + 1}" for i in range(len(df))]); plt.ylabel("ms per tick")
plt.axhline(1000, color="red", ls="--", label="1 s refresh budget"); plt.legend(); plt.title("Stress test: tick latency (orange = 36 channels)")
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
# ---------------- [교육용가상자료] 합성 CCTV 영상 생성기 ----------------
# 실제 영상 대신 코드로 그린 객체(사각형·점)로 기지 외곽 고정 카메라 화면을 흉내 낸다.
H, W, T = 64, 96, 10 # 화면 크기(px), 시퀀스 길이(프레임)
HORIZON, FENCE = 20, 46 # 하늘/지면 경계, 울타리 행
ROI_ROWS = (32, 48) # 울타리 접근 경계구역(ROI): 지면 32~47행
CLASSES = ["person", "vehicle", "drone", "animal", "bird"]
INTRUDER = {"person", "vehicle", "drone"} # 비인가 접근(경보 대상)
CONDS = ["day", "night", "rain", "fog"]
def background(rng):
bg = np.zeros((H, W))
bg[:HORIZON] = np.linspace(200, 170, HORIZON)[:, None] # 하늘
bg[HORIZON:] = 95 + rng.normal(0, 3, (H - HORIZON, W)) # 지면 질감
bg[26:30] = 115 # 외곽 도로
bg[FENCE] = 60; bg[FENCE - 3:FENCE + 1, ::8] = 60 # 울타리·기둥
return bg
def tree(img, t, wind):
"""좌측 수목 수관: 바람 세기에 따라 좌우로 흔들려 오경보를 만든다."""
s = int(round(wind * np.sin(t * 1.3)))
img[8:30, max(0, 2 + s):14 + s] = 55
def make_seq(seed, cond=None, objs=None, wind=None, camo=False):
rng = np.random.default_rng(seed)
cond = cond or str(rng.choice(CONDS))
wind = rng.uniform(0, 2) if wind is None else wind
bg = background(rng)
n = int(rng.integers(1, 4)) if objs is None else len(objs)
kinds = objs or list(rng.choice(CLASSES, n, p=[0.3, 0.15, 0.25, 0.15, 0.15]))
obs = []
for k in kinds:
if k == "person":
o = dict(y=rng.uniform(30, 36), x=rng.uniform(15, 85), vy=rng.uniform(0.8, 1.2), vx=rng.uniform(-0.6, 0.6), h=7, w=3,
val=72 if camo else rng.uniform(35, 50))
elif k == "vehicle":
o = dict(y=rng.uniform(33, 38), x=rng.uniform(10, 30), vy=0.2, vx=rng.uniform(2, 4), h=5, w=10, val=rng.uniform(150, 170))
elif k == "drone":
o = dict(y=rng.uniform(3, 15), x=rng.uniform(10, 80), vy=rng.uniform(0, 0.6), vx=rng.uniform(-2, 2), h=2, w=2 if camo else 3,
val=rng.uniform(80, 120))
elif k == "animal":
o = dict(y=rng.uniform(34, 42), x=rng.uniform(15, 80), vy=0, vx=rng.uniform(-1.5, 1.5), h=2, w=4, val=rng.uniform(60, 75))
else:
o = dict(y=rng.uniform(3, 15), x=rng.uniform(10, 80), vy=0, vx=rng.uniform(-3, 3), h=1, w=2, val=rng.uniform(100, 130))
o["cls"] = str(k); obs.append(o)
frames, labels = [], []
for t in range(-8, T): # 음수 시각 = 객체 없는 배경 학습 구간(사전 관측)
f = bg.copy(); tree(f, t, wind); lab = []
for oi, o in enumerate(obs if t >= 0 else []):
jit = rng.normal(0, 1.2, 2) if o["cls"] in ("animal", "bird") else (0, 0) # 불규칙 움직임
y, x = int(o["y"] + o["vy"] * t + jit[0]), int(o["x"] + o["vx"] * t + jit[1])
y0, x0, y1, x1 = max(y, 0), max(x, 0), min(y + o["h"], H), min(x + o["w"], W)
if y1 > y0 and x1 > x0:
f[y0:y1, x0:x1] = o["val"]
if o["cls"] == "person":
f[max(y0 - 2, 0):y0, x0 + 1:x0 + 2] = o["val"] # 머리
lab.append({"cls": o["cls"], "oid": oi, "bbox": (x0, y0, x1, y1), "intruder": o["cls"] in INTRUDER})
if cond == "night":
f = f * 0.5 + 10 + rng.normal(0, 3.5, f.shape)
else:
f = f + rng.normal(0, 2.5, f.shape)
if cond == "rain":
m = rng.random(f.shape) < 0.006
for d in range(3):
f[d:][m[:H - d]] += 45
if cond == "fog":
f = 0.45 * f + 0.55 * 150
frames.append(np.clip(f, 0, 255)); labels.append(lab)
fr = np.array(frames)
return {"id": f"CAM-{seed:04d}", "cond": cond, "wind": wind, "ref": fr[:8], "frames": fr[8:], "labels": labels[8:]}
# ---------------- 전처리·탐지 공통 함수(배경 차분 기반) ----------------
def enhance(seq):
"""조도 보정(백분위 신장) + 비 조건의 빗줄기 제거(3x3 중앙값). 배경 학습 구간(ref)과 같은 기준으로 보정한다."""
allf = np.concatenate([seq["ref"], seq["frames"]])
f = ndimage.median_filter(allf, size=(1, 3, 3)) if seq["cond"] == "rain" else allf
lo, hi = np.percentile(f, [1, 99.5])
f = np.clip((f - lo) / (hi - lo + 1e-6) * 255, 0, 255)
return f[:len(seq["ref"])], f[len(seq["ref"]):]
def detect(ref, frames, k=4.0, min_area=3, sky_k=None, smooth=False):
"""배경 학습 구간의 중앙값 배경과 차이를 강건 임계값(k x MAD)으로 이진화해 연결 성분을 객체 후보로 만든다.
smooth=True 이면 연속 3프레임 평균으로 시간 잡음을 줄여 저대비 객체를 살린다."""
bg = np.median(ref, axis=0)
if smooth:
frames = ndimage.uniform_filter1d(frames, 3, axis=0, mode="nearest")
out = []
for f in frames:
d = np.abs(f - bg)
mad = np.median(d) * 1.4826 + 1e-6
thr = np.full(d.shape, k * mad)
if sky_k is not None:
thr[:HORIZON] = sky_k * mad
lab, n = ndimage.label(d > thr)
dets = []
for i, sl in enumerate(ndimage.find_objects(lab), start=1):
area = int((lab[sl] == i).sum())
if area >= min_area:
dets.append({"bbox": (sl[1].start, sl[0].start, sl[1].stop, sl[0].stop), "area": area,
"contrast": float(d[sl][lab[sl] == i].mean() / mad)})
out.append(dets)
return out
def center(b):
return ((b[0] + b[2]) / 2, (b[1] + b[3]) / 2)
def match(dets, labels, dist=4.0):
"""탐지와 정답을 중심 거리로 1:1 대응시킨다. 반환: [(det, label or None)], 놓친 label 목록"""
used, pairs = set(), []
for d in dets:
cd = center(d["bbox"]); best, bj = dist, None
for j, l in enumerate(labels):
cl = center(l["bbox"]); e = np.hypot(cd[0] - cl[0], cd[1] - cl[1])
if j not in used and e < best:
best, bj = e, j
if bj is not None:
used.add(bj)
pairs.append((d, labels[bj] if bj is not None else None))
return pairs, [l for j, l in enumerate(labels) if j not in used]
def in_roi(b):
cy = (b[1] + b[3]) / 2
return cy < HORIZON or ROI_ROWS[0] <= cy < ROI_ROWS[1]
def rule_class(b, area):
"""Baseline 규칙: 위치(하늘/지면)·크기·종횡비로 1차 분류(학습 없음, 판단 근거가 그대로 드러난다)"""
w, h = b[2] - b[0], b[3] - b[1]
cy = (b[1] + b[3]) / 2
if cy < HORIZON:
return "drone" if 3 <= area <= 12 else "clutter"
if area >= 30 and w > h:
return "vehicle"
if h >= w and 6 <= area <= 30:
return "person"
return "clutter"
# ---------------- 프레임 간 추적과 궤적 특징 ----------------
def track(dets, gate=6.0):
"""최근접 이웃으로 프레임별 탐지를 궤적으로 잇는다."""
tracks = []
for t, ds in enumerate(dets):
free = list(range(len(ds)))
for tr in tracks:
if tr["last"] != t - 1 or not free:
continue
c0 = center(tr["b"][-1])
dd = [np.hypot(center(ds[i]["bbox"])[0] - c0[0], center(ds[i]["bbox"])[1] - c0[1]) for i in free]
j = int(np.argmin(dd))
if dd[j] < gate:
i = free.pop(j); tr["b"].append(ds[i]["bbox"]); tr["a"].append(ds[i]["area"]); tr["c"].append(ds[i]["contrast"])
tr["t"].append(t); tr["last"] = t
for i in free:
tracks.append({"b": [ds[i]["bbox"]], "a": [ds[i]["area"]], "c": [ds[i]["contrast"]], "t": [t], "last": t})
return tracks
TFEATS = ["n", "area", "aspect", "row", "sky", "speed", "straight", "disp", "contrast", "area_cv", "col"]
def track_features(tr):
c = np.array([center(b) for b in tr["b"]])
wh = np.array([(b[2] - b[0], b[3] - b[1]) for b in tr["b"]])
steps = np.hypot(*np.diff(c, axis=0).T) if len(c) > 1 else np.array([0.0])
disp = float(np.hypot(*(c[-1] - c[0])))
return {"n": len(c), "area": float(np.mean(tr["a"])), "aspect": float(np.mean(wh[:, 1] / wh[:, 0])), "row": float(c[:, 1].mean()),
"sky": float(c[:, 1].mean() < HORIZON), "speed": float(steps.mean()), "straight": disp / (steps.sum() + 1e-6),
"disp": disp, "contrast": float(np.mean(tr["c"])), "area_cv": float(np.std(tr["a"]) / (np.mean(tr["a"]) + 1e-6)),
"col": float(c[:, 0].mean())}
def track_truth(tr, labels):
"""궤적의 각 프레임을 정답과 대응시켜 다수결로 정체를 정한다(없으면 clutter)."""
votes = []
for b, t in zip(tr["b"], tr["t"]):
pairs, _ = match([{"bbox": b}], labels[t])
votes.append((pairs[0][1]["cls"], pairs[0][1]["oid"]) if pairs[0][1] else ("clutter", -1))
return max(set(votes), key=votes.count)
def seq_tracks(seq, k=4.0, min_area=3, sky_k=None, smooth=False):
ref, fr = enhance(seq)
tracks = track(detect(ref, fr, k, min_area, sky_k, smooth))
rows = []
for tr in tracks:
f = track_features(tr); f["truth"], f["oid"] = track_truth(tr, seq["labels"]); f["seq"] = seq["id"]; f["cond"] = seq["cond"]
f["tr"] = tr; rows.append(f)
return rows
TRAIN_SEEDS, OPS_SEEDS = range(0, 80), range(3000, 3060)
def tracks_of(seeds, k, wind=None):
out = []
for s in seeds:
q = make_seq(s, wind=wind)
for r in seq_tracks(q, k=k, min_area=2, sky_k=k - 0.5):
if in_roi(r["tr"]["b"][-1]):
out.append(r)
return out
def X_of(rows):
return np.array([[r[k] for k in TFEATS] for r in rows])
_CACHE = {}
def run_system(k):
if k not in _CACHE:
_CACHE[k] = _run_system(k)
return _CACHE[k]
def _run_system(k):
"""탐지 감도 k로 학습·운용 데이터를 처리하고, 운용 궤적마다 침입 확률(prob)을 붙인다."""
tr, te = tracks_of(TRAIN_SEEDS, k), tracks_of(OPS_SEEDS, k)
m = make_pipeline(StandardScaler(), LogisticRegression(class_weight="balanced", C=2.0, max_iter=2000, random_state=0))
m.fit(X_of(tr), [r["truth"] in INTRUDER for r in tr])
for r, p in zip(te, m.predict_proba(X_of(te))[:, 1]):
r["prob"] = float(p)
return te
def objects():
return {(q["id"], l["oid"]) for q in (make_seq(s) for s in OPS_SEEDS) for labs in q["labels"] for l in labs if l["intruder"]}
OBJS = objects()
def set_alerts(te, thr, min_n):
for r in te:
r["alert"] = r["n"] >= min_n and r["prob"] >= thr
return te
def summarize(te):
al = [r for r in te if r["alert"]]
far = sum(r["truth"] not in INTRUDER for r in al) / max(len(al), 1)
found = {(r["seq"], r["oid"]) for r in al if r["truth"] in INTRUDER}
rec = len(found & OBJS) / len(OBJS)
return {"경보": len(al), "오경보율": round(far, 3), "재현율": round(rec, 3), "종합정확도(F1)": round(2 * (1 - far) * rec / max(1 - far + rec, 1e-9), 3)}
def aar_log(te):
rows = []
for r in te:
if r["alert"] and r["truth"] not in INTRUDER:
where = "수목 구역(x<16)" if r["col"] < 16 else "그 외"
rows.append({"구분": "오경보", "원인": {"clutter": "수목·잡음", "animal": "야생동물", "bird": "조류"}[r["truth"]], "조건": r["cond"], "구역": where, "시퀀스": r["seq"]})
alerted = {(r["seq"], r["oid"]) for r in te if r["alert"] and r["truth"] in INTRUDER}
tracked = {(r["seq"], r["oid"]) for r in te if r["truth"] in INTRUDER}
for o in sorted(OBJS):
if o not in alerted:
rows.append({"구분": "미탐", "원인": "점수 미달·지속 부족" if o in tracked else "추적 실패(탐지 누락)", "조건": "-", "구역": "-", "시퀀스": o[0]})
return pd.DataFrame(rows, columns=["구분", "원인", "조건", "구역", "시퀀스"])
# 1) 초기 설정으로 모의 운용: '놓치지 않기' 우선(낮은 임계값, 짧은 지속 조건)
INIT = {"k": 3.0, "alert_thr": 0.3, "min_frames": 2}
te0 = set_alerts(run_system(INIT["k"]), INIT["alert_thr"], INIT["min_frames"])
before = summarize(te0)
log = aar_log(te0)
print("[1. 무엇이 일어났나] 초기 설정", INIT, "->", before)
print(log.groupby(["구분", "원인"]).size().rename("건수").to_string())
print("\n[2. 왜 일어났나] 오경보 조건 분포:", log[log["구분"] == "오경보"]["조건"].value_counts().to_dict(), "/ 구역:", log[log["구분"] == "오경보"]["구역"].value_counts().to_dict())
# 2) 재보정: 제약(오경보율 15% 이하, 재현율 85% 이상) 안에서 F1 최대 조합 탐색
grid = []
for k in (2.5, 3.0, 3.5):
te = run_system(k)
for thr in (0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8):
for mn in (2, 3, 4):
grid.append({"k": k, "경보임계값": thr, "최소프레임": mn, **summarize(set_alerts(te, thr, mn))})
g = pd.DataFrame(grid)
seen = set(map(tuple, log[log["구분"] == "오경보"][["시퀀스", "원인"]].values))
def repeats(row):
"""AAR에서 식별한 오경보(시퀀스·원인)가 이 설정에서 다시 나오는 건수"""
t = set_alerts(run_system(row["k"]), row["경보임계값"], int(row["최소프레임"]))
return len(seen & set(map(tuple, aar_log(t)[lambda d: d["구분"] == "오경보"][["시퀀스", "원인"]].values)))
ok = g[(g["오경보율"] <= 0.15) & (g["재현율"] >= 0.85)].copy()
ok["재발"] = ok.apply(repeats, axis=1)
ok = ok.sort_values(["재발", "종합정확도(F1)", "재현율"], ascending=[True, False, False]) # 재발 최소 -> F1 최대 순
best = ok.iloc[0]
print("\n[3. 무엇을 바꾸나] 제약 만족 조합 상위 5개(재발 적은 순) (전체 조합 %d개 중 %d개 만족)" % (len(g), len(ok)))
print(ok.head().to_string(index=False))
te1 = set_alerts(run_system(best["k"]), best["경보임계값"], int(best["최소프레임"]))
after = summarize(te1)
log1 = aar_log(te1)
gain = (after["종합정확도(F1)"] - before["종합정확도(F1)"]) * 100
print(f"\n재보정 후: {after}")
print(f"종합정확도 변화 {before['종합정확도(F1)']:.3f} -> {after['종합정확도(F1)']:.3f} ({gain:+.1f}%p, 목표 +10%p 이상: {'달성' if gain >= 10 else '미달'})")
rep = log[log["구분"] == "오경보"].groupby(["시퀀스", "원인"]).size().index.intersection(log1[log1["구분"] == "오경보"].groupby(["시퀀스", "원인"]).size().index)
print(f"AAR에서 식별한 오경보 {log['구분'].eq('오경보').sum()}건 -> 재보정 후 {log1['구분'].eq('오경보').sum()}건, 같은 시퀀스·같은 원인으로 재발한 오류 {len(rep)}건 (목표 0건: {'달성' if len(rep) == 0 else '미달'})")
# 3) 재보정 파라미터(yaml 내용 문자열)
tuned = f"detection_k: {best['k']}\nalert_threshold: {best['경보임계값']}\nmin_track_frames: {int(best['최소프레임'])}\nconstraints: {{max_far: 0.15, min_recall: 0.85}}"
print("\n[재보정 파라미터(sec01_tuned_params.yaml 내용)]\n" + tuned)
print("개선 조치: 탐지 감도 k를 올리고 최소 지속 프레임 조건을 강화해 수목·동물·조류 오경보를 줄이고, 재보정값은 현장 영상으로 재검증 후 적용한다.")
plt.figure(figsize=(6, 4))
for k in (2.5, 3.0, 3.5):
s = g[g["k"] == k]
plt.plot(s["오경보율"], s["재현율"], "o", alpha=0.6, label=f"k={k}")
plt.axvline(0.15, ls="--", c="gray"); plt.axhline(0.85, ls="--", c="gray")
plt.xlabel("false alarm rate"); plt.ylabel("intruder recall"); plt.legend(); plt.title("AAR re-tuning grid (synthetic)")
plt.show()
In [ ]:
"""데이터 검증, 센서 단절 보정·전처리, 객체 탐지·추적, 신뢰도 점수, HITL 검토 대기열, 관제 상황판, AAR 재보정값 적용까지 비인가 접근 탐지 전체 파이프라인을 한 파일로 통합 실행하고, 자체 점검(실행 라인 커버리지 포함)과 시설 방호 브리핑 요약을 출력한다."""
import dis
import hashlib
import json
import sys
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
# ---------------- [교육용가상자료] 합성 CCTV 영상 생성기 ----------------
# 실제 영상 대신 코드로 그린 객체(사각형·점)로 기지 외곽 고정 카메라 화면을 흉내 낸다.
H, W, T = 64, 96, 10 # 화면 크기(px), 시퀀스 길이(프레임)
HORIZON, FENCE = 20, 46 # 하늘/지면 경계, 울타리 행
ROI_ROWS = (32, 48) # 울타리 접근 경계구역(ROI): 지면 32~47행
CLASSES = ["person", "vehicle", "drone", "animal", "bird"]
INTRUDER = {"person", "vehicle", "drone"} # 비인가 접근(경보 대상)
CONDS = ["day", "night", "rain", "fog"]
def background(rng):
bg = np.zeros((H, W))
bg[:HORIZON] = np.linspace(200, 170, HORIZON)[:, None] # 하늘
bg[HORIZON:] = 95 + rng.normal(0, 3, (H - HORIZON, W)) # 지면 질감
bg[26:30] = 115 # 외곽 도로
bg[FENCE] = 60; bg[FENCE - 3:FENCE + 1, ::8] = 60 # 울타리·기둥
return bg
def tree(img, t, wind):
"""좌측 수목 수관: 바람 세기에 따라 좌우로 흔들려 오경보를 만든다."""
s = int(round(wind * np.sin(t * 1.3)))
img[8:30, max(0, 2 + s):14 + s] = 55
def make_seq(seed, cond=None, objs=None, wind=None, camo=False):
rng = np.random.default_rng(seed)
cond = cond or str(rng.choice(CONDS))
wind = rng.uniform(0, 2) if wind is None else wind
bg = background(rng)
n = int(rng.integers(1, 4)) if objs is None else len(objs)
kinds = objs or list(rng.choice(CLASSES, n, p=[0.3, 0.15, 0.25, 0.15, 0.15]))
obs = []
for k in kinds:
if k == "person":
o = dict(y=rng.uniform(30, 36), x=rng.uniform(15, 85), vy=rng.uniform(0.8, 1.2), vx=rng.uniform(-0.6, 0.6), h=7, w=3,
val=72 if camo else rng.uniform(35, 50))
elif k == "vehicle":
o = dict(y=rng.uniform(33, 38), x=rng.uniform(10, 30), vy=0.2, vx=rng.uniform(2, 4), h=5, w=10, val=rng.uniform(150, 170))
elif k == "drone":
o = dict(y=rng.uniform(3, 15), x=rng.uniform(10, 80), vy=rng.uniform(0, 0.6), vx=rng.uniform(-2, 2), h=2, w=2 if camo else 3,
val=rng.uniform(80, 120))
elif k == "animal":
o = dict(y=rng.uniform(34, 42), x=rng.uniform(15, 80), vy=0, vx=rng.uniform(-1.5, 1.5), h=2, w=4, val=rng.uniform(60, 75))
else:
o = dict(y=rng.uniform(3, 15), x=rng.uniform(10, 80), vy=0, vx=rng.uniform(-3, 3), h=1, w=2, val=rng.uniform(100, 130))
o["cls"] = str(k); obs.append(o)
frames, labels = [], []
for t in range(-8, T): # 음수 시각 = 객체 없는 배경 학습 구간(사전 관측)
f = bg.copy(); tree(f, t, wind); lab = []
for oi, o in enumerate(obs if t >= 0 else []):
jit = rng.normal(0, 1.2, 2) if o["cls"] in ("animal", "bird") else (0, 0) # 불규칙 움직임
y, x = int(o["y"] + o["vy"] * t + jit[0]), int(o["x"] + o["vx"] * t + jit[1])
y0, x0, y1, x1 = max(y, 0), max(x, 0), min(y + o["h"], H), min(x + o["w"], W)
if y1 > y0 and x1 > x0:
f[y0:y1, x0:x1] = o["val"]
if o["cls"] == "person":
f[max(y0 - 2, 0):y0, x0 + 1:x0 + 2] = o["val"] # 머리
lab.append({"cls": o["cls"], "oid": oi, "bbox": (x0, y0, x1, y1), "intruder": o["cls"] in INTRUDER})
if cond == "night":
f = f * 0.5 + 10 + rng.normal(0, 3.5, f.shape)
else:
f = f + rng.normal(0, 2.5, f.shape)
if cond == "rain":
m = rng.random(f.shape) < 0.006
for d in range(3):
f[d:][m[:H - d]] += 45
if cond == "fog":
f = 0.45 * f + 0.55 * 150
frames.append(np.clip(f, 0, 255)); labels.append(lab)
fr = np.array(frames)
return {"id": f"CAM-{seed:04d}", "cond": cond, "wind": wind, "ref": fr[:8], "frames": fr[8:], "labels": labels[8:]}
# ---------------- 전처리·탐지 공통 함수(배경 차분 기반) ----------------
def enhance(seq):
"""조도 보정(백분위 신장) + 비 조건의 빗줄기 제거(3x3 중앙값). 배경 학습 구간(ref)과 같은 기준으로 보정한다."""
allf = np.concatenate([seq["ref"], seq["frames"]])
f = ndimage.median_filter(allf, size=(1, 3, 3)) if seq["cond"] == "rain" else allf
lo, hi = np.percentile(f, [1, 99.5])
f = np.clip((f - lo) / (hi - lo + 1e-6) * 255, 0, 255)
return f[:len(seq["ref"])], f[len(seq["ref"]):]
def detect(ref, frames, k=4.0, min_area=3, sky_k=None, smooth=False):
"""배경 학습 구간의 중앙값 배경과 차이를 강건 임계값(k x MAD)으로 이진화해 연결 성분을 객체 후보로 만든다.
smooth=True 이면 연속 3프레임 평균으로 시간 잡음을 줄여 저대비 객체를 살린다."""
bg = np.median(ref, axis=0)
if smooth:
frames = ndimage.uniform_filter1d(frames, 3, axis=0, mode="nearest")
out = []
for f in frames:
d = np.abs(f - bg)
mad = np.median(d) * 1.4826 + 1e-6
thr = np.full(d.shape, k * mad)
if sky_k is not None:
thr[:HORIZON] = sky_k * mad
lab, n = ndimage.label(d > thr)
dets = []
for i, sl in enumerate(ndimage.find_objects(lab), start=1):
area = int((lab[sl] == i).sum())
if area >= min_area:
dets.append({"bbox": (sl[1].start, sl[0].start, sl[1].stop, sl[0].stop), "area": area,
"contrast": float(d[sl][lab[sl] == i].mean() / mad)})
out.append(dets)
return out
def center(b):
return ((b[0] + b[2]) / 2, (b[1] + b[3]) / 2)
def match(dets, labels, dist=4.0):
"""탐지와 정답을 중심 거리로 1:1 대응시킨다. 반환: [(det, label or None)], 놓친 label 목록"""
used, pairs = set(), []
for d in dets:
cd = center(d["bbox"]); best, bj = dist, None
for j, l in enumerate(labels):
cl = center(l["bbox"]); e = np.hypot(cd[0] - cl[0], cd[1] - cl[1])
if j not in used and e < best:
best, bj = e, j
if bj is not None:
used.add(bj)
pairs.append((d, labels[bj] if bj is not None else None))
return pairs, [l for j, l in enumerate(labels) if j not in used]
def in_roi(b):
cy = (b[1] + b[3]) / 2
return cy < HORIZON or ROI_ROWS[0] <= cy < ROI_ROWS[1]
def rule_class(b, area):
"""Baseline 규칙: 위치(하늘/지면)·크기·종횡비로 1차 분류(학습 없음, 판단 근거가 그대로 드러난다)"""
w, h = b[2] - b[0], b[3] - b[1]
cy = (b[1] + b[3]) / 2
if cy < HORIZON:
return "drone" if 3 <= area <= 12 else "clutter"
if area >= 30 and w > h:
return "vehicle"
if h >= w and 6 <= area <= 30:
return "person"
return "clutter"
# ---------------- 프레임 간 추적과 궤적 특징 ----------------
def track(dets, gate=6.0):
"""최근접 이웃으로 프레임별 탐지를 궤적으로 잇는다."""
tracks = []
for t, ds in enumerate(dets):
free = list(range(len(ds)))
for tr in tracks:
if tr["last"] != t - 1 or not free:
continue
c0 = center(tr["b"][-1])
dd = [np.hypot(center(ds[i]["bbox"])[0] - c0[0], center(ds[i]["bbox"])[1] - c0[1]) for i in free]
j = int(np.argmin(dd))
if dd[j] < gate:
i = free.pop(j); tr["b"].append(ds[i]["bbox"]); tr["a"].append(ds[i]["area"]); tr["c"].append(ds[i]["contrast"])
tr["t"].append(t); tr["last"] = t
for i in free:
tracks.append({"b": [ds[i]["bbox"]], "a": [ds[i]["area"]], "c": [ds[i]["contrast"]], "t": [t], "last": t})
return tracks
TFEATS = ["n", "area", "aspect", "row", "sky", "speed", "straight", "disp", "contrast", "area_cv", "col"]
def track_features(tr):
c = np.array([center(b) for b in tr["b"]])
wh = np.array([(b[2] - b[0], b[3] - b[1]) for b in tr["b"]])
steps = np.hypot(*np.diff(c, axis=0).T) if len(c) > 1 else np.array([0.0])
disp = float(np.hypot(*(c[-1] - c[0])))
return {"n": len(c), "area": float(np.mean(tr["a"])), "aspect": float(np.mean(wh[:, 1] / wh[:, 0])), "row": float(c[:, 1].mean()),
"sky": float(c[:, 1].mean() < HORIZON), "speed": float(steps.mean()), "straight": disp / (steps.sum() + 1e-6),
"disp": disp, "contrast": float(np.mean(tr["c"])), "area_cv": float(np.std(tr["a"]) / (np.mean(tr["a"]) + 1e-6)),
"col": float(c[:, 0].mean())}
def track_truth(tr, labels):
"""궤적의 각 프레임을 정답과 대응시켜 다수결로 정체를 정한다(없으면 clutter)."""
votes = []
for b, t in zip(tr["b"], tr["t"]):
pairs, _ = match([{"bbox": b}], labels[t])
votes.append((pairs[0][1]["cls"], pairs[0][1]["oid"]) if pairs[0][1] else ("clutter", -1))
return max(set(votes), key=votes.count)
def seq_tracks(seq, k=4.0, min_area=3, sky_k=None, smooth=False):
ref, fr = enhance(seq)
tracks = track(detect(ref, fr, k, min_area, sky_k, smooth))
rows = []
for tr in tracks:
f = track_features(tr); f["truth"], f["oid"] = track_truth(tr, seq["labels"]); f["seq"] = seq["id"]; f["cond"] = seq["cond"]
f["tr"] = tr; rows.append(f)
return rows
CONFIG = {"detection_k": 3.5, "alert_threshold": 0.6, "min_track_frames": 4} # 미션 9 재보정값
def validate(seq):
errs = []
if seq["cond"] not in CONDS:
errs.append("condition 허용값 아님")
if seq["frames"].ndim != 3 or seq["frames"].shape[1:] != (H, W):
errs.append("해상도 오류")
if not str(seq["id"]).startswith("CAM-"):
errs.append("camera_id 형식 오류")
return errs
def repair(seq):
"""센서 단절(전 화면 0) 프레임을 직전 프레임으로 대체하고 보정 건수를 돌려준다."""
fr = seq["frames"].copy(); fixed = 0
for t in range(len(fr)):
if fr[t].max() < 1:
fr[t] = fr[t - 1] if t > 0 else seq["ref"][-1]; fixed += 1
return {**seq, "frames": fr}, fixed
def X_of(rows):
return np.array([[r[k] for k in TFEATS] for r in rows])
def tracks_of(seq):
k = CONFIG["detection_k"]
return [r for r in seq_tracks(seq, k=k, min_area=2, sky_k=k - 0.5) if in_roi(r["tr"]["b"][-1])]
class Queue:
def __init__(self):
self.items, self.log = [], []
def submit(self, aid, score, grade):
prev = self.log[-1]["hash"] if self.log else "GENESIS"
body = {"event": "SUBMIT", "id": aid, "score": round(score, 3), "by": "AI"}
self.log.append({**body, "prev": prev, "hash": hashlib.sha256((prev + json.dumps(body, sort_keys=True)).encode()).hexdigest()})
self.items.append({"id": aid, "grade": grade, "status": "PENDING"})
def verify(self):
prev = "GENESIS"
for rec in self.log:
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, sort_keys=True)).encode()).hexdigest():
return False
prev = rec["hash"]
return True
def grade_of(p):
return "High" if p >= 0.8 else ("Medium" if p >= CONFIG["alert_threshold"] else "Low")
def process(seq, engine, queue):
"""시퀀스 1개: 단절 보정 -> 탐지·추적 -> 신뢰도 -> 경보 대기열 등록."""
seq, fixed = repair(seq)
trs = tracks_of(seq)
if trs:
for r, p in zip(trs, engine.predict_proba(X_of(trs))[:, 1]):
r["score"], r["grade"] = float(p), grade_of(float(p))
r["alert"] = r["n"] >= CONFIG["min_track_frames"] and r["score"] >= CONFIG["alert_threshold"]
if r["alert"]:
queue.submit(f"{seq['id']}-{len(queue.items):03d}", r["score"], r["grade"])
return trs, fixed
t_start = time.perf_counter()
# ---------------- 1. 학습(엔진 준비) ----------------
train = [r for s in range(0, 160) for r in tracks_of(make_seq(s))]
engine = make_pipeline(StandardScaler(), LogisticRegression(class_weight="balanced", C=2.0, max_iter=2000, random_state=0))
engine.fit(X_of(train), [r["truth"] in INTRUDER for r in train])
# ---------------- 2. 운용 데이터 처리(손상 데이터 2건 주입) ----------------
ops = [make_seq(5000 + i) for i in range(30)]
ops[3] = {**ops[3], "cond": "snow"} # 허용되지 않은 조건
ops[7] = {**ops[7], "frames": np.where((np.arange(T) % 5 == 2)[:, None, None], 0.0, ops[7]["frames"])} # 센서 단절 프레임
queue, rejected, all_tracks, runs, fixed_total = Queue(), {}, [], [], 0
trace_lines = set()
codes = {f.__code__ for f in (validate, repair, tracks_of, grade_of, process, X_of, Queue.submit, Queue.verify,
enhance, detect, match, in_roi, rule_class, track, track_features, track_truth, seq_tracks, center)}
lines_total = {c: {ln for _, ln in dis.findlinestarts(c) if ln != c.co_firstlineno} for c in codes}
def tracer(frame, event, arg):
if frame.f_code in codes:
def local(fr, ev, ar):
if ev == "line":
trace_lines.add((fr.f_code, fr.f_lineno))
return local
return local
sys.settrace(tracer)
for s in ops:
errs = validate(s)
if errs:
rejected[s["id"]] = errs; continue
t0 = time.perf_counter()
trs, fx = process(s, engine, queue)
fixed_total += fx
runs.append({"seq": s["id"], "ms_per_frame": (time.perf_counter() - t0) * 1000 / T, "alerts": sum(r["alert"] for r in trs)})
all_tracks += trs
queue.verify(); grade_of(0.9); X_of(all_tracks[:1])
sys.settrace(None)
covered = sum(len({ln for c2, ln in trace_lines if c2 is c} & lines_total[c]) for c in codes)
total_lines = sum(len(v) for v in lines_total.values())
# ---------------- 3. 성능 평가 ----------------
done_ids = {s["id"] for s in ops} - set(rejected)
objs = {(q["id"], l["oid"]) for q in ops if q["id"] in done_ids for labs in q["labels"] for l in labs if l["intruder"]}
alerts = [r for r in all_tracks if r["alert"]]
far = sum(r["truth"] not in INTRUDER for r in alerts) / max(len(alerts), 1)
recall = len({(r["seq"], r["oid"]) for r in alerts if r["truth"] in INTRUDER} & objs) / max(len(objs), 1)
ms = np.mean([r["ms_per_frame"] for r in runs])
# ---------------- 4. 자체 점검(통합 테스트) ----------------
cov = covered / total_lines
chk = {
"손상 데이터 차단": "CAM-5003" in rejected,
"처리 시퀀스 29개 전량 성공": len(runs) == 29,
"센서 단절 프레임 보정": fixed_total >= 1,
"침입 객체 재현율 85% 이상": recall >= 0.85,
"오경보율 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) and queue.verify(),
"프레임당 50ms 이내": ms < 50,
"실행 라인 커버리지 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)} 통과 ({passed / len(chk):.0%}), 파이프라인 실행 성공 {'예' if len(runs) + len(rejected) == len(ops) else '아니오'}, 핵심 모듈 실행 라인 커버리지 {cov:.1%}")
# ---------------- 5. 관제 상황판 ----------------
focus = max({r["seq"] for r in all_tracks}, key=lambda q: sum(r["alert"] for r in all_tracks if r["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", vmin=0, vmax=255)
for r in [r for r in all_tracks if r["seq"] == focus and r["alert"]]:
c = np.array([center(b) for b in r["tr"]["b"]]); ax[0, 0].plot(c[:, 0], c[:, 1], color=col[r["grade"]], lw=2)
ax[0, 0].set_title(f"{focus} alert tracks (color = grade)"); ax[0, 0].axis("off")
g = pd.Series([r["grade"] for r in alerts]).value_counts().reindex(["High", "Medium"]).fillna(0)
ax[0, 1].bar(g.index, g.values, color=[col[i] for i in g.index]); ax[0, 1].set_title("Alerts by grade (all pending human review)")
ax[1, 0].bar([r["seq"][-2:] for r in runs], [r["alerts"] for r in runs]); ax[1, 0].set_title("Alerts per camera sequence")
ax[1, 1].axis("off")
ax[1, 1].text(0, 0.9, f"Intruder 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("SEC01 integrated pipeline - synthetic training data only"); plt.tight_layout(); plt.show()
# ---------------- 6. 시설 방호 브리핑 요약 ----------------
print(f"""
[기지 CCTV 비인가 접근 탐지 브리핑 요약 - 교육용가상자료]
1. 처리: 시퀀스 {len(ops)}개 중 {len(runs)}개 처리, 손상 데이터 {len(rejected)}건 차단({', '.join(rejected)}), 센서 단절 프레임 {fixed_total}장 보정
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. 한계: 코드로 그린 합성 영상 기준 결과로, 실제 CCTV·기상·배경에서는 재학습과 현장 검증이 필요하다.""")