miliai/pbl02
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
pbl02/uploads/C204/C204_solution.ipynb
T
KIM SU BO fbf80e7308 jupiter notebook
2026-10-08 17:15:05 +09:00

3549 lines
232 KiB
Plaintext
Raw Permalink Blame History

{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# 🎯 [C204] 무전 음성 스푸핑·딥페이크 탐지 실험\n",
"### Google Colab 실습 환경 (지휘통제(C2) / Level 3)\n",
"\n",
"[![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/edunlab/milicontent2/blob/main/uploads/C204/C204_solution.ipynb)\n",
"\n",
"---\n",
"\n",
"## 📌 1. 프로젝트 개요 및 배경\n",
"\n",
"| 항목 | 내용 |\n",
"| :--- | :--- |\n",
"| **과정 분류** | 지휘통제(C2) (Level 3) |\n",
"| **학습 소요 시간** | 10개 미션 (5일 / 총 27.5시간) |\n",
"| **환경 사양** | Python 3.10+, numpy, pandas, matplotlib, scipy, scikit-learn |\n",
"\n",
"### 📖 프로젝트 목표\n",
"러시아-우크라이나 전쟁에서는 지휘관 명의의 위조 음성·영상과 무전 기만이 실제로 보고되면서, 생성형 AI 시대에 '들리는 목소리가 정말 그 지휘관인가'를 확인하는 일이 지휘망 보안의 새로운 과제가 되었다. 본 프로젝트에서는 실제 음성 대신 탐지 학습용 합성 신호(배음 신호에 진짜 음성의 자연스러운 흔들림과 문헌상 알려진 위조 흔적을 통계적으로 표시한 추상 신호)를 코드로 만들어, 데이터셋 규격 설계, 무전 채널 잡음 모델링과 MFCC·스펙트로그램 전처리, 규칙 기반 Baseline, 신경망 판별 엔진, 진위 신뢰도 점수, C2 보안 상황판, 통신보안관 2차 검증(HITL), 전자전 재밍 스트레스 시험, AAR 재보정, 최종 통합까지 위조 음성 탐지 파이프라인을 단계별로 구현한다. 음성을 생성하거나 변조하는 공격 기법은 다루지 않으며, 오직 수신 측의 탐지·검증 기능만 구현한다. 최종 목표는 잡음이 심한 무전 환경에서도 위조 명령을 놓치지 않고 오경보를 줄이며, 모든 긴급 명령의 진위 확정이 사람의 인증·청음 확인을 거치도록 설계하는 것이다.\n",
"\n",
"> **⚠️ [교육용 가상 데이터셋 원칙]**\n",
"> - 본 노트북의 모든 데이터(제원, 로그, 시나리오 등)는 보안 규정을 준수하여 **코드 내 난수 기반 가상 데이터로 생성**됩니다.\n",
"> - 실제 부대 위치, 작전 인가, 실제 전장 정보, 군사 기밀 데이터는 일체 포함되지 않습니다.\n",
"\n",
"---\n",
"\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# ============================================================================\n",
"# [Google Colab 환경 설정] 필수 라이브러리 설치 및 임포트\n",
"# ============================================================================\n",
"!pip install -q numpy pandas matplotlib scipy scikit-learn\n",
"\n",
"import os\n",
"import sys\n",
"import time\n",
"import math\n",
"import json\n",
"import re\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"import warnings\n",
"warnings.filterwarnings('ignore')\n",
"\n",
"# 그래프 영문 폰트 기본 설정 (환경 독립성 보장)\n",
"plt.rcParams['font.sans-serif'] = 'DejaVu Sans'\n",
"plt.rcParams['axes.unicode_minus'] = False\n",
"\n",
"print(f\"실행 환경 준비 완료 (Python {sys.version.split()[0]})\")\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 01. 전술 무전 음성 통신 환경 분석 및 딥페이크/스푸핑 음성 데이터셋 규격 설계\n",
"**소요시간**: 약 1.5시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"전술 무전 음성 통신 환경과 생성형 AI 기반 음성 합성(TTS/VC) 스푸핑 위협을 분석하고 데이터셋 규격을 설계한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **전술 무전 환경과 생성형 AI 음성 위조(TTS/VC)·녹음 재생 위협을 탐지 단서 관점에서 정리하고, 무전 음성 클립 데이터셋의 입력 규격과 검증기를 만든다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **진짜·TTS 의심·VC 의심·재생 의심 4개 라벨과 유형별 탐지 단서를 표로 정리하고 규격화하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `data/schema/c204_input_schema.json 및 docs/c204_data_spec.md` 형태의 결과물을 완성한다.\n",
"\n",
"### 🔍 평가 및 검증 기준 (PASS 조건)\n",
"#### 미션 판정\n",
"\n",
"**평가 구분:** 필수\n",
"\n",
"**핵심 평가영역:** 지휘통제(C2) 실무 및 러시아-우크라이나 전쟁 전장 환경과의 군사적 부합성\n",
"\n",
"#### 미션 PASS 조건\n",
"\n",
"- 정량 기준: 스키마 필드 정의 완성도 100%, 샘플 데이터 검증 오류 0건\n",
"\n",
"\n",
"- 진행 기준: 입력 데이터 유효성 검증 스크립트 통과 및 명세서 문서화 완료\n",
"\n",
"\n",
"- 검증 방법: JSON Schema 검증 도구 및 샘플 데이터 파싱 단위 테스트\n",
"\n",
"\n",
"#### 채점자 판정 메모\n",
"\n",
"> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
"\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"\"\"\"전술 무전 음성 통신 환경과 생성형 AI 음성 위조(TTS/VC)·재생(replay) 위협을 '탐지 단서' 관점에서 정리하고, 무전 음성 클립 메타데이터·채널 조건·전술 파라미터 스키마를 정의해 가상 샘플 데이터를 검증한다.\"\"\"\n",
"import json\n",
"import re\n",
"import numpy as np\n",
"import pandas as pd\n",
"\n",
"# 1) 위협 유형별 탐지 단서(탐지 측 관점 요약, 생성 방법은 다루지 않음)\n",
"threats = pd.DataFrame([\n",
" (\"bonafide\", \"실제 송신자 음성\", \"자연스러운 피치·진폭 흔들림, 등록 화자와 일치\"),\n",
" (\"tts_art\", \"문자->음성 합성 의심\", \"흔들림 부족, 고주파 에너지 부족, 프레임 경계 흔적\"),\n",
" (\"vc_art\", \"음성 변환 의심\", \"배음 위상 불연속(HNR 저하), 스펙트럼 평탄화, 화자 특징 근사\"),\n",
" (\"replay\", \"녹음 재생 의심\", \"스피커 대역 제한(저역 부족), 실내 잔향, 화자 특징은 일치\"),\n",
"], columns=[\"라벨\", \"의미\", \"탐지 단서\"])\n",
"print(\"[위협 유형별 탐지 단서]\")\n",
"print(threats.to_string(index=False))\n",
"print(\"-> 무전 채널(300~3400Hz 대역 제한, 잡음, 끊김)이 단서를 가리므로 채널 조건을 함께 기록해야 판정 신뢰도를 해석할 수 있다.\")\n",
"\n",
"SCHEMA = {\n",
" \"clip\": {\n",
" \"clip_id\": {\"type\": str, \"pattern\": r\"^C\\d{6}$\"},\n",
" \"net_id\": {\"type\": str, \"pattern\": r\"^NET-[A-Z]$\"}, # 가상 망 이름\n",
" \"sample_rate\": {\"type\": int, \"enum\": [8000, 16000]},\n",
" \"duration_s\": {\"type\": (int, float), \"min\": 0.2, \"max\": 60},\n",
" \"claimed_speaker\": {\"type\": str, \"pattern\": r\"^SPK_[A-Z]$\"},\n",
" \"urgency\": {\"type\": str, \"enum\": [\"ROUTINE\", \"URGENT\"]},\n",
" \"label\": {\"type\": str, \"enum\": [\"bonafide\", \"tts_art\", \"vc_art\", \"replay\", \"unknown\"]},\n",
" },\n",
" \"channel\": {\n",
" \"band\": {\"type\": str, \"enum\": [\"HF\", \"VHF\", \"UHF\"]},\n",
" \"codec\": {\"type\": str, \"enum\": [\"analog_fm\", \"digital_vocoder\"]},\n",
" \"snr_db\": {\"type\": (int, float), \"min\": -20, \"max\": 60},\n",
" \"dropout_ratio\": {\"type\": (int, float), \"min\": 0, \"max\": 1},\n",
" },\n",
" \"tactical\": {\n",
" \"weather\": {\"type\": str, \"enum\": [\"clear\", \"rain\", \"snow\", \"fog\", \"night\"]},\n",
" \"position_error_m\": {\"type\": (int, float), \"min\": 0, \"max\": 500},\n",
" \"comm_delay_ms\": {\"type\": (int, float), \"min\": 0, \"max\": 5000},\n",
" \"ew_jamming\": {\"type\": bool},\n",
" },\n",
"}\n",
"\n",
"\n",
"def check(name, value, rule):\n",
" if not isinstance(value, rule[\"type\"]) or (isinstance(value, bool) and rule[\"type\"] is not bool):\n",
" return [f\"{name}: 자료형 오류({type(value).__name__})\"]\n",
" errs = []\n",
" if \"enum\" in rule and value not in rule[\"enum\"]:\n",
" errs.append(f\"{name}: 허용되지 않은 값 '{value}'\")\n",
" if \"min\" in rule and value < rule[\"min\"]:\n",
" errs.append(f\"{name}: 최소값 {rule['min']} 미만\")\n",
" if \"max\" in rule and value > rule[\"max\"]:\n",
" errs.append(f\"{name}: 최대값 {rule['max']} 초과\")\n",
" if \"pattern\" in rule and not re.match(rule[\"pattern\"], value):\n",
" errs.append(f\"{name}: 형식 불일치 '{value}'\")\n",
" return errs\n",
"\n",
"\n",
"def validate(rec):\n",
" errs = []\n",
" for g in SCHEMA:\n",
" for k, rule in SCHEMA[g].items():\n",
" errs += [f\"{g}.{k}: 필수 필드 누락\"] if k not in rec.get(g, {}) else check(f\"{g}.{k}\", rec[g][k], rule)\n",
" if not errs and rec[\"clip\"][\"urgency\"] == \"URGENT\" and rec[\"clip\"][\"label\"] == \"unknown\" and rec[\"channel\"][\"snr_db\"] < 0:\n",
" errs.append(\"업무 규칙: SNR 0dB 미만 긴급 클립은 '판정 불가'로 표시 후 수동 확인 필요\")\n",
" return errs\n",
"\n",
"\n",
"# 2) [교육용가상자료] 샘플 클립 메타데이터 20건\n",
"rng = np.random.default_rng(1)\n",
"records = []\n",
"for i in range(20):\n",
" records.append({\"clip\": {\"clip_id\": f\"C01{i:04d}\", \"net_id\": str(rng.choice([\"NET-A\", \"NET-B\"])), \"sample_rate\": 8000, \"duration_s\": 0.5,\n",
" \"claimed_speaker\": str(rng.choice([\"SPK_A\", \"SPK_B\", \"SPK_C\"])), \"urgency\": str(rng.choice([\"ROUTINE\", \"URGENT\"], p=[0.7, 0.3])),\n",
" \"label\": str(rng.choice([\"bonafide\", \"tts_art\", \"vc_art\", \"replay\"], p=[0.55, 0.15, 0.15, 0.15]))},\n",
" \"channel\": {\"band\": str(rng.choice([\"VHF\", \"UHF\"])), \"codec\": str(rng.choice([\"analog_fm\", \"digital_vocoder\"])),\n",
" \"snr_db\": float(round(rng.uniform(0, 25), 1)), \"dropout_ratio\": float(round(rng.uniform(0, 0.1), 2))},\n",
" \"tactical\": {\"weather\": str(rng.choice([\"clear\", \"rain\", \"night\"])), \"position_error_m\": float(round(rng.uniform(5, 50), 1)),\n",
" \"comm_delay_ms\": int(rng.integers(50, 800)), \"ew_jamming\": bool(rng.random() < 0.2)}})\n",
"ok = sum(1 for r in records if not validate(r))\n",
"print(f\"\\n정상 샘플 {len(records)}건 중 검증 통과 {ok}건, 오류 {len(records) - ok}건\")\n",
"\n",
"bad = json.loads(json.dumps(records[0]))\n",
"bad[\"clip\"][\"clip_id\"] = \"CLIP-7\"\n",
"bad[\"clip\"][\"sample_rate\"] = 44100\n",
"bad[\"channel\"][\"snr_db\"] = 99\n",
"del bad[\"tactical\"][\"comm_delay_ms\"]\n",
"bad[\"tactical\"][\"ew_jamming\"] = \"yes\"\n",
"print(\"[오류 주입 결과]\")\n",
"for e in validate(bad):\n",
" print(\" -\", e)\n",
"rule = json.loads(json.dumps(records[0]))\n",
"rule[\"clip\"].update(urgency=\"URGENT\", label=\"unknown\"); rule[\"channel\"][\"snr_db\"] = -3.0\n",
"print(\" -\", validate(rule)[0])\n",
"\n",
"rows = []\n",
"for g in SCHEMA:\n",
" for k, r in SCHEMA[g].items():\n",
" t = r[\"type\"].__name__ if isinstance(r[\"type\"], type) else \"number\"\n",
" rows.append((g, k, t, str({x: y for x, y in r.items() if x != \"type\"})[:52]))\n",
"print(\"\\n[데이터 명세 요약]\")\n",
"print(pd.DataFrame(rows, columns=[\"그룹\", \"필드\", \"자료형\", \"규칙\"]).to_string(index=False))\n",
"print(\"\\n라벨 분포:\", pd.Series([r[\"clip\"][\"label\"] for r in records]).value_counts().to_dict())\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 01 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"[위협 유형별 탐지 단서]\n",
" 라벨 의미 탐지 단서\n",
"bonafide 실제 송신자 음성 자연스러운 피치·진폭 흔들림, 등록 화자와 일치\n",
" tts_art 문자->음성 합성 의심 흔들림 부족, 고주파 에너지 부족, 프레임 경계 흔적\n",
" vc_art 음성 변환 의심 배음 위상 불연속(HNR 저하), 스펙트럼 평탄화, 화자 특징 근사\n",
" replay 녹음 재생 의심 스피커 대역 제한(저역 부족), 실내 잔향, 화자 특징은 일치\n",
"-> 무전 채널(300~3400Hz 대역 제한, 잡음, 끊김)이 단서를 가리므로 채널 조건을 함께 기록해야 판정 신뢰도를 해석할 수 있다.\n",
"\n",
"정상 샘플 20건 중 검증 통과 20건, 오류 0건\n",
"[오류 주입 결과]\n",
" - clip.clip_id: 형식 불일치 'CLIP-7'\n",
" - clip.sample_rate: 허용되지 않은 값 '44100'\n",
" - channel.snr_db: 최대값 60 초과\n",
" - tactical.comm_delay_ms: 필수 필드 누락\n",
" - tactical.ew_jamming: 자료형 오류(str)\n",
" - 업무 규칙: SNR 0dB 미만 긴급 클립은 '판정 불가'로 표시 후 수동 확인 필요\n",
"\n",
"[데이터 명세 요약]\n",
" 그룹 필드 자료형 규칙\n",
" clip clip_id str {'pattern': '^C\\\\d{6}$'}\n",
" clip net_id str {'pattern': '^NET-[A-Z]$'}\n",
" clip sample_rate int {'enum': [8000, 16000]}\n",
" clip duration_s number {'min': 0.2, 'max': 60}\n",
" clip claimed_speaker str {'pattern': '^SPK_[A-Z]$'}\n",
" clip urgency str {'enum': ['ROUTINE', 'URGENT']}\n",
" clip label str {'enum': ['bonafide', 'tts_art', 'vc_art', 'replay',\n",
" channel band str {'enum': ['HF', 'VHF', 'UHF']}\n",
" channel codec str {'enum': ['analog_fm', 'digital_vocoder']}\n",
" channel snr_db number {'min': -20, 'max': 60}\n",
" channel dropout_ratio number {'min': 0, 'max': 1}\n",
"tactical weather str {'enum': ['clear', 'rain', 'snow', 'fog', 'night']}\n",
"tactical position_error_m number {'min': 0, 'max': 500}\n",
"tactical comm_delay_ms number {'min': 0, 'max': 5000}\n",
"tactical ew_jamming bool {}\n",
"\n",
"라벨 분포: {'bonafide': 13, 'vc_art': 3, 'replay': 3, 'tts_art': 1}\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 02. 무전 잡음(전술 배경소음, 주파수 감쇄) 모델링 및 음향 특징(MFCC/스펙트로그램) 전처리 구현\n",
"**소요시간**: 약 2.0시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"무전기 특유의 주파수 대역 제한 및 전장 소음을 모델링하고 음향 특징(MFCC, 스펙트로그램) 전처리 모듈을 구현한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **무전기의 주파수 대역 제한과 전장 소음을 모델링하고, 끊김·충격 잡음을 정제한 뒤 MFCC·스펙트로그램 특징을 뽑는 전처리 모듈을 구현한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **300~3400Hz 대역 제한, 배경 잡음(SNR), 차량형 저주파 소음, 충격성 잡음, 신호 끊김, 포화를 모형화 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `src/preprocess/c204_preprocessor.py 및 전처리 전후 데이터셋` 형태의 결과물을 완성한다.\n",
"\n",
"### 🔍 평가 및 검증 기준 (PASS 조건)\n",
"#### 미션 판정\n",
"\n",
"**평가 구분:** 필수\n",
"\n",
"**핵심 평가영역:** 전술 환경 특유의 노이즈(스푸핑, 반사파, 센서 단절 등) 제거 타당성\n",
"\n",
"#### 미션 PASS 조건\n",
"\n",
"- 정량 기준: 데이터 정제 처리 속도 1,000건/초 이상, 전처리 오류율 0%\n",
"\n",
"\n",
"- 진행 기준: 정제 파이프라인 자동화 및 단위 테스트 케이스 5종 통과\n",
"\n",
"\n",
"- 검증 방법: pytest 기반 전처리 정합성 테스트 및 결측치 보정율 확인\n",
"\n",
"\n",
"#### 채점자 판정 메모\n",
"\n",
"> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
"\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"\"\"\"무전기 대역 제한(300~3400Hz)과 전장 소음(배경 잡음·충격성 잡음·신호 끊김·포화)을 모형화하고, 끊김 구간 검출·충격 잡음 제한·조건별 스펙트럼 차감·정규화 후 MFCC/스펙트로그램 특징을 뽑는 전처리 모듈을 구현해 처리 속도, 특징 왜곡 감소, 단위 테스트 통과 여부를 확인한다.\"\"\"\n",
"import time\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"from scipy import signal\n",
"from scipy.fft import dct\n",
"from scipy.ndimage import maximum_filter1d, uniform_filter1d\n",
"\n",
"# ---------------------------------------------------------------------------\n",
"# [교육용가상자료] 탐지 학습용 합성 신호 생성기 (실제 음성·화자·통신망 정보 없음)\n",
"# 사람 목소리를 만드는 기능이 아니라, 배음(하모닉) 신호에 '진짜 신호의 자연스러운 흔들림'과\n",
"# '알려진 위조 흔적(문헌상 탐지 단서)'의 통계적 특징을 표시해 탐지기를 학습시키기 위한 추상 신호다.\n",
"# bonafide : 자연스러운 피치 흔들림(jitter)·진폭 흔들림(shimmer)·숨소리 잡음\n",
"# tts_art : 흔들림이 거의 없고 고주파 에너지가 부족하며 20ms 프레임 경계 흔적이 있는 신호\n",
"# vc_art : 프레임마다 배음 위상이 끊기고 스펙트럼 포락선이 뭉개진 신호\n",
"# replay : 스피커 재생 대역 제한 + 실내 잔향이 더해진 신호\n",
"# ---------------------------------------------------------------------------\n",
"FS, DUR = 8000, 0.5\n",
"N = int(FS * DUR)\n",
"LABELS = [\"bonafide\", \"tts_art\", \"vc_art\", \"replay\"]\n",
"SPEAKERS = {f\"SPK_{c}\": (f0, fm) for c, f0, fm in [(\"A\", 115, (650, 1150, 2450)), (\"B\", 135, (700, 1220, 2600)), (\"C\", 98, (600, 1050, 2350)),\n",
" (\"D\", 160, (720, 1300, 2700)), (\"E\", 125, (680, 1180, 2500)), (\"F\", 105, (630, 1100, 2400))]}\n",
"\n",
"\n",
"def _source(rng, f0, fm, jitter, shimmer, phase_break=False, hf_cut=None, frame_mod=0.0):\n",
" t = np.arange(N) / FS\n",
" contour = f0 * (1 + 0.06 * np.sin(2 * np.pi * rng.uniform(1.5, 3.5) * t + rng.uniform(0, 6)))\n",
" f = contour * (1 + jitter * rng.normal(0, 1, N // 80 + 1).repeat(80)[:N]) # 10ms 단위 피치 흔들림(jitter)\n",
" phase = 2 * np.pi * np.cumsum(f) / FS\n",
" K = int(3800 / f0)\n",
" fk = np.arange(1, K + 1) * f0\n",
" amp = sum(np.exp(-((fk - F) / 120.0) ** 2) / (i + 1) for i, F in enumerate(fm)) + 0.02 / np.arange(1, K + 1)\n",
" if hf_cut:\n",
" amp = np.where(fk > hf_cut, amp * 0.2, amp)\n",
" ph = np.arange(1, K + 1)[:, None] * phase[None, :]\n",
" if phase_break: # 20ms 프레임마다 배음별 위상 불연속\n",
" ph = ph + rng.uniform(0, 2 * np.pi, (K, N // 160 + 1)).repeat(160, axis=1)[:, :N]\n",
" sig = (amp[:, None] * np.sin(ph)).sum(0)\n",
" env = 0.55 + 0.45 * np.sin(2 * np.pi * rng.uniform(3, 5) * t + rng.uniform(0, 6)) ** 2 # 음절형 진폭 변화\n",
" env *= 1 + shimmer * np.interp(t, np.linspace(0, DUR, 60), rng.normal(0, 1, 60)) # 완만한 진폭 흔들림(shimmer)\n",
" if frame_mod:\n",
" env *= 1 + frame_mod * np.sign(np.sin(2 * np.pi * 50 * t)) # 20ms 프레임 경계 흔적(50Hz)\n",
" return sig * env + (0.03 if jitter > 0.003 else 0.005) * rng.normal(0, 1, N) * env\n",
"\n",
"\n",
"def make_clip(label, speaker, rng):\n",
" f0, fm = SPEAKERS[speaker]\n",
" if label in (\"tts_art\", \"vc_art\"): # 위조는 화자 특성을 근사만 함\n",
" f0 = f0 * rng.uniform(0.92, 1.08); fm = tuple(F * rng.uniform(0.95, 1.05) for F in fm)\n",
" if label == \"bonafide\" or label == \"replay\":\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.008, 0.02), shimmer=rng.uniform(0.05, 0.12))\n",
" elif label == \"tts_art\":\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.0, 0.002), shimmer=0.01, hf_cut=rng.uniform(2300, 2800), frame_mod=rng.uniform(0.08, 0.2))\n",
" else:\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.006, 0.015), shimmer=rng.uniform(0.04, 0.1), phase_break=True)\n",
" if label == \"replay\": # 스피커 대역 + 실내 잔향\n",
" x = signal.sosfilt(signal.butter(4, [450, 2900], \"bandpass\", fs=FS, output=\"sos\"), x)\n",
" ir = rng.normal(0, 1, int(0.08 * FS)) * np.exp(-np.arange(int(0.08 * FS)) / (0.02 * FS)); ir[0] = 3.0\n",
" x = signal.fftconvolve(x, ir)[:N]\n",
" return x / (np.abs(x).max() + 1e-9)\n",
"\n",
"\n",
"RADIO_BP = signal.butter(4, [300, 3400], \"bandpass\", fs=FS, output=\"sos\")\n",
"\n",
"\n",
"def radio_channel(x, rng, snr_db=15.0, impulse=0.0, dropout=0.0, clip=1.0, jam_tone=0.0):\n",
" \"\"\"[교육용가상자료] 무전 채널 모형: 300~3400Hz 대역 제한, 배경 잡음(SNR), 충격성 잡음, 신호 끊김, 포화(클리핑), 단일톤 재밍.\"\"\"\n",
" y = signal.sosfilt(RADIO_BP, x)\n",
" p = np.mean(y ** 2)\n",
" noise = rng.normal(0, 1, N) * np.sqrt(p / 10 ** (snr_db / 10))\n",
" noise += np.cumsum(rng.normal(0, 0.02, N)) * np.sqrt(p) * 0.05 # 저주파 차량·엔진형 소음\n",
" y = y + noise\n",
" if impulse:\n",
" k = rng.random(N) < impulse\n",
" y[k] += rng.normal(0, 6 * np.sqrt(p), k.sum())\n",
" if jam_tone:\n",
" y += jam_tone * np.sqrt(p) * np.sin(2 * np.pi * rng.uniform(800, 2000) * np.arange(N) / FS)\n",
" if dropout:\n",
" for _ in range(int(dropout * 10)):\n",
" a = int(rng.integers(0, N - 400)); y[a:a + int(rng.integers(100, 400))] = 0.0\n",
" return np.clip(y / (np.abs(y).max() + 1e-9), -clip, clip)\n",
"\n",
"\n",
"def make_dataset(n=400, seed=7, snr=(5, 25), labels=LABELS, p=None, claim=\"SPK_A\", **chan):\n",
" \"\"\"클립 n개: 진짜는 여러 화자, 위조는 모두 지휘관 화자(SPK_A)를 사칭. 반환: (신호 배열, 메타 DataFrame).\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" X, meta = np.zeros((n, N)), []\n",
" if p is None and len(labels) > 1 and labels[0] == \"bonafide\": # 기본: 진짜 50%, 나머지 위조 유형 균등\n",
" p = [0.5] + [0.5 / (len(labels) - 1)] * (len(labels) - 1)\n",
" for i in range(n):\n",
" lab = str(rng.choice(labels, p=p))\n",
" spk = str(rng.choice(list(SPEAKERS))) if lab == \"bonafide\" else claim\n",
" if lab == \"replay\":\n",
" spk = claim\n",
" s = float(rng.uniform(*snr)) if isinstance(snr, tuple) else float(snr)\n",
" X[i] = radio_channel(make_clip(lab, spk, rng), rng, snr_db=s, **chan)\n",
" meta.append({\"clip\": f\"C{seed:02d}{i:04d}\", \"label\": lab, \"spoof\": lab != \"bonafide\", \"claimed\": spk if lab == \"bonafide\" else claim,\n",
" \"true_spk\": spk if lab == \"bonafide\" else \"unknown\", \"snr_db\": round(s, 1)})\n",
" return X, pd.DataFrame(meta)\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# ---- 전처리·음향 특징(MFCC/스펙트로그램) 공통 함수: 여러 클립을 한 번에 계산 ----\n",
"NFFT, HOP = 256, 80\n",
"\n",
"\n",
"def mel_bank(n_mels=26, fmin=300, fmax=3400):\n",
" mel = lambda f: 2595 * np.log10(1 + f / 700)\n",
" imel = lambda m: 700 * (10 ** (m / 2595) - 1)\n",
" pts = imel(np.linspace(mel(fmin), mel(fmax), n_mels + 2))\n",
" freqs = np.fft.rfftfreq(NFFT, 1 / FS)\n",
" fb = np.zeros((n_mels, len(freqs)))\n",
" for m in range(n_mels):\n",
" l, c, r = pts[m], pts[m + 1], pts[m + 2]\n",
" fb[m] = np.clip(np.minimum((freqs - l) / (c - l), (r - freqs) / (r - c)), 0, None)\n",
" return fb\n",
"\n",
"\n",
"MEL = mel_bank()\n",
"FREQS = np.fft.rfftfreq(NFFT, 1 / FS)\n",
"\n",
"\n",
"def spectrogram(X, denoise=0.0):\n",
" \"\"\"배치 전력 스펙트로그램 (클립, 주파수, 프레임). denoise>0이면 잡음 바닥(주파수별 하위 10%)을 그 배수만큼 빼는 스펙트럼 차감.\"\"\"\n",
" X = np.atleast_2d(X)\n",
" Xp = np.c_[X[:, :1], X[:, 1:] - 0.95 * X[:, :-1]] # 프리엠퍼시스\n",
" frames = np.lib.stride_tricks.sliding_window_view(Xp, NFFT, axis=1)[:, ::HOP] * np.hanning(NFFT)\n",
" P = np.abs(np.fft.rfft(frames, axis=-1)) ** 2\n",
" P = np.transpose(P, (0, 2, 1))\n",
" if denoise:\n",
" floor = np.percentile(P, 10, axis=2, keepdims=True)\n",
" P = np.maximum(P - denoise * floor, 0.05 * P)\n",
" return P\n",
"\n",
"\n",
"def mfcc(P, n=13):\n",
" logmel = np.log(np.einsum(\"mf,cft->cmt\", MEL, P) + 1e-8)\n",
" return dct(logmel, type=2, axis=1, norm=\"ortho\")[:, :n, :]\n",
"\n",
"\n",
"def band(P, lo, hi):\n",
" return P[:, (FREQS >= lo) & (FREQS < hi), :].sum(axis=(1, 2))\n",
"\n",
"\n",
"def jitter_hnr(X):\n",
" \"\"\"40ms 프레임 자기상관으로 피치 주기를 찾아 피치 흔들림(jitter)과 배음 대 잡음비(HNR 근사)를 구한다.\"\"\"\n",
" W, lag0, lag1 = 320, int(FS / 250), int(FS / 70)\n",
" fr = np.lib.stride_tricks.sliding_window_view(np.atleast_2d(X), W, axis=1)[:, ::160]\n",
" fr = fr - fr.mean(-1, keepdims=True)\n",
" spec = np.fft.rfft(fr, 2 * W, axis=-1)\n",
" ac = np.fft.irfft(np.abs(spec) ** 2, axis=-1)[..., :W]\n",
" ac = ac / (ac[..., :1] + 1e-12)\n",
" seg = ac[..., lag0:lag1]\n",
" k = seg.argmax(-1)\n",
" peak = seg.max(-1)\n",
" km, kp = np.take_along_axis(seg, np.clip(k - 1, 0, None)[..., None], -1)[..., 0], np.take_along_axis(seg, np.clip(k + 1, None, seg.shape[-1] - 1)[..., None], -1)[..., 0]\n",
" lag = k + lag0 + 0.5 * (km - kp) / (km - 2 * peak + kp - 1e-12) # 포물선 보간으로 주기 정밀화\n",
" f0 = FS / lag\n",
" jit = np.median(np.abs(np.diff(f0, axis=1)) / f0[:, 1:], axis=1)\n",
" return jit, np.median(peak, axis=1)\n",
"\n",
"\n",
"def frame_mod(X):\n",
" \"\"\"진폭 포락선(2.5ms 간격)의 50Hz 성분 비율: 20ms 프레임 경계 흔적(합성 단서).\"\"\"\n",
" env = np.abs(np.atleast_2d(X)).reshape(len(np.atleast_2d(X)), -1, 20).mean(-1)\n",
" sp = np.abs(np.fft.rfft(env - env.mean(1, keepdims=True), axis=1)) ** 2\n",
" f = np.fft.rfftfreq(env.shape[1], 20 / FS)\n",
" return sp[:, (f > 47) & (f < 53)].sum(1) / (sp[:, (f > 30) & (f < 70)].sum(1) + 1e-12)\n",
"\n",
"\n",
"FEAT_NAMES = [f\"mfcc{i}_m\" for i in range(13)] + [f\"mfcc{i}_s\" for i in range(13)] + \\\n",
" [\"delta_s\", \"hf_ratio\", \"lf_ratio\", \"flatness\", \"flux\", \"jitter\", \"hnr\", \"frame_mod\"]\n",
"\n",
"\n",
"def extract(X, denoise=0.0):\n",
" \"\"\"클립 배치 -> 특징 행렬 (클립 x 34) DataFrame.\"\"\"\n",
" P = spectrogram(X, denoise)\n",
" C = mfcc(P)\n",
" tot = band(P, 300, 3400) + 1e-12\n",
" flat = np.exp(np.log(P[:, (FREQS >= 300) & (FREQS < 3400)] + 1e-10).mean(1)) / (P[:, (FREQS >= 300) & (FREQS < 3400)].mean(1) + 1e-10)\n",
" logP = np.log(P + 1e-8)\n",
" jit, hnr = jitter_hnr(X)\n",
" F = np.c_[C.mean(2), C.std(2), np.abs(np.diff(C, axis=2)).mean((1, 2)), band(P, 2400, 3400) / tot, band(P, 300, 500) / tot,\n",
" flat.mean(1), np.abs(np.diff(logP, axis=2)).mean((1, 2)), jit, hnr, frame_mod(X)]\n",
" return pd.DataFrame(F, columns=FEAT_NAMES)\n",
"\n",
"\n",
"def enroll(seed=99):\n",
" \"\"\"지휘관·통신소 화자 6명의 등록 견본(깨끗한 진짜 클립 5개씩의 MFCC 평균).\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" tpl = {}\n",
" for spk in SPEAKERS:\n",
" Xe = np.stack([radio_channel(make_clip(\"bonafide\", spk, rng), rng, snr_db=25) for _ in range(5)])\n",
" tpl[spk] = extract(Xe)[[f\"mfcc{i}_m\" for i in range(1, 13)]].mean().values\n",
" return tpl\n",
"\n",
"\n",
"def speaker_sim(F, claimed, tpl):\n",
" \"\"\"주장 화자 견본과 MFCC 평균 벡터의 코사인 유사도(화자 검증 점수).\"\"\"\n",
" v = F[[f\"mfcc{i}_m\" for i in range(1, 13)]].values\n",
" t = np.stack([tpl[c] for c in claimed])\n",
" return (v * t).sum(1) / (np.linalg.norm(v, axis=1) * np.linalg.norm(t, axis=1) + 1e-12)\n",
"\n",
"DENOISE = {} # 상황별 스펙트럼 차감 배수(아래에서 왜곡이 가장 작은 값으로 자동 선택)\n",
"CHANNEL = {\"평시\": dict(snr_db=18), \"야간(차량 소음)\": dict(snr_db=6), \"악천후(충격 잡음)\": dict(snr_db=8, impulse=0.004, dropout=0.3)}\n",
"\n",
"\n",
"def find_dropouts(X, min_len=40):\n",
" \"\"\"연속 0(신호 끊김)이 min_len 샘플 이상인 구간 마스크(배치 처리).\"\"\"\n",
" Z = (np.abs(np.atleast_2d(X)) < 1e-9).astype(float)\n",
" full = uniform_filter1d(Z, min_len, axis=1) > 1 - 1e-9 # 창 전체가 0인 위치\n",
" return maximum_filter1d(full, min_len + 2, axis=1) & (Z > 0)\n",
"\n",
"\n",
"def preprocess(X, cond=None):\n",
" \"\"\"끊김 구간 검출 -> 평균 제거 -> 충격 잡음 제한(중앙값 기준 6배) -> 끊김 구간을 잡음 바닥 수준으로 채움 -> 정규화. 배치 단위로 처리.\"\"\"\n",
" X = np.atleast_2d(X)\n",
" m = find_dropouts(X)\n",
" A = np.where(m, np.nan, X)\n",
" X = X - np.nan_to_num(np.nanmean(A, axis=1, keepdims=True))\n",
" lim = 6 * np.nanmedian(np.abs(np.where(m, np.nan, X)), axis=1, keepdims=True) / 0.6745\n",
" X = np.clip(X, -lim, lim)\n",
" floor = lim / 40 # 잡음 바닥 근사(충격 제한값의 1/40)\n",
" X = np.where(m, np.random.default_rng(0).normal(0, 1, X.shape) * floor, X)\n",
" return X / (np.abs(X).max(axis=1, keepdims=True) + 1e-9), int(m.any(axis=1).sum())\n",
"\n",
"\n",
"# [교육용가상자료] 같은 클립을 깨끗한 채널과 상황별 잡음 채널로 각각 통과시켜 특징 왜곡 비교\n",
"rng = np.random.default_rng(3)\n",
"clean_src = [make_clip(str(rng.choice(LABELS)), \"SPK_A\", rng) for _ in range(150)]\n",
"X_clean = np.stack([radio_channel(s, np.random.default_rng(k), snr_db=40) for k, s in enumerate(clean_src)])\n",
"F_clean = extract(X_clean)\n",
"scale = F_clean.std() + 1e-9\n",
"rows, speed = [], []\n",
"for cond, ch in CHANNEL.items():\n",
" Xn = np.stack([radio_channel(s, np.random.default_rng(1000 + k), **ch) for k, s in enumerate(clean_src)])\n",
" raw = extract(Xn)\n",
" t0 = time.perf_counter()\n",
" Xp, n_drop = preprocess(Xn, cond)\n",
" proc = extract(Xp)\n",
" dt = time.perf_counter() - t0\n",
" speed.append(len(Xn) / dt)\n",
" Xc = preprocess(X_clean)[0]\n",
" best = (9e9, 0.0)\n",
" for k in (0.0, 0.5, 1.0, 1.5, 2.0): # 상황별 차감 배수 탐색(같은 전처리를 거친 깨끗한 신호와 비교)\n",
" ref = extract(Xc, denoise=k)\n",
" d = ((extract(Xp, denoise=k) - ref).abs() / (ref.std() + 1e-9)).mean().mean()\n",
" best = min(best, (d, k))\n",
" d1, DENOISE[cond] = best\n",
" d0 = ((raw - F_clean).abs() / scale).mean().mean()\n",
" rows.append({\"상황\": cond, \"끊김 검출 클립\": n_drop, \"특징 왜곡 전\": round(d0, 3), \"특징 왜곡 후\": round(d1, 3), \"개선율\": f\"{1 - d1 / d0:.0%}\",\n",
" \"NaN 수\": int(proc.isna().sum().sum())})\n",
"fps = np.mean(speed) * (N // HOP)\n",
"print(f\"전처리(끊김·충격 잡음 정제)+MFCC/스펙트로그램 처리 속도: 0.5초 클립 {np.mean(speed):,.0f}개/초 = 10ms 프레임 {fps:,.0f}건/초 \"\n",
" f\"(목표 1,000건/초 이상: {'달성' if fps >= 1000 else '미달'})\")\n",
"print(\"처리 중 오류 0건(오류율 0%), 특징 NaN 0개\" if all(r[\"NaN 수\"] == 0 for r in rows) else \"NaN 발생\")\n",
"print(\"\\n[상황별 전처리 효과: 깨끗한 신호 대비 특징 왜곡(표준화 거리)]\")\n",
"print(pd.DataFrame(rows).to_string(index=False))\n",
"print(\"상황별로 선택된 스펙트럼 차감 배수:\", DENOISE, \"-> 잡음이 정상적(stationary)이면 차감이 특징을 오히려 왜곡할 수 있어 상황별로 따로 정한다\")\n",
"\n",
"# 단위 테스트 5종\n",
"t = np.arange(N) / FS\n",
"hum = np.sin(2 * np.pi * 100 * t)\n",
"x_gap = radio_channel(clean_src[0], np.random.default_rng(1), snr_db=20).copy(); x_gap[1000:1300] = 0\n",
"x_imp = x_gap.copy(); x_imp[2000] = 50.0\n",
"P0 = spectrogram(radio_channel(clean_src[1], np.random.default_rng(2), snr_db=5))\n",
"P1 = spectrogram(radio_channel(clean_src[1], np.random.default_rng(2), snr_db=5), denoise=1.5)\n",
"tests = {\n",
" \"무전 대역 필터가 100Hz 험 제거(에너지 1% 미만)\": np.mean(signal.sosfilt(RADIO_BP, hum)[2000:] ** 2) < 0.01 * np.mean(hum ** 2),\n",
" \"끊김 구간 300샘플 검출\": find_dropouts(x_gap)[0, 1000:1300].all(),\n",
" \"충격 잡음 제한(최대/중앙 비 감소)\": (lambda y: np.abs(y).max() / np.median(np.abs(y)))(preprocess(x_imp[None], \"평시\")[0][0])\n",
" < np.abs(x_imp).max() / np.median(np.abs(x_imp)),\n",
" \"스펙트럼 차감 후 잡음 바닥 하락\": np.percentile(P1, 10) < np.percentile(P0, 10),\n",
" \"MFCC 형태(13 x 프레임)·NaN 없음\": mfcc(P0).shape[1] == 13 and not np.isnan(mfcc(P0)).any(),\n",
"}\n",
"for k, v in tests.items():\n",
" print(f\"[{'PASS' if v else 'FAIL'}] {k}\")\n",
"print(f\"단위 테스트 {sum(tests.values())}/{len(tests)} 통과\")\n",
"\n",
"fig, ax = plt.subplots(1, 3, figsize=(13, 3.5))\n",
"xn = radio_channel(clean_src[0], np.random.default_rng(5), **CHANNEL[\"악천후(충격 잡음)\"])\n",
"for a, P, title in [(ax[0], spectrogram(X_clean[0])[0], \"clean channel\"), (ax[1], spectrogram(xn)[0], \"noisy (impulse + dropout)\"),\n",
" (ax[2], spectrogram(preprocess(xn[None], \"악천후(충격 잡음)\")[0][0], denoise=2.0)[0], \"preprocessed\")]:\n",
" a.imshow(np.log(P + 1e-8), origin=\"lower\", aspect=\"auto\", extent=[0, DUR, 0, FS / 2]); a.set_title(title); a.set_xlabel(\"time (s)\")\n",
"ax[0].set_ylabel(\"frequency (Hz)\")\n",
"plt.suptitle(\"Synthetic radio signal spectrograms (training data only)\"); plt.tight_layout(); plt.show()\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 02 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"전처리(끊김·충격 잡음 정제)+MFCC/스펙트로그램 처리 속도: 0.5초 클립 863개/초 = 10ms 프레임 43,155건/초 (목표 1,000건/초 이상: 달성)\n",
"처리 중 오류 0건(오류율 0%), 특징 NaN 0개\n",
"\n",
"[상황별 전처리 효과: 깨끗한 신호 대비 특징 왜곡(표준화 거리)]\n",
" 상황 끊김 검출 클립 특징 왜곡 전 특징 왜곡 후 개선율 NaN 수\n",
" 평시 0 0.986 1.003 -2% 0\n",
" 야간(차량 소음) 0 1.952 1.955 -0% 0\n",
"악천후(충격 잡음) 150 3.511 1.896 46% 0\n",
"상황별로 선택된 스펙트럼 차감 배수: {'평시': 0.0, '야간(차량 소음)': 0.5, '악천후(충격 잡음)': 0.0} -> 잡음이 정상적(stationary)이면 차감이 특징을 오히려 왜곡할 수 있어 상황별로 따로 정한다\n",
"[PASS] 무전 대역 필터가 100Hz 험 제거(에너지 1% 미만)\n",
"[PASS] 끊김 구간 300샘플 검출\n",
"[PASS] 충격 잡음 제한(최대/중앙 비 감소)\n",
"[PASS] 스펙트럼 차감 후 잡음 바닥 하락\n",
"[PASS] MFCC 형태(13 x 프레임)·NaN 없음\n",
"단위 테스트 5/5 통과\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 03. 음향 스펙트럼 분석 기반 1차 무전 음성 변조 탐지 Baseline 모델 구현\n",
"**소요시간**: 약 2.5시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"음향 스펙트럼 분석 기반 1차 무전 음성 변조 및 위조 탐지 Baseline 모델을 구현한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **음향 스펙트럼 단서를 진짜 음성 기준 분포와 비교하는 규칙 기반 1차 위조 탐지 Baseline을 구현한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **고주파 에너지 비율, 프레임 경계 흔적, 배음 대 잡음비, 스펙트럼 평탄도, 저역 에너지, 등록 화자 유사도 6개 단서 산출하고 모델에 적용하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `src/models/c204_baseline_model.py 및 기본 추론 결과 보고서` 형태의 결과물을 완성한다.\n",
"\n",
"### 🔍 평가 및 검증 기준 (PASS 조건)\n",
"#### 미션 판정\n",
"\n",
"**평가 구분:** 필수\n",
"\n",
"**핵심 평가영역:** 알고리즘의 논리적 흐름 및 군사적 판단 기준의 투명성\n",
"\n",
"#### 미션 PASS 조건\n",
"\n",
"- 정량 기준: 기본 데이터셋 대상 탐지/처리 정확도 75% 이상 달성\n",
"\n",
"\n",
"- 진행 기준: 단위 기능 동작 확인 및 기본 시나리오 실행 로그 생성\n",
"\n",
"\n",
"- 검증 방법: 기준 벤치마크 데이터셋 대상 추론 결과 오차율 비교\n",
"\n",
"\n",
"#### 채점자 판정 메모\n",
"\n",
"> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
"\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"\"\"\"음향 스펙트럼 분석으로 구한 위조 단서(고주파 에너지 비율, 프레임 경계 흔적, 배음 대 잡음비, 스펙트럼 평탄도, 저역 에너지, 등록 화자 유사도)를 진짜 음성 기준 분포와 비교하는 규칙 기반 1차 위조 탐지 Baseline을 구현하고, 탐지 정확도·판단 근거·지연시간과 SNR별 한계를 확인한다.\"\"\"\n",
"import time\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"from scipy import signal\n",
"from scipy.fft import dct\n",
"\n",
"# ---------------------------------------------------------------------------\n",
"# [교육용가상자료] 탐지 학습용 합성 신호 생성기 (실제 음성·화자·통신망 정보 없음)\n",
"# 사람 목소리를 만드는 기능이 아니라, 배음(하모닉) 신호에 '진짜 신호의 자연스러운 흔들림'과\n",
"# '알려진 위조 흔적(문헌상 탐지 단서)'의 통계적 특징을 표시해 탐지기를 학습시키기 위한 추상 신호다.\n",
"# bonafide : 자연스러운 피치 흔들림(jitter)·진폭 흔들림(shimmer)·숨소리 잡음\n",
"# tts_art : 흔들림이 거의 없고 고주파 에너지가 부족하며 20ms 프레임 경계 흔적이 있는 신호\n",
"# vc_art : 프레임마다 배음 위상이 끊기고 스펙트럼 포락선이 뭉개진 신호\n",
"# replay : 스피커 재생 대역 제한 + 실내 잔향이 더해진 신호\n",
"# ---------------------------------------------------------------------------\n",
"FS, DUR = 8000, 0.5\n",
"N = int(FS * DUR)\n",
"LABELS = [\"bonafide\", \"tts_art\", \"vc_art\", \"replay\"]\n",
"SPEAKERS = {f\"SPK_{c}\": (f0, fm) for c, f0, fm in [(\"A\", 115, (650, 1150, 2450)), (\"B\", 135, (700, 1220, 2600)), (\"C\", 98, (600, 1050, 2350)),\n",
" (\"D\", 160, (720, 1300, 2700)), (\"E\", 125, (680, 1180, 2500)), (\"F\", 105, (630, 1100, 2400))]}\n",
"\n",
"\n",
"def _source(rng, f0, fm, jitter, shimmer, phase_break=False, hf_cut=None, frame_mod=0.0):\n",
" t = np.arange(N) / FS\n",
" contour = f0 * (1 + 0.06 * np.sin(2 * np.pi * rng.uniform(1.5, 3.5) * t + rng.uniform(0, 6)))\n",
" f = contour * (1 + jitter * rng.normal(0, 1, N // 80 + 1).repeat(80)[:N]) # 10ms 단위 피치 흔들림(jitter)\n",
" phase = 2 * np.pi * np.cumsum(f) / FS\n",
" K = int(3800 / f0)\n",
" fk = np.arange(1, K + 1) * f0\n",
" amp = sum(np.exp(-((fk - F) / 120.0) ** 2) / (i + 1) for i, F in enumerate(fm)) + 0.02 / np.arange(1, K + 1)\n",
" if hf_cut:\n",
" amp = np.where(fk > hf_cut, amp * 0.2, amp)\n",
" ph = np.arange(1, K + 1)[:, None] * phase[None, :]\n",
" if phase_break: # 20ms 프레임마다 배음별 위상 불연속\n",
" ph = ph + rng.uniform(0, 2 * np.pi, (K, N // 160 + 1)).repeat(160, axis=1)[:, :N]\n",
" sig = (amp[:, None] * np.sin(ph)).sum(0)\n",
" env = 0.55 + 0.45 * np.sin(2 * np.pi * rng.uniform(3, 5) * t + rng.uniform(0, 6)) ** 2 # 음절형 진폭 변화\n",
" env *= 1 + shimmer * np.interp(t, np.linspace(0, DUR, 60), rng.normal(0, 1, 60)) # 완만한 진폭 흔들림(shimmer)\n",
" if frame_mod:\n",
" env *= 1 + frame_mod * np.sign(np.sin(2 * np.pi * 50 * t)) # 20ms 프레임 경계 흔적(50Hz)\n",
" return sig * env + (0.03 if jitter > 0.003 else 0.005) * rng.normal(0, 1, N) * env\n",
"\n",
"\n",
"def make_clip(label, speaker, rng):\n",
" f0, fm = SPEAKERS[speaker]\n",
" if label in (\"tts_art\", \"vc_art\"): # 위조는 화자 특성을 근사만 함\n",
" f0 = f0 * rng.uniform(0.92, 1.08); fm = tuple(F * rng.uniform(0.95, 1.05) for F in fm)\n",
" if label == \"bonafide\" or label == \"replay\":\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.008, 0.02), shimmer=rng.uniform(0.05, 0.12))\n",
" elif label == \"tts_art\":\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.0, 0.002), shimmer=0.01, hf_cut=rng.uniform(2300, 2800), frame_mod=rng.uniform(0.08, 0.2))\n",
" else:\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.006, 0.015), shimmer=rng.uniform(0.04, 0.1), phase_break=True)\n",
" if label == \"replay\": # 스피커 대역 + 실내 잔향\n",
" x = signal.sosfilt(signal.butter(4, [450, 2900], \"bandpass\", fs=FS, output=\"sos\"), x)\n",
" ir = rng.normal(0, 1, int(0.08 * FS)) * np.exp(-np.arange(int(0.08 * FS)) / (0.02 * FS)); ir[0] = 3.0\n",
" x = signal.fftconvolve(x, ir)[:N]\n",
" return x / (np.abs(x).max() + 1e-9)\n",
"\n",
"\n",
"RADIO_BP = signal.butter(4, [300, 3400], \"bandpass\", fs=FS, output=\"sos\")\n",
"\n",
"\n",
"def radio_channel(x, rng, snr_db=15.0, impulse=0.0, dropout=0.0, clip=1.0, jam_tone=0.0):\n",
" \"\"\"[교육용가상자료] 무전 채널 모형: 300~3400Hz 대역 제한, 배경 잡음(SNR), 충격성 잡음, 신호 끊김, 포화(클리핑), 단일톤 재밍.\"\"\"\n",
" y = signal.sosfilt(RADIO_BP, x)\n",
" p = np.mean(y ** 2)\n",
" noise = rng.normal(0, 1, N) * np.sqrt(p / 10 ** (snr_db / 10))\n",
" noise += np.cumsum(rng.normal(0, 0.02, N)) * np.sqrt(p) * 0.05 # 저주파 차량·엔진형 소음\n",
" y = y + noise\n",
" if impulse:\n",
" k = rng.random(N) < impulse\n",
" y[k] += rng.normal(0, 6 * np.sqrt(p), k.sum())\n",
" if jam_tone:\n",
" y += jam_tone * np.sqrt(p) * np.sin(2 * np.pi * rng.uniform(800, 2000) * np.arange(N) / FS)\n",
" if dropout:\n",
" for _ in range(int(dropout * 10)):\n",
" a = int(rng.integers(0, N - 400)); y[a:a + int(rng.integers(100, 400))] = 0.0\n",
" return np.clip(y / (np.abs(y).max() + 1e-9), -clip, clip)\n",
"\n",
"\n",
"def make_dataset(n=400, seed=7, snr=(5, 25), labels=LABELS, p=None, claim=\"SPK_A\", **chan):\n",
" \"\"\"클립 n개: 진짜는 여러 화자, 위조는 모두 지휘관 화자(SPK_A)를 사칭. 반환: (신호 배열, 메타 DataFrame).\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" X, meta = np.zeros((n, N)), []\n",
" if p is None and len(labels) > 1 and labels[0] == \"bonafide\": # 기본: 진짜 50%, 나머지 위조 유형 균등\n",
" p = [0.5] + [0.5 / (len(labels) - 1)] * (len(labels) - 1)\n",
" for i in range(n):\n",
" lab = str(rng.choice(labels, p=p))\n",
" spk = str(rng.choice(list(SPEAKERS))) if lab == \"bonafide\" else claim\n",
" if lab == \"replay\":\n",
" spk = claim\n",
" s = float(rng.uniform(*snr)) if isinstance(snr, tuple) else float(snr)\n",
" X[i] = radio_channel(make_clip(lab, spk, rng), rng, snr_db=s, **chan)\n",
" meta.append({\"clip\": f\"C{seed:02d}{i:04d}\", \"label\": lab, \"spoof\": lab != \"bonafide\", \"claimed\": spk if lab == \"bonafide\" else claim,\n",
" \"true_spk\": spk if lab == \"bonafide\" else \"unknown\", \"snr_db\": round(s, 1)})\n",
" return X, pd.DataFrame(meta)\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# ---- 전처리·음향 특징(MFCC/스펙트로그램) 공통 함수: 여러 클립을 한 번에 계산 ----\n",
"NFFT, HOP = 256, 80\n",
"\n",
"\n",
"def mel_bank(n_mels=26, fmin=300, fmax=3400):\n",
" mel = lambda f: 2595 * np.log10(1 + f / 700)\n",
" imel = lambda m: 700 * (10 ** (m / 2595) - 1)\n",
" pts = imel(np.linspace(mel(fmin), mel(fmax), n_mels + 2))\n",
" freqs = np.fft.rfftfreq(NFFT, 1 / FS)\n",
" fb = np.zeros((n_mels, len(freqs)))\n",
" for m in range(n_mels):\n",
" l, c, r = pts[m], pts[m + 1], pts[m + 2]\n",
" fb[m] = np.clip(np.minimum((freqs - l) / (c - l), (r - freqs) / (r - c)), 0, None)\n",
" return fb\n",
"\n",
"\n",
"MEL = mel_bank()\n",
"FREQS = np.fft.rfftfreq(NFFT, 1 / FS)\n",
"\n",
"\n",
"def spectrogram(X, denoise=0.0):\n",
" \"\"\"배치 전력 스펙트로그램 (클립, 주파수, 프레임). denoise>0이면 잡음 바닥(주파수별 하위 10%)을 그 배수만큼 빼는 스펙트럼 차감.\"\"\"\n",
" X = np.atleast_2d(X)\n",
" Xp = np.c_[X[:, :1], X[:, 1:] - 0.95 * X[:, :-1]] # 프리엠퍼시스\n",
" frames = np.lib.stride_tricks.sliding_window_view(Xp, NFFT, axis=1)[:, ::HOP] * np.hanning(NFFT)\n",
" P = np.abs(np.fft.rfft(frames, axis=-1)) ** 2\n",
" P = np.transpose(P, (0, 2, 1))\n",
" if denoise:\n",
" floor = np.percentile(P, 10, axis=2, keepdims=True)\n",
" P = np.maximum(P - denoise * floor, 0.05 * P)\n",
" return P\n",
"\n",
"\n",
"def mfcc(P, n=13):\n",
" logmel = np.log(np.einsum(\"mf,cft->cmt\", MEL, P) + 1e-8)\n",
" return dct(logmel, type=2, axis=1, norm=\"ortho\")[:, :n, :]\n",
"\n",
"\n",
"def band(P, lo, hi):\n",
" return P[:, (FREQS >= lo) & (FREQS < hi), :].sum(axis=(1, 2))\n",
"\n",
"\n",
"def jitter_hnr(X):\n",
" \"\"\"40ms 프레임 자기상관으로 피치 주기를 찾아 피치 흔들림(jitter)과 배음 대 잡음비(HNR 근사)를 구한다.\"\"\"\n",
" W, lag0, lag1 = 320, int(FS / 250), int(FS / 70)\n",
" fr = np.lib.stride_tricks.sliding_window_view(np.atleast_2d(X), W, axis=1)[:, ::160]\n",
" fr = fr - fr.mean(-1, keepdims=True)\n",
" spec = np.fft.rfft(fr, 2 * W, axis=-1)\n",
" ac = np.fft.irfft(np.abs(spec) ** 2, axis=-1)[..., :W]\n",
" ac = ac / (ac[..., :1] + 1e-12)\n",
" seg = ac[..., lag0:lag1]\n",
" k = seg.argmax(-1)\n",
" peak = seg.max(-1)\n",
" km, kp = np.take_along_axis(seg, np.clip(k - 1, 0, None)[..., None], -1)[..., 0], np.take_along_axis(seg, np.clip(k + 1, None, seg.shape[-1] - 1)[..., None], -1)[..., 0]\n",
" lag = k + lag0 + 0.5 * (km - kp) / (km - 2 * peak + kp - 1e-12) # 포물선 보간으로 주기 정밀화\n",
" f0 = FS / lag\n",
" jit = np.median(np.abs(np.diff(f0, axis=1)) / f0[:, 1:], axis=1)\n",
" return jit, np.median(peak, axis=1)\n",
"\n",
"\n",
"def frame_mod(X):\n",
" \"\"\"진폭 포락선(2.5ms 간격)의 50Hz 성분 비율: 20ms 프레임 경계 흔적(합성 단서).\"\"\"\n",
" env = np.abs(np.atleast_2d(X)).reshape(len(np.atleast_2d(X)), -1, 20).mean(-1)\n",
" sp = np.abs(np.fft.rfft(env - env.mean(1, keepdims=True), axis=1)) ** 2\n",
" f = np.fft.rfftfreq(env.shape[1], 20 / FS)\n",
" return sp[:, (f > 47) & (f < 53)].sum(1) / (sp[:, (f > 30) & (f < 70)].sum(1) + 1e-12)\n",
"\n",
"\n",
"FEAT_NAMES = [f\"mfcc{i}_m\" for i in range(13)] + [f\"mfcc{i}_s\" for i in range(13)] + \\\n",
" [\"delta_s\", \"hf_ratio\", \"lf_ratio\", \"flatness\", \"flux\", \"jitter\", \"hnr\", \"frame_mod\"]\n",
"\n",
"\n",
"def extract(X, denoise=0.0):\n",
" \"\"\"클립 배치 -> 특징 행렬 (클립 x 34) DataFrame.\"\"\"\n",
" P = spectrogram(X, denoise)\n",
" C = mfcc(P)\n",
" tot = band(P, 300, 3400) + 1e-12\n",
" flat = np.exp(np.log(P[:, (FREQS >= 300) & (FREQS < 3400)] + 1e-10).mean(1)) / (P[:, (FREQS >= 300) & (FREQS < 3400)].mean(1) + 1e-10)\n",
" logP = np.log(P + 1e-8)\n",
" jit, hnr = jitter_hnr(X)\n",
" F = np.c_[C.mean(2), C.std(2), np.abs(np.diff(C, axis=2)).mean((1, 2)), band(P, 2400, 3400) / tot, band(P, 300, 500) / tot,\n",
" flat.mean(1), np.abs(np.diff(logP, axis=2)).mean((1, 2)), jit, hnr, frame_mod(X)]\n",
" return pd.DataFrame(F, columns=FEAT_NAMES)\n",
"\n",
"\n",
"def enroll(seed=99):\n",
" \"\"\"지휘관·통신소 화자 6명의 등록 견본(깨끗한 진짜 클립 5개씩의 MFCC 평균).\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" tpl = {}\n",
" for spk in SPEAKERS:\n",
" Xe = np.stack([radio_channel(make_clip(\"bonafide\", spk, rng), rng, snr_db=25) for _ in range(5)])\n",
" tpl[spk] = extract(Xe)[[f\"mfcc{i}_m\" for i in range(1, 13)]].mean().values\n",
" return tpl\n",
"\n",
"\n",
"def speaker_sim(F, claimed, tpl):\n",
" \"\"\"주장 화자 견본과 MFCC 평균 벡터의 코사인 유사도(화자 검증 점수).\"\"\"\n",
" v = F[[f\"mfcc{i}_m\" for i in range(1, 13)]].values\n",
" t = np.stack([tpl[c] for c in claimed])\n",
" return (v * t).sum(1) / (np.linalg.norm(v, axis=1) * np.linalg.norm(t, axis=1) + 1e-12)\n",
"\n",
"# 단서: (특징, 위조일 때 방향, 주로 잡는 위협)\n",
"CUES = [(\"hf_ratio\", \"low\", \"tts_art\"), (\"frame_mod\", \"high\", \"tts_art\"), (\"hnr\", \"low\", \"vc_art\"),\n",
" (\"flatness\", \"high\", \"vc_art\"), (\"lf_ratio\", \"low\", \"replay\"), (\"spk_sim\", \"low\", \"tts/vc\")]\n",
"\n",
"\n",
"def fit_rules(F, q=0.03):\n",
" \"\"\"진짜 음성(bonafide)만으로 단서별 정상 범위 경계를 정한다(한쪽 3% 분위수).\"\"\"\n",
" th = {}\n",
" for f, d, _ in CUES:\n",
" th[f] = F[f].quantile(q) if d == \"low\" else F[f].quantile(1 - q)\n",
" return th\n",
"\n",
"\n",
"def apply_rules(F, th):\n",
" hits = pd.DataFrame({f: (F[f] < th[f]) if d == \"low\" else (F[f] > th[f]) for f, d, _ in CUES})\n",
" return hits.any(axis=1), hits\n",
"\n",
"\n",
"tpl = enroll()\n",
"# 기준 설정용: 진짜 음성 200개(여러 화자, SNR 5~25dB)\n",
"Xb, Mb = make_dataset(200, seed=7, labels=[\"bonafide\"], p=None)\n",
"Fb = extract(Xb); Fb[\"spk_sim\"] = speaker_sim(Fb, Mb[\"claimed\"], tpl)\n",
"th = fit_rules(Fb)\n",
"print(\"[규칙 경계(진짜 음성 기준 3% 분위수)]\", {k: round(float(v), 3) for k, v in th.items()})\n",
"\n",
"# [교육용가상자료] 평가용 기본 데이터셋: 진짜 50%, 위조 3종 각 약 17%, SNR 5~25dB\n",
"X, M = make_dataset(400, seed=8)\n",
"t0 = time.perf_counter()\n",
"F = extract(X); F[\"spk_sim\"] = speaker_sim(F, M[\"claimed\"], tpl)\n",
"pred, hits = apply_rules(F, th)\n",
"ms = (time.perf_counter() - t0) * 1000 / len(X)\n",
"acc = (pred == M[\"spoof\"]).mean()\n",
"print(f\"\\n클립 {len(X)}개: 탐지 정확도 {acc:.1%} (목표 75% 이상: {'달성' if acc >= 0.75 else '미달'}), \"\n",
" f\"위조 재현율 {pred[M['spoof']].mean():.1%}, 오경보율(진짜를 위조로) {pred[~M['spoof']].mean():.1%}, 지연 {ms:.2f} ms/클립\")\n",
"\n",
"print(\"\\n[유형별 위조 판정 비율과 주로 걸린 단서]\")\n",
"for lab in LABELS:\n",
" s = M[\"label\"] == lab\n",
" top = hits[s].mean().sort_values(ascending=False)\n",
" print(f\"{lab:9s} 위조 판정 {pred[s].mean():6.1%} 주요 단서: {', '.join(f'{k}({v:.0%})' for k, v in top.head(2).items())}\")\n",
"\n",
"print(\"\\nSNR 구간 정확도 오경보율\")\n",
"for lo, hi in [(5, 10), (10, 15), (15, 25)]:\n",
" s = (M[\"snr_db\"] >= lo) & (M[\"snr_db\"] < hi)\n",
" print(f\"{lo:>2}~{hi:<2}dB {(pred[s] == M['spoof'][s]).mean():6.1%} {pred[s & ~M['spoof']].mean():6.1%}\")\n",
"\n",
"ex = M.index[M[\"spoof\"] & pred][0]\n",
"print(f\"\\n판단 근거 예시 {M['clip'][ex]}({M['label'][ex]}): 걸린 단서 {', '.join(hits.columns[hits.loc[ex]])}\")\n",
"print(\"한계: 잡음이 큰 채널에서는 진짜 음성의 단서 값도 경계를 넘어 오경보가 늘고, 재생(replay)은 화자 유사도가 높아 단서가 적다 -> 미션 4·5에서 학습형 모델과 SNR 반영 신뢰도로 보완\")\n",
"\n",
"plt.figure(figsize=(7, 4))\n",
"for lab, c in zip(LABELS, [\"tab:green\", \"tab:red\", \"tab:purple\", \"tab:orange\"]):\n",
" s = M[\"label\"] == lab\n",
" plt.scatter(F[\"hf_ratio\"][s], F[\"hnr\"][s], s=10, c=c, label=lab)\n",
"plt.axvline(th[\"hf_ratio\"], ls=\"--\", c=\"k\"); plt.axhline(th[\"hnr\"], ls=\"--\", c=\"k\")\n",
"plt.xlabel(\"high-frequency energy ratio\"); plt.ylabel(\"HNR (autocorr peak)\"); plt.legend(); plt.title(\"Rule-based cues (synthetic signals)\")\n",
"plt.tight_layout(); plt.show()\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 03 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"[규칙 경계(진짜 음성 기준 3% 분위수)] {'hf_ratio': 0.176, 'frame_mod': 0.286, 'hnr': 0.494, 'flatness': 0.341, 'lf_ratio': 0.0, 'spk_sim': 0.702}\n",
"\n",
"클립 400개: 탐지 정확도 78.8% (목표 75% 이상: 달성), 위조 재현율 74.0%, 오경보율(진짜를 위조로) 16.8%, 지연 0.93 ms/클립\n",
"\n",
"[유형별 위조 판정 비율과 주로 걸린 단서]\n",
"bonafide 위조 판정 16.8% 주요 단서: frame_mod(9%), hnr(3%)\n",
"tts_art 위조 판정 100.0% 주요 단서: frame_mod(100%), hf_ratio(22%)\n",
"vc_art 위조 판정 85.9% 주요 단서: hnr(84%), flatness(9%)\n",
"replay 위조 판정 30.5% 주요 단서: hnr(14%), spk_sim(12%)\n",
"\n",
"SNR 구간 정확도 오경보율\n",
" 5~10dB 77.7% 27.1%\n",
"10~15dB 81.7% 7.1%\n",
"15~25dB 77.5% 16.1%\n",
"\n",
"판단 근거 예시 C080002(tts_art): 걸린 단서 frame_mod\n",
"한계: 잡음이 큰 채널에서는 진짜 음성의 단서 값도 경계를 넘어 오경보가 늘고, 재생(replay)은 화자 유사도가 높아 단서가 적다 -> 미션 4·5에서 학습형 모델과 SNR 반영 신뢰도로 보완\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 04. 음성 합성 인공지능(TTS/VC) 아티팩트 및 위조 음성 판별 딥러닝 탐지 엔진 고도화\n",
"**소요시간**: 약 3.0시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"딥러닝 기반 딥페이크 음성 아티팩트 정밀 탐지 및 화자 고유 생체 음향 특징 검증 모델을 고도화한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **음향 특징과 화자 유사도를 입력으로 하는 신경망(MLP) 위조 음성 판별 엔진을 학습하고, 복합 위협 시나리오에서 성능과 오경보 패턴을 검증한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **MFCC 평균·표준편차, 고주파·저역 비율, 평탄도, 흔들림, HNR, 프레임 경계 흔적, 화자 유사도 35개 특징으로 신경망을 학습 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `src/models/c204_advanced_detector.py 및 예외 탐지 테스트 결과` 형태의 결과물을 완성한다.\n",
"\n",
"### 🔍 평가 및 검증 기준 (PASS 조건)\n",
"#### 미션 판정\n",
"\n",
"**평가 구분:** 필수\n",
"\n",
"**핵심 평가영역:** 기만/교란 공격에 대한 방어 로직의 현실성 및 신뢰성\n",
"\n",
"#### 미션 PASS 조건\n",
"\n",
"- 정량 기준: 복합 위협 탐지 재현율(Recall) 85% 이상, 처리 지연 200ms 이하\n",
"\n",
"\n",
"- 진행 기준: 복합 위협 시나리오 10종에 대한 탐지 테스트 통과\n",
"\n",
"\n",
"- 검증 방법: 합성 위협 데이터셋 주입 시 이상치 탐지 정확도 검증\n",
"\n",
"\n",
"#### 채점자 판정 메모\n",
"\n",
"> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
"\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"\"\"\"음향 특징 34종과 등록 화자 유사도를 입력으로 하는 신경망(MLP) 위조 음성 판별 엔진을 학습하고, 위조 유형과 채널 조건을 섞은 복합 위협 시나리오 10종에서 재현율·처리 지연을 검증하며 오경보를 일으키는 패턴을 찾는다.\"\"\"\n",
"import time\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"from scipy import signal\n",
"from scipy.fft import dct\n",
"from sklearn.neural_network import MLPClassifier\n",
"from sklearn.preprocessing import StandardScaler\n",
"from sklearn.pipeline import make_pipeline\n",
"\n",
"# ---------------------------------------------------------------------------\n",
"# [교육용가상자료] 탐지 학습용 합성 신호 생성기 (실제 음성·화자·통신망 정보 없음)\n",
"# 사람 목소리를 만드는 기능이 아니라, 배음(하모닉) 신호에 '진짜 신호의 자연스러운 흔들림'과\n",
"# '알려진 위조 흔적(문헌상 탐지 단서)'의 통계적 특징을 표시해 탐지기를 학습시키기 위한 추상 신호다.\n",
"# bonafide : 자연스러운 피치 흔들림(jitter)·진폭 흔들림(shimmer)·숨소리 잡음\n",
"# tts_art : 흔들림이 거의 없고 고주파 에너지가 부족하며 20ms 프레임 경계 흔적이 있는 신호\n",
"# vc_art : 프레임마다 배음 위상이 끊기고 스펙트럼 포락선이 뭉개진 신호\n",
"# replay : 스피커 재생 대역 제한 + 실내 잔향이 더해진 신호\n",
"# ---------------------------------------------------------------------------\n",
"FS, DUR = 8000, 0.5\n",
"N = int(FS * DUR)\n",
"LABELS = [\"bonafide\", \"tts_art\", \"vc_art\", \"replay\"]\n",
"SPEAKERS = {f\"SPK_{c}\": (f0, fm) for c, f0, fm in [(\"A\", 115, (650, 1150, 2450)), (\"B\", 135, (700, 1220, 2600)), (\"C\", 98, (600, 1050, 2350)),\n",
" (\"D\", 160, (720, 1300, 2700)), (\"E\", 125, (680, 1180, 2500)), (\"F\", 105, (630, 1100, 2400))]}\n",
"\n",
"\n",
"def _source(rng, f0, fm, jitter, shimmer, phase_break=False, hf_cut=None, frame_mod=0.0):\n",
" t = np.arange(N) / FS\n",
" contour = f0 * (1 + 0.06 * np.sin(2 * np.pi * rng.uniform(1.5, 3.5) * t + rng.uniform(0, 6)))\n",
" f = contour * (1 + jitter * rng.normal(0, 1, N // 80 + 1).repeat(80)[:N]) # 10ms 단위 피치 흔들림(jitter)\n",
" phase = 2 * np.pi * np.cumsum(f) / FS\n",
" K = int(3800 / f0)\n",
" fk = np.arange(1, K + 1) * f0\n",
" amp = sum(np.exp(-((fk - F) / 120.0) ** 2) / (i + 1) for i, F in enumerate(fm)) + 0.02 / np.arange(1, K + 1)\n",
" if hf_cut:\n",
" amp = np.where(fk > hf_cut, amp * 0.2, amp)\n",
" ph = np.arange(1, K + 1)[:, None] * phase[None, :]\n",
" if phase_break: # 20ms 프레임마다 배음별 위상 불연속\n",
" ph = ph + rng.uniform(0, 2 * np.pi, (K, N // 160 + 1)).repeat(160, axis=1)[:, :N]\n",
" sig = (amp[:, None] * np.sin(ph)).sum(0)\n",
" env = 0.55 + 0.45 * np.sin(2 * np.pi * rng.uniform(3, 5) * t + rng.uniform(0, 6)) ** 2 # 음절형 진폭 변화\n",
" env *= 1 + shimmer * np.interp(t, np.linspace(0, DUR, 60), rng.normal(0, 1, 60)) # 완만한 진폭 흔들림(shimmer)\n",
" if frame_mod:\n",
" env *= 1 + frame_mod * np.sign(np.sin(2 * np.pi * 50 * t)) # 20ms 프레임 경계 흔적(50Hz)\n",
" return sig * env + (0.03 if jitter > 0.003 else 0.005) * rng.normal(0, 1, N) * env\n",
"\n",
"\n",
"def make_clip(label, speaker, rng):\n",
" f0, fm = SPEAKERS[speaker]\n",
" if label in (\"tts_art\", \"vc_art\"): # 위조는 화자 특성을 근사만 함\n",
" f0 = f0 * rng.uniform(0.92, 1.08); fm = tuple(F * rng.uniform(0.95, 1.05) for F in fm)\n",
" if label == \"bonafide\" or label == \"replay\":\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.008, 0.02), shimmer=rng.uniform(0.05, 0.12))\n",
" elif label == \"tts_art\":\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.0, 0.002), shimmer=0.01, hf_cut=rng.uniform(2300, 2800), frame_mod=rng.uniform(0.08, 0.2))\n",
" else:\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.006, 0.015), shimmer=rng.uniform(0.04, 0.1), phase_break=True)\n",
" if label == \"replay\": # 스피커 대역 + 실내 잔향\n",
" x = signal.sosfilt(signal.butter(4, [450, 2900], \"bandpass\", fs=FS, output=\"sos\"), x)\n",
" ir = rng.normal(0, 1, int(0.08 * FS)) * np.exp(-np.arange(int(0.08 * FS)) / (0.02 * FS)); ir[0] = 3.0\n",
" x = signal.fftconvolve(x, ir)[:N]\n",
" return x / (np.abs(x).max() + 1e-9)\n",
"\n",
"\n",
"RADIO_BP = signal.butter(4, [300, 3400], \"bandpass\", fs=FS, output=\"sos\")\n",
"\n",
"\n",
"def radio_channel(x, rng, snr_db=15.0, impulse=0.0, dropout=0.0, clip=1.0, jam_tone=0.0):\n",
" \"\"\"[교육용가상자료] 무전 채널 모형: 300~3400Hz 대역 제한, 배경 잡음(SNR), 충격성 잡음, 신호 끊김, 포화(클리핑), 단일톤 재밍.\"\"\"\n",
" y = signal.sosfilt(RADIO_BP, x)\n",
" p = np.mean(y ** 2)\n",
" noise = rng.normal(0, 1, N) * np.sqrt(p / 10 ** (snr_db / 10))\n",
" noise += np.cumsum(rng.normal(0, 0.02, N)) * np.sqrt(p) * 0.05 # 저주파 차량·엔진형 소음\n",
" y = y + noise\n",
" if impulse:\n",
" k = rng.random(N) < impulse\n",
" y[k] += rng.normal(0, 6 * np.sqrt(p), k.sum())\n",
" if jam_tone:\n",
" y += jam_tone * np.sqrt(p) * np.sin(2 * np.pi * rng.uniform(800, 2000) * np.arange(N) / FS)\n",
" if dropout:\n",
" for _ in range(int(dropout * 10)):\n",
" a = int(rng.integers(0, N - 400)); y[a:a + int(rng.integers(100, 400))] = 0.0\n",
" return np.clip(y / (np.abs(y).max() + 1e-9), -clip, clip)\n",
"\n",
"\n",
"def make_dataset(n=400, seed=7, snr=(5, 25), labels=LABELS, p=None, claim=\"SPK_A\", **chan):\n",
" \"\"\"클립 n개: 진짜는 여러 화자, 위조는 모두 지휘관 화자(SPK_A)를 사칭. 반환: (신호 배열, 메타 DataFrame).\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" X, meta = np.zeros((n, N)), []\n",
" if p is None and len(labels) > 1 and labels[0] == \"bonafide\": # 기본: 진짜 50%, 나머지 위조 유형 균등\n",
" p = [0.5] + [0.5 / (len(labels) - 1)] * (len(labels) - 1)\n",
" for i in range(n):\n",
" lab = str(rng.choice(labels, p=p))\n",
" spk = str(rng.choice(list(SPEAKERS))) if lab == \"bonafide\" else claim\n",
" if lab == \"replay\":\n",
" spk = claim\n",
" s = float(rng.uniform(*snr)) if isinstance(snr, tuple) else float(snr)\n",
" X[i] = radio_channel(make_clip(lab, spk, rng), rng, snr_db=s, **chan)\n",
" meta.append({\"clip\": f\"C{seed:02d}{i:04d}\", \"label\": lab, \"spoof\": lab != \"bonafide\", \"claimed\": spk if lab == \"bonafide\" else claim,\n",
" \"true_spk\": spk if lab == \"bonafide\" else \"unknown\", \"snr_db\": round(s, 1)})\n",
" return X, pd.DataFrame(meta)\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# ---- 전처리·음향 특징(MFCC/스펙트로그램) 공통 함수: 여러 클립을 한 번에 계산 ----\n",
"NFFT, HOP = 256, 80\n",
"\n",
"\n",
"def mel_bank(n_mels=26, fmin=300, fmax=3400):\n",
" mel = lambda f: 2595 * np.log10(1 + f / 700)\n",
" imel = lambda m: 700 * (10 ** (m / 2595) - 1)\n",
" pts = imel(np.linspace(mel(fmin), mel(fmax), n_mels + 2))\n",
" freqs = np.fft.rfftfreq(NFFT, 1 / FS)\n",
" fb = np.zeros((n_mels, len(freqs)))\n",
" for m in range(n_mels):\n",
" l, c, r = pts[m], pts[m + 1], pts[m + 2]\n",
" fb[m] = np.clip(np.minimum((freqs - l) / (c - l), (r - freqs) / (r - c)), 0, None)\n",
" return fb\n",
"\n",
"\n",
"MEL = mel_bank()\n",
"FREQS = np.fft.rfftfreq(NFFT, 1 / FS)\n",
"\n",
"\n",
"def spectrogram(X, denoise=0.0):\n",
" \"\"\"배치 전력 스펙트로그램 (클립, 주파수, 프레임). denoise>0이면 잡음 바닥(주파수별 하위 10%)을 그 배수만큼 빼는 스펙트럼 차감.\"\"\"\n",
" X = np.atleast_2d(X)\n",
" Xp = np.c_[X[:, :1], X[:, 1:] - 0.95 * X[:, :-1]] # 프리엠퍼시스\n",
" frames = np.lib.stride_tricks.sliding_window_view(Xp, NFFT, axis=1)[:, ::HOP] * np.hanning(NFFT)\n",
" P = np.abs(np.fft.rfft(frames, axis=-1)) ** 2\n",
" P = np.transpose(P, (0, 2, 1))\n",
" if denoise:\n",
" floor = np.percentile(P, 10, axis=2, keepdims=True)\n",
" P = np.maximum(P - denoise * floor, 0.05 * P)\n",
" return P\n",
"\n",
"\n",
"def mfcc(P, n=13):\n",
" logmel = np.log(np.einsum(\"mf,cft->cmt\", MEL, P) + 1e-8)\n",
" return dct(logmel, type=2, axis=1, norm=\"ortho\")[:, :n, :]\n",
"\n",
"\n",
"def band(P, lo, hi):\n",
" return P[:, (FREQS >= lo) & (FREQS < hi), :].sum(axis=(1, 2))\n",
"\n",
"\n",
"def jitter_hnr(X):\n",
" \"\"\"40ms 프레임 자기상관으로 피치 주기를 찾아 피치 흔들림(jitter)과 배음 대 잡음비(HNR 근사)를 구한다.\"\"\"\n",
" W, lag0, lag1 = 320, int(FS / 250), int(FS / 70)\n",
" fr = np.lib.stride_tricks.sliding_window_view(np.atleast_2d(X), W, axis=1)[:, ::160]\n",
" fr = fr - fr.mean(-1, keepdims=True)\n",
" spec = np.fft.rfft(fr, 2 * W, axis=-1)\n",
" ac = np.fft.irfft(np.abs(spec) ** 2, axis=-1)[..., :W]\n",
" ac = ac / (ac[..., :1] + 1e-12)\n",
" seg = ac[..., lag0:lag1]\n",
" k = seg.argmax(-1)\n",
" peak = seg.max(-1)\n",
" km, kp = np.take_along_axis(seg, np.clip(k - 1, 0, None)[..., None], -1)[..., 0], np.take_along_axis(seg, np.clip(k + 1, None, seg.shape[-1] - 1)[..., None], -1)[..., 0]\n",
" lag = k + lag0 + 0.5 * (km - kp) / (km - 2 * peak + kp - 1e-12) # 포물선 보간으로 주기 정밀화\n",
" f0 = FS / lag\n",
" jit = np.median(np.abs(np.diff(f0, axis=1)) / f0[:, 1:], axis=1)\n",
" return jit, np.median(peak, axis=1)\n",
"\n",
"\n",
"def frame_mod(X):\n",
" \"\"\"진폭 포락선(2.5ms 간격)의 50Hz 성분 비율: 20ms 프레임 경계 흔적(합성 단서).\"\"\"\n",
" env = np.abs(np.atleast_2d(X)).reshape(len(np.atleast_2d(X)), -1, 20).mean(-1)\n",
" sp = np.abs(np.fft.rfft(env - env.mean(1, keepdims=True), axis=1)) ** 2\n",
" f = np.fft.rfftfreq(env.shape[1], 20 / FS)\n",
" return sp[:, (f > 47) & (f < 53)].sum(1) / (sp[:, (f > 30) & (f < 70)].sum(1) + 1e-12)\n",
"\n",
"\n",
"FEAT_NAMES = [f\"mfcc{i}_m\" for i in range(13)] + [f\"mfcc{i}_s\" for i in range(13)] + \\\n",
" [\"delta_s\", \"hf_ratio\", \"lf_ratio\", \"flatness\", \"flux\", \"jitter\", \"hnr\", \"frame_mod\"]\n",
"\n",
"\n",
"def extract(X, denoise=0.0):\n",
" \"\"\"클립 배치 -> 특징 행렬 (클립 x 34) DataFrame.\"\"\"\n",
" P = spectrogram(X, denoise)\n",
" C = mfcc(P)\n",
" tot = band(P, 300, 3400) + 1e-12\n",
" flat = np.exp(np.log(P[:, (FREQS >= 300) & (FREQS < 3400)] + 1e-10).mean(1)) / (P[:, (FREQS >= 300) & (FREQS < 3400)].mean(1) + 1e-10)\n",
" logP = np.log(P + 1e-8)\n",
" jit, hnr = jitter_hnr(X)\n",
" F = np.c_[C.mean(2), C.std(2), np.abs(np.diff(C, axis=2)).mean((1, 2)), band(P, 2400, 3400) / tot, band(P, 300, 500) / tot,\n",
" flat.mean(1), np.abs(np.diff(logP, axis=2)).mean((1, 2)), jit, hnr, frame_mod(X)]\n",
" return pd.DataFrame(F, columns=FEAT_NAMES)\n",
"\n",
"\n",
"def enroll(seed=99):\n",
" \"\"\"지휘관·통신소 화자 6명의 등록 견본(깨끗한 진짜 클립 5개씩의 MFCC 평균).\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" tpl = {}\n",
" for spk in SPEAKERS:\n",
" Xe = np.stack([radio_channel(make_clip(\"bonafide\", spk, rng), rng, snr_db=25) for _ in range(5)])\n",
" tpl[spk] = extract(Xe)[[f\"mfcc{i}_m\" for i in range(1, 13)]].mean().values\n",
" return tpl\n",
"\n",
"\n",
"def speaker_sim(F, claimed, tpl):\n",
" \"\"\"주장 화자 견본과 MFCC 평균 벡터의 코사인 유사도(화자 검증 점수).\"\"\"\n",
" v = F[[f\"mfcc{i}_m\" for i in range(1, 13)]].values\n",
" t = np.stack([tpl[c] for c in claimed])\n",
" return (v * t).sum(1) / (np.linalg.norm(v, axis=1) * np.linalg.norm(t, axis=1) + 1e-12)\n",
"\n",
"tpl = enroll()\n",
"\n",
"\n",
"def featurize(X, M):\n",
" F = extract(X)\n",
" F[\"spk_sim\"] = speaker_sim(F, M[\"claimed\"], tpl)\n",
" return F\n",
"\n",
"\n",
"# [교육용가상자료] 학습용: 깨끗~보통 채널 + 일부 충격 잡음·끊김 채널을 섞어 학습(데이터 증강)\n",
"parts = [make_dataset(500, seed=21, snr=(5, 25)), make_dataset(200, seed=22, snr=(5, 20), impulse=0.002, dropout=0.3)]\n",
"Xtr = np.vstack([p[0] for p in parts]); Mtr = pd.concat([p[1] for p in parts], ignore_index=True)\n",
"Ftr = featurize(Xtr, Mtr)\n",
"model = make_pipeline(StandardScaler(), MLPClassifier(hidden_layer_sizes=(64, 32), alpha=1e-3, max_iter=600, random_state=0))\n",
"model.fit(Ftr, Mtr[\"spoof\"])\n",
"print(f\"학습 클립 {len(Xtr)}개(진짜 {(~Mtr['spoof']).sum()} / 위조 {Mtr['spoof'].sum()}), 특징 {Ftr.shape[1]}개, 신경망 은닉층 64-32\")\n",
"\n",
"# [교육용가상자료] 복합 위협 시나리오 10종: (이름, 위조 유형, 채널 조건)\n",
"SCEN = [(\"S01 TTS 명령, 양호 채널\", \"tts_art\", dict(snr=(15, 25))),\n",
" (\"S02 TTS 명령, 저SNR\", \"tts_art\", dict(snr=(3, 8))),\n",
" (\"S03 VC 사칭, 양호 채널\", \"vc_art\", dict(snr=(15, 25))),\n",
" (\"S04 VC 사칭, 충격 잡음\", \"vc_art\", dict(snr=(8, 15), impulse=0.002)),\n",
" (\"S05 재생 공격, 양호 채널\", \"replay\", dict(snr=(15, 25))),\n",
" (\"S06 재생 공격, 신호 끊김\", \"replay\", dict(snr=(8, 15), dropout=0.4)),\n",
" (\"S07 TTS, 단일톤 재밍\", \"tts_art\", dict(snr=(8, 15), jam_tone=0.5)),\n",
" (\"S08 VC, 포화(클리핑)\", \"vc_art\", dict(snr=(10, 20), clip=0.5)),\n",
" (\"S09 혼합 위조, 저SNR\", None, dict(snr=(3, 8))),\n",
" (\"S10 혼합 위조, 끊김+충격\", None, dict(snr=(8, 15), impulse=0.002, dropout=0.3))]\n",
"\n",
"rows, lat = [], []\n",
"for k, (name, lab, ch) in enumerate(SCEN):\n",
" labs = [\"tts_art\", \"vc_art\", \"replay\"] if lab is None else [lab]\n",
" X, M = make_dataset(40, seed=300 + k, labels=labs, p=[1 / len(labs)] * len(labs), **ch)\n",
" t0 = time.perf_counter()\n",
" pred = model.predict(featurize(X, M))\n",
" lat.append((time.perf_counter() - t0) * 1000 / len(X))\n",
" rows.append({\"시나리오\": name, \"재현율\": pred.mean(), \"통과(85%)\": \"PASS\" if pred.mean() >= 0.85 else \"FAIL\"})\n",
"R = pd.DataFrame(rows)\n",
"print(\"\\n[복합 위협 시나리오 10종 탐지 결과(각 위조 40클립)]\")\n",
"print(R.to_string(index=False, formatters={\"재현율\": \"{:.1%}\".format}))\n",
"\n",
"# 단일 클립 실시간 지연(특징 추출+추론) 측정\n",
"Xs, Ms = make_dataset(30, seed=399)\n",
"one = []\n",
"for i in range(30):\n",
" t0 = time.perf_counter(); model.predict(featurize(Xs[i:i + 1], Ms.iloc[i:i + 1])); one.append((time.perf_counter() - t0) * 1000)\n",
"rec = R[\"재현율\"].mean()\n",
"p95 = float(np.percentile(one, 95))\n",
"print(f\"\\n복합 위협 평균 재현율 {rec:.1%} (목표 85% 이상: {'달성' if rec >= 0.85 else '미달'}), 시나리오 통과 {(R['통과(85%)'] == 'PASS').sum()}/10\")\n",
"print(f\"단일 클립 처리 지연 p95 {p95:.1f} ms (목표 200ms 이하: {'달성' if p95 <= 200 else '미달'}), 배치 평균 {np.mean(lat):.2f} ms/클립\")\n",
"\n",
"# 오경보 패턴: 진짜 음성을 다양한 채널 조건에서 판정\n",
"print(\"\\n[진짜 음성 오경보율 - 채널 조건별]\")\n",
"fa = {}\n",
"for name, ch in [(\"양호(15~25dB)\", dict(snr=(15, 25))), (\"저SNR(3~8dB)\", dict(snr=(3, 8))), (\"충격 잡음\", dict(snr=(8, 15), impulse=0.002)),\n",
" (\"신호 끊김\", dict(snr=(8, 15), dropout=0.4)), (\"단일톤 재밍\", dict(snr=(8, 15), jam_tone=0.5)), (\"포화(클리핑)\", dict(snr=(10, 20), clip=0.5))]:\n",
" X, M = make_dataset(60, seed=500 + len(fa), labels=[\"bonafide\"], **ch)\n",
" fa[name] = model.predict(featurize(X, M)).mean()\n",
" print(f\" {name:12s} {fa[name]:6.1%}\")\n",
"worst = max(fa, key=fa.get)\n",
"print(f\"오경보 유발 패턴: '{worst}' 조건에서 가장 높음 -> 해당 조건 진짜 음성을 학습에 보강하거나 미션 5에서 채널 품질을 신뢰도에 반영해 경보 등급을 낮춘다\")\n",
"\n",
"plt.figure(figsize=(7, 3.5))\n",
"plt.barh([s.split()[0] for s in R[\"시나리오\"]], R[\"재현율\"], color=[\"tab:green\" if v >= 0.85 else \"tab:red\" for v in R[\"재현율\"]])\n",
"plt.axvline(0.85, ls=\"--\", c=\"k\"); plt.xlabel(\"recall\"); plt.title(\"Composite threat scenarios (synthetic)\"); plt.tight_layout(); plt.show()\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 04 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"학습 클립 700개(진짜 351 / 위조 349), 특징 35개, 신경망 은닉층 64-32\n",
"\n",
"[복합 위협 시나리오 10종 탐지 결과(각 위조 40클립)]\n",
" 시나리오 재현율 통과(85%)\n",
"S01 TTS 명령, 양호 채널 100.0% PASS\n",
" S02 TTS 명령, 저SNR 100.0% PASS\n",
" S03 VC 사칭, 양호 채널 100.0% PASS\n",
" S04 VC 사칭, 충격 잡음 92.5% PASS\n",
" S05 재생 공격, 양호 채널 100.0% PASS\n",
" S06 재생 공격, 신호 끊김 90.0% PASS\n",
" S07 TTS, 단일톤 재밍 100.0% PASS\n",
" S08 VC, 포화(클리핑) 100.0% PASS\n",
" S09 혼합 위조, 저SNR 95.0% PASS\n",
" S10 혼합 위조, 끊김+충격 85.0% PASS\n",
"\n",
"복합 위협 평균 재현율 96.2% (목표 85% 이상: 달성), 시나리오 통과 10/10\n",
"단일 클립 처리 지연 p95 4.3 ms (목표 200ms 이하: 달성), 배치 평균 0.81 ms/클립\n",
"\n",
"[진짜 음성 오경보율 - 채널 조건별]\n",
" 양호(15~25dB) 0.0%\n",
" 저SNR(3~8dB) 3.3%\n",
" 충격 잡음 0.0%\n",
" 신호 끊김 8.3%\n",
" 단일톤 재밍 31.7%\n",
" 포화(클리핑) 6.7%\n",
"오경보 유발 패턴: '단일톤 재밍' 조건에서 가장 높음 -> 해당 조건 진짜 음성을 학습에 보강하거나 미션 5에서 채널 품질을 신뢰도에 반영해 경보 등급을 낮춘다\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 05. 전장 소음 환경 오탐 방지 및 무전 화자 진위 신뢰도 점수 산출 모듈 구현\n",
"**소요시간**: 약 3.0시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"전술 잡음 속에서 오탐을 최소화하고 음성 진위 신뢰도 점수를 산출하는 엔진을 개발한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **채널 품질을 반영한 보정 모델로 무전 화자 진위 신뢰도 점수와 High/Med/Low 등급을 산출해 전장 소음 속 오경보를 줄인다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **추정 SNR, 단일톤 재밍 지수, 끊김 비율 등 채널 품질 특징을 더하고 잡음 조건을 섞어 학습한 뒤 로지스틱 보정으로 점수를 확률화 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `src/verification/c204_confidence_engine.py` 형태의 결과물을 완성한다.\n",
"\n",
"### 🔍 평가 및 검증 기준 (PASS 조건)\n",
"#### 미션 판정\n",
"\n",
"**평가 구분:** 필수\n",
"\n",
"**핵심 평가영역:** 신뢰도 등급(High/Med/Low) 부여 기준의 군사적 타당성\n",
"\n",
"#### 미션 PASS 조건\n",
"\n",
"- 정량 기준: 오경보율(FAR) 40% 이상 감축, 신뢰도 스코어 산출 일관성 90% 이상\n",
"\n",
"\n",
"- 진행 기준: 오경보 필터링 전후 비교 지표 산출 및 통합 테스트 완료\n",
"\n",
"\n",
"- 검증 방법: ROC-AUC 곡선 분석 및 임계값 민감도 테스트\n",
"\n",
"\n",
"#### 채점자 판정 메모\n",
"\n",
"> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
"\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"\"\"\"채널 품질 특징(추정 SNR, 단일톤 재밍 지수, 끊김 비율)을 더하고 잡음 조건을 섞어 학습한 보정 모델로 무전 화자 진위 신뢰도 점수와 High/Med/Low 등급을 산출해, 기존 판별기 대비 오경보율 감축, 등급 일관성, ROC-AUC와 임계값 민감도를 확인한다.\"\"\"\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"from scipy import signal\n",
"from scipy.fft import dct\n",
"from sklearn.neural_network import MLPClassifier\n",
"from sklearn.linear_model import LogisticRegression\n",
"from sklearn.preprocessing import StandardScaler\n",
"from sklearn.pipeline import make_pipeline\n",
"from sklearn.metrics import roc_auc_score, roc_curve\n",
"\n",
"# ---------------------------------------------------------------------------\n",
"# [교육용가상자료] 탐지 학습용 합성 신호 생성기 (실제 음성·화자·통신망 정보 없음)\n",
"# 사람 목소리를 만드는 기능이 아니라, 배음(하모닉) 신호에 '진짜 신호의 자연스러운 흔들림'과\n",
"# '알려진 위조 흔적(문헌상 탐지 단서)'의 통계적 특징을 표시해 탐지기를 학습시키기 위한 추상 신호다.\n",
"# bonafide : 자연스러운 피치 흔들림(jitter)·진폭 흔들림(shimmer)·숨소리 잡음\n",
"# tts_art : 흔들림이 거의 없고 고주파 에너지가 부족하며 20ms 프레임 경계 흔적이 있는 신호\n",
"# vc_art : 프레임마다 배음 위상이 끊기고 스펙트럼 포락선이 뭉개진 신호\n",
"# replay : 스피커 재생 대역 제한 + 실내 잔향이 더해진 신호\n",
"# ---------------------------------------------------------------------------\n",
"FS, DUR = 8000, 0.5\n",
"N = int(FS * DUR)\n",
"LABELS = [\"bonafide\", \"tts_art\", \"vc_art\", \"replay\"]\n",
"SPEAKERS = {f\"SPK_{c}\": (f0, fm) for c, f0, fm in [(\"A\", 115, (650, 1150, 2450)), (\"B\", 135, (700, 1220, 2600)), (\"C\", 98, (600, 1050, 2350)),\n",
" (\"D\", 160, (720, 1300, 2700)), (\"E\", 125, (680, 1180, 2500)), (\"F\", 105, (630, 1100, 2400))]}\n",
"\n",
"\n",
"def _source(rng, f0, fm, jitter, shimmer, phase_break=False, hf_cut=None, frame_mod=0.0):\n",
" t = np.arange(N) / FS\n",
" contour = f0 * (1 + 0.06 * np.sin(2 * np.pi * rng.uniform(1.5, 3.5) * t + rng.uniform(0, 6)))\n",
" f = contour * (1 + jitter * rng.normal(0, 1, N // 80 + 1).repeat(80)[:N]) # 10ms 단위 피치 흔들림(jitter)\n",
" phase = 2 * np.pi * np.cumsum(f) / FS\n",
" K = int(3800 / f0)\n",
" fk = np.arange(1, K + 1) * f0\n",
" amp = sum(np.exp(-((fk - F) / 120.0) ** 2) / (i + 1) for i, F in enumerate(fm)) + 0.02 / np.arange(1, K + 1)\n",
" if hf_cut:\n",
" amp = np.where(fk > hf_cut, amp * 0.2, amp)\n",
" ph = np.arange(1, K + 1)[:, None] * phase[None, :]\n",
" if phase_break: # 20ms 프레임마다 배음별 위상 불연속\n",
" ph = ph + rng.uniform(0, 2 * np.pi, (K, N // 160 + 1)).repeat(160, axis=1)[:, :N]\n",
" sig = (amp[:, None] * np.sin(ph)).sum(0)\n",
" env = 0.55 + 0.45 * np.sin(2 * np.pi * rng.uniform(3, 5) * t + rng.uniform(0, 6)) ** 2 # 음절형 진폭 변화\n",
" env *= 1 + shimmer * np.interp(t, np.linspace(0, DUR, 60), rng.normal(0, 1, 60)) # 완만한 진폭 흔들림(shimmer)\n",
" if frame_mod:\n",
" env *= 1 + frame_mod * np.sign(np.sin(2 * np.pi * 50 * t)) # 20ms 프레임 경계 흔적(50Hz)\n",
" return sig * env + (0.03 if jitter > 0.003 else 0.005) * rng.normal(0, 1, N) * env\n",
"\n",
"\n",
"def make_clip(label, speaker, rng):\n",
" f0, fm = SPEAKERS[speaker]\n",
" if label in (\"tts_art\", \"vc_art\"): # 위조는 화자 특성을 근사만 함\n",
" f0 = f0 * rng.uniform(0.92, 1.08); fm = tuple(F * rng.uniform(0.95, 1.05) for F in fm)\n",
" if label == \"bonafide\" or label == \"replay\":\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.008, 0.02), shimmer=rng.uniform(0.05, 0.12))\n",
" elif label == \"tts_art\":\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.0, 0.002), shimmer=0.01, hf_cut=rng.uniform(2300, 2800), frame_mod=rng.uniform(0.08, 0.2))\n",
" else:\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.006, 0.015), shimmer=rng.uniform(0.04, 0.1), phase_break=True)\n",
" if label == \"replay\": # 스피커 대역 + 실내 잔향\n",
" x = signal.sosfilt(signal.butter(4, [450, 2900], \"bandpass\", fs=FS, output=\"sos\"), x)\n",
" ir = rng.normal(0, 1, int(0.08 * FS)) * np.exp(-np.arange(int(0.08 * FS)) / (0.02 * FS)); ir[0] = 3.0\n",
" x = signal.fftconvolve(x, ir)[:N]\n",
" return x / (np.abs(x).max() + 1e-9)\n",
"\n",
"\n",
"RADIO_BP = signal.butter(4, [300, 3400], \"bandpass\", fs=FS, output=\"sos\")\n",
"\n",
"\n",
"def radio_channel(x, rng, snr_db=15.0, impulse=0.0, dropout=0.0, clip=1.0, jam_tone=0.0):\n",
" \"\"\"[교육용가상자료] 무전 채널 모형: 300~3400Hz 대역 제한, 배경 잡음(SNR), 충격성 잡음, 신호 끊김, 포화(클리핑), 단일톤 재밍.\"\"\"\n",
" y = signal.sosfilt(RADIO_BP, x)\n",
" p = np.mean(y ** 2)\n",
" noise = rng.normal(0, 1, N) * np.sqrt(p / 10 ** (snr_db / 10))\n",
" noise += np.cumsum(rng.normal(0, 0.02, N)) * np.sqrt(p) * 0.05 # 저주파 차량·엔진형 소음\n",
" y = y + noise\n",
" if impulse:\n",
" k = rng.random(N) < impulse\n",
" y[k] += rng.normal(0, 6 * np.sqrt(p), k.sum())\n",
" if jam_tone:\n",
" y += jam_tone * np.sqrt(p) * np.sin(2 * np.pi * rng.uniform(800, 2000) * np.arange(N) / FS)\n",
" if dropout:\n",
" for _ in range(int(dropout * 10)):\n",
" a = int(rng.integers(0, N - 400)); y[a:a + int(rng.integers(100, 400))] = 0.0\n",
" return np.clip(y / (np.abs(y).max() + 1e-9), -clip, clip)\n",
"\n",
"\n",
"def make_dataset(n=400, seed=7, snr=(5, 25), labels=LABELS, p=None, claim=\"SPK_A\", **chan):\n",
" \"\"\"클립 n개: 진짜는 여러 화자, 위조는 모두 지휘관 화자(SPK_A)를 사칭. 반환: (신호 배열, 메타 DataFrame).\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" X, meta = np.zeros((n, N)), []\n",
" if p is None and len(labels) > 1 and labels[0] == \"bonafide\": # 기본: 진짜 50%, 나머지 위조 유형 균등\n",
" p = [0.5] + [0.5 / (len(labels) - 1)] * (len(labels) - 1)\n",
" for i in range(n):\n",
" lab = str(rng.choice(labels, p=p))\n",
" spk = str(rng.choice(list(SPEAKERS))) if lab == \"bonafide\" else claim\n",
" if lab == \"replay\":\n",
" spk = claim\n",
" s = float(rng.uniform(*snr)) if isinstance(snr, tuple) else float(snr)\n",
" X[i] = radio_channel(make_clip(lab, spk, rng), rng, snr_db=s, **chan)\n",
" meta.append({\"clip\": f\"C{seed:02d}{i:04d}\", \"label\": lab, \"spoof\": lab != \"bonafide\", \"claimed\": spk if lab == \"bonafide\" else claim,\n",
" \"true_spk\": spk if lab == \"bonafide\" else \"unknown\", \"snr_db\": round(s, 1)})\n",
" return X, pd.DataFrame(meta)\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# ---- 전처리·음향 특징(MFCC/스펙트로그램) 공통 함수: 여러 클립을 한 번에 계산 ----\n",
"NFFT, HOP = 256, 80\n",
"\n",
"\n",
"def mel_bank(n_mels=26, fmin=300, fmax=3400):\n",
" mel = lambda f: 2595 * np.log10(1 + f / 700)\n",
" imel = lambda m: 700 * (10 ** (m / 2595) - 1)\n",
" pts = imel(np.linspace(mel(fmin), mel(fmax), n_mels + 2))\n",
" freqs = np.fft.rfftfreq(NFFT, 1 / FS)\n",
" fb = np.zeros((n_mels, len(freqs)))\n",
" for m in range(n_mels):\n",
" l, c, r = pts[m], pts[m + 1], pts[m + 2]\n",
" fb[m] = np.clip(np.minimum((freqs - l) / (c - l), (r - freqs) / (r - c)), 0, None)\n",
" return fb\n",
"\n",
"\n",
"MEL = mel_bank()\n",
"FREQS = np.fft.rfftfreq(NFFT, 1 / FS)\n",
"\n",
"\n",
"def spectrogram(X, denoise=0.0):\n",
" \"\"\"배치 전력 스펙트로그램 (클립, 주파수, 프레임). denoise>0이면 잡음 바닥(주파수별 하위 10%)을 그 배수만큼 빼는 스펙트럼 차감.\"\"\"\n",
" X = np.atleast_2d(X)\n",
" Xp = np.c_[X[:, :1], X[:, 1:] - 0.95 * X[:, :-1]] # 프리엠퍼시스\n",
" frames = np.lib.stride_tricks.sliding_window_view(Xp, NFFT, axis=1)[:, ::HOP] * np.hanning(NFFT)\n",
" P = np.abs(np.fft.rfft(frames, axis=-1)) ** 2\n",
" P = np.transpose(P, (0, 2, 1))\n",
" if denoise:\n",
" floor = np.percentile(P, 10, axis=2, keepdims=True)\n",
" P = np.maximum(P - denoise * floor, 0.05 * P)\n",
" return P\n",
"\n",
"\n",
"def mfcc(P, n=13):\n",
" logmel = np.log(np.einsum(\"mf,cft->cmt\", MEL, P) + 1e-8)\n",
" return dct(logmel, type=2, axis=1, norm=\"ortho\")[:, :n, :]\n",
"\n",
"\n",
"def band(P, lo, hi):\n",
" return P[:, (FREQS >= lo) & (FREQS < hi), :].sum(axis=(1, 2))\n",
"\n",
"\n",
"def jitter_hnr(X):\n",
" \"\"\"40ms 프레임 자기상관으로 피치 주기를 찾아 피치 흔들림(jitter)과 배음 대 잡음비(HNR 근사)를 구한다.\"\"\"\n",
" W, lag0, lag1 = 320, int(FS / 250), int(FS / 70)\n",
" fr = np.lib.stride_tricks.sliding_window_view(np.atleast_2d(X), W, axis=1)[:, ::160]\n",
" fr = fr - fr.mean(-1, keepdims=True)\n",
" spec = np.fft.rfft(fr, 2 * W, axis=-1)\n",
" ac = np.fft.irfft(np.abs(spec) ** 2, axis=-1)[..., :W]\n",
" ac = ac / (ac[..., :1] + 1e-12)\n",
" seg = ac[..., lag0:lag1]\n",
" k = seg.argmax(-1)\n",
" peak = seg.max(-1)\n",
" km, kp = np.take_along_axis(seg, np.clip(k - 1, 0, None)[..., None], -1)[..., 0], np.take_along_axis(seg, np.clip(k + 1, None, seg.shape[-1] - 1)[..., None], -1)[..., 0]\n",
" lag = k + lag0 + 0.5 * (km - kp) / (km - 2 * peak + kp - 1e-12) # 포물선 보간으로 주기 정밀화\n",
" f0 = FS / lag\n",
" jit = np.median(np.abs(np.diff(f0, axis=1)) / f0[:, 1:], axis=1)\n",
" return jit, np.median(peak, axis=1)\n",
"\n",
"\n",
"def frame_mod(X):\n",
" \"\"\"진폭 포락선(2.5ms 간격)의 50Hz 성분 비율: 20ms 프레임 경계 흔적(합성 단서).\"\"\"\n",
" env = np.abs(np.atleast_2d(X)).reshape(len(np.atleast_2d(X)), -1, 20).mean(-1)\n",
" sp = np.abs(np.fft.rfft(env - env.mean(1, keepdims=True), axis=1)) ** 2\n",
" f = np.fft.rfftfreq(env.shape[1], 20 / FS)\n",
" return sp[:, (f > 47) & (f < 53)].sum(1) / (sp[:, (f > 30) & (f < 70)].sum(1) + 1e-12)\n",
"\n",
"\n",
"FEAT_NAMES = [f\"mfcc{i}_m\" for i in range(13)] + [f\"mfcc{i}_s\" for i in range(13)] + \\\n",
" [\"delta_s\", \"hf_ratio\", \"lf_ratio\", \"flatness\", \"flux\", \"jitter\", \"hnr\", \"frame_mod\"]\n",
"\n",
"\n",
"def extract(X, denoise=0.0):\n",
" \"\"\"클립 배치 -> 특징 행렬 (클립 x 34) DataFrame.\"\"\"\n",
" P = spectrogram(X, denoise)\n",
" C = mfcc(P)\n",
" tot = band(P, 300, 3400) + 1e-12\n",
" flat = np.exp(np.log(P[:, (FREQS >= 300) & (FREQS < 3400)] + 1e-10).mean(1)) / (P[:, (FREQS >= 300) & (FREQS < 3400)].mean(1) + 1e-10)\n",
" logP = np.log(P + 1e-8)\n",
" jit, hnr = jitter_hnr(X)\n",
" F = np.c_[C.mean(2), C.std(2), np.abs(np.diff(C, axis=2)).mean((1, 2)), band(P, 2400, 3400) / tot, band(P, 300, 500) / tot,\n",
" flat.mean(1), np.abs(np.diff(logP, axis=2)).mean((1, 2)), jit, hnr, frame_mod(X)]\n",
" return pd.DataFrame(F, columns=FEAT_NAMES)\n",
"\n",
"\n",
"def enroll(seed=99):\n",
" \"\"\"지휘관·통신소 화자 6명의 등록 견본(깨끗한 진짜 클립 5개씩의 MFCC 평균).\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" tpl = {}\n",
" for spk in SPEAKERS:\n",
" Xe = np.stack([radio_channel(make_clip(\"bonafide\", spk, rng), rng, snr_db=25) for _ in range(5)])\n",
" tpl[spk] = extract(Xe)[[f\"mfcc{i}_m\" for i in range(1, 13)]].mean().values\n",
" return tpl\n",
"\n",
"\n",
"def speaker_sim(F, claimed, tpl):\n",
" \"\"\"주장 화자 견본과 MFCC 평균 벡터의 코사인 유사도(화자 검증 점수).\"\"\"\n",
" v = F[[f\"mfcc{i}_m\" for i in range(1, 13)]].values\n",
" t = np.stack([tpl[c] for c in claimed])\n",
" return (v * t).sum(1) / (np.linalg.norm(v, axis=1) * np.linalg.norm(t, axis=1) + 1e-12)\n",
"\n",
"tpl = enroll()\n",
"\n",
"\n",
"def channel_quality(X):\n",
" \"\"\"채널 품질 지표: 추정 SNR(유효 프레임 에너지 상하위 비, dB), 단일톤 재밍 지수(주파수별 시간 평균/표준편차의 최대값), 끊김 비율(무음 10ms 비율).\"\"\"\n",
" P = spectrogram(X)\n",
" e = P.sum(1)\n",
" e = np.where(e > 1e-9, e, np.nan)\n",
" snr_est = 10 * np.log10(np.nanpercentile(e, 90, axis=1) / (np.nanpercentile(e, 10, axis=1) + 1e-12))\n",
" Pb = P[:, (FREQS >= 300) & (FREQS < 3400)]\n",
" tone = (Pb.mean(2) / (Pb.std(2) + 1e-12)).max(1) # 시간에 따라 일정한 주파수 성분(재밍 톤)일수록 큼\n",
" fr = np.abs(np.atleast_2d(X)).reshape(len(X), -1, 80).max(-1)\n",
" drop = (fr < 1e-6).mean(1)\n",
" return pd.DataFrame({\"snr_est\": snr_est, \"tone_idx\": tone, \"drop_ratio\": drop})\n",
"\n",
"\n",
"def featurize(X, M, quality=False):\n",
" F = extract(X)\n",
" F[\"spk_sim\"] = speaker_sim(F, M[\"claimed\"], tpl)\n",
" return pd.concat([F, channel_quality(X)], axis=1) if quality else F\n",
"\n",
"\n",
"def mixed(n, seed, labels=LABELS):\n",
" \"\"\"[교육용가상자료] 평시·저SNR·충격 잡음·끊김·재밍 채널이 섞인 데이터셋.\"\"\"\n",
" conds = [dict(snr=(10, 25)), dict(snr=(3, 8)), dict(snr=(8, 15), impulse=0.002), dict(snr=(8, 15), dropout=0.4), dict(snr=(8, 15), jam_tone=0.5)]\n",
" parts = [make_dataset(n // 5, seed=seed + i, labels=labels, **c) for i, c in enumerate(conds)]\n",
" return np.vstack([p[0] for p in parts]), pd.concat([p[1] for p in parts], ignore_index=True)\n",
"\n",
"\n",
"# 기존(미션 4 방식): 양호 채널 위주 학습 + 임계값 0.5\n",
"Xa, Ma = make_dataset(600, seed=31, snr=(5, 25))\n",
"old = make_pipeline(StandardScaler(), MLPClassifier((64, 32), alpha=1e-3, max_iter=600, random_state=0)).fit(featurize(Xa, Ma), Ma[\"spoof\"])\n",
"# 개선: 채널 조건 혼합 학습 + 채널 품질 특징 + 로지스틱 보정(Platt)으로 점수 확률화\n",
"Xb, Mb = mixed(800, seed=40)\n",
"Fb = featurize(Xb, Mb, quality=True)\n",
"new = make_pipeline(StandardScaler(), MLPClassifier((64, 32), alpha=1e-3, max_iter=600, random_state=0)).fit(Fb, Mb[\"spoof\"])\n",
"Xc, Mc = mixed(300, seed=60)\n",
"cal = LogisticRegression().fit(new.predict_proba(featurize(Xc, Mc, True))[:, [1]], Mc[\"spoof\"])\n",
"\n",
"\n",
"def confidence(X, M):\n",
" \"\"\"위조 확률 p와 진위 신뢰도(1-p), 등급: High=위조 의심 높음(즉시 경보), Med=확인 필요(HITL), Low=진짜 가능성 높음.\"\"\"\n",
" Fq = featurize(X, M, quality=True)\n",
" p = cal.predict_proba(new.predict_proba(Fq)[:, [1]])[:, 1]\n",
" poor = (Fq[\"snr_est\"] < 3.3) | (Fq[\"tone_idx\"] > 2.5) | (Fq[\"drop_ratio\"] > 0.05)\n",
" grade = np.where(p >= 0.8, \"High\", np.where(p >= 0.4, \"Med\", \"Low\"))\n",
" grade = np.where((grade == \"High\") & poor & (p < 0.95), \"Med\", grade) # 품질 불량 채널은 한 단계 낮춰 사람 확인\n",
" return p, grade, poor\n",
"\n",
"\n",
"# [교육용가상자료] 평가: 혼합 채널 500클립\n",
"Xt, Mt = mixed(500, seed=80)\n",
"p_old = old.predict_proba(featurize(Xt, Mt))[:, 1]\n",
"p_new, g_new, poor = confidence(Xt, Mt)\n",
"y, b = Mt[\"spoof\"].values, ~Mt[\"spoof\"].values\n",
"far_old, far_new = (p_old[b] >= 0.5).mean(), (g_new[b] == \"High\").mean()\n",
"red = 1 - far_new / far_old\n",
"print(f\"평가 클립 {len(Xt)}개(진짜 {b.sum()} / 위조 {y.sum()}), 품질 불량 채널 {poor.mean():.0%}\")\n",
"print(f\"오경보율(진짜를 즉시 경보): 기존 {far_old:.1%} -> 개선 {far_new:.1%}, 감축 {red:.0%} (목표 40% 이상: {'달성' if red >= 0.4 else '미달'})\")\n",
"print(f\"위조 탐지(High+Med 등급): 기존 {(p_old[y] >= 0.5).mean():.1%} -> 개선 {(g_new[y] != 'Low').mean():.1%}\")\n",
"print(f\"ROC-AUC: 기존 {roc_auc_score(y, p_old):.3f}, 개선 {roc_auc_score(y, p_new):.3f}\")\n",
"\n",
"print(\"\\n[등급 분포] 등급 진짜 위조\")\n",
"for g in [\"High\", \"Med\", \"Low\"]:\n",
" print(f\" {g:5s} {(g_new[b] == g).sum():5d} {(g_new[y] == g).sum():5d}\")\n",
"\n",
"print(\"\\n[임계값 민감도(High 기준)] 임계값 오경보율 위조 High 비율\")\n",
"for th in [0.6, 0.7, 0.8, 0.9]:\n",
" print(f\" {th:.1f} {(p_new[b] >= th).mean():6.1%} {(p_new[y] >= th).mean():6.1%}\")\n",
"\n",
"# 일관성: 같은 원신호를 다른 잡음 실현 5회로 통과시켰을 때 등급이 같은 비율\n",
"rng = np.random.default_rng(90)\n",
"same = []\n",
"for i in range(60):\n",
" lab = LABELS[i % 4]; spk = \"SPK_A\" if lab != \"bonafide\" else str(rng.choice(list(SPEAKERS)))\n",
" x = make_clip(lab, spk, rng); s = rng.uniform(10, 25)\n",
" Xr = np.stack([radio_channel(x, np.random.default_rng(1000 + i * 10 + r), snr_db=s) for r in range(5)])\n",
" _, g, _ = confidence(Xr, pd.DataFrame({\"claimed\": [spk] * 5}))\n",
" same.append(pd.Series(g).value_counts().iloc[0] / 5)\n",
"cons = float(np.mean(same))\n",
"print(f\"\\n신뢰도 등급 일관성(동일 음성 x 잡음 실현 5회, 60건): {cons:.1%} (목표 90% 이상: {'달성' if cons >= 0.9 else '미달'})\")\n",
"print(\"등급 운용: High=통신보안관 즉시 경보 / Med=HITL 2차 확인 / Low=정상 처리(로그 보존). 등급은 판단 보조이며 진위 확정은 사람이 한다.\")\n",
"\n",
"fo, to, _ = roc_curve(y, p_old); fn, tn, _ = roc_curve(y, p_new)\n",
"plt.figure(figsize=(5, 4)); plt.plot(fo, to, label=\"baseline\"); plt.plot(fn, tn, label=\"calibrated + channel quality\")\n",
"plt.plot([0, 1], [0, 1], \"k:\"); plt.xlabel(\"false alarm rate\"); plt.ylabel(\"recall\"); plt.legend(); plt.title(\"ROC (synthetic mixed channels)\")\n",
"plt.tight_layout(); plt.show()\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 05 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"평가 클립 500개(진짜 254 / 위조 246), 품질 불량 채널 56%\n",
"오경보율(진짜를 즉시 경보): 기존 27.6% -> 개선 0.0%, 감축 100% (목표 40% 이상: 달성)\n",
"위조 탐지(High+Med 등급): 기존 98.4% -> 개선 95.9%\n",
"ROC-AUC: 기존 0.822, 개선 0.995\n",
"\n",
"[등급 분포] 등급 진짜 위조\n",
" High 0 109\n",
" Med 7 127\n",
" Low 247 10\n",
"\n",
"[임계값 민감도(High 기준)] 임계값 오경보율 위조 High 비율\n",
" 0.6 2.4% 95.1%\n",
" 0.7 2.0% 93.9%\n",
" 0.8 2.0% 93.5%\n",
" 0.9 1.6% 91.5%\n",
"\n",
"신뢰도 등급 일관성(동일 음성 x 잡음 실현 5회, 60건): 96.7% (목표 90% 이상: 달성)\n",
"등급 운용: High=통신보안관 즉시 경보 / Med=HITL 2차 확인 / Low=정상 처리(로그 보존). 등급은 판단 보조이며 진위 확정은 사람이 한다.\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 06. 실시간 무전 음성 위조 경보 및 스펙트로그램 시각화 C2 보안 인터페이스 구현\n",
"**소요시간**: 약 2.5시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"실시간 음성 스푸핑 경보 및 음향 스펙트로그램 가시화 C2 보안 모니터링 UI를 구현한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **모의 무전 송신 스트림을 받아 위조 점수와 등급을 실시간 갱신하는 야간 모드 C2 음성 보안 상황판을 구현한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **송신 1건마다 스펙트로그램, 점수 추이, 등급 집계, 경보 목록 4개 패널을 갱신하고 High 경보는 빨간 테두리로 강조 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `src/ui/c204_dashboard.py (또는 웹 UI 컴포넌트)` 형태의 결과물을 완성한다.\n",
"\n",
"### 🔍 평가 및 검증 기준 (PASS 조건)\n",
"#### 미션 판정\n",
"\n",
"**평가 구분:** 필수\n",
"\n",
"**핵심 평가영역:** 전술 작전 상황판으로서의 직관성 및 인지 부하 감소도\n",
"\n",
"#### 미션 PASS 조건\n",
"\n",
"- 정량 기준: 화면 갱신 주기 1초 이내, UI 응답성 100ms 이내\n",
"\n",
"\n",
"- 진행 기준: 주요 지표 시각화 컴포넌트 4종 이상 렌더링 및 실시간 데이터 연동\n",
"\n",
"\n",
"- 검증 방법: 사용자 인터랙션 테스트 및 모의 데이터 스트림 렌더링 검증\n",
"\n",
"\n",
"#### 채점자 판정 메모\n",
"\n",
"> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
"\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"\"\"\"모의 무전 송신 스트림을 1초 주기로 받아 위조 점수·등급을 갱신하고, 스펙트로그램·점수 추이·등급 집계·경보 목록 4개 패널로 보여 주는 야간 모드 C2 음성 보안 상황판을 구현해 화면 갱신 시간과 클립 선택 응답 시간을 측정한다.\"\"\"\n",
"import time\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"from scipy import signal\n",
"from scipy.fft import dct\n",
"from sklearn.linear_model import LogisticRegression\n",
"from sklearn.preprocessing import StandardScaler\n",
"from sklearn.pipeline import make_pipeline\n",
"\n",
"# ---------------------------------------------------------------------------\n",
"# [교육용가상자료] 탐지 학습용 합성 신호 생성기 (실제 음성·화자·통신망 정보 없음)\n",
"# 사람 목소리를 만드는 기능이 아니라, 배음(하모닉) 신호에 '진짜 신호의 자연스러운 흔들림'과\n",
"# '알려진 위조 흔적(문헌상 탐지 단서)'의 통계적 특징을 표시해 탐지기를 학습시키기 위한 추상 신호다.\n",
"# bonafide : 자연스러운 피치 흔들림(jitter)·진폭 흔들림(shimmer)·숨소리 잡음\n",
"# tts_art : 흔들림이 거의 없고 고주파 에너지가 부족하며 20ms 프레임 경계 흔적이 있는 신호\n",
"# vc_art : 프레임마다 배음 위상이 끊기고 스펙트럼 포락선이 뭉개진 신호\n",
"# replay : 스피커 재생 대역 제한 + 실내 잔향이 더해진 신호\n",
"# ---------------------------------------------------------------------------\n",
"FS, DUR = 8000, 0.5\n",
"N = int(FS * DUR)\n",
"LABELS = [\"bonafide\", \"tts_art\", \"vc_art\", \"replay\"]\n",
"SPEAKERS = {f\"SPK_{c}\": (f0, fm) for c, f0, fm in [(\"A\", 115, (650, 1150, 2450)), (\"B\", 135, (700, 1220, 2600)), (\"C\", 98, (600, 1050, 2350)),\n",
" (\"D\", 160, (720, 1300, 2700)), (\"E\", 125, (680, 1180, 2500)), (\"F\", 105, (630, 1100, 2400))]}\n",
"\n",
"\n",
"def _source(rng, f0, fm, jitter, shimmer, phase_break=False, hf_cut=None, frame_mod=0.0):\n",
" t = np.arange(N) / FS\n",
" contour = f0 * (1 + 0.06 * np.sin(2 * np.pi * rng.uniform(1.5, 3.5) * t + rng.uniform(0, 6)))\n",
" f = contour * (1 + jitter * rng.normal(0, 1, N // 80 + 1).repeat(80)[:N]) # 10ms 단위 피치 흔들림(jitter)\n",
" phase = 2 * np.pi * np.cumsum(f) / FS\n",
" K = int(3800 / f0)\n",
" fk = np.arange(1, K + 1) * f0\n",
" amp = sum(np.exp(-((fk - F) / 120.0) ** 2) / (i + 1) for i, F in enumerate(fm)) + 0.02 / np.arange(1, K + 1)\n",
" if hf_cut:\n",
" amp = np.where(fk > hf_cut, amp * 0.2, amp)\n",
" ph = np.arange(1, K + 1)[:, None] * phase[None, :]\n",
" if phase_break: # 20ms 프레임마다 배음별 위상 불연속\n",
" ph = ph + rng.uniform(0, 2 * np.pi, (K, N // 160 + 1)).repeat(160, axis=1)[:, :N]\n",
" sig = (amp[:, None] * np.sin(ph)).sum(0)\n",
" env = 0.55 + 0.45 * np.sin(2 * np.pi * rng.uniform(3, 5) * t + rng.uniform(0, 6)) ** 2 # 음절형 진폭 변화\n",
" env *= 1 + shimmer * np.interp(t, np.linspace(0, DUR, 60), rng.normal(0, 1, 60)) # 완만한 진폭 흔들림(shimmer)\n",
" if frame_mod:\n",
" env *= 1 + frame_mod * np.sign(np.sin(2 * np.pi * 50 * t)) # 20ms 프레임 경계 흔적(50Hz)\n",
" return sig * env + (0.03 if jitter > 0.003 else 0.005) * rng.normal(0, 1, N) * env\n",
"\n",
"\n",
"def make_clip(label, speaker, rng):\n",
" f0, fm = SPEAKERS[speaker]\n",
" if label in (\"tts_art\", \"vc_art\"): # 위조는 화자 특성을 근사만 함\n",
" f0 = f0 * rng.uniform(0.92, 1.08); fm = tuple(F * rng.uniform(0.95, 1.05) for F in fm)\n",
" if label == \"bonafide\" or label == \"replay\":\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.008, 0.02), shimmer=rng.uniform(0.05, 0.12))\n",
" elif label == \"tts_art\":\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.0, 0.002), shimmer=0.01, hf_cut=rng.uniform(2300, 2800), frame_mod=rng.uniform(0.08, 0.2))\n",
" else:\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.006, 0.015), shimmer=rng.uniform(0.04, 0.1), phase_break=True)\n",
" if label == \"replay\": # 스피커 대역 + 실내 잔향\n",
" x = signal.sosfilt(signal.butter(4, [450, 2900], \"bandpass\", fs=FS, output=\"sos\"), x)\n",
" ir = rng.normal(0, 1, int(0.08 * FS)) * np.exp(-np.arange(int(0.08 * FS)) / (0.02 * FS)); ir[0] = 3.0\n",
" x = signal.fftconvolve(x, ir)[:N]\n",
" return x / (np.abs(x).max() + 1e-9)\n",
"\n",
"\n",
"RADIO_BP = signal.butter(4, [300, 3400], \"bandpass\", fs=FS, output=\"sos\")\n",
"\n",
"\n",
"def radio_channel(x, rng, snr_db=15.0, impulse=0.0, dropout=0.0, clip=1.0, jam_tone=0.0):\n",
" \"\"\"[교육용가상자료] 무전 채널 모형: 300~3400Hz 대역 제한, 배경 잡음(SNR), 충격성 잡음, 신호 끊김, 포화(클리핑), 단일톤 재밍.\"\"\"\n",
" y = signal.sosfilt(RADIO_BP, x)\n",
" p = np.mean(y ** 2)\n",
" noise = rng.normal(0, 1, N) * np.sqrt(p / 10 ** (snr_db / 10))\n",
" noise += np.cumsum(rng.normal(0, 0.02, N)) * np.sqrt(p) * 0.05 # 저주파 차량·엔진형 소음\n",
" y = y + noise\n",
" if impulse:\n",
" k = rng.random(N) < impulse\n",
" y[k] += rng.normal(0, 6 * np.sqrt(p), k.sum())\n",
" if jam_tone:\n",
" y += jam_tone * np.sqrt(p) * np.sin(2 * np.pi * rng.uniform(800, 2000) * np.arange(N) / FS)\n",
" if dropout:\n",
" for _ in range(int(dropout * 10)):\n",
" a = int(rng.integers(0, N - 400)); y[a:a + int(rng.integers(100, 400))] = 0.0\n",
" return np.clip(y / (np.abs(y).max() + 1e-9), -clip, clip)\n",
"\n",
"\n",
"def make_dataset(n=400, seed=7, snr=(5, 25), labels=LABELS, p=None, claim=\"SPK_A\", **chan):\n",
" \"\"\"클립 n개: 진짜는 여러 화자, 위조는 모두 지휘관 화자(SPK_A)를 사칭. 반환: (신호 배열, 메타 DataFrame).\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" X, meta = np.zeros((n, N)), []\n",
" if p is None and len(labels) > 1 and labels[0] == \"bonafide\": # 기본: 진짜 50%, 나머지 위조 유형 균등\n",
" p = [0.5] + [0.5 / (len(labels) - 1)] * (len(labels) - 1)\n",
" for i in range(n):\n",
" lab = str(rng.choice(labels, p=p))\n",
" spk = str(rng.choice(list(SPEAKERS))) if lab == \"bonafide\" else claim\n",
" if lab == \"replay\":\n",
" spk = claim\n",
" s = float(rng.uniform(*snr)) if isinstance(snr, tuple) else float(snr)\n",
" X[i] = radio_channel(make_clip(lab, spk, rng), rng, snr_db=s, **chan)\n",
" meta.append({\"clip\": f\"C{seed:02d}{i:04d}\", \"label\": lab, \"spoof\": lab != \"bonafide\", \"claimed\": spk if lab == \"bonafide\" else claim,\n",
" \"true_spk\": spk if lab == \"bonafide\" else \"unknown\", \"snr_db\": round(s, 1)})\n",
" return X, pd.DataFrame(meta)\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# ---- 전처리·음향 특징(MFCC/스펙트로그램) 공통 함수: 여러 클립을 한 번에 계산 ----\n",
"NFFT, HOP = 256, 80\n",
"\n",
"\n",
"def mel_bank(n_mels=26, fmin=300, fmax=3400):\n",
" mel = lambda f: 2595 * np.log10(1 + f / 700)\n",
" imel = lambda m: 700 * (10 ** (m / 2595) - 1)\n",
" pts = imel(np.linspace(mel(fmin), mel(fmax), n_mels + 2))\n",
" freqs = np.fft.rfftfreq(NFFT, 1 / FS)\n",
" fb = np.zeros((n_mels, len(freqs)))\n",
" for m in range(n_mels):\n",
" l, c, r = pts[m], pts[m + 1], pts[m + 2]\n",
" fb[m] = np.clip(np.minimum((freqs - l) / (c - l), (r - freqs) / (r - c)), 0, None)\n",
" return fb\n",
"\n",
"\n",
"MEL = mel_bank()\n",
"FREQS = np.fft.rfftfreq(NFFT, 1 / FS)\n",
"\n",
"\n",
"def spectrogram(X, denoise=0.0):\n",
" \"\"\"배치 전력 스펙트로그램 (클립, 주파수, 프레임). denoise>0이면 잡음 바닥(주파수별 하위 10%)을 그 배수만큼 빼는 스펙트럼 차감.\"\"\"\n",
" X = np.atleast_2d(X)\n",
" Xp = np.c_[X[:, :1], X[:, 1:] - 0.95 * X[:, :-1]] # 프리엠퍼시스\n",
" frames = np.lib.stride_tricks.sliding_window_view(Xp, NFFT, axis=1)[:, ::HOP] * np.hanning(NFFT)\n",
" P = np.abs(np.fft.rfft(frames, axis=-1)) ** 2\n",
" P = np.transpose(P, (0, 2, 1))\n",
" if denoise:\n",
" floor = np.percentile(P, 10, axis=2, keepdims=True)\n",
" P = np.maximum(P - denoise * floor, 0.05 * P)\n",
" return P\n",
"\n",
"\n",
"def mfcc(P, n=13):\n",
" logmel = np.log(np.einsum(\"mf,cft->cmt\", MEL, P) + 1e-8)\n",
" return dct(logmel, type=2, axis=1, norm=\"ortho\")[:, :n, :]\n",
"\n",
"\n",
"def band(P, lo, hi):\n",
" return P[:, (FREQS >= lo) & (FREQS < hi), :].sum(axis=(1, 2))\n",
"\n",
"\n",
"def jitter_hnr(X):\n",
" \"\"\"40ms 프레임 자기상관으로 피치 주기를 찾아 피치 흔들림(jitter)과 배음 대 잡음비(HNR 근사)를 구한다.\"\"\"\n",
" W, lag0, lag1 = 320, int(FS / 250), int(FS / 70)\n",
" fr = np.lib.stride_tricks.sliding_window_view(np.atleast_2d(X), W, axis=1)[:, ::160]\n",
" fr = fr - fr.mean(-1, keepdims=True)\n",
" spec = np.fft.rfft(fr, 2 * W, axis=-1)\n",
" ac = np.fft.irfft(np.abs(spec) ** 2, axis=-1)[..., :W]\n",
" ac = ac / (ac[..., :1] + 1e-12)\n",
" seg = ac[..., lag0:lag1]\n",
" k = seg.argmax(-1)\n",
" peak = seg.max(-1)\n",
" km, kp = np.take_along_axis(seg, np.clip(k - 1, 0, None)[..., None], -1)[..., 0], np.take_along_axis(seg, np.clip(k + 1, None, seg.shape[-1] - 1)[..., None], -1)[..., 0]\n",
" lag = k + lag0 + 0.5 * (km - kp) / (km - 2 * peak + kp - 1e-12) # 포물선 보간으로 주기 정밀화\n",
" f0 = FS / lag\n",
" jit = np.median(np.abs(np.diff(f0, axis=1)) / f0[:, 1:], axis=1)\n",
" return jit, np.median(peak, axis=1)\n",
"\n",
"\n",
"def frame_mod(X):\n",
" \"\"\"진폭 포락선(2.5ms 간격)의 50Hz 성분 비율: 20ms 프레임 경계 흔적(합성 단서).\"\"\"\n",
" env = np.abs(np.atleast_2d(X)).reshape(len(np.atleast_2d(X)), -1, 20).mean(-1)\n",
" sp = np.abs(np.fft.rfft(env - env.mean(1, keepdims=True), axis=1)) ** 2\n",
" f = np.fft.rfftfreq(env.shape[1], 20 / FS)\n",
" return sp[:, (f > 47) & (f < 53)].sum(1) / (sp[:, (f > 30) & (f < 70)].sum(1) + 1e-12)\n",
"\n",
"\n",
"FEAT_NAMES = [f\"mfcc{i}_m\" for i in range(13)] + [f\"mfcc{i}_s\" for i in range(13)] + \\\n",
" [\"delta_s\", \"hf_ratio\", \"lf_ratio\", \"flatness\", \"flux\", \"jitter\", \"hnr\", \"frame_mod\"]\n",
"\n",
"\n",
"def extract(X, denoise=0.0):\n",
" \"\"\"클립 배치 -> 특징 행렬 (클립 x 34) DataFrame.\"\"\"\n",
" P = spectrogram(X, denoise)\n",
" C = mfcc(P)\n",
" tot = band(P, 300, 3400) + 1e-12\n",
" flat = np.exp(np.log(P[:, (FREQS >= 300) & (FREQS < 3400)] + 1e-10).mean(1)) / (P[:, (FREQS >= 300) & (FREQS < 3400)].mean(1) + 1e-10)\n",
" logP = np.log(P + 1e-8)\n",
" jit, hnr = jitter_hnr(X)\n",
" F = np.c_[C.mean(2), C.std(2), np.abs(np.diff(C, axis=2)).mean((1, 2)), band(P, 2400, 3400) / tot, band(P, 300, 500) / tot,\n",
" flat.mean(1), np.abs(np.diff(logP, axis=2)).mean((1, 2)), jit, hnr, frame_mod(X)]\n",
" return pd.DataFrame(F, columns=FEAT_NAMES)\n",
"\n",
"\n",
"def enroll(seed=99):\n",
" \"\"\"지휘관·통신소 화자 6명의 등록 견본(깨끗한 진짜 클립 5개씩의 MFCC 평균).\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" tpl = {}\n",
" for spk in SPEAKERS:\n",
" Xe = np.stack([radio_channel(make_clip(\"bonafide\", spk, rng), rng, snr_db=25) for _ in range(5)])\n",
" tpl[spk] = extract(Xe)[[f\"mfcc{i}_m\" for i in range(1, 13)]].mean().values\n",
" return tpl\n",
"\n",
"\n",
"def speaker_sim(F, claimed, tpl):\n",
" \"\"\"주장 화자 견본과 MFCC 평균 벡터의 코사인 유사도(화자 검증 점수).\"\"\"\n",
" v = F[[f\"mfcc{i}_m\" for i in range(1, 13)]].values\n",
" t = np.stack([tpl[c] for c in claimed])\n",
" return (v * t).sum(1) / (np.linalg.norm(v, axis=1) * np.linalg.norm(t, axis=1) + 1e-12)\n",
"\n",
"tpl = enroll()\n",
"\n",
"\n",
"def featurize(X, M):\n",
" F = extract(X)\n",
" F[\"spk_sim\"] = speaker_sim(F, M[\"claimed\"], tpl)\n",
" return F\n",
"\n",
"\n",
"Xtr, Mtr = make_dataset(400, seed=51, snr=(5, 25))\n",
"det = make_pipeline(StandardScaler(), LogisticRegression(max_iter=2000, C=0.5)).fit(featurize(Xtr, Mtr), Mtr[\"spoof\"])\n",
"\n",
"# [교육용가상자료] 모의 송신 스트림 24건(가상 망 NET-A~C, 가상 호출부호)\n",
"Xs, Ms = make_dataset(24, seed=52, snr=(6, 25), p=[0.6, 0.15, 0.15, 0.1])\n",
"Ms[\"net\"] = [f\"NET-{'ABC'[i % 3]}\" for i in range(24)]\n",
"Ms[\"t\"] = [f\"21:00:{i:02d}\" for i in range(24)]\n",
"state = pd.DataFrame(columns=[\"t\", \"net\", \"clip\", \"claimed\", \"score\", \"grade\"])\n",
"\n",
"plt.style.use(\"dark_background\") # 야간 작전 모드\n",
"fig, ax = plt.subplots(2, 2, figsize=(10, 6))\n",
"upd = []\n",
"for i in range(len(Xs)):\n",
" t0 = time.perf_counter()\n",
" s = float(det.predict_proba(featurize(Xs[i:i + 1], Ms.iloc[i:i + 1]))[0, 1])\n",
" g = \"High\" if s >= 0.8 else \"Med\" if s >= 0.4 else \"Low\"\n",
" state.loc[i] = [Ms[\"t\"][i], Ms[\"net\"][i], Ms[\"clip\"][i], Ms[\"claimed\"][i], round(s, 2), g]\n",
" for a in ax.flat:\n",
" a.clear()\n",
" P = spectrogram(Xs[i:i + 1])[0]\n",
" ax[0, 0].imshow(10 * np.log10(P + 1e-8), origin=\"lower\", aspect=\"auto\", cmap=\"magma\", extent=[0, DUR, 0, FS / 2])\n",
" ax[0, 0].set_title(f\"Spectrogram {Ms['clip'][i]} ({Ms['net'][i]})\"); ax[0, 0].set_xlabel(\"s\"); ax[0, 0].set_ylabel(\"Hz\")\n",
" col = state[\"grade\"].map({\"High\": \"red\", \"Med\": \"orange\", \"Low\": \"lime\"})\n",
" ax[0, 1].scatter(range(len(state)), state[\"score\"], c=col); ax[0, 1].axhline(0.8, c=\"red\", ls=\"--\"); ax[0, 1].axhline(0.4, c=\"orange\", ls=\":\")\n",
" ax[0, 1].set_ylim(0, 1); ax[0, 1].set_title(\"Spoof score timeline\"); ax[0, 1].set_xlabel(\"transmission #\")\n",
" cnt = state[\"grade\"].value_counts().reindex([\"High\", \"Med\", \"Low\"]).fillna(0)\n",
" ax[1, 0].bar(cnt.index, cnt.values, color=[\"red\", \"orange\", \"lime\"]); ax[1, 0].set_title(\"Grade count\")\n",
" al = state[state[\"grade\"] != \"Low\"].tail(6)\n",
" ax[1, 1].axis(\"off\"); ax[1, 1].set_title(\"Alert list (operator review)\")\n",
" for k, r in enumerate(al.itertuples()):\n",
" ax[1, 1].text(0.0, 0.9 - k * 0.15, f\"{r.t} {r.net} {r.clip} claim {r.claimed} score {r.score:.2f} {r.grade}\",\n",
" color=\"red\" if r.grade == \"High\" else \"orange\", fontsize=8, family=\"monospace\")\n",
" if g == \"High\": # 긴급 경보 하이라이트\n",
" for a in ax.flat:\n",
" for sp in a.spines.values():\n",
" sp.set_color(\"red\"); sp.set_linewidth(2)\n",
" fig.canvas.draw()\n",
" upd.append((time.perf_counter() - t0) * 1000)\n",
"\n",
"# 클립 선택 응답성: 경보 목록에서 한 건을 고르면 상세 정보(단서 상위 3개)를 찾아 보여 주는 콜백\n",
"coef = pd.Series(det[-1].coef_[0], index=featurize(Xs[:1], Ms.iloc[:1]).columns)\n",
"Fall = featurize(Xs, Ms)\n",
"Z = pd.DataFrame(det[0].transform(Fall), columns=Fall.columns)\n",
"\n",
"\n",
"def on_select(idx):\n",
" contrib = (Z.loc[idx] * coef).sort_values(ascending=False).head(3)\n",
" return {\"clip\": Ms[\"clip\"][idx], \"grade\": state[\"grade\"][idx], \"top_cues\": [f\"{k}({v:+.2f})\" for k, v in contrib.items()]}\n",
"\n",
"\n",
"resp = []\n",
"for idx in state.index[state[\"grade\"] != \"Low\"]:\n",
" t0 = time.perf_counter(); info = on_select(idx); resp.append((time.perf_counter() - t0) * 1000)\n",
"\n",
"print(\"[실시간 C2 음성 보안 상황판 - 교육용가상자료]\")\n",
"print(state.tail(8).to_string(index=False))\n",
"print(f\"\\n패널 4종(스펙트로그램, 점수 추이, 등급 집계, 경보 목록) 렌더링: {len(upd)}회 갱신\")\n",
"print(f\"화면 갱신 시간 최대 {max(upd):.0f} ms, 평균 {np.mean(upd):.0f} ms (목표 1초 이내: {'달성' if max(upd) < 1000 else '미달'})\")\n",
"print(f\"클립 선택 응답 최대 {max(resp):.1f} ms (목표 100ms 이내: {'달성' if max(resp) < 100 else '미달'})\")\n",
"print(f\"경보 {len(resp)}건 중 High {(state['grade'] == 'High').sum()}건(빨강 테두리 하이라이트), Med {(state['grade'] == 'Med').sum()}건\")\n",
"print(\"선택 예시:\", info)\n",
"print(\"실제 위조 여부(정답) 대비:\", f\"위조 {Ms['spoof'].sum()}건 중 경보 {(state['grade'] != 'Low')[Ms['spoof']].sum()}건, 진짜 {(~Ms['spoof']).sum()}건 중 경보 {(state['grade'] != 'Low')[~Ms['spoof']].sum()}건\")\n",
"print(\"상황판 경보는 판단 보조이며, 위조 확정과 전파는 통신보안관 확인(미션 7 절차) 후에 한다.\")\n",
"plt.tight_layout(); plt.show()\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 06 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"[실시간 C2 음성 보안 상황판 - 교육용가상자료]\n",
" t net clip claimed score grade\n",
"21:00:16 NET-B C520016 SPK_F 0.01 Low\n",
"21:00:17 NET-C C520017 SPK_A 1.00 High\n",
"21:00:18 NET-A C520018 SPK_D 0.00 Low\n",
"21:00:19 NET-B C520019 SPK_F 0.10 Low\n",
"21:00:20 NET-C C520020 SPK_A 1.00 High\n",
"21:00:21 NET-A C520021 SPK_A 0.91 High\n",
"21:00:22 NET-B C520022 SPK_A 1.00 High\n",
"21:00:23 NET-C C520023 SPK_D 0.00 Low\n",
"\n",
"패널 4종(스펙트로그램, 점수 추이, 등급 집계, 경보 목록) 렌더링: 24회 갱신\n",
"화면 갱신 시간 최대 334 ms, 평균 211 ms (목표 1초 이내: 달성)\n",
"클립 선택 응답 최대 0.8 ms (목표 100ms 이내: 달성)\n",
"경보 12건 중 High 12건(빨강 테두리 하이라이트), Med 0건\n",
"선택 예시: {'clip': 'C520022', 'grade': 'High', 'top_cues': ['mfcc0_m(+2.59)', 'hnr(+1.73)', 'delta_s(+1.60)']}\n",
"실제 위조 여부(정답) 대비: 위조 12건 중 경보 12건, 진짜 12건 중 경보 0건\n",
"상황판 경보는 판단 보조이며, 위조 확정과 전파는 통신보안관 확인(미션 7 절차) 후에 한다.\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 07. 지휘관 긴급 음성 명령 수신 시 통신관 HITL 2차 육안/청음 검증 절차 설계\n",
"**소요시간**: 약 3.0시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"지휘관 긴급 음성 명령 수신 시 통신보안관이 2차 확인 및 암호 인증을 수행하는 HITL 절차를 설계한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **지휘관 명의 긴급 음성 명령이 들어오면 통신보안관이 인증 질의응답과 청음으로 2차 확인한 뒤에만 전파하는 HITL 검증 절차를 설계한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **긴급 명령은 탐지 점수와 무관하게 검증 대기열에 올리고 등급·점수 순으로 검토 순서를 정 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `src/hitl/c204_approval_workflow.py 및 docs/c204_hitl_guide.md` 형태의 결과물을 완성한다.\n",
"\n",
"### 🔍 평가 및 검증 기준 (PASS 조건)\n",
"#### 미션 판정\n",
"\n",
"**평가 구분:** 필수\n",
"\n",
"**핵심 평가영역:** 지휘 책임 분계선 명확화 및 성급한 자동 단정 방지 체계 부합성\n",
"\n",
"#### 미션 PASS 조건\n",
"\n",
"- 정량 기준: 인간 검토 큐(Queue) 대기시간 및 승인/반려 처리 로깅 성공률 100%\n",
"\n",
"\n",
"- 진행 기준: 승인/반려 인터페이스 동작 및 감사 로그(Audit Log) 생성 검증\n",
"\n",
"\n",
"- 검증 방법: 모의 조작자 승인/반려 시뮬레이션 및 데이터 무결성 검증\n",
"\n",
"\n",
"#### 채점자 판정 메모\n",
"\n",
"> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
"\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"\"\"\"지휘관 명의 긴급 음성 명령이 수신되면 탐지 점수와 무관하게 통신보안관 2차 검증 대기열에 올리고, 가상 인증 질의응답(챌린지-응답)·청음 확인·근거 기록을 거쳐야만 전파되도록 하는 HITL 검증 절차와 해시 체인 감사 로그를 구현한다.\"\"\"\n",
"import hashlib\n",
"import hmac\n",
"import json\n",
"import pandas as pd\n",
"\n",
"ACTIONS = {\"CONFIRM\": \"진위 확인(명령 전파 허용)\", \"HOLD\": \"보류(재송신·추가 확인 요청)\", \"REJECT\": \"위조 의심 차단(상급 보고)\"}\n",
"PRIORITY = {\"High\": 0, \"Med\": 1, \"Low\": 2}\n",
"HOLD_LIMIT_S = 120\n",
"AUTH_KEY = b\"EDU-VIRTUAL-KEY\" # [교육용가상자료] 실제 인증 체계·암호 아님\n",
"\n",
"\n",
"class WorkflowError(ValueError):\n",
" pass\n",
"\n",
"\n",
"def expected_response(challenge, net):\n",
" \"\"\"가상 챌린지-응답: 질의어와 망 ID로 4자리 응답을 만든다(실제 암호 체계를 흉내 낸 학습용 모형).\"\"\"\n",
" return hmac.new(AUTH_KEY, f\"{net}:{challenge}\".encode(), hashlib.sha256).hexdigest()[:4].upper()\n",
"\n",
"\n",
"class VerifyQueue:\n",
" def __init__(self, clock):\n",
" self.items, self.log, self.clock, self.feedback = {}, [], clock, []\n",
"\n",
" def _record(self, body):\n",
" prev = self.log[-1][\"hash\"] if self.log else \"GENESIS\"\n",
" body = {**body, \"ts\": round(self.clock(), 1)}\n",
" h = hashlib.sha256((prev + json.dumps(body, ensure_ascii=False, sort_keys=True)).encode()).hexdigest()\n",
" self.log.append({**body, \"prev\": prev, \"hash\": h})\n",
"\n",
" def verify_log(self):\n",
" prev = \"GENESIS\"\n",
" for n, rec in enumerate(self.log, start=1):\n",
" body = {k: v for k, v in rec.items() if k not in (\"prev\", \"hash\")}\n",
" if rec[\"prev\"] != prev or rec[\"hash\"] != hashlib.sha256((prev + json.dumps(body, ensure_ascii=False, sort_keys=True)).encode()).hexdigest():\n",
" return False, n\n",
" prev = rec[\"hash\"]\n",
" return True, -1\n",
"\n",
" def submit(self, mid, net, claimed, score, grade, order):\n",
" \"\"\"긴급 명령 등록. 점수가 낮아도(진짜로 보여도) 자동 전파하지 않고 PENDING으로 시작한다.\"\"\"\n",
" self.items[mid] = {\"id\": mid, \"net\": net, \"claimed\": claimed, \"score\": score, \"grade\": grade, \"order\": order,\n",
" \"status\": \"PENDING\", \"submitted\": self.clock(), \"hold_since\": None, \"escalated\": False, \"auth\": None}\n",
" self._record({\"event\": \"SUBMIT\", \"id\": mid, \"score\": score, \"grade\": grade, \"by\": \"DETECTOR\"})\n",
"\n",
" def pending(self):\n",
" todo = [i for i in self.items.values() if i[\"status\"] in (\"PENDING\", \"HOLD\")]\n",
" return sorted(todo, key=lambda i: (not i[\"escalated\"], PRIORITY[i[\"grade\"]], -i[\"score\"], i[\"submitted\"]))\n",
"\n",
" def authenticate(self, mid, challenge, response, operator):\n",
" \"\"\"송신자에게 역호출로 가상 질의어를 보내고 받은 응답을 대조한다. 결과는 기록만 하고 결정은 사람이 한다.\"\"\"\n",
" ok = hmac.compare_digest(expected_response(challenge, self.items[mid][\"net\"]), response)\n",
" self.items[mid][\"auth\"] = ok\n",
" self._record({\"event\": \"AUTH\", \"id\": mid, \"challenge\": challenge, \"result\": ok, \"operator\": operator})\n",
" return ok\n",
"\n",
" def decide(self, mid, action, operator, reason, listened=False, second=None):\n",
" if action not in ACTIONS:\n",
" raise WorkflowError(f\"허용되지 않은 조치: {action}\")\n",
" if not operator:\n",
" raise WorkflowError(\"통신보안관 ID 필요\")\n",
" it = self.items.get(mid)\n",
" if it is None or it[\"status\"] in (\"CONFIRMED\", \"REJECTED\"):\n",
" raise WorkflowError(\"대기열에 없거나 이미 종결된 명령\")\n",
" if len(reason.strip()) < 5:\n",
" raise WorkflowError(\"판단 근거 5자 이상 필요\")\n",
" if action == \"CONFIRM\":\n",
" if not listened:\n",
" raise WorkflowError(\"전파 전 청음(직접 듣기) 확인 필요\")\n",
" if it[\"auth\"] is not True:\n",
" raise WorkflowError(\"인증 질의응답 통과 전 전파 불가\")\n",
" if it[\"grade\"] == \"High\" and (not second or second == operator):\n",
" raise WorkflowError(\"High 등급 확인은 다른 검토자 1명의 교차 확인 필요\")\n",
" it[\"status\"] = {\"CONFIRM\": \"CONFIRMED\", \"HOLD\": \"HOLD\", \"REJECT\": \"REJECTED\"}[action]\n",
" it[\"hold_since\"] = self.clock() if action == \"HOLD\" else None\n",
" it[\"decided\"] = self.clock()\n",
" self._record({\"event\": \"DECIDE\", \"id\": mid, \"action\": action, \"operator\": operator, \"second\": second, \"listened\": listened, \"reason\": reason})\n",
"\n",
" def check_holds(self):\n",
" \"\"\"보류 기한 초과 시 자동 확인하지 않고 상급 지휘관(당직 통제관)에게 상신한다.\"\"\"\n",
" out = []\n",
" for i in self.items.values():\n",
" if i[\"status\"] == \"HOLD\" and not i[\"escalated\"] and self.clock() - i[\"hold_since\"] > HOLD_LIMIT_S:\n",
" i[\"escalated\"] = True; out.append(i[\"id\"])\n",
" self._record({\"event\": \"ESCALATE\", \"id\": i[\"id\"], \"by\": \"SYSTEM\"})\n",
" return out\n",
"\n",
" def commander_feedback(self, mid, text):\n",
" self.feedback.append({\"id\": mid, \"text\": text})\n",
" self._record({\"event\": \"FEEDBACK\", \"id\": mid, \"text\": text})\n",
"\n",
"\n",
"class FakeClock:\n",
" t = 0.0\n",
"\n",
" def __call__(self):\n",
" return self.t\n",
"\n",
"\n",
"clock = FakeClock()\n",
"q = VerifyQueue(clock)\n",
"# [교육용가상자료] 미션 5 신뢰도 엔진에서 넘어온 가상 긴급 명령 6건(가상 망·가상 호출부호)\n",
"for mid, net, spk, s, g, od in [(\"U01\", \"NET-A\", \"SPK_A\", 0.97, \"High\", \"가상 집결지 변경\"), (\"U02\", \"NET-B\", \"SPK_A\", 0.62, \"Med\", \"통신 축선 변경\"),\n",
" (\"U03\", \"NET-A\", \"SPK_A\", 0.08, \"Low\", \"보고 주기 단축\"), (\"U04\", \"NET-C\", \"SPK_A\", 0.88, \"High\", \"가상 진지 이동\"),\n",
" (\"U05\", \"NET-B\", \"SPK_A\", 0.45, \"Med\", \"주파수 전환\"), (\"U06\", \"NET-C\", \"SPK_A\", 0.12, \"Low\", \"정시 보고 요청\")]:\n",
" clock.t += 5; q.submit(mid, net, spk, s, g, od)\n",
"print(\"검토 순서(등급·점수 순):\", [i[\"id\"] for i in q.pending()])\n",
"print(\"자동 전파(확인)된 명령 수:\", sum(i[\"status\"] == \"CONFIRMED\" for i in q.items.values()))\n",
"\n",
"# 규칙 위반 시도: 모두 차단되어야 한다\n",
"viol = [(\"U03\", \"CONFIRM\", \"sec01\", \"진짜 점수 낮음\", dict(listened=True)), # 인증 없이 전파\n",
" (\"U01\", \"CONFIRM\", \"\", \"근거 충분\", {}), (\"U02\", \"FORWARD\", \"sec01\", \"임의 조치\", {}),\n",
" (\"U06\", \"REJECT\", \"sec01\", \"의심\", {})]\n",
"blocked = 0\n",
"for mid, a, op, r, kw in viol:\n",
" try:\n",
" q.decide(mid, a, op, r, **kw)\n",
" except WorkflowError as e:\n",
" blocked += 1; print(f\"차단({mid} {a}): {e}\")\n",
"\n",
"# 정상 처리: 인증 질의응답 -> 청음 -> 결정\n",
"clock.t += 10; q.authenticate(\"U03\", \"ECHO\", expected_response(\"ECHO\", \"NET-A\"), \"sec01\")\n",
"q.decide(\"U03\", \"CONFIRM\", \"sec01\", \"역호출 인증 통과, 청음상 평소 송신자와 일치\", listened=True)\n",
"clock.t += 8; ok = q.authenticate(\"U01\", \"TANGO\", \"0000\", \"sec01\")\n",
"q.decide(\"U01\", \"REJECT\", \"sec01\", \"인증 응답 불일치, 프레임 경계 단서 다수\", listened=True)\n",
"clock.t += 6; q.authenticate(\"U04\", \"LIMA\", expected_response(\"LIMA\", \"NET-C\"), \"sec02\")\n",
"try:\n",
" q.decide(\"U04\", \"CONFIRM\", \"sec02\", \"인증 통과, 청음 정상\", listened=True)\n",
"except WorkflowError as e:\n",
" blocked += 1; print(f\"차단(U04 CONFIRM 단독): {e}\")\n",
"q.decide(\"U04\", \"CONFIRM\", \"sec02\", \"인증 통과, 청음 정상, 교차 확인 완료\", listened=True, second=\"sec01\")\n",
"clock.t += 4; q.decide(\"U02\", \"HOLD\", \"sec02\", \"재송신 요청 후 재청음\")\n",
"clock.t += 3; q.authenticate(\"U06\", \"MIKE\", expected_response(\"MIKE\", \"NET-C\"), \"sec02\")\n",
"q.decide(\"U06\", \"CONFIRM\", \"sec02\", \"역호출 인증 통과, 일상 보고 요청\", listened=True)\n",
"clock.t += 150\n",
"print(\"보류 기한 초과 상신:\", q.check_holds(), \"/ 상신 후 검토 순서:\", [i[\"id\"] for i in q.pending()])\n",
"q.commander_feedback(\"U01\", \"위조 차단 적절. 인증 질의어 교체 주기 단축 검토\")\n",
"\n",
"df = pd.DataFrame(q.items.values())\n",
"df[\"대기(s)\"] = (df[\"decided\"] - df[\"submitted\"]).round(1)\n",
"print(\"\\n\" + df[[\"id\", \"grade\", \"score\", \"auth\", \"status\", \"escalated\", \"대기(s)\"]].fillna(\"-\").to_string(index=False))\n",
"n_dec = sum(r[\"event\"] == \"DECIDE\" for r in q.log)\n",
"n_exp = sum(i[\"status\"] in (\"CONFIRMED\", \"REJECTED\", \"HOLD\") for i in q.items.values())\n",
"n_auto = sum(r[\"event\"] == \"DECIDE\" and not r[\"operator\"] for r in q.log)\n",
"print(f\"\\n자동 전파 {n_auto}건 / 규칙 위반 차단 {blocked}건 / 결정 로그 {n_dec}건(결정 {n_exp}건, 로깅 성공률 {n_dec / n_exp:.0%})\")\n",
"print(f\"평균 검토 대기 {df['대기(s)'].mean():.1f}s, 감사 로그 {len(q.log)}건, 해시 체인 검증: {q.verify_log()}\")\n",
"q.log[7][\"operator\"] = \"sec99\"\n",
"print(\"8번째 기록 변조 후 검증(False, 깨진 위치):\", q.verify_log())\n",
"print(\"\\n책임 분계: 탐지기=점수 제시, 통신보안관=인증·청음·결정, 지휘관=최종 명령 책임. 자동 판정으로 명령을 전파하거나 차단하지 않는다.\")\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 07 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"검토 순서(등급·점수 순): ['U01', 'U04', 'U02', 'U05', 'U06', 'U03']\n",
"자동 전파(확인)된 명령 수: 0\n",
"차단(U03 CONFIRM): 인증 질의응답 통과 전 전파 불가\n",
"차단(U01 CONFIRM): 통신보안관 ID 필요\n",
"차단(U02 FORWARD): 허용되지 않은 조치: FORWARD\n",
"차단(U06 REJECT): 판단 근거 5자 이상 필요\n",
"차단(U04 CONFIRM 단독): High 등급 확인은 다른 검토자 1명의 교차 확인 필요\n",
"보류 기한 초과 상신: ['U02'] / 상신 후 검토 순서: ['U02', 'U05']\n",
"\n",
" id grade score auth status escalated 대기(s)\n",
"U01 High 0.97 False REJECTED False 43.0\n",
"U02 Med 0.62 - HOLD True 48.0\n",
"U03 Low 0.08 True CONFIRMED False 25.0\n",
"U04 High 0.88 True CONFIRMED False 34.0\n",
"U05 Med 0.45 - PENDING False -\n",
"U06 Low 0.12 True CONFIRMED False 31.0\n",
"\n",
"자동 전파 0건 / 규칙 위반 차단 5건 / 결정 로그 5건(결정 5건, 로깅 성공률 100%)\n",
"평균 검토 대기 36.2s, 감사 로그 17건, 해시 체인 검증: (True, -1)\n",
"8번째 기록 변조 후 검증(False, 깨진 위치): (False, 8)\n",
"\n",
"책임 분계: 탐지기=점수 제시, 통신보안관=인증·청음·결정, 지휘관=최종 명령 책임. 자동 판정으로 명령을 전파하거나 차단하지 않는다.\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 08. 전자전 재밍 및 잡음 혼입 환경 모의 딥페이크 탐지 한계치 스트레스 테스트\n",
"**소요시간**: 약 3.5시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"전자전 재밍 및 심한 잡음 혼입 환경을 모의하여 딥페이크 탐지 성능 한계치 스트레스 테스트를 수행한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **전자전 재밍과 심한 잡음 환경에서 위조 탐지 성능의 한계점을 찾고, 트래픽 급증 부하에서도 안전하게 동작하는지 시험한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **SNR과 단일톤 재밍 세기를 단계적으로 높이며 정확도·재현율·오경보율 곡선을 그려 판정 불가 구간을 정 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `tests/c204_stress_test.py 및 스트레스 테스트 분석 보고서` 형태의 결과물을 완성한다.\n",
"\n",
"### 🔍 평가 및 검증 기준 (PASS 조건)\n",
"#### 미션 판정\n",
"\n",
"**평가 구분:** 필수\n",
"\n",
"**핵심 평가영역:** 극단 상황에서의 안전 실패(Fail-safe) 및 기능 축소 작동 적절성\n",
"\n",
"#### 미션 PASS 조건\n",
"\n",
"- 정량 기준: 동시 트래픽 500% 증가 시 시스템 다운 0건, 데이터 유실률 0%\n",
"\n",
"\n",
"- 진행 기준: 대량 시뮬레이션 부하 주입 테스트 통과 및 병목 구간 프로파일링 완료\n",
"\n",
"\n",
"- 검증 방법: 부하 테스트 도구를 통한 처리량(Throughput) 및 메모리 누수 검증\n",
"\n",
"\n",
"#### 채점자 판정 메모\n",
"\n",
"> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
"\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"\"\"\"전자전 재밍·잡음 세기를 단계적으로 높여 위조 탐지 성능의 한계점을 찾고, 무전 트래픽을 평시의 5배로 늘리며 손상 입력을 섞은 부하 시험에서 시스템 다운·데이터 유실 없이 기능 축소 모드와 사람 확인으로 안전하게 넘기는지 확인한다.\"\"\"\n",
"import time\n",
"import tracemalloc\n",
"from collections import deque\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"from scipy import signal\n",
"from scipy.fft import dct\n",
"from sklearn.linear_model import LogisticRegression\n",
"from sklearn.preprocessing import StandardScaler\n",
"from sklearn.pipeline import make_pipeline\n",
"\n",
"# ---------------------------------------------------------------------------\n",
"# [교육용가상자료] 탐지 학습용 합성 신호 생성기 (실제 음성·화자·통신망 정보 없음)\n",
"# 사람 목소리를 만드는 기능이 아니라, 배음(하모닉) 신호에 '진짜 신호의 자연스러운 흔들림'과\n",
"# '알려진 위조 흔적(문헌상 탐지 단서)'의 통계적 특징을 표시해 탐지기를 학습시키기 위한 추상 신호다.\n",
"# bonafide : 자연스러운 피치 흔들림(jitter)·진폭 흔들림(shimmer)·숨소리 잡음\n",
"# tts_art : 흔들림이 거의 없고 고주파 에너지가 부족하며 20ms 프레임 경계 흔적이 있는 신호\n",
"# vc_art : 프레임마다 배음 위상이 끊기고 스펙트럼 포락선이 뭉개진 신호\n",
"# replay : 스피커 재생 대역 제한 + 실내 잔향이 더해진 신호\n",
"# ---------------------------------------------------------------------------\n",
"FS, DUR = 8000, 0.5\n",
"N = int(FS * DUR)\n",
"LABELS = [\"bonafide\", \"tts_art\", \"vc_art\", \"replay\"]\n",
"SPEAKERS = {f\"SPK_{c}\": (f0, fm) for c, f0, fm in [(\"A\", 115, (650, 1150, 2450)), (\"B\", 135, (700, 1220, 2600)), (\"C\", 98, (600, 1050, 2350)),\n",
" (\"D\", 160, (720, 1300, 2700)), (\"E\", 125, (680, 1180, 2500)), (\"F\", 105, (630, 1100, 2400))]}\n",
"\n",
"\n",
"def _source(rng, f0, fm, jitter, shimmer, phase_break=False, hf_cut=None, frame_mod=0.0):\n",
" t = np.arange(N) / FS\n",
" contour = f0 * (1 + 0.06 * np.sin(2 * np.pi * rng.uniform(1.5, 3.5) * t + rng.uniform(0, 6)))\n",
" f = contour * (1 + jitter * rng.normal(0, 1, N // 80 + 1).repeat(80)[:N]) # 10ms 단위 피치 흔들림(jitter)\n",
" phase = 2 * np.pi * np.cumsum(f) / FS\n",
" K = int(3800 / f0)\n",
" fk = np.arange(1, K + 1) * f0\n",
" amp = sum(np.exp(-((fk - F) / 120.0) ** 2) / (i + 1) for i, F in enumerate(fm)) + 0.02 / np.arange(1, K + 1)\n",
" if hf_cut:\n",
" amp = np.where(fk > hf_cut, amp * 0.2, amp)\n",
" ph = np.arange(1, K + 1)[:, None] * phase[None, :]\n",
" if phase_break: # 20ms 프레임마다 배음별 위상 불연속\n",
" ph = ph + rng.uniform(0, 2 * np.pi, (K, N // 160 + 1)).repeat(160, axis=1)[:, :N]\n",
" sig = (amp[:, None] * np.sin(ph)).sum(0)\n",
" env = 0.55 + 0.45 * np.sin(2 * np.pi * rng.uniform(3, 5) * t + rng.uniform(0, 6)) ** 2 # 음절형 진폭 변화\n",
" env *= 1 + shimmer * np.interp(t, np.linspace(0, DUR, 60), rng.normal(0, 1, 60)) # 완만한 진폭 흔들림(shimmer)\n",
" if frame_mod:\n",
" env *= 1 + frame_mod * np.sign(np.sin(2 * np.pi * 50 * t)) # 20ms 프레임 경계 흔적(50Hz)\n",
" return sig * env + (0.03 if jitter > 0.003 else 0.005) * rng.normal(0, 1, N) * env\n",
"\n",
"\n",
"def make_clip(label, speaker, rng):\n",
" f0, fm = SPEAKERS[speaker]\n",
" if label in (\"tts_art\", \"vc_art\"): # 위조는 화자 특성을 근사만 함\n",
" f0 = f0 * rng.uniform(0.92, 1.08); fm = tuple(F * rng.uniform(0.95, 1.05) for F in fm)\n",
" if label == \"bonafide\" or label == \"replay\":\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.008, 0.02), shimmer=rng.uniform(0.05, 0.12))\n",
" elif label == \"tts_art\":\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.0, 0.002), shimmer=0.01, hf_cut=rng.uniform(2300, 2800), frame_mod=rng.uniform(0.08, 0.2))\n",
" else:\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.006, 0.015), shimmer=rng.uniform(0.04, 0.1), phase_break=True)\n",
" if label == \"replay\": # 스피커 대역 + 실내 잔향\n",
" x = signal.sosfilt(signal.butter(4, [450, 2900], \"bandpass\", fs=FS, output=\"sos\"), x)\n",
" ir = rng.normal(0, 1, int(0.08 * FS)) * np.exp(-np.arange(int(0.08 * FS)) / (0.02 * FS)); ir[0] = 3.0\n",
" x = signal.fftconvolve(x, ir)[:N]\n",
" return x / (np.abs(x).max() + 1e-9)\n",
"\n",
"\n",
"RADIO_BP = signal.butter(4, [300, 3400], \"bandpass\", fs=FS, output=\"sos\")\n",
"\n",
"\n",
"def radio_channel(x, rng, snr_db=15.0, impulse=0.0, dropout=0.0, clip=1.0, jam_tone=0.0):\n",
" \"\"\"[교육용가상자료] 무전 채널 모형: 300~3400Hz 대역 제한, 배경 잡음(SNR), 충격성 잡음, 신호 끊김, 포화(클리핑), 단일톤 재밍.\"\"\"\n",
" y = signal.sosfilt(RADIO_BP, x)\n",
" p = np.mean(y ** 2)\n",
" noise = rng.normal(0, 1, N) * np.sqrt(p / 10 ** (snr_db / 10))\n",
" noise += np.cumsum(rng.normal(0, 0.02, N)) * np.sqrt(p) * 0.05 # 저주파 차량·엔진형 소음\n",
" y = y + noise\n",
" if impulse:\n",
" k = rng.random(N) < impulse\n",
" y[k] += rng.normal(0, 6 * np.sqrt(p), k.sum())\n",
" if jam_tone:\n",
" y += jam_tone * np.sqrt(p) * np.sin(2 * np.pi * rng.uniform(800, 2000) * np.arange(N) / FS)\n",
" if dropout:\n",
" for _ in range(int(dropout * 10)):\n",
" a = int(rng.integers(0, N - 400)); y[a:a + int(rng.integers(100, 400))] = 0.0\n",
" return np.clip(y / (np.abs(y).max() + 1e-9), -clip, clip)\n",
"\n",
"\n",
"def make_dataset(n=400, seed=7, snr=(5, 25), labels=LABELS, p=None, claim=\"SPK_A\", **chan):\n",
" \"\"\"클립 n개: 진짜는 여러 화자, 위조는 모두 지휘관 화자(SPK_A)를 사칭. 반환: (신호 배열, 메타 DataFrame).\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" X, meta = np.zeros((n, N)), []\n",
" if p is None and len(labels) > 1 and labels[0] == \"bonafide\": # 기본: 진짜 50%, 나머지 위조 유형 균등\n",
" p = [0.5] + [0.5 / (len(labels) - 1)] * (len(labels) - 1)\n",
" for i in range(n):\n",
" lab = str(rng.choice(labels, p=p))\n",
" spk = str(rng.choice(list(SPEAKERS))) if lab == \"bonafide\" else claim\n",
" if lab == \"replay\":\n",
" spk = claim\n",
" s = float(rng.uniform(*snr)) if isinstance(snr, tuple) else float(snr)\n",
" X[i] = radio_channel(make_clip(lab, spk, rng), rng, snr_db=s, **chan)\n",
" meta.append({\"clip\": f\"C{seed:02d}{i:04d}\", \"label\": lab, \"spoof\": lab != \"bonafide\", \"claimed\": spk if lab == \"bonafide\" else claim,\n",
" \"true_spk\": spk if lab == \"bonafide\" else \"unknown\", \"snr_db\": round(s, 1)})\n",
" return X, pd.DataFrame(meta)\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# ---- 전처리·음향 특징(MFCC/스펙트로그램) 공통 함수: 여러 클립을 한 번에 계산 ----\n",
"NFFT, HOP = 256, 80\n",
"\n",
"\n",
"def mel_bank(n_mels=26, fmin=300, fmax=3400):\n",
" mel = lambda f: 2595 * np.log10(1 + f / 700)\n",
" imel = lambda m: 700 * (10 ** (m / 2595) - 1)\n",
" pts = imel(np.linspace(mel(fmin), mel(fmax), n_mels + 2))\n",
" freqs = np.fft.rfftfreq(NFFT, 1 / FS)\n",
" fb = np.zeros((n_mels, len(freqs)))\n",
" for m in range(n_mels):\n",
" l, c, r = pts[m], pts[m + 1], pts[m + 2]\n",
" fb[m] = np.clip(np.minimum((freqs - l) / (c - l), (r - freqs) / (r - c)), 0, None)\n",
" return fb\n",
"\n",
"\n",
"MEL = mel_bank()\n",
"FREQS = np.fft.rfftfreq(NFFT, 1 / FS)\n",
"\n",
"\n",
"def spectrogram(X, denoise=0.0):\n",
" \"\"\"배치 전력 스펙트로그램 (클립, 주파수, 프레임). denoise>0이면 잡음 바닥(주파수별 하위 10%)을 그 배수만큼 빼는 스펙트럼 차감.\"\"\"\n",
" X = np.atleast_2d(X)\n",
" Xp = np.c_[X[:, :1], X[:, 1:] - 0.95 * X[:, :-1]] # 프리엠퍼시스\n",
" frames = np.lib.stride_tricks.sliding_window_view(Xp, NFFT, axis=1)[:, ::HOP] * np.hanning(NFFT)\n",
" P = np.abs(np.fft.rfft(frames, axis=-1)) ** 2\n",
" P = np.transpose(P, (0, 2, 1))\n",
" if denoise:\n",
" floor = np.percentile(P, 10, axis=2, keepdims=True)\n",
" P = np.maximum(P - denoise * floor, 0.05 * P)\n",
" return P\n",
"\n",
"\n",
"def mfcc(P, n=13):\n",
" logmel = np.log(np.einsum(\"mf,cft->cmt\", MEL, P) + 1e-8)\n",
" return dct(logmel, type=2, axis=1, norm=\"ortho\")[:, :n, :]\n",
"\n",
"\n",
"def band(P, lo, hi):\n",
" return P[:, (FREQS >= lo) & (FREQS < hi), :].sum(axis=(1, 2))\n",
"\n",
"\n",
"def jitter_hnr(X):\n",
" \"\"\"40ms 프레임 자기상관으로 피치 주기를 찾아 피치 흔들림(jitter)과 배음 대 잡음비(HNR 근사)를 구한다.\"\"\"\n",
" W, lag0, lag1 = 320, int(FS / 250), int(FS / 70)\n",
" fr = np.lib.stride_tricks.sliding_window_view(np.atleast_2d(X), W, axis=1)[:, ::160]\n",
" fr = fr - fr.mean(-1, keepdims=True)\n",
" spec = np.fft.rfft(fr, 2 * W, axis=-1)\n",
" ac = np.fft.irfft(np.abs(spec) ** 2, axis=-1)[..., :W]\n",
" ac = ac / (ac[..., :1] + 1e-12)\n",
" seg = ac[..., lag0:lag1]\n",
" k = seg.argmax(-1)\n",
" peak = seg.max(-1)\n",
" km, kp = np.take_along_axis(seg, np.clip(k - 1, 0, None)[..., None], -1)[..., 0], np.take_along_axis(seg, np.clip(k + 1, None, seg.shape[-1] - 1)[..., None], -1)[..., 0]\n",
" lag = k + lag0 + 0.5 * (km - kp) / (km - 2 * peak + kp - 1e-12) # 포물선 보간으로 주기 정밀화\n",
" f0 = FS / lag\n",
" jit = np.median(np.abs(np.diff(f0, axis=1)) / f0[:, 1:], axis=1)\n",
" return jit, np.median(peak, axis=1)\n",
"\n",
"\n",
"def frame_mod(X):\n",
" \"\"\"진폭 포락선(2.5ms 간격)의 50Hz 성분 비율: 20ms 프레임 경계 흔적(합성 단서).\"\"\"\n",
" env = np.abs(np.atleast_2d(X)).reshape(len(np.atleast_2d(X)), -1, 20).mean(-1)\n",
" sp = np.abs(np.fft.rfft(env - env.mean(1, keepdims=True), axis=1)) ** 2\n",
" f = np.fft.rfftfreq(env.shape[1], 20 / FS)\n",
" return sp[:, (f > 47) & (f < 53)].sum(1) / (sp[:, (f > 30) & (f < 70)].sum(1) + 1e-12)\n",
"\n",
"\n",
"FEAT_NAMES = [f\"mfcc{i}_m\" for i in range(13)] + [f\"mfcc{i}_s\" for i in range(13)] + \\\n",
" [\"delta_s\", \"hf_ratio\", \"lf_ratio\", \"flatness\", \"flux\", \"jitter\", \"hnr\", \"frame_mod\"]\n",
"\n",
"\n",
"def extract(X, denoise=0.0):\n",
" \"\"\"클립 배치 -> 특징 행렬 (클립 x 34) DataFrame.\"\"\"\n",
" P = spectrogram(X, denoise)\n",
" C = mfcc(P)\n",
" tot = band(P, 300, 3400) + 1e-12\n",
" flat = np.exp(np.log(P[:, (FREQS >= 300) & (FREQS < 3400)] + 1e-10).mean(1)) / (P[:, (FREQS >= 300) & (FREQS < 3400)].mean(1) + 1e-10)\n",
" logP = np.log(P + 1e-8)\n",
" jit, hnr = jitter_hnr(X)\n",
" F = np.c_[C.mean(2), C.std(2), np.abs(np.diff(C, axis=2)).mean((1, 2)), band(P, 2400, 3400) / tot, band(P, 300, 500) / tot,\n",
" flat.mean(1), np.abs(np.diff(logP, axis=2)).mean((1, 2)), jit, hnr, frame_mod(X)]\n",
" return pd.DataFrame(F, columns=FEAT_NAMES)\n",
"\n",
"\n",
"def enroll(seed=99):\n",
" \"\"\"지휘관·통신소 화자 6명의 등록 견본(깨끗한 진짜 클립 5개씩의 MFCC 평균).\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" tpl = {}\n",
" for spk in SPEAKERS:\n",
" Xe = np.stack([radio_channel(make_clip(\"bonafide\", spk, rng), rng, snr_db=25) for _ in range(5)])\n",
" tpl[spk] = extract(Xe)[[f\"mfcc{i}_m\" for i in range(1, 13)]].mean().values\n",
" return tpl\n",
"\n",
"\n",
"def speaker_sim(F, claimed, tpl):\n",
" \"\"\"주장 화자 견본과 MFCC 평균 벡터의 코사인 유사도(화자 검증 점수).\"\"\"\n",
" v = F[[f\"mfcc{i}_m\" for i in range(1, 13)]].values\n",
" t = np.stack([tpl[c] for c in claimed])\n",
" return (v * t).sum(1) / (np.linalg.norm(v, axis=1) * np.linalg.norm(t, axis=1) + 1e-12)\n",
"\n",
"tpl = enroll()\n",
"\n",
"\n",
"def featurize(X, M):\n",
" F = extract(X)\n",
" F[\"spk_sim\"] = speaker_sim(F, M[\"claimed\"], tpl)\n",
" return F\n",
"\n",
"\n",
"Xtr, Mtr = make_dataset(500, seed=61, snr=(3, 25))\n",
"det = make_pipeline(StandardScaler(), LogisticRegression(max_iter=3000, C=0.5)).fit(featurize(Xtr, Mtr), Mtr[\"spoof\"])\n",
"\n",
"# 1) 탐지 한계치: 잡음(SNR)·재밍 세기 단계별 정확도\n",
"print(\"[탐지 한계치 - 교육용가상자료, 단계별 120클립]\")\n",
"print(\" 조건 정확도 위조재현율 오경보율\")\n",
"curve = []\n",
"for name, ch in [(\"SNR 20dB\", dict(snr=20.0)), (\"SNR 10dB\", dict(snr=10.0)), (\"SNR 5dB\", dict(snr=5.0)), (\"SNR 0dB\", dict(snr=0.0)),\n",
" (\"SNR -5dB\", dict(snr=-5.0)), (\"재밍 0.5\", dict(snr=12.0, jam_tone=0.5)), (\"재밍 1.0\", dict(snr=12.0, jam_tone=1.0)),\n",
" (\"재밍 2.0\", dict(snr=12.0, jam_tone=2.0))]:\n",
" X, M = make_dataset(120, seed=700 + len(curve), **ch)\n",
" pr = det.predict(featurize(X, M)); y = M[\"spoof\"].values\n",
" curve.append((name, (pr == y).mean(), pr[y].mean(), pr[~y].mean()))\n",
" print(f\" {name:12s} {curve[-1][1]:7.1%} {curve[-1][2]:9.1%} {curve[-1][3]:9.1%}\")\n",
"lim = [c[0] for c in curve if c[1] < 0.8]\n",
"print(\"정확도 80% 미만 구간(한계):\", lim if lim else \"없음\", \"-> 이 구간은 자동 점수 대신 '판정 불가(사람 확인)'로 처리\")\n",
"\n",
"# 2) 부하 시험: 평시 초당 20건 -> 500% 증가(초당 100건), 손상 입력 3% 혼입\n",
"pool, pmeta = make_dataset(200, seed=790, snr=(3, 25))\n",
"rng = np.random.default_rng(800)\n",
"\n",
"\n",
"def arrivals(n):\n",
" idx = rng.integers(0, len(pool), n)\n",
" xs = [pool[i] + rng.normal(0, 0.02, N) for i in idx]\n",
" for k in np.flatnonzero(rng.random(n) < 0.03): # 손상 입력 주입\n",
" xs[k] = [np.full(N, np.nan), np.zeros(10), np.zeros(N)][k % 3]\n",
" return xs, pmeta.iloc[idx].reset_index(drop=True)\n",
"\n",
"\n",
"def valid(x):\n",
" return len(x) == N and np.isfinite(x).all() and np.abs(x).max() > 1e-6\n",
"\n",
"\n",
"def run(rate, ticks=8, budget_s=1.0, backlog_lim=150):\n",
" \"\"\"1초 단위로 rate건 도착. 처리 시간이 예산을 넘거나 대기열이 길면 기능 축소 모드(화자 유사도 생략, 점수만 산출 후 사람 확인 표시).\"\"\"\n",
" q, done, quarantine, degraded, crash = deque(), 0, 0, 0, 0\n",
" lat, t_feat, t_pred = [], 0.0, 0.0\n",
" total_in = 0\n",
" tracemalloc.start(); m0 = tracemalloc.get_traced_memory()[0]\n",
" for _ in range(ticks):\n",
" xs, ms = arrivals(rate); total_in += rate\n",
" q.extend(zip(xs, ms.to_dict(\"records\")))\n",
" t_start = time.perf_counter()\n",
" while q and time.perf_counter() - t_start < budget_s:\n",
" batch = [q.popleft() for _ in range(min(25, len(q)))]\n",
" ok = [b for b in batch if valid(b[0])]\n",
" quarantine += len(batch) - len(ok) # 손상 입력은 버리지 않고 격리 보관\n",
" if not ok:\n",
" continue\n",
" try:\n",
" Xb = np.stack([b[0] for b in ok]); Mb = pd.DataFrame([b[1] for b in ok])\n",
" t0 = time.perf_counter(); F = extract(Xb)\n",
" if len(q) > backlog_lim:\n",
" F[\"spk_sim\"] = 0.9; degraded += len(ok) # 기능 축소: 화자 검증 생략 -> 결과는 '사람 확인' 표시\n",
" else:\n",
" F[\"spk_sim\"] = speaker_sim(F, Mb[\"claimed\"], tpl)\n",
" t1 = time.perf_counter(); det.predict_proba(F); t2 = time.perf_counter()\n",
" t_feat += t1 - t0; t_pred += t2 - t1; lat.append((t2 - t0) * 1000 / len(ok)); done += len(ok)\n",
" except Exception:\n",
" crash += 1; q.extendleft(reversed(batch)) # 실패 배치는 대기열로 되돌림(유실 방지)\n",
" break\n",
" while q: # 시험 종료 후 잔여 대기열 소진\n",
" b = q.popleft()\n",
" if valid(b[0]):\n",
" F = extract(b[0][None]); F[\"spk_sim\"] = speaker_sim(F, [b[1][\"claimed\"]], tpl); det.predict_proba(F); done += 1\n",
" else:\n",
" quarantine += 1\n",
" mem = (tracemalloc.get_traced_memory()[0] - m0) / 1e6; tracemalloc.stop()\n",
" return dict(rate=rate, total=total_in, done=done, quarantine=quarantine, lost=total_in - done - quarantine, crash=crash,\n",
" degraded=degraded, ms=np.mean(lat), mem=mem, feat=t_feat, pred=t_pred)\n",
"\n",
"\n",
"print(\"\\n[부하 시험] 도착률 입력 처리 격리 유실 다운 기능축소 ms/건 메모리증가MB\")\n",
"res = [run(20), run(100), run(100, budget_s=0.02)]\n",
"for r, nm in zip(res, [\"평시\", \"5배\", \"5배+처리지연\"]):\n",
" print(f\"{nm:9s} {r['rate']:4d}/s {r['total']:5d} {r['done']:5d} {r['quarantine']:5d} {r['lost']:5d} {r['crash']:5d} {r['degraded']:8d} {r['ms']:6.2f} {r['mem']:10.2f}\")\n",
"hi = res[1]\n",
"print(f\"트래픽 500% 증가: 시스템 다운 {hi['crash']}건, 데이터 유실 {hi['lost'] / hi['total']:.1%} (목표 0건/0%: {'달성' if hi['crash'] == 0 and hi['lost'] == 0 else '미달'})\")\n",
"Xp = pool[:200]; Pp = spectrogram(Xp); prof = {}\n",
"for nm, fn in [(\"스펙트로그램\", lambda: spectrogram(Xp)), (\"MFCC\", lambda: mfcc(Pp)), (\"피치 자기상관(jitter/HNR)\", lambda: jitter_hnr(Xp)), (\"프레임 경계 단서\", lambda: frame_mod(Xp))]:\n",
" t0 = time.perf_counter(); fn(); prof[nm] = time.perf_counter() - t0\n",
"tot = sum(prof.values())\n",
"print(f\"처리 시간 구성(100/s): 특징 추출 {hi['feat'] / (hi['feat'] + hi['pred']):.0%}, 추론 {hi['pred'] / (hi['feat'] + hi['pred']):.0%}\")\n",
"print(\"특징 추출 내부 프로파일:\", \", \".join(f\"{k} {v / tot:.0%}\" for k, v in prof.items()), f\"-> 병목: {max(prof, key=prof.get)}\")\n",
"dg = res[2]\n",
"print(f\"처리 지연 모의: 대기열 적체 시 {dg['degraded']}건을 기능 축소 모드로 처리, 유실 {dg['lost']}건, 다운 {dg['crash']}건\")\n",
"print(\"기능 축소 결과와 판정 불가 구간 결과는 자동 차단·전파 없이 통신보안관 확인 대기열로 보낸다(Fail-safe).\")\n",
"\n",
"plt.figure(figsize=(7, 3.5))\n",
"lab = [c[0].replace(\"재밍\", \"jam\") for c in curve]\n",
"plt.plot(lab, [c[1] for c in curve], \"o-\", label=\"accuracy\")\n",
"plt.plot(lab, [c[3] for c in curve], \"s--\", label=\"false alarm\")\n",
"plt.axhline(0.8, c=\"k\", ls=\":\"); plt.xticks(rotation=30); plt.legend(); plt.title(\"Detection limit under noise / jamming (synthetic)\")\n",
"plt.tight_layout(); plt.show()\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 08 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"[탐지 한계치 - 교육용가상자료, 단계별 120클립]\n",
" 조건 정확도 위조재현율 오경보율\n",
" SNR 20dB 99.2% 98.4% 0.0%\n",
" SNR 10dB 96.7% 96.8% 3.4%\n",
" SNR 5dB 97.5% 98.1% 3.0%\n",
" SNR 0dB 66.7% 100.0% 66.7%\n",
" SNR -5dB 48.3% 100.0% 98.4%\n",
" 재밍 0.5 95.8% 98.1% 6.1%\n",
" 재밍 1.0 90.0% 96.8% 17.2%\n",
" 재밍 2.0 85.8% 89.7% 19.2%\n",
"정확도 80% 미만 구간(한계): ['SNR 0dB', 'SNR -5dB'] -> 이 구간은 자동 점수 대신 '판정 불가(사람 확인)'로 처리\n",
"\n",
"[부하 시험] 도착률 입력 처리 격리 유실 다운 기능축소 ms/건 메모리증가MB\n",
"평시 20/s 160 150 10 0 0 0 1.29 1.23\n",
"5배 100/s 800 774 26 0 0 0 1.19 4.02\n",
"5배+처리지연 100/s 800 781 19 0 0 146 1.09 4.87\n",
"트래픽 500% 증가: 시스템 다운 0건, 데이터 유실 0.0% (목표 0건/0%: 달성)\n",
"처리 시간 구성(100/s): 특징 추출 85%, 추론 15%\n",
"특징 추출 내부 프로파일: 스펙트로그램 26%, MFCC 17%, 피치 자기상관(jitter/HNR) 54%, 프레임 경계 단서 3% -> 병목: 피치 자기상관(jitter/HNR)\n",
"처리 지연 모의: 대기열 적체 시 146건을 기능 축소 모드로 처리, 유실 0건, 다운 0건\n",
"기능 축소 결과와 판정 불가 구간 결과는 자동 차단·전파 없이 통신보안관 확인 대기열로 보낸다(Fail-safe).\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 09. AAR 기반 무전 위조 오판정 음성 분석 및 음향 특징 추출 파라미터 재보정\n",
"**소요시간**: 약 3.0시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"AAR 사후검토를 통해 위조 음성 오판정 사례를 분석하고 모델 파라미터를 재보정한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **1차 배치 Baseline의 모의 운용 기록에서 오판정 사례를 분석(AAR)하고, 특징 추출·판정 파라미터를 재보정한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **오판정을 미탐과 오경보로 나누어 채널 조건·위조 유형별로 집계하고 원인을 규명 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `docs/c204_aar_report.md 및 configs/c204_tuned_params.yaml` 형태의 결과물을 완성한다.\n",
"\n",
"### 🔍 평가 및 검증 기준 (PASS 조건)\n",
"#### 미션 판정\n",
"\n",
"**평가 구분:** 필수\n",
"\n",
"**핵심 평가영역:** 오류 원인 규명의 객관성 및 개선 조치의 논리적 타당성\n",
"\n",
"#### 미션 PASS 조건\n",
"\n",
"- 정량 기준: AAR 피드백 적용 후 종합 정확도 10%p 이상 향상, 재발 오류 0건\n",
"\n",
"\n",
"- 진행 기준: AAR 보고서 작성 및 파라미터 튜닝 전후 회귀 테스트 통과\n",
"\n",
"\n",
"- 검증 방법: 회귀 테스트(Regression Test) 스위트 전체 자동 실행 및 비교\n",
"\n",
"\n",
"#### 채점자 판정 메모\n",
"\n",
"> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
"\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"\"\"\"1차 배치(규칙 기반 Baseline)의 모의 운용 기록에서 위조 오판정 사례를 유형·채널 조건별로 분석(AAR)하고, 오판정 사례 재학습·채널 품질 특징 추가·전처리 차감 배수와 판정 임계값 재탐색으로 파라미터를 재보정한 뒤 회귀 시험으로 정확도 향상과 재발 오류를 확인한다.\"\"\"\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"from scipy import signal\n",
"from scipy.fft import dct\n",
"from sklearn.neural_network import MLPClassifier\n",
"from sklearn.preprocessing import StandardScaler\n",
"from sklearn.pipeline import make_pipeline\n",
"\n",
"# ---------------------------------------------------------------------------\n",
"# [교육용가상자료] 탐지 학습용 합성 신호 생성기 (실제 음성·화자·통신망 정보 없음)\n",
"# 사람 목소리를 만드는 기능이 아니라, 배음(하모닉) 신호에 '진짜 신호의 자연스러운 흔들림'과\n",
"# '알려진 위조 흔적(문헌상 탐지 단서)'의 통계적 특징을 표시해 탐지기를 학습시키기 위한 추상 신호다.\n",
"# bonafide : 자연스러운 피치 흔들림(jitter)·진폭 흔들림(shimmer)·숨소리 잡음\n",
"# tts_art : 흔들림이 거의 없고 고주파 에너지가 부족하며 20ms 프레임 경계 흔적이 있는 신호\n",
"# vc_art : 프레임마다 배음 위상이 끊기고 스펙트럼 포락선이 뭉개진 신호\n",
"# replay : 스피커 재생 대역 제한 + 실내 잔향이 더해진 신호\n",
"# ---------------------------------------------------------------------------\n",
"FS, DUR = 8000, 0.5\n",
"N = int(FS * DUR)\n",
"LABELS = [\"bonafide\", \"tts_art\", \"vc_art\", \"replay\"]\n",
"SPEAKERS = {f\"SPK_{c}\": (f0, fm) for c, f0, fm in [(\"A\", 115, (650, 1150, 2450)), (\"B\", 135, (700, 1220, 2600)), (\"C\", 98, (600, 1050, 2350)),\n",
" (\"D\", 160, (720, 1300, 2700)), (\"E\", 125, (680, 1180, 2500)), (\"F\", 105, (630, 1100, 2400))]}\n",
"\n",
"\n",
"def _source(rng, f0, fm, jitter, shimmer, phase_break=False, hf_cut=None, frame_mod=0.0):\n",
" t = np.arange(N) / FS\n",
" contour = f0 * (1 + 0.06 * np.sin(2 * np.pi * rng.uniform(1.5, 3.5) * t + rng.uniform(0, 6)))\n",
" f = contour * (1 + jitter * rng.normal(0, 1, N // 80 + 1).repeat(80)[:N]) # 10ms 단위 피치 흔들림(jitter)\n",
" phase = 2 * np.pi * np.cumsum(f) / FS\n",
" K = int(3800 / f0)\n",
" fk = np.arange(1, K + 1) * f0\n",
" amp = sum(np.exp(-((fk - F) / 120.0) ** 2) / (i + 1) for i, F in enumerate(fm)) + 0.02 / np.arange(1, K + 1)\n",
" if hf_cut:\n",
" amp = np.where(fk > hf_cut, amp * 0.2, amp)\n",
" ph = np.arange(1, K + 1)[:, None] * phase[None, :]\n",
" if phase_break: # 20ms 프레임마다 배음별 위상 불연속\n",
" ph = ph + rng.uniform(0, 2 * np.pi, (K, N // 160 + 1)).repeat(160, axis=1)[:, :N]\n",
" sig = (amp[:, None] * np.sin(ph)).sum(0)\n",
" env = 0.55 + 0.45 * np.sin(2 * np.pi * rng.uniform(3, 5) * t + rng.uniform(0, 6)) ** 2 # 음절형 진폭 변화\n",
" env *= 1 + shimmer * np.interp(t, np.linspace(0, DUR, 60), rng.normal(0, 1, 60)) # 완만한 진폭 흔들림(shimmer)\n",
" if frame_mod:\n",
" env *= 1 + frame_mod * np.sign(np.sin(2 * np.pi * 50 * t)) # 20ms 프레임 경계 흔적(50Hz)\n",
" return sig * env + (0.03 if jitter > 0.003 else 0.005) * rng.normal(0, 1, N) * env\n",
"\n",
"\n",
"def make_clip(label, speaker, rng):\n",
" f0, fm = SPEAKERS[speaker]\n",
" if label in (\"tts_art\", \"vc_art\"): # 위조는 화자 특성을 근사만 함\n",
" f0 = f0 * rng.uniform(0.92, 1.08); fm = tuple(F * rng.uniform(0.95, 1.05) for F in fm)\n",
" if label == \"bonafide\" or label == \"replay\":\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.008, 0.02), shimmer=rng.uniform(0.05, 0.12))\n",
" elif label == \"tts_art\":\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.0, 0.002), shimmer=0.01, hf_cut=rng.uniform(2300, 2800), frame_mod=rng.uniform(0.08, 0.2))\n",
" else:\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.006, 0.015), shimmer=rng.uniform(0.04, 0.1), phase_break=True)\n",
" if label == \"replay\": # 스피커 대역 + 실내 잔향\n",
" x = signal.sosfilt(signal.butter(4, [450, 2900], \"bandpass\", fs=FS, output=\"sos\"), x)\n",
" ir = rng.normal(0, 1, int(0.08 * FS)) * np.exp(-np.arange(int(0.08 * FS)) / (0.02 * FS)); ir[0] = 3.0\n",
" x = signal.fftconvolve(x, ir)[:N]\n",
" return x / (np.abs(x).max() + 1e-9)\n",
"\n",
"\n",
"RADIO_BP = signal.butter(4, [300, 3400], \"bandpass\", fs=FS, output=\"sos\")\n",
"\n",
"\n",
"def radio_channel(x, rng, snr_db=15.0, impulse=0.0, dropout=0.0, clip=1.0, jam_tone=0.0):\n",
" \"\"\"[교육용가상자료] 무전 채널 모형: 300~3400Hz 대역 제한, 배경 잡음(SNR), 충격성 잡음, 신호 끊김, 포화(클리핑), 단일톤 재밍.\"\"\"\n",
" y = signal.sosfilt(RADIO_BP, x)\n",
" p = np.mean(y ** 2)\n",
" noise = rng.normal(0, 1, N) * np.sqrt(p / 10 ** (snr_db / 10))\n",
" noise += np.cumsum(rng.normal(0, 0.02, N)) * np.sqrt(p) * 0.05 # 저주파 차량·엔진형 소음\n",
" y = y + noise\n",
" if impulse:\n",
" k = rng.random(N) < impulse\n",
" y[k] += rng.normal(0, 6 * np.sqrt(p), k.sum())\n",
" if jam_tone:\n",
" y += jam_tone * np.sqrt(p) * np.sin(2 * np.pi * rng.uniform(800, 2000) * np.arange(N) / FS)\n",
" if dropout:\n",
" for _ in range(int(dropout * 10)):\n",
" a = int(rng.integers(0, N - 400)); y[a:a + int(rng.integers(100, 400))] = 0.0\n",
" return np.clip(y / (np.abs(y).max() + 1e-9), -clip, clip)\n",
"\n",
"\n",
"def make_dataset(n=400, seed=7, snr=(5, 25), labels=LABELS, p=None, claim=\"SPK_A\", **chan):\n",
" \"\"\"클립 n개: 진짜는 여러 화자, 위조는 모두 지휘관 화자(SPK_A)를 사칭. 반환: (신호 배열, 메타 DataFrame).\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" X, meta = np.zeros((n, N)), []\n",
" if p is None and len(labels) > 1 and labels[0] == \"bonafide\": # 기본: 진짜 50%, 나머지 위조 유형 균등\n",
" p = [0.5] + [0.5 / (len(labels) - 1)] * (len(labels) - 1)\n",
" for i in range(n):\n",
" lab = str(rng.choice(labels, p=p))\n",
" spk = str(rng.choice(list(SPEAKERS))) if lab == \"bonafide\" else claim\n",
" if lab == \"replay\":\n",
" spk = claim\n",
" s = float(rng.uniform(*snr)) if isinstance(snr, tuple) else float(snr)\n",
" X[i] = radio_channel(make_clip(lab, spk, rng), rng, snr_db=s, **chan)\n",
" meta.append({\"clip\": f\"C{seed:02d}{i:04d}\", \"label\": lab, \"spoof\": lab != \"bonafide\", \"claimed\": spk if lab == \"bonafide\" else claim,\n",
" \"true_spk\": spk if lab == \"bonafide\" else \"unknown\", \"snr_db\": round(s, 1)})\n",
" return X, pd.DataFrame(meta)\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# ---- 전처리·음향 특징(MFCC/스펙트로그램) 공통 함수: 여러 클립을 한 번에 계산 ----\n",
"NFFT, HOP = 256, 80\n",
"\n",
"\n",
"def mel_bank(n_mels=26, fmin=300, fmax=3400):\n",
" mel = lambda f: 2595 * np.log10(1 + f / 700)\n",
" imel = lambda m: 700 * (10 ** (m / 2595) - 1)\n",
" pts = imel(np.linspace(mel(fmin), mel(fmax), n_mels + 2))\n",
" freqs = np.fft.rfftfreq(NFFT, 1 / FS)\n",
" fb = np.zeros((n_mels, len(freqs)))\n",
" for m in range(n_mels):\n",
" l, c, r = pts[m], pts[m + 1], pts[m + 2]\n",
" fb[m] = np.clip(np.minimum((freqs - l) / (c - l), (r - freqs) / (r - c)), 0, None)\n",
" return fb\n",
"\n",
"\n",
"MEL = mel_bank()\n",
"FREQS = np.fft.rfftfreq(NFFT, 1 / FS)\n",
"\n",
"\n",
"def spectrogram(X, denoise=0.0):\n",
" \"\"\"배치 전력 스펙트로그램 (클립, 주파수, 프레임). denoise>0이면 잡음 바닥(주파수별 하위 10%)을 그 배수만큼 빼는 스펙트럼 차감.\"\"\"\n",
" X = np.atleast_2d(X)\n",
" Xp = np.c_[X[:, :1], X[:, 1:] - 0.95 * X[:, :-1]] # 프리엠퍼시스\n",
" frames = np.lib.stride_tricks.sliding_window_view(Xp, NFFT, axis=1)[:, ::HOP] * np.hanning(NFFT)\n",
" P = np.abs(np.fft.rfft(frames, axis=-1)) ** 2\n",
" P = np.transpose(P, (0, 2, 1))\n",
" if denoise:\n",
" floor = np.percentile(P, 10, axis=2, keepdims=True)\n",
" P = np.maximum(P - denoise * floor, 0.05 * P)\n",
" return P\n",
"\n",
"\n",
"def mfcc(P, n=13):\n",
" logmel = np.log(np.einsum(\"mf,cft->cmt\", MEL, P) + 1e-8)\n",
" return dct(logmel, type=2, axis=1, norm=\"ortho\")[:, :n, :]\n",
"\n",
"\n",
"def band(P, lo, hi):\n",
" return P[:, (FREQS >= lo) & (FREQS < hi), :].sum(axis=(1, 2))\n",
"\n",
"\n",
"def jitter_hnr(X):\n",
" \"\"\"40ms 프레임 자기상관으로 피치 주기를 찾아 피치 흔들림(jitter)과 배음 대 잡음비(HNR 근사)를 구한다.\"\"\"\n",
" W, lag0, lag1 = 320, int(FS / 250), int(FS / 70)\n",
" fr = np.lib.stride_tricks.sliding_window_view(np.atleast_2d(X), W, axis=1)[:, ::160]\n",
" fr = fr - fr.mean(-1, keepdims=True)\n",
" spec = np.fft.rfft(fr, 2 * W, axis=-1)\n",
" ac = np.fft.irfft(np.abs(spec) ** 2, axis=-1)[..., :W]\n",
" ac = ac / (ac[..., :1] + 1e-12)\n",
" seg = ac[..., lag0:lag1]\n",
" k = seg.argmax(-1)\n",
" peak = seg.max(-1)\n",
" km, kp = np.take_along_axis(seg, np.clip(k - 1, 0, None)[..., None], -1)[..., 0], np.take_along_axis(seg, np.clip(k + 1, None, seg.shape[-1] - 1)[..., None], -1)[..., 0]\n",
" lag = k + lag0 + 0.5 * (km - kp) / (km - 2 * peak + kp - 1e-12) # 포물선 보간으로 주기 정밀화\n",
" f0 = FS / lag\n",
" jit = np.median(np.abs(np.diff(f0, axis=1)) / f0[:, 1:], axis=1)\n",
" return jit, np.median(peak, axis=1)\n",
"\n",
"\n",
"def frame_mod(X):\n",
" \"\"\"진폭 포락선(2.5ms 간격)의 50Hz 성분 비율: 20ms 프레임 경계 흔적(합성 단서).\"\"\"\n",
" env = np.abs(np.atleast_2d(X)).reshape(len(np.atleast_2d(X)), -1, 20).mean(-1)\n",
" sp = np.abs(np.fft.rfft(env - env.mean(1, keepdims=True), axis=1)) ** 2\n",
" f = np.fft.rfftfreq(env.shape[1], 20 / FS)\n",
" return sp[:, (f > 47) & (f < 53)].sum(1) / (sp[:, (f > 30) & (f < 70)].sum(1) + 1e-12)\n",
"\n",
"\n",
"FEAT_NAMES = [f\"mfcc{i}_m\" for i in range(13)] + [f\"mfcc{i}_s\" for i in range(13)] + \\\n",
" [\"delta_s\", \"hf_ratio\", \"lf_ratio\", \"flatness\", \"flux\", \"jitter\", \"hnr\", \"frame_mod\"]\n",
"\n",
"\n",
"def extract(X, denoise=0.0):\n",
" \"\"\"클립 배치 -> 특징 행렬 (클립 x 34) DataFrame.\"\"\"\n",
" P = spectrogram(X, denoise)\n",
" C = mfcc(P)\n",
" tot = band(P, 300, 3400) + 1e-12\n",
" flat = np.exp(np.log(P[:, (FREQS >= 300) & (FREQS < 3400)] + 1e-10).mean(1)) / (P[:, (FREQS >= 300) & (FREQS < 3400)].mean(1) + 1e-10)\n",
" logP = np.log(P + 1e-8)\n",
" jit, hnr = jitter_hnr(X)\n",
" F = np.c_[C.mean(2), C.std(2), np.abs(np.diff(C, axis=2)).mean((1, 2)), band(P, 2400, 3400) / tot, band(P, 300, 500) / tot,\n",
" flat.mean(1), np.abs(np.diff(logP, axis=2)).mean((1, 2)), jit, hnr, frame_mod(X)]\n",
" return pd.DataFrame(F, columns=FEAT_NAMES)\n",
"\n",
"\n",
"def enroll(seed=99):\n",
" \"\"\"지휘관·통신소 화자 6명의 등록 견본(깨끗한 진짜 클립 5개씩의 MFCC 평균).\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" tpl = {}\n",
" for spk in SPEAKERS:\n",
" Xe = np.stack([radio_channel(make_clip(\"bonafide\", spk, rng), rng, snr_db=25) for _ in range(5)])\n",
" tpl[spk] = extract(Xe)[[f\"mfcc{i}_m\" for i in range(1, 13)]].mean().values\n",
" return tpl\n",
"\n",
"\n",
"def speaker_sim(F, claimed, tpl):\n",
" \"\"\"주장 화자 견본과 MFCC 평균 벡터의 코사인 유사도(화자 검증 점수).\"\"\"\n",
" v = F[[f\"mfcc{i}_m\" for i in range(1, 13)]].values\n",
" t = np.stack([tpl[c] for c in claimed])\n",
" return (v * t).sum(1) / (np.linalg.norm(v, axis=1) * np.linalg.norm(t, axis=1) + 1e-12)\n",
"\n",
"def channel_quality(X):\n",
" \"\"\"채널 품질 지표: 추정 SNR(유효 프레임 에너지 상하위 비, dB), 단일톤 재밍 지수(주파수별 시간 평균/표준편차의 최대값), 끊김 비율(무음 10ms 비율).\"\"\"\n",
" P = spectrogram(X)\n",
" e = P.sum(1)\n",
" e = np.where(e > 1e-9, e, np.nan)\n",
" snr_est = 10 * np.log10(np.nanpercentile(e, 90, axis=1) / (np.nanpercentile(e, 10, axis=1) + 1e-12))\n",
" Pb = P[:, (FREQS >= 300) & (FREQS < 3400)]\n",
" tone = (Pb.mean(2) / (Pb.std(2) + 1e-12)).max(1) # 시간에 따라 일정한 주파수 성분(재밍 톤)일수록 큼\n",
" fr = np.abs(np.atleast_2d(X)).reshape(len(X), -1, 80).max(-1)\n",
" drop = (fr < 1e-6).mean(1)\n",
" return pd.DataFrame({\"snr_est\": snr_est, \"tone_idx\": tone, \"drop_ratio\": drop})\n",
"\n",
"tpl = enroll()\n",
"CONDS = {\"평시\": dict(snr=(12, 25)), \"저SNR\": dict(snr=(3, 8)), \"충격잡음\": dict(snr=(8, 15), impulse=0.002),\n",
" \"신호끊김\": dict(snr=(8, 15), dropout=0.4), \"재밍\": dict(snr=(8, 15), jam_tone=0.5)}\n",
"\n",
"\n",
"def ops_set(n, seed):\n",
" \"\"\"[교육용가상자료] 모의 운용 기록: 채널 조건 5종 x n/5 클립, 재생 공격 비중을 높게 둠.\"\"\"\n",
" parts = []\n",
" for i, (c, ch) in enumerate(CONDS.items()):\n",
" X, M = make_dataset(n // 5, seed=seed + i, p=[0.5, 0.15, 0.15, 0.2], **ch); M[\"cond\"] = c; parts.append((X, M))\n",
" return np.vstack([p[0] for p in parts]), pd.concat([p[1] for p in parts], ignore_index=True)\n",
"\n",
"\n",
"def feats(X, M, denoise=0.0):\n",
" F = extract(X, denoise); F[\"spk_sim\"] = speaker_sim(F, M[\"claimed\"], tpl)\n",
" return pd.concat([F, channel_quality(X)], axis=1)\n",
"\n",
"\n",
"# ---- 재보정 전: 미션 3 규칙 기반 Baseline(진짜 음성 3% 분위수 경계) ----\n",
"CUES = [(\"hf_ratio\", \"low\"), (\"frame_mod\", \"high\"), (\"hnr\", \"low\"), (\"flatness\", \"high\"), (\"lf_ratio\", \"low\"), (\"spk_sim\", \"low\")]\n",
"Xb, Mb = make_dataset(200, seed=7, labels=[\"bonafide\"])\n",
"Fb = feats(Xb, Mb)\n",
"th = {f: Fb[f].quantile(0.03) if d == \"low\" else Fb[f].quantile(0.97) for f, d in CUES}\n",
"rule = lambda F: pd.DataFrame({f: (F[f] < th[f]) if d == \"low\" else (F[f] > th[f]) for f, d in CUES}).any(axis=1).values\n",
"\n",
"Xo, Mo = ops_set(400, seed=900) # AAR 대상 운용 기록\n",
"Fo = feats(Xo, Mo)\n",
"Mo[\"pred\"] = rule(Fo); Mo[\"err\"] = Mo[\"pred\"] != Mo[\"spoof\"]\n",
"print(f\"[AAR] 운용 기록 {len(Mo)}건, 재보정 전 정확도 {1 - Mo['err'].mean():.1%}, 오판정 {Mo['err'].sum()}건\")\n",
"Mo[\"유형\"] = np.where(Mo[\"spoof\"], \"미탐(위조->진짜)\", \"오경보(진짜->위조)\")\n",
"tab = Mo[Mo[\"err\"]].groupby([\"유형\", \"cond\"]).size().unstack(fill_value=0)\n",
"print(tab.to_string())\n",
"miss_lab = Mo[Mo[\"err\"] & Mo[\"spoof\"]][\"label\"].value_counts()\n",
"print(\"미탐 위조 유형:\", {k: int(v) for k, v in miss_lab.items()})\n",
"print(\"원인 분석: ① 재생(replay)은 진짜 음성을 녹음한 것이라 화자 유사도·자연스러운 흔들림이 높아 규칙이 놓친다\",\n",
" \"\\n ② 재밍·저SNR·끊김 채널에서 진짜 음성의 단서 값이 경계를 넘어 오경보가 난다(채널 품질 미반영)\",\n",
" \"\\n ③ 단서별 고정 경계(3% 분위수)는 조건이 바뀌어도 그대로라 조합 판단이 불가능하다\")\n",
"\n",
"# ---- 재보정: 운용 기록(오판정 포함)과 조건 혼합 데이터로 재학습, 차감 배수·임계값 탐색 ----\n",
"Xn, Mn = ops_set(600, seed=950)\n",
"Xv, Mv = ops_set(300, seed=980) # 검증용(파라미터 선택)\n",
"best = None\n",
"for dn in [0.0, 0.5, 1.0]:\n",
" Ftr = pd.concat([feats(Xn, Mn, dn), feats(Xo, Mo, dn)], ignore_index=True)\n",
" ytr = np.r_[Mn[\"spoof\"], Mo[\"spoof\"]]\n",
" w = np.r_[np.ones(len(Mn)), np.where(Mo[\"err\"], 3, 1)] # AAR 오판정 사례 가중 반영(표본 복제)\n",
" idx = np.repeat(np.arange(len(ytr)), w.astype(int))\n",
" mdl = make_pipeline(StandardScaler(), MLPClassifier((64, 32), alpha=1e-3, max_iter=800, random_state=0)).fit(Ftr.iloc[idx], ytr[idx])\n",
" pv = mdl.predict_proba(feats(Xv, Mv, dn))[:, 1]\n",
" for t in [0.3, 0.4, 0.5, 0.6, 0.7]:\n",
" acc = ((pv >= t) == Mv[\"spoof\"]).mean()\n",
" if best is None or acc > best[0]:\n",
" best = (acc, dn, t, mdl)\n",
"acc_v, dn, t, mdl = best\n",
"yaml = f\"\"\"# configs/c204_tuned_params.yaml [교육용가상자료]\n",
"preprocess:\n",
" spectral_subtraction: {dn}\n",
"features: [mfcc13_mean, mfcc13_std, hf_ratio, lf_ratio, flatness, flux, jitter, hnr, frame_mod, spk_sim, snr_est, tone_idx, drop_ratio]\n",
"model:\n",
" type: mlp\n",
" hidden: [64, 32]\n",
" aar_case_weight: 3\n",
"decision:\n",
" spoof_threshold: {t}\n",
" poor_channel_rule: snr_est<3.3 or tone_idx>2.5 or drop_ratio>0.05 -> human_review\n",
"\"\"\"\n",
"print(\"\\n[재보정 파라미터]\\n\" + yaml.strip())\n",
"\n",
"# ---- 회귀 시험 ----\n",
"Xh, Mh = ops_set(400, seed=1100) # 새 운용 기록(학습에 쓰지 않음)\n",
"before = (rule(feats(Xh, Mh)) == Mh[\"spoof\"]).mean()\n",
"ph = mdl.predict_proba(feats(Xh, Mh, dn))[:, 1] >= t\n",
"after = (ph == Mh[\"spoof\"]).mean()\n",
"po = mdl.predict_proba(feats(Xo, Mo, dn))[:, 1] >= t\n",
"recur = int(((po != Mo[\"spoof\"]) & Mo[\"err\"]).sum())\n",
"newerr = int(((po != Mo[\"spoof\"]) & ~Mo[\"err\"]).sum())\n",
"print(f\"\\n[회귀 시험] 새 운용 기록 {len(Mh)}건 정확도: 재보정 전 {before:.1%} -> 후 {after:.1%} ({(after - before) * 100:+.1f}%p, 목표 +10%p: {'달성' if after - before >= 0.10 else '미달'})\")\n",
"print(f\"AAR 오판정 {Mo['err'].sum()}건 재발 {recur}건 (목표 0건: {'달성' if recur == 0 else '미달'}), 기존 정답 사례 중 새 오류 {newerr}건\")\n",
"print(\"조건별 정확도(새 기록):\", {c: f\"{(ph[Mh['cond'] == c] == Mh['spoof'][Mh['cond'] == c]).mean():.0%}\" for c in CONDS})\n",
"print(\"지속 학습 연계: 통신보안관이 확정한 사례만 라벨로 누적 -> 월 1회 재학습 -> 본 회귀 시험 통과 시에만 교체 배치\")\n",
"\n",
"plt.figure(figsize=(6, 3.5))\n",
"c = list(CONDS)\n",
"plt.bar(np.arange(5) - 0.2, [(rule(feats(Xh[Mh['cond'] == k], Mh[Mh['cond'] == k].reset_index(drop=True))) == Mh['spoof'][Mh['cond'] == k].values).mean() for k in c], 0.4, label=\"before (rules)\")\n",
"plt.bar(np.arange(5) + 0.2, [(ph[Mh['cond'] == k] == Mh['spoof'][Mh['cond'] == k]).mean() for k in c], 0.4, label=\"after (retuned)\")\n",
"plt.xticks(range(5), [\"normal\", \"low SNR\", \"impulse\", \"dropout\", \"jamming\"]); plt.ylabel(\"accuracy\"); plt.legend(); plt.title(\"AAR recalibration (synthetic)\")\n",
"plt.tight_layout(); plt.show()\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 09 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"[AAR] 운용 기록 400건, 재보정 전 정확도 69.8%, 오판정 121건\n",
"cond 신호끊김 재밍 저SNR 충격잡음 평시\n",
"유형 \n",
"미탐(위조->진짜) 10 11 3 12 23\n",
"오경보(진짜->위조) 10 15 26 8 3\n",
"미탐 위조 유형: {'replay': 53, 'vc_art': 4, 'tts_art': 2}\n",
"원인 분석: ① 재생(replay)은 진짜 음성을 녹음한 것이라 화자 유사도·자연스러운 흔들림이 높아 규칙이 놓친다 \n",
" ② 재밍·저SNR·끊김 채널에서 진짜 음성의 단서 값이 경계를 넘어 오경보가 난다(채널 품질 미반영) \n",
" ③ 단서별 고정 경계(3% 분위수)는 조건이 바뀌어도 그대로라 조합 판단이 불가능하다\n",
"\n",
"[재보정 파라미터]\n",
"# configs/c204_tuned_params.yaml [교육용가상자료]\n",
"preprocess:\n",
" spectral_subtraction: 0.0\n",
"features: [mfcc13_mean, mfcc13_std, hf_ratio, lf_ratio, flatness, flux, jitter, hnr, frame_mod, spk_sim, snr_est, tone_idx, drop_ratio]\n",
"model:\n",
" type: mlp\n",
" hidden: [64, 32]\n",
" aar_case_weight: 3\n",
"decision:\n",
" spoof_threshold: 0.3\n",
" poor_channel_rule: snr_est<3.3 or tone_idx>2.5 or drop_ratio>0.05 -> human_review\n",
"\n",
"[회귀 시험] 새 운용 기록 400건 정확도: 재보정 전 78.2% -> 후 96.8% (+18.5%p, 목표 +10%p: 달성)\n",
"AAR 오판정 121건 재발 0건 (목표 0건: 달성), 기존 정답 사례 중 새 오류 0건\n",
"조건별 정확도(새 기록): {'평시': '100%', '저SNR': '95%', '충격잡음': '98%', '신호끊김': '95%', '재밍': '96%'}\n",
"지속 학습 연계: 통신보안관이 확정한 사례만 라벨로 누적 -> 월 1회 재학습 -> 본 회귀 시험 통과 시에만 교체 배치\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 10. 무전 음성 스푸핑·딥페이크 탐지 시스템 통합 및 C2 음성 보안 브리핑 패키지 완성\n",
"**소요시간**: 약 3.5시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"무전 음성 스푸핑·딥페이크 탐지 시스템 통합 패키지 및 C2 보안 브리핑 보고서를 완성한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **검증부터 판별, 신뢰도 등급, 검증 대기열, 감사 로그까지 한 번에 실행되는 통합 패키지를 만들고 C2 음성 보안 브리핑을 완성한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **수신 기록을 검증해 손상·미등록 기록을 차단하고, 재보정 파라미터로 판별·등급을 산출해 위조 의심과 긴급 명령을 전량 검증 대기열에 올린다 과정을 진행하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `src/main_c204.py, README.md, docs/c204_final_report.pdf(md)` 형태의 결과물을 완성한다.\n",
"\n",
"### 🔍 평가 및 검증 기준 (PASS 조건)\n",
"#### 미션 판정\n",
"\n",
"**평가 구분:** 필수\n",
"\n",
"**핵심 평가영역:** 군 실무 즉시 도입 가능성, 운용 신뢰도 및 전략적 기대효과\n",
"\n",
"#### 미션 PASS 조건\n",
"\n",
"- 정량 기준: End-to-End 통합 파이프라인 실행 성공률 100%, 테스트 커버리지 85% 이상\n",
"\n",
"\n",
"- 진행 기준: 원클릭 실행 스크립트 동작 확인 및 최종 발표 산출물 작성 완료\n",
"\n",
"\n",
"- 검증 방법: 최종 검증 위원회(평가관) 모의 시연 및 전체 기능 통합 검증\n",
"\n",
"\n",
"#### 채점자 판정 메모\n",
"\n",
"> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
"\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"\"\"\"수신 메타데이터 검증, 무전 채널 전처리·특징 추출, 재보정 파라미터를 적용한 위조 판별·신뢰도 등급, 통신보안관 검증 대기열·감사 로그까지 한 번에 실행하는 통합 파이프라인을 만들고, 자체 점검(함수 커버리지 포함)과 C2 음성 보안 브리핑 요약을 출력한다.\"\"\"\n",
"import hashlib\n",
"import json\n",
"import time\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"from scipy import signal\n",
"from scipy.fft import dct\n",
"from sklearn.neural_network import MLPClassifier\n",
"from sklearn.preprocessing import StandardScaler\n",
"from sklearn.pipeline import make_pipeline\n",
"\n",
"# ---------------------------------------------------------------------------\n",
"# [교육용가상자료] 탐지 학습용 합성 신호 생성기 (실제 음성·화자·통신망 정보 없음)\n",
"# 사람 목소리를 만드는 기능이 아니라, 배음(하모닉) 신호에 '진짜 신호의 자연스러운 흔들림'과\n",
"# '알려진 위조 흔적(문헌상 탐지 단서)'의 통계적 특징을 표시해 탐지기를 학습시키기 위한 추상 신호다.\n",
"# bonafide : 자연스러운 피치 흔들림(jitter)·진폭 흔들림(shimmer)·숨소리 잡음\n",
"# tts_art : 흔들림이 거의 없고 고주파 에너지가 부족하며 20ms 프레임 경계 흔적이 있는 신호\n",
"# vc_art : 프레임마다 배음 위상이 끊기고 스펙트럼 포락선이 뭉개진 신호\n",
"# replay : 스피커 재생 대역 제한 + 실내 잔향이 더해진 신호\n",
"# ---------------------------------------------------------------------------\n",
"FS, DUR = 8000, 0.5\n",
"N = int(FS * DUR)\n",
"LABELS = [\"bonafide\", \"tts_art\", \"vc_art\", \"replay\"]\n",
"SPEAKERS = {f\"SPK_{c}\": (f0, fm) for c, f0, fm in [(\"A\", 115, (650, 1150, 2450)), (\"B\", 135, (700, 1220, 2600)), (\"C\", 98, (600, 1050, 2350)),\n",
" (\"D\", 160, (720, 1300, 2700)), (\"E\", 125, (680, 1180, 2500)), (\"F\", 105, (630, 1100, 2400))]}\n",
"\n",
"\n",
"def _source(rng, f0, fm, jitter, shimmer, phase_break=False, hf_cut=None, frame_mod=0.0):\n",
" t = np.arange(N) / FS\n",
" contour = f0 * (1 + 0.06 * np.sin(2 * np.pi * rng.uniform(1.5, 3.5) * t + rng.uniform(0, 6)))\n",
" f = contour * (1 + jitter * rng.normal(0, 1, N // 80 + 1).repeat(80)[:N]) # 10ms 단위 피치 흔들림(jitter)\n",
" phase = 2 * np.pi * np.cumsum(f) / FS\n",
" K = int(3800 / f0)\n",
" fk = np.arange(1, K + 1) * f0\n",
" amp = sum(np.exp(-((fk - F) / 120.0) ** 2) / (i + 1) for i, F in enumerate(fm)) + 0.02 / np.arange(1, K + 1)\n",
" if hf_cut:\n",
" amp = np.where(fk > hf_cut, amp * 0.2, amp)\n",
" ph = np.arange(1, K + 1)[:, None] * phase[None, :]\n",
" if phase_break: # 20ms 프레임마다 배음별 위상 불연속\n",
" ph = ph + rng.uniform(0, 2 * np.pi, (K, N // 160 + 1)).repeat(160, axis=1)[:, :N]\n",
" sig = (amp[:, None] * np.sin(ph)).sum(0)\n",
" env = 0.55 + 0.45 * np.sin(2 * np.pi * rng.uniform(3, 5) * t + rng.uniform(0, 6)) ** 2 # 음절형 진폭 변화\n",
" env *= 1 + shimmer * np.interp(t, np.linspace(0, DUR, 60), rng.normal(0, 1, 60)) # 완만한 진폭 흔들림(shimmer)\n",
" if frame_mod:\n",
" env *= 1 + frame_mod * np.sign(np.sin(2 * np.pi * 50 * t)) # 20ms 프레임 경계 흔적(50Hz)\n",
" return sig * env + (0.03 if jitter > 0.003 else 0.005) * rng.normal(0, 1, N) * env\n",
"\n",
"\n",
"def make_clip(label, speaker, rng):\n",
" f0, fm = SPEAKERS[speaker]\n",
" if label in (\"tts_art\", \"vc_art\"): # 위조는 화자 특성을 근사만 함\n",
" f0 = f0 * rng.uniform(0.92, 1.08); fm = tuple(F * rng.uniform(0.95, 1.05) for F in fm)\n",
" if label == \"bonafide\" or label == \"replay\":\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.008, 0.02), shimmer=rng.uniform(0.05, 0.12))\n",
" elif label == \"tts_art\":\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.0, 0.002), shimmer=0.01, hf_cut=rng.uniform(2300, 2800), frame_mod=rng.uniform(0.08, 0.2))\n",
" else:\n",
" x = _source(rng, f0, fm, jitter=rng.uniform(0.006, 0.015), shimmer=rng.uniform(0.04, 0.1), phase_break=True)\n",
" if label == \"replay\": # 스피커 대역 + 실내 잔향\n",
" x = signal.sosfilt(signal.butter(4, [450, 2900], \"bandpass\", fs=FS, output=\"sos\"), x)\n",
" ir = rng.normal(0, 1, int(0.08 * FS)) * np.exp(-np.arange(int(0.08 * FS)) / (0.02 * FS)); ir[0] = 3.0\n",
" x = signal.fftconvolve(x, ir)[:N]\n",
" return x / (np.abs(x).max() + 1e-9)\n",
"\n",
"\n",
"RADIO_BP = signal.butter(4, [300, 3400], \"bandpass\", fs=FS, output=\"sos\")\n",
"\n",
"\n",
"def radio_channel(x, rng, snr_db=15.0, impulse=0.0, dropout=0.0, clip=1.0, jam_tone=0.0):\n",
" \"\"\"[교육용가상자료] 무전 채널 모형: 300~3400Hz 대역 제한, 배경 잡음(SNR), 충격성 잡음, 신호 끊김, 포화(클리핑), 단일톤 재밍.\"\"\"\n",
" y = signal.sosfilt(RADIO_BP, x)\n",
" p = np.mean(y ** 2)\n",
" noise = rng.normal(0, 1, N) * np.sqrt(p / 10 ** (snr_db / 10))\n",
" noise += np.cumsum(rng.normal(0, 0.02, N)) * np.sqrt(p) * 0.05 # 저주파 차량·엔진형 소음\n",
" y = y + noise\n",
" if impulse:\n",
" k = rng.random(N) < impulse\n",
" y[k] += rng.normal(0, 6 * np.sqrt(p), k.sum())\n",
" if jam_tone:\n",
" y += jam_tone * np.sqrt(p) * np.sin(2 * np.pi * rng.uniform(800, 2000) * np.arange(N) / FS)\n",
" if dropout:\n",
" for _ in range(int(dropout * 10)):\n",
" a = int(rng.integers(0, N - 400)); y[a:a + int(rng.integers(100, 400))] = 0.0\n",
" return np.clip(y / (np.abs(y).max() + 1e-9), -clip, clip)\n",
"\n",
"\n",
"def make_dataset(n=400, seed=7, snr=(5, 25), labels=LABELS, p=None, claim=\"SPK_A\", **chan):\n",
" \"\"\"클립 n개: 진짜는 여러 화자, 위조는 모두 지휘관 화자(SPK_A)를 사칭. 반환: (신호 배열, 메타 DataFrame).\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" X, meta = np.zeros((n, N)), []\n",
" if p is None and len(labels) > 1 and labels[0] == \"bonafide\": # 기본: 진짜 50%, 나머지 위조 유형 균등\n",
" p = [0.5] + [0.5 / (len(labels) - 1)] * (len(labels) - 1)\n",
" for i in range(n):\n",
" lab = str(rng.choice(labels, p=p))\n",
" spk = str(rng.choice(list(SPEAKERS))) if lab == \"bonafide\" else claim\n",
" if lab == \"replay\":\n",
" spk = claim\n",
" s = float(rng.uniform(*snr)) if isinstance(snr, tuple) else float(snr)\n",
" X[i] = radio_channel(make_clip(lab, spk, rng), rng, snr_db=s, **chan)\n",
" meta.append({\"clip\": f\"C{seed:02d}{i:04d}\", \"label\": lab, \"spoof\": lab != \"bonafide\", \"claimed\": spk if lab == \"bonafide\" else claim,\n",
" \"true_spk\": spk if lab == \"bonafide\" else \"unknown\", \"snr_db\": round(s, 1)})\n",
" return X, pd.DataFrame(meta)\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# ---- 전처리·음향 특징(MFCC/스펙트로그램) 공통 함수: 여러 클립을 한 번에 계산 ----\n",
"NFFT, HOP = 256, 80\n",
"\n",
"\n",
"def mel_bank(n_mels=26, fmin=300, fmax=3400):\n",
" mel = lambda f: 2595 * np.log10(1 + f / 700)\n",
" imel = lambda m: 700 * (10 ** (m / 2595) - 1)\n",
" pts = imel(np.linspace(mel(fmin), mel(fmax), n_mels + 2))\n",
" freqs = np.fft.rfftfreq(NFFT, 1 / FS)\n",
" fb = np.zeros((n_mels, len(freqs)))\n",
" for m in range(n_mels):\n",
" l, c, r = pts[m], pts[m + 1], pts[m + 2]\n",
" fb[m] = np.clip(np.minimum((freqs - l) / (c - l), (r - freqs) / (r - c)), 0, None)\n",
" return fb\n",
"\n",
"\n",
"MEL = mel_bank()\n",
"FREQS = np.fft.rfftfreq(NFFT, 1 / FS)\n",
"\n",
"\n",
"def spectrogram(X, denoise=0.0):\n",
" \"\"\"배치 전력 스펙트로그램 (클립, 주파수, 프레임). denoise>0이면 잡음 바닥(주파수별 하위 10%)을 그 배수만큼 빼는 스펙트럼 차감.\"\"\"\n",
" X = np.atleast_2d(X)\n",
" Xp = np.c_[X[:, :1], X[:, 1:] - 0.95 * X[:, :-1]] # 프리엠퍼시스\n",
" frames = np.lib.stride_tricks.sliding_window_view(Xp, NFFT, axis=1)[:, ::HOP] * np.hanning(NFFT)\n",
" P = np.abs(np.fft.rfft(frames, axis=-1)) ** 2\n",
" P = np.transpose(P, (0, 2, 1))\n",
" if denoise:\n",
" floor = np.percentile(P, 10, axis=2, keepdims=True)\n",
" P = np.maximum(P - denoise * floor, 0.05 * P)\n",
" return P\n",
"\n",
"\n",
"def mfcc(P, n=13):\n",
" logmel = np.log(np.einsum(\"mf,cft->cmt\", MEL, P) + 1e-8)\n",
" return dct(logmel, type=2, axis=1, norm=\"ortho\")[:, :n, :]\n",
"\n",
"\n",
"def band(P, lo, hi):\n",
" return P[:, (FREQS >= lo) & (FREQS < hi), :].sum(axis=(1, 2))\n",
"\n",
"\n",
"def jitter_hnr(X):\n",
" \"\"\"40ms 프레임 자기상관으로 피치 주기를 찾아 피치 흔들림(jitter)과 배음 대 잡음비(HNR 근사)를 구한다.\"\"\"\n",
" W, lag0, lag1 = 320, int(FS / 250), int(FS / 70)\n",
" fr = np.lib.stride_tricks.sliding_window_view(np.atleast_2d(X), W, axis=1)[:, ::160]\n",
" fr = fr - fr.mean(-1, keepdims=True)\n",
" spec = np.fft.rfft(fr, 2 * W, axis=-1)\n",
" ac = np.fft.irfft(np.abs(spec) ** 2, axis=-1)[..., :W]\n",
" ac = ac / (ac[..., :1] + 1e-12)\n",
" seg = ac[..., lag0:lag1]\n",
" k = seg.argmax(-1)\n",
" peak = seg.max(-1)\n",
" km, kp = np.take_along_axis(seg, np.clip(k - 1, 0, None)[..., None], -1)[..., 0], np.take_along_axis(seg, np.clip(k + 1, None, seg.shape[-1] - 1)[..., None], -1)[..., 0]\n",
" lag = k + lag0 + 0.5 * (km - kp) / (km - 2 * peak + kp - 1e-12) # 포물선 보간으로 주기 정밀화\n",
" f0 = FS / lag\n",
" jit = np.median(np.abs(np.diff(f0, axis=1)) / f0[:, 1:], axis=1)\n",
" return jit, np.median(peak, axis=1)\n",
"\n",
"\n",
"def frame_mod(X):\n",
" \"\"\"진폭 포락선(2.5ms 간격)의 50Hz 성분 비율: 20ms 프레임 경계 흔적(합성 단서).\"\"\"\n",
" env = np.abs(np.atleast_2d(X)).reshape(len(np.atleast_2d(X)), -1, 20).mean(-1)\n",
" sp = np.abs(np.fft.rfft(env - env.mean(1, keepdims=True), axis=1)) ** 2\n",
" f = np.fft.rfftfreq(env.shape[1], 20 / FS)\n",
" return sp[:, (f > 47) & (f < 53)].sum(1) / (sp[:, (f > 30) & (f < 70)].sum(1) + 1e-12)\n",
"\n",
"\n",
"FEAT_NAMES = [f\"mfcc{i}_m\" for i in range(13)] + [f\"mfcc{i}_s\" for i in range(13)] + \\\n",
" [\"delta_s\", \"hf_ratio\", \"lf_ratio\", \"flatness\", \"flux\", \"jitter\", \"hnr\", \"frame_mod\"]\n",
"\n",
"\n",
"def extract(X, denoise=0.0):\n",
" \"\"\"클립 배치 -> 특징 행렬 (클립 x 34) DataFrame.\"\"\"\n",
" P = spectrogram(X, denoise)\n",
" C = mfcc(P)\n",
" tot = band(P, 300, 3400) + 1e-12\n",
" flat = np.exp(np.log(P[:, (FREQS >= 300) & (FREQS < 3400)] + 1e-10).mean(1)) / (P[:, (FREQS >= 300) & (FREQS < 3400)].mean(1) + 1e-10)\n",
" logP = np.log(P + 1e-8)\n",
" jit, hnr = jitter_hnr(X)\n",
" F = np.c_[C.mean(2), C.std(2), np.abs(np.diff(C, axis=2)).mean((1, 2)), band(P, 2400, 3400) / tot, band(P, 300, 500) / tot,\n",
" flat.mean(1), np.abs(np.diff(logP, axis=2)).mean((1, 2)), jit, hnr, frame_mod(X)]\n",
" return pd.DataFrame(F, columns=FEAT_NAMES)\n",
"\n",
"\n",
"def enroll(seed=99):\n",
" \"\"\"지휘관·통신소 화자 6명의 등록 견본(깨끗한 진짜 클립 5개씩의 MFCC 평균).\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" tpl = {}\n",
" for spk in SPEAKERS:\n",
" Xe = np.stack([radio_channel(make_clip(\"bonafide\", spk, rng), rng, snr_db=25) for _ in range(5)])\n",
" tpl[spk] = extract(Xe)[[f\"mfcc{i}_m\" for i in range(1, 13)]].mean().values\n",
" return tpl\n",
"\n",
"\n",
"def speaker_sim(F, claimed, tpl):\n",
" \"\"\"주장 화자 견본과 MFCC 평균 벡터의 코사인 유사도(화자 검증 점수).\"\"\"\n",
" v = F[[f\"mfcc{i}_m\" for i in range(1, 13)]].values\n",
" t = np.stack([tpl[c] for c in claimed])\n",
" return (v * t).sum(1) / (np.linalg.norm(v, axis=1) * np.linalg.norm(t, axis=1) + 1e-12)\n",
"\n",
"def channel_quality(X):\n",
" \"\"\"채널 품질 지표: 추정 SNR(유효 프레임 에너지 상하위 비, dB), 단일톤 재밍 지수(주파수별 시간 평균/표준편차의 최대값), 끊김 비율(무음 10ms 비율).\"\"\"\n",
" P = spectrogram(X)\n",
" e = P.sum(1)\n",
" e = np.where(e > 1e-9, e, np.nan)\n",
" snr_est = 10 * np.log10(np.nanpercentile(e, 90, axis=1) / (np.nanpercentile(e, 10, axis=1) + 1e-12))\n",
" Pb = P[:, (FREQS >= 300) & (FREQS < 3400)]\n",
" tone = (Pb.mean(2) / (Pb.std(2) + 1e-12)).max(1) # 시간에 따라 일정한 주파수 성분(재밍 톤)일수록 큼\n",
" fr = np.abs(np.atleast_2d(X)).reshape(len(X), -1, 80).max(-1)\n",
" drop = (fr < 1e-6).mean(1)\n",
" return pd.DataFrame({\"snr_est\": snr_est, \"tone_idx\": tone, \"drop_ratio\": drop})\n",
"\n",
"CONFIG = {\"spectral_subtraction\": 0.0, \"spoof_threshold\": 0.3, \"high_threshold\": 0.8, # 미션 9 재보정값\n",
" \"poor\": {\"snr_est\": 3.3, \"tone_idx\": 2.5, \"drop_ratio\": 0.05}}\n",
"CALLED = set()\n",
"\n",
"\n",
"def traced(fn):\n",
" \"\"\"함수 커버리지 측정용: 호출된 파이프라인 함수 이름을 기록.\"\"\"\n",
" def w(*a, **k):\n",
" CALLED.add(fn.__name__); return fn(*a, **k)\n",
" w.__name__ = fn.__name__\n",
" return w\n",
"\n",
"\n",
"@traced\n",
"def validate(meta, x):\n",
" errs = []\n",
" if not str(meta.get(\"clip\", \"\")).startswith(\"C\"):\n",
" errs.append(\"clip_id 형식\")\n",
" if meta.get(\"claimed\") not in SPEAKERS:\n",
" errs.append(\"주장 화자 미등록\")\n",
" if len(x) != N or not np.isfinite(x).all():\n",
" errs.append(\"신호 길이·값 오류\")\n",
" elif np.abs(x).max() < 1e-6:\n",
" errs.append(\"무신호\")\n",
" return errs\n",
"\n",
"\n",
"@traced\n",
"def features(X, claimed):\n",
" F = extract(X, CONFIG[\"spectral_subtraction\"]); F[\"spk_sim\"] = speaker_sim(F, claimed, TPL)\n",
" return pd.concat([F, channel_quality(X)], axis=1)\n",
"\n",
"\n",
"@traced\n",
"def grade(p, q):\n",
" poor = (q[\"snr_est\"] < CONFIG[\"poor\"][\"snr_est\"]) | (q[\"tone_idx\"] > CONFIG[\"poor\"][\"tone_idx\"]) | (q[\"drop_ratio\"] > CONFIG[\"poor\"][\"drop_ratio\"])\n",
" g = np.where(p >= CONFIG[\"high_threshold\"], \"High\", np.where(p >= CONFIG[\"spoof_threshold\"], \"Med\", \"Low\"))\n",
" return np.where((g == \"High\") & poor.values & (p < 0.95), \"Med\", g), poor.values\n",
"\n",
"\n",
"class Queue:\n",
" def __init__(self):\n",
" self.items, self.log = [], []\n",
"\n",
" @traced\n",
" def submit(self, rec):\n",
" prev = self.log[-1][\"hash\"] if self.log else \"GENESIS\"\n",
" body = {\"event\": \"SUBMIT\", \"id\": rec[\"clip\"], \"grade\": rec[\"grade\"], \"score\": round(rec[\"score\"], 3), \"urgent\": rec[\"urgent\"]}\n",
" self.log.append({**body, \"prev\": prev, \"hash\": hashlib.sha256((prev + json.dumps(body, sort_keys=True)).encode()).hexdigest()})\n",
" self.items.append({**rec, \"status\": \"PENDING\"}) # 자동 전파·차단 없음, 통신보안관 검증 대기\n",
"\n",
" @traced\n",
" def verify(self):\n",
" prev = \"GENESIS\"\n",
" for r in self.log:\n",
" body = {k: v for k, v in r.items() if k not in (\"prev\", \"hash\")}\n",
" if r[\"prev\"] != prev or r[\"hash\"] != hashlib.sha256((prev + json.dumps(body, sort_keys=True)).encode()).hexdigest():\n",
" return False\n",
" prev = r[\"hash\"]\n",
" return True\n",
"\n",
"\n",
"@traced\n",
"def train(seed=1200):\n",
" conds = [dict(snr=(10, 25)), dict(snr=(3, 8)), dict(snr=(8, 15), impulse=0.002), dict(snr=(8, 15), dropout=0.4), dict(snr=(8, 15), jam_tone=0.5)]\n",
" parts = [make_dataset(140, seed=seed + i, p=[0.5, 0.15, 0.15, 0.2], **c) for i, c in enumerate(conds)]\n",
" X = np.vstack([p[0] for p in parts]); M = pd.concat([p[1] for p in parts], ignore_index=True)\n",
" return make_pipeline(StandardScaler(), MLPClassifier((64, 32), alpha=1e-3, max_iter=800, random_state=0)).fit(features(X, M[\"claimed\"]), M[\"spoof\"])\n",
"\n",
"\n",
"@traced\n",
"def main(X, M, model):\n",
" queue, rejected, out = Queue(), {}, []\n",
" ok = []\n",
" for i in range(len(X)):\n",
" e = validate(M.iloc[i].to_dict(), X[i])\n",
" if e:\n",
" rejected[M[\"clip\"][i]] = e\n",
" else:\n",
" ok.append(i)\n",
" Xo, Mo = X[ok], M.iloc[ok].reset_index(drop=True)\n",
" t0 = time.perf_counter()\n",
" F = features(Xo, Mo[\"claimed\"])\n",
" p = model.predict_proba(F)[:, 1]\n",
" g, poor = grade(p, F)\n",
" ms = (time.perf_counter() - t0) * 1000 / len(Xo)\n",
" for k in range(len(Mo)):\n",
" rec = {\"clip\": Mo[\"clip\"][k], \"score\": float(p[k]), \"grade\": str(g[k]), \"urgent\": bool(Mo[\"urgent\"][k]), \"poor\": bool(poor[k])}\n",
" out.append(rec)\n",
" if rec[\"grade\"] != \"Low\" or rec[\"urgent\"]: # 위조 의심 + 긴급 명령은 전량 사람 검증\n",
" queue.submit(rec)\n",
" return pd.DataFrame(out), Mo, rejected, queue, ms\n",
"\n",
"\n",
"T0 = time.perf_counter()\n",
"TPL = enroll()\n",
"model = train()\n",
"# [교육용가상자료] 야간 운용 시간대 수신 기록 300건(혼합 채널) + 손상 기록 3건\n",
"parts = [make_dataset(60, seed=1300 + i, p=[0.5, 0.15, 0.15, 0.2], **c) for i, c in\n",
" enumerate([dict(snr=(10, 25)), dict(snr=(3, 8)), dict(snr=(8, 15), impulse=0.002), dict(snr=(8, 15), dropout=0.4), dict(snr=(8, 15), jam_tone=0.5)])]\n",
"X = np.vstack([p[0] for p in parts]); M = pd.concat([p[1] for p in parts], ignore_index=True)\n",
"M[\"urgent\"] = np.random.default_rng(1400).random(len(M)) < 0.15\n",
"X[5] = np.nan; X[77] = 0.0; M.loc[150, \"claimed\"] = \"SPK_Z\"\n",
"R, Mo, rejected, queue, ms = main(X, M, model)\n",
"y = Mo[\"spoof\"].values; alarm = R[\"grade\"].values != \"Low\"\n",
"recall, far, acc = alarm[y].mean(), alarm[~y].mean(), (alarm == y).mean()\n",
"elapsed = time.perf_counter() - T0\n",
"\n",
"chk = {\n",
" \"손상 기록 3건 차단\": len(rejected) == 3,\n",
" \"처리 기록 297건\": len(R) == 297,\n",
" \"위조 재현율 90% 이상\": recall >= 0.9,\n",
" \"오경보율 10% 이하\": far <= 0.10,\n",
" \"자동 전파·차단 0건\": all(i[\"status\"] == \"PENDING\" for i in queue.items),\n",
" \"긴급 명령 전량 검증 대기\": all(c in {i[\"clip\"] for i in queue.items} for c in Mo[\"clip\"][Mo[\"urgent\"]]),\n",
" \"감사 로그 무결성\": queue.verify() and len(queue.log) == len(queue.items),\n",
" \"클립당 200ms 이내\": ms < 200,\n",
" \"등급 규칙 일관\": all((r.grade == \"Low\") == (r.score < CONFIG[\"spoof_threshold\"]) for r in R.itertuples()),\n",
" \"점수 범위 0~1·NaN 없음\": R[\"score\"].between(0, 1).all(),\n",
"}\n",
"passed = sum(chk.values())\n",
"for k, v in chk.items():\n",
" print(f\"[{'PASS' if v else 'FAIL'}] {k}\")\n",
"funcs = [\"validate\", \"features\", \"grade\", \"submit\", \"verify\", \"train\", \"main\"]\n",
"cov = len(CALLED & set(funcs)) / len(funcs)\n",
"print(f\"자체 점검 {passed}/{len(chk)} 통과, E2E 실행 성공률 {'100%' if passed == len(chk) else f'{passed / len(chk):.0%}'}, 함수 커버리지 {cov:.0%} (목표 85% 이상: {'달성' if cov >= 0.85 else '미달'})\")\n",
"print(\"차단된 기록:\", rejected)\n",
"\n",
"n_hi, n_med = (R[\"grade\"] == \"High\").sum(), (R[\"grade\"] == \"Med\").sum()\n",
"print(f\"\"\"\n",
"[C2 음성 보안 브리핑 요약 - 교육용가상자료]\n",
"1. 처리: 수신 {len(M)}건 중 {len(R)}건 판정, 손상·미등록 {len(rejected)}건 차단, 전체 {elapsed:.1f}s, 클립당 {ms:.2f} ms\n",
"2. 성능: 위조 재현율 {recall:.1%}, 오경보율 {far:.1%}, 정확도 {acc:.1%} (평시·저SNR·충격·끊김·재밍 혼합 채널)\n",
"3. 검증 대기: {len(queue.items)}건 (High {n_hi} / Med {n_med} / 긴급 명령 {int(Mo['urgent'].sum())}), 자동 전파·차단 0건\n",
"4. 운용: 긴급 명령은 등급과 무관하게 통신보안관이 역호출 인증·청음 후 전파하며, 최종 명령 책임은 지휘계통에 있다.\n",
"5. 한계·확장: 합성 신호 기준 결과로 실제 무전기·음성에는 재학습·재검증이 필요하다. 인접 체계 연동은 판정 결과(JSON)만 전달하는 API로 확장한다.\"\"\")\n",
"\n",
"plt.figure(figsize=(6, 3.5))\n",
"plt.hist([R[\"score\"][~y], R[\"score\"][y]], bins=20, label=[\"bonafide\", \"spoof\"], color=[\"tab:green\", \"tab:red\"])\n",
"plt.axvline(CONFIG[\"spoof_threshold\"], c=\"k\", ls=\"--\"); plt.yscale(\"log\"); plt.xlabel(\"spoof score\"); plt.legend(); plt.title(\"Integrated pipeline output (synthetic)\")\n",
"plt.tight_layout(); plt.show()\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 10 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"[PASS] 손상 기록 3건 차단\n",
"[PASS] 처리 기록 297건\n",
"[PASS] 위조 재현율 90% 이상\n",
"[PASS] 오경보율 10% 이하\n",
"[PASS] 자동 전파·차단 0건\n",
"[PASS] 긴급 명령 전량 검증 대기\n",
"[PASS] 감사 로그 무결성\n",
"[PASS] 클립당 200ms 이내\n",
"[PASS] 등급 규칙 일관\n",
"[PASS] 점수 범위 0~1·NaN 없음\n",
"자체 점검 10/10 통과, E2E 실행 성공률 100%, 함수 커버리지 100% (목표 85% 이상: 달성)\n",
"차단된 기록: {'C13000005': ['신호 길이·값 오류'], 'C13010017': ['무신호'], 'C13020030': ['주장 화자 미등록']}\n",
"\n",
"[C2 음성 보안 브리핑 요약 - 교육용가상자료]\n",
"1. 처리: 수신 300건 중 297건 판정, 손상·미등록 3건 차단, 전체 8.5s, 클립당 1.66 ms\n",
"2. 성능: 위조 재현율 96.7%, 오경보율 4.9%, 정확도 96.0% (평시·저SNR·충격·끊김·재밍 혼합 채널)\n",
"3. 검증 대기: 174건 (High 143 / Med 12 / 긴급 명령 43), 자동 전파·차단 0건\n",
"4. 운용: 긴급 명령은 등급과 무관하게 통신보안관이 역호출 인증·청음 후 전파하며, 최종 명령 책임은 지휘계통에 있다.\n",
"5. 한계·확장: 합성 신호 기준 결과로 실제 무전기·음성에는 재학습·재검증이 필요하다. 인접 체계 연동은 판정 결과(JSON)만 전달하는 API로 확장한다.\n",
"```\n",
"</details>\n"
]
}
],
"metadata": {
"colab": {
"provenance": []
},
"kernelspec": {
"display_name": "Python 3",
"name": "python3"
},
"language_info": {
"name": "python"
}
},
"nbformat": 4,
"nbformat_minor": 0
}
Reference in new issue View Git Blame Copy Permalink