# SEC01 정답 코드 - 기지 CCTV 드론·비인가 접근 객체 탐지 프로토타입 > [교육용가상자료] 모든 데이터는 코드 안에서 난수로 만든 가상 자료이며 실제 부대·장비·인원·작전 정보와 무관합니다. ## 실행 환경 - Python 3.10 이상, numpy, pandas, scipy, scikit-learn, matplotlib - 미션별 코드는 각각 한 파일로 독립 실행됩니다. 외부 파일 없이 코드 안에서 가상 데이터를 생성합니다. - 그래프의 글자는 한글 글꼴이 없는 실행 환경에서도 깨지지 않도록 영문으로 표기했습니다. 표·출력 결과는 한글입니다. - 난수 시드를 고정해 두어 같은 코드를 실행하면 같은 결과가 나옵니다. ## 미션 1. 군사기지 CCTV 영상 규격 분석 및 비인가 침입 객체(드론/차량/인원) 데이터셋 정의 - 난이도: 초급 / 예상 소요시간: 1.5시간 - 코드 설명: 기지 외곽 CCTV 영상 메타데이터와 비인가 침입 객체(드론·차량·인원) 라벨의 스키마(카메라 규격, 전술 파라미터, 경계구역, 바운딩 박스)를 정의하고, 코드로 그린 합성 영상에서 만든 샘플 20건을 검증한다. ```python """기지 외곽 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()) ``` 실행 결과 예시 ```text 정상 샘플 20건 중 검증 통과 20건, 오류 0건 -> 목표(검증 오류 0건): 달성 [오류 주입 결과] 주입 5종 -> 검출 5건 - camera_id: 형식 불일치 'CAM-1' - camera.fps: 최대값 60 초과 - tactical.comm_delay_ms: 필수 필드 누락 - roi: row_start는 row_end보다 작아야 함 - labels[0].class: 허용되지 않은 값 'ufo' 주입 오류 전부 검출: 예 [데이터 명세 요약] 필드 수 15 그룹 필드 자료형 규칙 기본 camera_id str {'pattern': '^CAM-\\d{4}$'} 기본 n_frames int {'min': 1, 'max': 3000} camera mode str {'enum': ['EO', 'IR']} camera resolution list {'length': 2} camera fps number {'min': 1, 'max': 60} tactical condition str {'enum': ['day', 'night', 'rain', 'fog']} tactical wind_level number {'min': 0, 'max': 10} tactical position_error_m number {'min': 0, 'max': 100} tactical comm_delay_ms number {'min': 0, 'max': 10000} roi row_start int {'min': 0, 'max': 200} roi row_end int {'min': 1, 'max': 200} label class str {'enum': ['person', 'vehicle', 'drone', 'animal', 'bird']} label bbox list {'length': 4} label frame int {'min': 0} label intruder bool {} 침입 대상 클래스: ['drone', 'person', 'vehicle'] / 오경보 유발 클래스: animal, bird, 수목 흔들림(clutter) 클래스별 라벨 수: {'person': 57, 'vehicle': 27, 'drone': 21, 'bird': 15, 'animal': 15} ``` ## 미션 2. 주/야간 조도 변화 보정, 기상 잡음 필터링 및 침입 경계구역(ROI) 설정 전처리 구현 - 난이도: 초급~중급 / 예상 소요시간: 2.0시간 - 코드 설명: 주/야간 조도 보정, 비·잡음 필터, 센서 단절 프레임 보정, 울타리 경계구역(ROI) 마스크를 묶은 CCTV 전처리 모듈을 구현하고, 처리 속도·단위 테스트 5종·조건별 전처리 전후 탐지 성능을 측정한다. ```python """주/야간 조도 보정, 비·잡음 필터, 센서 단절 프레임 보정, 울타리 경계구역(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() ``` 실행 결과 예시 ```text 전처리 720프레임 / 0.38초 = 1,890프레임/초 (목표 1,000 이상: 달성) 전처리 오류(NaN·범위 이탈) 0건 -> 오류율 0% (목표 0%) [PASS] 단절 프레임 보정 [PASS] NaN 프레임 보정 [PASS] 야간 밝기 범위 [PASS] ROI 마스크 경계 [PASS] 출력 형상 보존 단위 테스트 5/5 통과 [조건별 전처리 전후 탐지 F1] (전: 원본 그대로 / 후: 단절 보정+조도 보정+필터 / ROI: 경계구역 밖 탐지 제외) 조건 시퀀스 전 F1 후 F1 후 재현율 후+ROI 정밀도 후+ROI F1 day 13 0.623 0.639 0.956 0.493 0.649 night 14 0.484 0.493 0.868 0.488 0.625 rain 7 0.083 0.303 0.792 0.343 0.479 fog 6 0.389 0.387 0.940 0.247 0.390 전체 40 0.247 0.463 0.886 0.419 0.568 전술 노이즈 대응: 센서 단절 프레임은 직전 프레임으로 대체, 비는 3x3 중앙값, 야간 조도는 백분위 신장으로 보정하며 스푸핑 영상 판별은 본 과제 범위 밖이다. ``` ## 미션 3. 기지 외곽 울타리 접근 객체 탐지 Baseline 비전 모델 구현 - 난이도: 중급 / 예상 소요시간: 2.5시간 - 코드 설명: 기지 외곽 고정 CCTV 합성 영상에서 배경 차분과 위치·크기·종횡비 규칙으로 울타리 접근 객체(인원·차량·드론)를 1차 감지하는 Baseline 비전 모델을 구현하고, 프레임 단위 탐지 정확도와 오경보를 측정한다. ```python """기지 외곽 고정 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() ``` 실행 결과 예시 ```text [Baseline] 합성 CCTV 시퀀스 40개(프레임 400장), 침입 객체 출현 530회 침입 객체 탐지·분류 정확도 91.7% (목표 75% 이상) -> 달성 침입 경보 발생률 90.6%, 오경보 109건 (원인: {'clutter': 107, 'animal': 2}) [정체별 Baseline 판정] (행: 실제, 열: 판정) pred clutter drone missed person vehicle truth animal 92 0 16 2 0 bird 0 0 110 0 0 clutter 901 94 0 93 0 drone 0 134 36 0 0 person 6 0 1 243 0 vehicle 0 0 1 0 109 [조건별 침입 객체 정확도] cond day 99.0% (100회) fog 100.0% (130회) night 99.1% (110회) rain 77.9% (190회) 한계: 규칙은 위장 인원·소형 드론(저대비·소면적)과 수목·동물 오경보에 약하다 -> 미션 4·5에서 보완한다. ``` ## 미션 4. 저고도 침투 소형 드론 및 은폐 침입자 정밀 식별 복합 비전 모델 고도화 - 난이도: 중급 / 예상 소요시간: 3.0시간 - 코드 설명: 저고도 소형 드론과 위장(저대비) 침입자를 식별하도록, 하늘 영역 민감도 상향·프레임 간 추적·궤적 특징 기반 랜덤포레스트 분류를 결합한 복합 탐지 모델로 고도화하고, 객체 단위 재현율과 프레임당 처리 지연을 Baseline과 비교한다. ```python """저고도 소형 드론과 위장(저대비) 침입자를 식별하도록, 하늘 영역 민감도 상향·프레임 간 추적·궤적 특징 기반 랜덤포레스트 분류를 결합한 복합 탐지 모델로 고도화하고, 객체 단위 재현율과 프레임당 처리 지연을 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)으로 한다.") ``` 실행 결과 예시 ```text [객체 단위 재현율 비교] (2프레임 이상 올바르게 식별되면 탐지로 인정) 시험셋 정체 객체 수 Baseline 재현율 고도화 재현율 지연(ms/프레임) 일반 person 18 1.000 1.000 - 일반 vehicle 9 1.000 1.000 - 일반 drone 21 0.952 1.000 - 일반 전체 48 0.979 1.000 9.738 위장·소형 person 11 0.727 0.909 - 위장·소형 vehicle 7 1.000 0.857 - 위장·소형 drone 9 0.889 1.000 - 위장·소형 전체 27 0.852 0.926 25.494 궤적 특징 중요도 상위: n 0.18, disp 0.14, row 0.12, area 0.10, aspect 0.10 복합 위협 탐지 재현율(최저 시험셋) 92.6% (목표 85% 이상) -> 달성 처리 지연 최대 25.5 ms/프레임 (목표 200 ms 이하) -> 달성 기만 대응: 한 프레임만 나타나는 섬광·빗줄기는 2프레임 이상 지속 조건으로 걸러지며, 최종 확인은 경계병 육안 확인(미션 7)으로 한다. ``` ## 미션 5. 수목 흔들림/야생동물 오경보 필터링 및 침입 위험 신뢰도 점수 엔진 개발 - 난이도: 중급~고급 / 예상 소요시간: 3.0시간 - 코드 설명: 수목 흔들림·야생동물·조류로 인한 오경보를 걸러내기 위해 궤적 특징(크기·종횡비·속도·직진성·대비 등)으로 침입 위험 신뢰도 점수와 등급(High/Medium/Low)을 계산하고, 기존 방식 대비 오경보율(FAR) 감축·점수 일관성·ROC-AUC를 검증한다. ```python """수목 흔들림·야생동물·조류로 인한 오경보를 걸러내기 위해 궤적 특징(크기·종횡비·속도·직진성·대비 등)으로 침입 위험 신뢰도 점수와 등급(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() ``` 실행 결과 예시 ```text 시험셋 후보 수 기존 FAR 엔진 FAR FAR 감축 침입 재현율 ROC-AUC 일반 시험셋 553 0.870 0.162 81.3% 0.931 0.984 강풍 시험셋 194 0.784 0.162 79.3% 0.738 0.931 오경보율 감축 81.3% (목표 40% 이상: 달성) 신뢰도 등급 일관성 99.0% (같은 입력 재실행 100%, 목표 90% 이상: 달성) [실제 정체별 등급 분포] (행: 실제 정체) 등급 High Low Medium 실제 animal 0 25 0 bird 4 10 2 clutter 2 433 5 drone 30 1 0 person 25 2 1 vehicle 10 2 1 [임계값 민감도] 임계값 0.3: 오경보율 0.272, 침입 재현율 0.931 임계값 0.5: 오경보율 0.163, 침입 재현율 0.931 임계값 0.7: 오경보율 0.132, 침입 재현율 0.917 임계값 0.9: 오경보율 0.016, 침입 재현율 0.847 최고 점수 후보 CAM-1034(drone, 점수 1.00) 근거: n(+4.27), row(+2.23), straight(+1.85) 등급 기준: High(0.8 이상)는 경계병 즉시 확인, Medium(0.5~0.8)은 영상 재확인, Low(0.5 미만)는 기록 후 감시 유지. 점수는 확인 순서를 정하는 보조 정보이며 침입 확정이 아니다. ``` ## 미션 6. CCTV 다채널 실시간 침입 경보 및 침투 경로 추적 대시보드 구현 - 난이도: 중급 / 예상 소요시간: 2.5시간 - 코드 설명: 6채널 CCTV 합성 영상을 1초 주기로 갱신하는 관제 상황판을 구현한다. 채널별 상태(경보/주시/정상), 침입 객체 궤적(침투 경로), 경보 목록을 한 화면에 모으고 갱신 처리시간을 측정한다. ```python """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() ``` 실행 결과 예시 ```text tick 원시 탐지 경보 항목 갱신 ms 1 17 0 16.8 2 29 0 17.2 3 33 2 20.6 4 26 3 24.2 5 31 4 22.4 6 15 6 21.1 7 31 6 22.2 8 21 7 22.0 9 28 6 22.7 10 32 6 23.6 6채널 갱신 평균 21.3 ms, 최대 24.2 ms (목표: 갱신 주기 1초 이내 -> 달성) 인지 부하: 마지막 틱 원시 탐지 32건 -> 경보 항목 6건으로 요약 [채널 상태판] 채널 조건 상태 경보 분류 주시 항목 CH1 day 경보 person 1 CH2 night 경보 vehicle 0 CH4 fog 경보 drone,person 0 CH6 night 경보 drone,vehicle 0 CH3 rain 주시 - 3 CH5 day 주시 - 1 한 화면 렌더링 246 ms (UI 응답 목표 100 ms: 데이터 갱신 21.3 ms는 달성, 렌더링은 실행 환경에 따라 달라짐) 상황판 경보는 판단 보조 정보이며, 출동·경보 해제 결심은 경계병 육안 확인과 지휘계통(미션 7)에 따른다. ``` ## 미션 7. 경계병 육안 확인 및 경보 해제/출동 결심을 위한 HITL 승인 워크플로우 설계 - 난이도: 고급 / 예상 소요시간: 3.0시간 - 코드 설명: CCTV 침입 경보를 경계병이 육안으로 확인한 뒤 경보 해제·출동 결심 요청·보류를 결정하는 인간 개입(HITL) 승인 워크플로우를 구현한다. AI는 자동으로 해제하거나 출동을 결정하지 않으며, 모든 결정은 담당자 ID·역할·근거와 함께 위·변조 확인이 가능한 감사 로그에 기록된다. ```python """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는 당직 지휘관), 출동 승인은 당직 지휘관이다. 실제 출동·대응은 경계 규정과 지휘계통에 따른다.") ``` 실행 결과 예시 ```text 검토 순서: ['A01', 'A04', 'A02', 'A05', 'A03', 'A06'] 자동 해제·자동 출동된 경보 수: 0 차단: High 등급 경보 해제는 당직 지휘관만 가능 차단: 판단 근거 5자 이상 필요 차단: 담당자 ID와 역할(sentry/duty_officer) 필요 차단: 허용되지 않은 조치: SHOOT 차단: 대기열에 없는 경보 보류 기한 초과 상신: ['A04'] 상신 후 검토 순서: ['A04', 'A02', 'A06'] id camera grade score status escalated 대기시간(s) A01 CH1 High 0.97 DISPATCHED False 7.0 A02 CH4 Medium 0.66 PENDING False - A03 CH3 Low 0.42 RELEASED False 16.0 A04 CH6 High 0.91 HOLD True - A05 CH2 Medium 0.58 RELEASED False 19.0 A06 CH5 Low 0.35 PENDING False - 처리 결정 5건 중 로그 기록 5건(성공률 100%), 전체 감사 로그 12건, 해시 체인 검증: (True, -1) 8번째 기록 변조 후 검증(False, 깨진 위치): (False, 8) 책임 분계: AI는 경보 제안만, 해제는 경계병(High는 당직 지휘관), 출동 승인은 당직 지휘관이다. 실제 출동·대응은 경계 규정과 지휘계통에 따른다. ``` ## 미션 8. 야간 악천후 및 다지점 동시 침입 상황 모의 CCTV 영상 스트레스 테스트 - 난이도: 고급 / 예상 소요시간: 3.5시간 - 코드 설명: 야간·비·안개의 악천후에서 CCTV 채널 수를 평시 대비 최대 6배(트래픽 500% 증가)로 늘리고 센서 단절·프레임 손실·전송 손실·처리 오류를 주입해, 시스템 다운 0건·경보 유실 0건과 안전 실패(Fail-safe)·기능 축소 동작을 검증한다. ```python """야간·비·안개의 악천후에서 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() ``` 실행 결과 예시 ```text 상황 채널 배율 다운 생성 경보 기록 경보 유실률 처리오류 격리 축소모드 호출 감시공백 틱 평균 ms 평시 6채널 6 x1 0 54 54 0% 0 0 0 15.4 x2 12채널 12 x2 0 120 120 0% 0 0 0 35.2 x4 24채널 24 x4 0 271 271 0% 0 40 0 71.9 x6 36채널(트래픽 500% 증가) 36 x6 0 521 521 0% 0 160 0 102.7 x6 + 프레임손실 15% + 전송손실 30% 36 x6 0 422 422 0% 0 135 49 84.1 x6 + 센서두절 4개 + 처리오류 5% 36 x6 0 444 444 0% 19 126 40 90.1 시스템 다운 0건(목표 0건), 최대 유실률 0%(목표 0%) -> 달성 기록 경보 유형(최악 상황): {'LIGHT': 256, 'FULL': 129, 'BLIND': 40, 'FAULT': 19} 채널·틱당 경보: 정밀 모드 0.97건, 축소 모드 2.07건(운용자 부담 2.1배), 축소 모드 침입 객체 프레임 탐지율 81.1% 안전 실패 원칙: 처리 용량을 넘으면 정밀도를 낮춰 경보를 더 내는 쪽(축소 모드)으로, 센서 두절·오류는 '정상'이 아니라 '감시 공백 경보'로 처리한다. 경보는 어느 경우에도 버리지 않고 전송 확인 후 삭제한다. ``` ## 미션 9. AAR 기반 오경보 유발 객체 분석 및 경계구역 탐지 감도 파라미터 재보정 - 난이도: 고급 / 예상 소요시간: 3.0시간 - 코드 설명: AAR(사후검토) 일지 형식으로 모의 운용의 오경보·미탐을 원인별(수목 흔들림·동물·조류·추적 실패 등)로 분류하고, 탐지 감도 k·최소 지속 프레임·경보 임계값을 제약 조건 안에서 격자 탐색으로 재보정해 개선 전후 성능과 재발 여부를 비교한다. ```python """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() ``` 실행 결과 예시 ```text [1. 무엇이 일어났나] 초기 설정 {'k': 3.0, 'alert_thr': 0.3, 'min_frames': 2} -> {'경보': 135, '오경보율': 0.43, '재현율': 0.963, '종합정확도(F1)': 0.716} 구분 원인 미탐 점수 미달·지속 부족 2 추적 실패(탐지 누락) 1 오경보 수목·잡음 31 야생동물 13 조류 14 [2. 왜 일어났나] 오경보 조건 분포: {'rain': 37, 'fog': 10, 'night': 8, 'day': 3} / 구역: {'그 외': 43, '수목 구역(x<16)': 15} [3. 무엇을 바꾸나] 제약 만족 조합 상위 5개(재발 적은 순) (전체 조합 63개 중 9개 만족) k 경보임계값 최소프레임 경보 오경보율 재현율 종합정확도(F1) 재발 3.5 0.6 4 84 0.083 0.963 0.939 7 3.5 0.7 4 83 0.084 0.950 0.933 7 3.5 0.8 4 82 0.085 0.938 0.926 7 3.5 0.7 3 84 0.095 0.950 0.927 8 3.5 0.8 3 83 0.096 0.938 0.920 8 재보정 후: {'경보': 84, '오경보율': 0.083, '재현율': 0.963, '종합정확도(F1)': 0.939} 종합정확도 변화 0.716 -> 0.939 (+22.3%p, 목표 +10%p 이상: 달성) AAR에서 식별한 오경보 58건 -> 재보정 후 7건, 같은 시퀀스·같은 원인으로 재발한 오류 7건 (목표 0건: 미달) [재보정 파라미터(sec01_tuned_params.yaml 내용)] detection_k: 3.5 alert_threshold: 0.6 min_track_frames: 4 constraints: {max_far: 0.15, min_recall: 0.85} 개선 조치: 탐지 감도 k를 올리고 최소 지속 프레임 조건을 강화해 수목·동물·조류 오경보를 줄이고, 재보정값은 현장 영상으로 재검증 후 적용한다. ``` ## 미션 10. 기지 CCTV 비인가 접근 탐지 시스템 통합 및 시설 방호 브리핑 패키지 완성 - 난이도: 최고급 / 예상 소요시간: 3.5시간 - 코드 설명: 데이터 검증, 센서 단절 보정·전처리, 객체 탐지·추적, 신뢰도 점수, HITL 검토 대기열, 관제 상황판, AAR 재보정값 적용까지 비인가 접근 탐지 전체 파이프라인을 한 파일로 통합 실행하고, 자체 점검(실행 라인 커버리지 포함)과 시설 방호 브리핑 요약을 출력한다. ```python """데이터 검증, 센서 단절 보정·전처리, 객체 탐지·추적, 신뢰도 점수, 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·기상·배경에서는 재학습과 현장 검증이 필요하다.""") ``` 실행 결과 예시 ```text [PASS] 손상 데이터 차단 [PASS] 처리 시퀀스 29개 전량 성공 [PASS] 센서 단절 프레임 보정 [PASS] 침입 객체 재현율 85% 이상 [PASS] 오경보율 15% 이하 [PASS] 자동 해제·출동 0건 [PASS] 경보 전량 대기열 등록 [PASS] 감사 로그 건수 일치·무결 [PASS] 프레임당 50ms 이내 [PASS] 실행 라인 커버리지 85% 이상 자체 점검 10/10 통과 (100%), 파이프라인 실행 성공 예, 핵심 모듈 실행 라인 커버리지 89.1% [기지 CCTV 비인가 접근 탐지 브리핑 요약 - 교육용가상자료] 1. 처리: 시퀀스 30개 중 29개 처리, 손상 데이터 1건 차단(CAM-5003), 센서 단절 프레임 2장 보정 2. 성능: 침입 객체 재현율 90.5%, 오경보율 11.6%, 프레임당 3.48 ms, 전체 소요 6.5 s 3. 경보: 43건 경계병 육안 확인 대기 (High 41 / Medium 2), 자동 해제·출동 0건 4. 운용 유의: 경보 해제·출동 결심은 경계병과 당직 지휘관의 승인 후에만 이루어지며, 실제 대응은 경계 규정과 지휘계통에 따른다. 5. 한계: 코드로 그린 합성 영상 기준 결과로, 실제 CCTV·기상·배경에서는 재학습과 현장 검증이 필요하다. ```