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

3085 lines
208 KiB
Plaintext
Raw Permalink Blame History

{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# 🎯 [C202] 대역폭 제한 시 메시지 우선순위 큐·재전송 프로토타입\n",
"### Google Colab 실습 환경 (지휘통제(C2) / Level 2)\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/C202/C202_solution.ipynb)\n",
"\n",
"---\n",
"\n",
"## 📌 1. 프로젝트 개요 및 배경\n",
"\n",
"| 항목 | 내용 |\n",
"| :--- | :--- |\n",
"| **과정 분류** | 지휘통제(C2) (Level 2) |\n",
"| **학습 소요 시간** | 10개 미션 (5일 / 총 27.5시간) |\n",
"| **환경 사양** | Python 3.10+, numpy, pandas, matplotlib, scipy, scikit-learn |\n",
"\n",
"### 📖 프로젝트 목표\n",
"러시아-우크라이나 전쟁에서는 전자전 재밍과 통신 장비 부족으로 협대역 무선망의 대역폭이 크게 줄어드는 상황이 잦았고, 이때 적 접촉 보고나 의무후송 요청 같은 긴급 전문이 행정 전문 뒤에 묶여 늦게 전달되는 문제가 지휘통제의 약점이 되었다. 본 프로젝트에서는 가상 협대역 링크(2.4kbps)와 가상 전술 전문을 코드로 직접 만들어, 전문 구조 정의, 패킷 파싱·압축 전처리, 긴급도 기반 우선순위 큐, 채널 상태에 따른 지능형 재전송, 링크 신뢰도 산출, 통신 관제 화면, 통신관 HITL 통제, 대역폭 급감 스트레스 시험, AAR 재조정, 최종 통합까지 메시지 우선순위 큐·재전송 프로토타입을 단계별로 구현한다. 최종 목표는 대역폭이 제한된 상황에서도 FLASH·IMMEDIATE 전문을 기한 안에 전달하고, 어떤 전문도 잃어버리지 않으며, 순서 조정의 최종 판단은 통신관이 내리도록 설계하는 것이다.\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. 협대역 전술 무선망(VHF/UHF) 대역폭 제약 조건 분석 및 전술 메시지 구조 정의\n",
"**소요시간**: 약 1.5시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"협대역 전술 무전망(VHF/UHF)의 전송 대역폭 제약과 전술 전문(Message) 구조를 분석한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **협대역 전술 무선망의 대역폭 제약이 전문 전송시간에 주는 영향을 분석하고, 전문 헤더·긴급도·링크 조건·전술 파라미터를 담은 데이터 스키마와 검증기를 만든다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **전문 유형(접촉 보고, 의무후송 요청, 상황 보고, 위치 보고, 보급 요청, 행정)별 평균 크기를 1.2~16kbps 링크에서 보낼 때 걸리는 시간을 표로 정리하고 규격화하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `data/schema/c202_input_schema.json 및 docs/c202_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": [
"\"\"\"협대역 전술 무선망(VHF/UHF)의 대역폭 제약을 전문 유형별 전송시간으로 분석하고, 전술 전문의 헤더·긴급도·링크·전술 파라미터 스키마를 정의해 가상 샘플 데이터를 검증한다.\"\"\"\n",
"import json\n",
"import re\n",
"import numpy as np\n",
"import pandas as pd\n",
"\n",
"# ---------------------------------------------------------------------------\n",
"# [교육용가상자료] 가상 전술 전문(메시지) 생성기 (실제 부대·통신망·제원 정보 없음)\n",
"# 긴급도 4단계(FLASH/IMMEDIATE/PRIORITY/ROUTINE)와 전문 유형별 크기를 가진 전문을 만든다.\n",
"# ---------------------------------------------------------------------------\n",
"PREC = [\"FLASH\", \"IMMEDIATE\", \"PRIORITY\", \"ROUTINE\"]\n",
"RANK = {p: i for i, p in enumerate(PREC)}\n",
"DEADLINE = {\"FLASH\": 10, \"IMMEDIATE\": 30, \"PRIORITY\": 120, \"ROUTINE\": 600} # 가상 전달 목표 시간(초)\n",
"MSG_TYPES = { # 유형: (긴급도 후보, 확률, 본문 크기 범위 bytes, 발생 비율)\n",
" \"CONTACT\": ([\"FLASH\", \"IMMEDIATE\"], [0.7, 0.3], (120, 300), 0.08),\n",
" \"MEDEVAC\": ([\"FLASH\", \"IMMEDIATE\"], [0.4, 0.6], (150, 350), 0.05),\n",
" \"SITREP\": ([\"IMMEDIATE\", \"PRIORITY\"], [0.2, 0.8], (600, 2500), 0.20),\n",
" \"POSREP\": ([\"PRIORITY\", \"ROUTINE\"], [0.5, 0.5], (60, 120), 0.35),\n",
" \"LOGREQ\": ([\"PRIORITY\", \"ROUTINE\"], [0.4, 0.6], (300, 1200), 0.14),\n",
" \"ADMIN\": ([\"ROUTINE\"], [1.0], (1000, 5000), 0.18),\n",
"}\n",
"\n",
"\n",
"def make_messages(duration=600, rate=0.12, seed=7, t_start=0.0, prefix=\"M\"):\n",
" \"\"\"duration초 동안 평균 rate건/초로 발생하는 가상 전문 목록.\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" n = int(rng.poisson(rate * duration))\n",
" times = np.sort(rng.uniform(t_start, t_start + duration, n))\n",
" types = list(MSG_TYPES)\n",
" msgs = []\n",
" for i, t0 in enumerate(times):\n",
" ty = types[rng.choice(len(types), p=[MSG_TYPES[k][3] for k in types])]\n",
" cands, p, (lo, hi), _ = MSG_TYPES[ty]\n",
" msgs.append({\"id\": f\"{prefix}{i:04d}\", \"t0\": float(t0), \"type\": ty, \"prec\": str(rng.choice(cands, p=p)),\n",
" \"size\": int(rng.integers(lo, hi + 1)), \"src\": f\"U{int(rng.integers(1, 9)):02d}\"})\n",
" return msgs\n",
"\n",
"\n",
"class Link:\n",
" \"\"\"[교육용가상자료] 가상 협대역 무선 링크(기본 2.4kbps). 초 단위 상태(good/bad/jam)에 따라\n",
" 비트오류율이 바뀌고(Gilbert-Elliott 모델), 구간별 대역폭 감소도 줄 수 있다.\"\"\"\n",
" BER = {\"good\": 1e-5, \"bad\": 2e-4, \"jam\": 1.2e-3}\n",
"\n",
" def __init__(self, horizon=4000, bw_bps=2400, seed=1, p_gb=0.02, p_bg=0.15, jam=(), bw_drop=(), rtt=(1.2, 0.4)):\n",
" rng = np.random.default_rng(seed)\n",
" st, s = [], \"good\"\n",
" for _ in range(int(horizon) + 1):\n",
" s = (\"bad\" if rng.random() < p_gb else \"good\") if s == \"good\" else (\"good\" if rng.random() < p_bg else \"bad\")\n",
" st.append(s)\n",
" for a, b in jam:\n",
" for k in range(int(a), min(int(b), len(st))):\n",
" st[k] = \"jam\"\n",
" self.states, self.bw_bps, self.bw_drop, self.rtt_par = st, bw_bps, list(bw_drop), rtt\n",
"\n",
" def state(self, t):\n",
" return self.states[min(int(t), len(self.states) - 1)]\n",
"\n",
" def bw(self, t):\n",
" f = 1.0\n",
" for a, b, fac in self.bw_drop:\n",
" if a <= t < b:\n",
" f = fac\n",
" return self.bw_bps * f\n",
"\n",
" def p_loss(self, nbytes, t):\n",
" return 1 - (1 - self.BER[self.state(t)]) ** (8 * nbytes)\n",
"\n",
" def rtt(self, rng):\n",
" return self.rtt_par[0] + rng.exponential(self.rtt_par[1])\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# 1) 대역폭 제약 분석: 링크 속도별 전문 1건 전송시간(헤더 32bytes/조각 256bytes 기준, 재전송 제외)\n",
"print(\"[대역폭별 전문 1건 전송시간(초) - 가상 링크]\")\n",
"rows = []\n",
"for ty, (_, _, (lo, hi), _) in MSG_TYPES.items():\n",
" size = (lo + hi) / 2\n",
" nbytes = size + np.ceil(size / 256) * 32\n",
" rows.append({\"유형\": ty, \"평균 크기(B)\": int(size), **{f\"{bw / 1000:g}kbps\": round(nbytes * 8 / bw, 1) for bw in (1200, 2400, 9600, 16000)}})\n",
"print(pd.DataFrame(rows).to_string(index=False))\n",
"print(\"-> 2.4kbps에서 행정전문 1건이 약 10초 이상 채널을 점유한다. FLASH 전달 목표(10초)를 지키려면 우선순위 큐와 조각 단위 전송이 필요하다.\")\n",
"\n",
"# 2) 스키마 정의: 전문 헤더, 링크 조건, 전술 파라미터(기상·위치오차·통신지연)\n",
"SCHEMA = {\n",
" \"required\": [\"msg_id\", \"precedence\", \"msg_type\", \"src\", \"created_s\", \"payload_bytes\", \"link\", \"tactical\"],\n",
" \"fields\": {\n",
" \"msg_id\": {\"type\": str, \"pattern\": r\"^M\\d{4}$\"},\n",
" \"precedence\": {\"type\": str, \"enum\": PREC},\n",
" \"msg_type\": {\"type\": str, \"enum\": list(MSG_TYPES)},\n",
" \"src\": {\"type\": str, \"pattern\": r\"^U\\d{2}$\"},\n",
" \"created_s\": {\"type\": (int, float), \"min\": 0},\n",
" \"payload_bytes\": {\"type\": int, \"min\": 1, \"max\": 8000},\n",
" },\n",
" \"link\": {\n",
" \"band\": {\"type\": str, \"enum\": [\"VHF\", \"UHF\"]},\n",
" \"bandwidth_bps\": {\"type\": int, \"enum\": [1200, 2400, 9600, 16000]},\n",
" \"snr_db\": {\"type\": (int, float), \"min\": -10, \"max\": 40},\n",
" \"mtu_bytes\": {\"type\": int, \"min\": 32, \"max\": 1024},\n",
" },\n",
" \"tactical\": {\n",
" \"weather\": {\"type\": str, \"enum\": [\"clear\", \"rain\", \"fog\", \"snow\", \"night\"]},\n",
" \"position_error_m\": {\"type\": (int, float), \"min\": 0, \"max\": 500},\n",
" \"comm_delay_ms\": {\"type\": (int, float), \"min\": 0, \"max\": 30000},\n",
" \"relay_hops\": {\"type\": int, \"min\": 0, \"max\": 5},\n",
" },\n",
"}\n",
"\n",
"\n",
"def check(name, value, rule):\n",
" \"\"\"값 하나를 규칙(type, enum, min, max, pattern)으로 검사해 오류 목록을 돌려준다.\"\"\"\n",
" # [TODO : 학생 실습 1] 협대역 전술 무선망(VHF/UHF) 대역폭 제약 조건 분석 및 전술 메시지 구조 정의 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def validate(rec):\n",
" # [TODO : 학생 실습 1] 협대역 전술 무선망(VHF/UHF) 대역폭 제약 조건 분석 및 전술 메시지 구조 정의 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"rng = np.random.default_rng(1)\n",
"records = [{\"msg_id\": m[\"id\"], \"precedence\": m[\"prec\"], \"msg_type\": m[\"type\"], \"src\": m[\"src\"],\n",
" \"created_s\": round(m[\"t0\"], 1), \"payload_bytes\": m[\"size\"],\n",
" \"link\": {\"band\": str(rng.choice([\"VHF\", \"UHF\"])), \"bandwidth_bps\": 2400, \"snr_db\": float(round(rng.uniform(3, 25), 1)), \"mtu_bytes\": 256},\n",
" \"tactical\": {\"weather\": str(rng.choice([\"clear\", \"rain\", \"fog\", \"snow\", \"night\"])), \"position_error_m\": float(round(rng.uniform(3, 50), 1)),\n",
" \"comm_delay_ms\": int(rng.integers(300, 3000)), \"relay_hops\": int(rng.integers(0, 3))}}\n",
" for m in make_messages(400, 0.12, seed=3)[:20]]\n",
"ok = sum(1 for r in records if not validate(r))\n",
"print(f\"\\n정상 샘플 {len(records)}건 중 검증 통과 {ok}건, 오류 {len(records) - ok}건\")\n",
"\n",
"# 4) 오류 주입 시험\n",
"bad = json.loads(json.dumps(records[0]))\n",
"bad[\"msg_id\"] = \"MSG-1\"\n",
"bad[\"precedence\"] = \"URGENT\"\n",
"bad[\"link\"][\"bandwidth_bps\"] = 3000\n",
"del bad[\"tactical\"][\"comm_delay_ms\"]\n",
"bad[\"tactical\"][\"relay_hops\"] = 9\n",
"print(\"[오류 주입 결과]\")\n",
"for e in validate(bad):\n",
" print(\" -\", e)\n",
"flash_big = {**records[0], \"precedence\": \"FLASH\", \"payload_bytes\": 4000}\n",
"print(\" -\", validate(flash_big)[0])\n",
"\n",
"# 5) 스키마 명세 요약\n",
"rows = []\n",
"for g in (\"fields\", \"link\", \"tactical\"):\n",
" for k, r in SCHEMA[g].items():\n",
" t = r[\"type\"].__name__ if isinstance(r[\"type\"], type) else \"number\"\n",
" rows.append((\"기본\" if g == \"fields\" else g, k, t, str({x: y for x, y in r.items() if x != \"type\"})[:58]))\n",
"print(\"\\n[데이터 명세 요약]\")\n",
"print(pd.DataFrame(rows, columns=[\"그룹\", \"필드\", \"자료형\", \"규칙\"]).to_string(index=False))\n",
"print(\"\\n긴급도별 전달 목표(초):\", DEADLINE)\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 01 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"[대역폭별 전문 1건 전송시간(초) - 가상 링크]\n",
" 유형 평균 크기(B) 1.2kbps 2.4kbps 9.6kbps 16kbps\n",
"CONTACT 210 1.6 0.8 0.2 0.1\n",
"MEDEVAC 250 1.9 0.9 0.2 0.1\n",
" SITREP 1550 11.8 5.9 1.5 0.9\n",
" POSREP 90 0.8 0.4 0.1 0.1\n",
" LOGREQ 750 5.6 2.8 0.7 0.4\n",
" ADMIN 3000 22.6 11.3 2.8 1.7\n",
"-> 2.4kbps에서 행정전문 1건이 약 10초 이상 채널을 점유한다. FLASH 전달 목표(10초)를 지키려면 우선순위 큐와 조각 단위 전송이 필요하다.\n",
"\n",
"정상 샘플 20건 중 검증 통과 20건, 오류 0건\n",
"[오류 주입 결과]\n",
" - msg_id: 형식 불일치 'MSG-1'\n",
" - precedence: 허용되지 않은 값 'URGENT'\n",
" - link.bandwidth_bps: 허용되지 않은 값 '3000'\n",
" - tactical.comm_delay_ms: 필수 필드 누락\n",
" - tactical.relay_hops: 최대값 5 초과\n",
" - 업무 규칙: FLASH 전문은 2,000bytes 이하로 작성\n",
"\n",
"[데이터 명세 요약]\n",
" 그룹 필드 자료형 규칙\n",
" 기본 msg_id str {'pattern': '^M\\\\d{4}$'}\n",
" 기본 precedence str {'enum': ['FLASH', 'IMMEDIATE', 'PRIORITY', 'ROUTINE']}\n",
" 기본 msg_type str {'enum': ['CONTACT', 'MEDEVAC', 'SITREP', 'POSREP', 'LOGRE\n",
" 기본 src str {'pattern': '^U\\\\d{2}$'}\n",
" 기본 created_s number {'min': 0}\n",
" 기본 payload_bytes int {'min': 1, 'max': 8000}\n",
" link band str {'enum': ['VHF', 'UHF']}\n",
" link bandwidth_bps int {'enum': [1200, 2400, 9600, 16000]}\n",
" link snr_db number {'min': -10, 'max': 40}\n",
" link mtu_bytes int {'min': 32, 'max': 1024}\n",
"tactical weather str {'enum': ['clear', 'rain', 'fog', 'snow', 'night']}\n",
"tactical position_error_m number {'min': 0, 'max': 500}\n",
"tactical comm_delay_ms number {'min': 0, 'max': 30000}\n",
"tactical relay_hops int {'min': 0, 'max': 5}\n",
"\n",
"긴급도별 전달 목표(초): {'FLASH': 10, 'IMMEDIATE': 30, 'PRIORITY': 120, 'ROUTINE': 600}\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 02. 전술 메시지(전문) 패킷 파싱, 크기 압축 및 헤더 정보 전처리 모듈 구현\n",
"**소요시간**: 약 2.0시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"전술 메시지 패킷 파싱, 헤더 압축 및 페이로드 데이터 정규화 전처리 모듈을 구현한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **잡음이 섞인 원시 전문 패킷을 파싱하고 긴급도 표기 정규화, CRC 검사, 중복 제거, 결측 시각 보정, 헤더 이진 압축, 본문 압축을 수행하는 전처리 모듈을 구현한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **긴급도 표기 혼재, 재전송으로 인한 중복 수신, 비트 오류, 전송 중 잘림, 미등록 송신국(스푸핑 의심) 패킷을 포함한 가상 원시 패킷 작성하여 원인을 분석하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `src/preprocess/c202_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": [
"\"\"\"가상 전술 전문 패킷(텍스트 헤더+본문)을 파싱하고 긴급도 표기 정규화·CRC 검사·중복 제거·결측 시각 보정·헤더 이진 압축·본문 압축을 수행하는 전처리 모듈을 구현해 처리 속도와 단위 테스트 통과 여부를 확인한다.\"\"\"\n",
"import struct\n",
"import time\n",
"import zlib\n",
"import numpy as np\n",
"import pandas as pd\n",
"\n",
"# ---------------------------------------------------------------------------\n",
"# [교육용가상자료] 가상 전술 전문(메시지) 생성기 (실제 부대·통신망·제원 정보 없음)\n",
"# 긴급도 4단계(FLASH/IMMEDIATE/PRIORITY/ROUTINE)와 전문 유형별 크기를 가진 전문을 만든다.\n",
"# ---------------------------------------------------------------------------\n",
"PREC = [\"FLASH\", \"IMMEDIATE\", \"PRIORITY\", \"ROUTINE\"]\n",
"RANK = {p: i for i, p in enumerate(PREC)}\n",
"DEADLINE = {\"FLASH\": 10, \"IMMEDIATE\": 30, \"PRIORITY\": 120, \"ROUTINE\": 600} # 가상 전달 목표 시간(초)\n",
"MSG_TYPES = { # 유형: (긴급도 후보, 확률, 본문 크기 범위 bytes, 발생 비율)\n",
" \"CONTACT\": ([\"FLASH\", \"IMMEDIATE\"], [0.7, 0.3], (120, 300), 0.08),\n",
" \"MEDEVAC\": ([\"FLASH\", \"IMMEDIATE\"], [0.4, 0.6], (150, 350), 0.05),\n",
" \"SITREP\": ([\"IMMEDIATE\", \"PRIORITY\"], [0.2, 0.8], (600, 2500), 0.20),\n",
" \"POSREP\": ([\"PRIORITY\", \"ROUTINE\"], [0.5, 0.5], (60, 120), 0.35),\n",
" \"LOGREQ\": ([\"PRIORITY\", \"ROUTINE\"], [0.4, 0.6], (300, 1200), 0.14),\n",
" \"ADMIN\": ([\"ROUTINE\"], [1.0], (1000, 5000), 0.18),\n",
"}\n",
"\n",
"\n",
"def make_messages(duration=600, rate=0.12, seed=7, t_start=0.0, prefix=\"M\"):\n",
" \"\"\"duration초 동안 평균 rate건/초로 발생하는 가상 전문 목록.\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" n = int(rng.poisson(rate * duration))\n",
" times = np.sort(rng.uniform(t_start, t_start + duration, n))\n",
" types = list(MSG_TYPES)\n",
" msgs = []\n",
" for i, t0 in enumerate(times):\n",
" ty = types[rng.choice(len(types), p=[MSG_TYPES[k][3] for k in types])]\n",
" cands, p, (lo, hi), _ = MSG_TYPES[ty]\n",
" msgs.append({\"id\": f\"{prefix}{i:04d}\", \"t0\": float(t0), \"type\": ty, \"prec\": str(rng.choice(cands, p=p)),\n",
" \"size\": int(rng.integers(lo, hi + 1)), \"src\": f\"U{int(rng.integers(1, 9)):02d}\"})\n",
" return msgs\n",
"\n",
"\n",
"class Link:\n",
" \"\"\"[교육용가상자료] 가상 협대역 무선 링크(기본 2.4kbps). 초 단위 상태(good/bad/jam)에 따라\n",
" 비트오류율이 바뀌고(Gilbert-Elliott 모델), 구간별 대역폭 감소도 줄 수 있다.\"\"\"\n",
" BER = {\"good\": 1e-5, \"bad\": 2e-4, \"jam\": 1.2e-3}\n",
"\n",
" def __init__(self, horizon=4000, bw_bps=2400, seed=1, p_gb=0.02, p_bg=0.15, jam=(), bw_drop=(), rtt=(1.2, 0.4)):\n",
" rng = np.random.default_rng(seed)\n",
" st, s = [], \"good\"\n",
" for _ in range(int(horizon) + 1):\n",
" s = (\"bad\" if rng.random() < p_gb else \"good\") if s == \"good\" else (\"good\" if rng.random() < p_bg else \"bad\")\n",
" st.append(s)\n",
" for a, b in jam:\n",
" for k in range(int(a), min(int(b), len(st))):\n",
" st[k] = \"jam\"\n",
" self.states, self.bw_bps, self.bw_drop, self.rtt_par = st, bw_bps, list(bw_drop), rtt\n",
"\n",
" def state(self, t):\n",
" return self.states[min(int(t), len(self.states) - 1)]\n",
"\n",
" def bw(self, t):\n",
" f = 1.0\n",
" for a, b, fac in self.bw_drop:\n",
" if a <= t < b:\n",
" f = fac\n",
" return self.bw_bps * f\n",
"\n",
" def p_loss(self, nbytes, t):\n",
" return 1 - (1 - self.BER[self.state(t)]) ** (8 * nbytes)\n",
"\n",
" def rtt(self, rng):\n",
" return self.rtt_par[0] + rng.exponential(self.rtt_par[1])\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"PREC_ALIAS = {\"Z\": \"FLASH\", \"FLASH\": \"FLASH\", \"O\": \"IMMEDIATE\", \"IMMED\": \"IMMEDIATE\", \"IMMEDIATE\": \"IMMEDIATE\",\n",
" \"P\": \"PRIORITY\", \"PRIORITY\": \"PRIORITY\", \"R\": \"ROUTINE\", \"ROUTINE\": \"ROUTINE\"}\n",
"ROSTER = {f\"U{i:02d}\" for i in range(1, 9)} # 가상 망 가입국 목록\n",
"TYPES = list(MSG_TYPES)\n",
"\n",
"\n",
"def make_body(m, rng):\n",
" \"\"\"[교육용가상자료] 'KEY=VALUE;' 형식의 정형 본문(실제 전문 양식 아님).\"\"\"\n",
" lines, k = [], 0\n",
" while sum(map(len, lines)) < m[\"size\"]:\n",
" lines.append(f\"L{k:02d}={m['type']} UNIT {m['src']} ITEM {int(rng.integers(1, 40)):02d} QTY {int(rng.integers(0, 99)):02d} STAT OK;\")\n",
" k += 1\n",
" return \"\".join(lines)[:m[\"size\"]]\n",
"\n",
"\n",
"def make_raw_packets(msgs, seed=5):\n",
" \"\"\"전술 환경 잡음을 넣은 원시 패킷 줄: 긴급도 표기 혼재, 중복 수신, 비트 오류(CRC 불일치), 시각 누락, 잘림, 미등록 송신국.\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" out = []\n",
" for m in msgs:\n",
" body = make_body(m, rng)\n",
" prec = rng.choice([PREC[RANK[m[\"prec\"]]], PREC[RANK[m[\"prec\"]]].lower(), \"ZOPR\"[RANK[m[\"prec\"]]], \" \" + \"ZOPR\"[RANK[m[\"prec\"]]] + \" \"])\n",
" dtg = \"\" if rng.random() < 0.05 else f\"{m['t0']:.1f}\"\n",
" src = m[\"src\"] if rng.random() > 0.02 else \"X77\"\n",
" line = f\"ZCZC|ID={m['id']}|PREC={prec}|TYPE={m['type']}|SRC={src}|DTG={dtg}|LEN={len(body)}|CRC={zlib.crc32(body.encode()):08x}|BODY={body}\"\n",
" r = rng.random()\n",
" if r < 0.03: # 비트 오류: 본문 일부 변형\n",
" line = line[:-5] + \"#\" + line[-4:]\n",
" elif r < 0.05: # 전송 중 잘림\n",
" line = line[:int(len(line) * 0.4)]\n",
" out.append(line)\n",
" if rng.random() < 0.06: # 재전송으로 인한 중복 수신\n",
" out.append(line)\n",
" return out\n",
"\n",
"\n",
"class ParseError(ValueError):\n",
" pass\n",
"\n",
"\n",
"def parse(line):\n",
" # [TODO : 학생 실습 2] 전술 메시지(전문) 패킷 파싱, 크기 압축 및 헤더 정보 전처리 모듈 구현 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def pack_header(r):\n",
" \"\"\"헤더 이진 압축: ID(2) + 긴급도·유형·송신국(2) + 시각(4, 0.1초) + 길이(2) + CRC(4) = 14bytes.\"\"\"\n",
" # [TODO : 학생 실습 2] 전술 메시지(전문) 패킷 파싱, 크기 압축 및 헤더 정보 전처리 모듈 구현 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def unpack_header(b):\n",
" i, flags, dtg, ln, crc = struct.unpack(\">HHIHI\", b)\n",
" return {\"id\": f\"M{i:04d}\", \"prec\": PREC[flags >> 12], \"type\": TYPES[(flags >> 8) & 0xF], \"src\": f\"U{flags & 0xFF:02d}\", \"dtg\": dtg / 10, \"len\": ln}\n",
"\n",
"\n",
"ZLEVEL = {1200: 9, 2400: 9, 9600: 6, 16000: 1} # 대역폭이 좁을수록 압축을 강하게\n",
"\n",
"\n",
"def preprocess(lines, bw_bps=2400):\n",
" good, quarantine, seen = [], [], set()\n",
" for ln in lines:\n",
" try:\n",
" r = parse(ln)\n",
" except (ParseError, ValueError) as e:\n",
" quarantine.append(str(e)); continue\n",
" if r[\"id\"] in seen:\n",
" quarantine.append(\"중복 수신(제거)\"); continue\n",
" seen.add(r[\"id\"]); good.append(r)\n",
" good.sort(key=lambda r: int(r[\"id\"][1:]))\n",
" dtg = pd.Series([r[\"dtg\"] for r in good], dtype=float)\n",
" miss = int(dtg.isna().sum())\n",
" filled = dtg.interpolate(limit_direction=\"both\") # 앞뒤 전문 시각으로 선형 보간\n",
" for r, v in zip(good, filled):\n",
" r[\"dtg_filled\"] = bool(np.isnan(r[\"dtg\"])); r[\"dtg\"] = float(v)\n",
" r[\"hdr_packed\"] = pack_header(r)\n",
" r[\"body_z\"] = zlib.compress(r[\"body\"].encode(), ZLEVEL[bw_bps])\n",
" return good, quarantine, miss\n",
"\n",
"\n",
"msgs = make_messages(1200, 0.21, seed=7)\n",
"lines = make_raw_packets(msgs)\n",
"t0 = time.perf_counter()\n",
"good, quarantine, miss = preprocess(lines)\n",
"dt = time.perf_counter() - t0\n",
"print(f\"원시 패킷 {len(lines)}줄 -> 정상 {len(good)}건, 격리 {len(quarantine)}건 / 처리 속도 {len(lines) / dt:,.0f}건/초 (목표 1,000건/초 이상)\")\n",
"print(\"처리 중 예외로 멈춘 건수 0건(오류율 0%) - 비정상 패킷은 사유와 함께 격리\")\n",
"print(\"격리 사유:\", pd.Series(quarantine).value_counts().to_dict())\n",
"print(f\"DTG 결측 {miss}건 중 보정 {sum(r['dtg_filled'] for r in good)}건 (보정율 {sum(r['dtg_filled'] for r in good) / max(miss, 1):.0%})\")\n",
"\n",
"# 전처리 전후 크기 비교(전문 유형별 평균 bytes)\n",
"df = pd.DataFrame([{\"유형\": r[\"type\"], \"원본 헤더\": r[\"hdr_bytes\"], \"압축 헤더\": len(r[\"hdr_packed\"]),\n",
" \"원본 본문\": len(r[\"body\"]), \"압축 본문\": len(r[\"body_z\"])} for r in good])\n",
"tab = df.groupby(\"유형\").mean().round(0).astype(int)\n",
"tab[\"전체 절감률\"] = (1 - (tab[\"압축 헤더\"] + tab[\"압축 본문\"]) / (tab[\"원본 헤더\"] + tab[\"원본 본문\"])).round(3)\n",
"print(\"\\n[전처리 전후 평균 크기(bytes)]\")\n",
"print(tab.to_string())\n",
"tot0, tot1 = (df[\"원본 헤더\"] + df[\"원본 본문\"]).sum(), (df[\"압축 헤더\"] + df[\"압축 본문\"]).sum()\n",
"print(f\"전체 {tot0:,}B -> {tot1:,}B, 2.4kbps 기준 전송시간 {tot0 * 8 / 2400 / 60:.1f}분 -> {tot1 * 8 / 2400 / 60:.1f}분\")\n",
"print(\"상황별 압축 수준(대역폭 bps: zlib 수준):\", ZLEVEL)\n",
"print(\"주의: 가상 본문은 정형 반복 문구라 압축률이 높게 나온다. 자유 서술 본문은 절감률이 이보다 낮다.\")\n",
"\n",
"# 단위 테스트 5종(+1): 조건을 알고 있는 시험용 패킷으로 확인\n",
"def mk(prec=\"Z\", src=\"U01\", body=\"L00=TEST UNIT U01;\", crc=None):\n",
" return f\"ZCZC|ID=M9999|PREC={prec}|TYPE=CONTACT|SRC={src}|DTG=1.0|LEN={len(body)}|CRC={crc or f'{zlib.crc32(body.encode()):08x}'}|BODY={body}\"\n",
"\n",
"\n",
"def raises(line):\n",
" try:\n",
" parse(line)\n",
" return False\n",
" except ParseError:\n",
" return True\n",
"\n",
"\n",
"r0 = good[0]\n",
"tests = {\n",
" \"정상 패킷 파싱\": parse(mk())[\"prec\"] == \"FLASH\",\n",
" \"긴급도 표기 정규화(' o ' -> IMMEDIATE)\": parse(mk(prec=\" o \"))[\"prec\"] == \"IMMEDIATE\",\n",
" \"CRC 불일치 검출\": raises(mk(crc=\"00000000\")),\n",
" \"미등록 송신국 격리\": raises(mk(src=\"X77\")),\n",
" \"중복 수신 제거\": len(preprocess([mk(), mk()])[0]) == 1 and len({r[\"id\"] for r in good}) == len(good),\n",
" \"결측 시각 보간 + 헤더 압축 왕복 일치\": not any(np.isnan(r[\"dtg\"]) for r in good) and unpack_header(r0[\"hdr_packed\"])[\"id\"] == r0[\"id\"],\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"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 02 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"원시 패킷 278줄 -> 정상 241건, 격리 37건 / 처리 속도 19,781건/초 (목표 1,000건/초 이상)\n",
"처리 중 예외로 멈춘 건수 0건(오류율 0%) - 비정상 패킷은 사유와 함께 격리\n",
"격리 사유: {'중복 수신(제거)': 18, 'CRC/길이 불일치(비트 오류)': 11, '미등록 송신국(스푸핑 의심)': 5, '필드 누락(BODY)': 3}\n",
"DTG 결측 15건 중 보정 15건 (보정율 100%)\n",
"\n",
"[전처리 전후 평균 크기(bytes)]\n",
" 원본 헤더 압축 헤더 원본 본문 압축 본문 전체 절감률\n",
"유형 \n",
"ADMIN 80 14 3506 555 0.841\n",
"CONTACT 81 14 227 93 0.653\n",
"LOGREQ 81 14 760 178 0.772\n",
"MEDEVAC 80 14 206 91 0.633\n",
"POSREP 80 14 89 63 0.544\n",
"SITREP 82 14 1666 303 0.819\n",
"전체 276,195B -> 54,134B, 2.4kbps 기준 전송시간 15.3분 -> 3.0분\n",
"상황별 압축 수준(대역폭 bps: zlib 수준): {1200: 9, 2400: 9, 9600: 6, 16000: 1}\n",
"주의: 가상 본문은 정형 반복 문구라 압축률이 높게 나온다. 자유 서술 본문은 절감률이 이보다 낮다.\n",
"[PASS] 정상 패킷 파싱\n",
"[PASS] 긴급도 표기 정규화(' o ' -> IMMEDIATE)\n",
"[PASS] CRC 불일치 검출\n",
"[PASS] 미등록 송신국 격리\n",
"[PASS] 중복 수신 제거\n",
"[PASS] 결측 시각 보간 + 헤더 압축 왕복 일치\n",
"단위 테스트 6/6 통과\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 03. 메시지 긴급도(Flash/Immediate/Priority/Routine) 기반 1차 우선순위 큐(Queue) 구현\n",
"**소요시간**: 약 2.5시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"전문 긴급도(Flash/Immediate/Priority/Routine) 기반 1차 메시지 우선순위 큐(Queue)를 구현한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **전문 긴급도(FLASH/IMMEDIATE/PRIORITY/ROUTINE)에 따라 먼저 보낼 전문을 고르는 1차 우선순위 큐를 구현하고 도착순 전송과 성능을 비교한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **heapq로 긴급도 우선·같은 긴급도는 도착순을 지키는 큐를 만들고 동작 확인하고 유효성을 점검하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `src/models/c202_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": [
"\"\"\"전문 긴급도(FLASH/IMMEDIATE/PRIORITY/ROUTINE) 기반 1차 우선순위 큐(heapq)를 구현하고, 가상 협대역 링크 전송 시뮬레이션으로 도착순(FIFO) 전송과 긴급도별 기한 내 전달률(처리 정확도)을 비교한다.\"\"\"\n",
"import heapq\n",
"import itertools\n",
"import time\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"\n",
"# ---------------------------------------------------------------------------\n",
"# [교육용가상자료] 가상 전술 전문(메시지) 생성기 (실제 부대·통신망·제원 정보 없음)\n",
"# 긴급도 4단계(FLASH/IMMEDIATE/PRIORITY/ROUTINE)와 전문 유형별 크기를 가진 전문을 만든다.\n",
"# ---------------------------------------------------------------------------\n",
"PREC = [\"FLASH\", \"IMMEDIATE\", \"PRIORITY\", \"ROUTINE\"]\n",
"RANK = {p: i for i, p in enumerate(PREC)}\n",
"DEADLINE = {\"FLASH\": 10, \"IMMEDIATE\": 30, \"PRIORITY\": 120, \"ROUTINE\": 600} # 가상 전달 목표 시간(초)\n",
"MSG_TYPES = { # 유형: (긴급도 후보, 확률, 본문 크기 범위 bytes, 발생 비율)\n",
" \"CONTACT\": ([\"FLASH\", \"IMMEDIATE\"], [0.7, 0.3], (120, 300), 0.08),\n",
" \"MEDEVAC\": ([\"FLASH\", \"IMMEDIATE\"], [0.4, 0.6], (150, 350), 0.05),\n",
" \"SITREP\": ([\"IMMEDIATE\", \"PRIORITY\"], [0.2, 0.8], (600, 2500), 0.20),\n",
" \"POSREP\": ([\"PRIORITY\", \"ROUTINE\"], [0.5, 0.5], (60, 120), 0.35),\n",
" \"LOGREQ\": ([\"PRIORITY\", \"ROUTINE\"], [0.4, 0.6], (300, 1200), 0.14),\n",
" \"ADMIN\": ([\"ROUTINE\"], [1.0], (1000, 5000), 0.18),\n",
"}\n",
"\n",
"\n",
"def make_messages(duration=600, rate=0.12, seed=7, t_start=0.0, prefix=\"M\"):\n",
" \"\"\"duration초 동안 평균 rate건/초로 발생하는 가상 전문 목록.\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" n = int(rng.poisson(rate * duration))\n",
" times = np.sort(rng.uniform(t_start, t_start + duration, n))\n",
" types = list(MSG_TYPES)\n",
" msgs = []\n",
" for i, t0 in enumerate(times):\n",
" ty = types[rng.choice(len(types), p=[MSG_TYPES[k][3] for k in types])]\n",
" cands, p, (lo, hi), _ = MSG_TYPES[ty]\n",
" msgs.append({\"id\": f\"{prefix}{i:04d}\", \"t0\": float(t0), \"type\": ty, \"prec\": str(rng.choice(cands, p=p)),\n",
" \"size\": int(rng.integers(lo, hi + 1)), \"src\": f\"U{int(rng.integers(1, 9)):02d}\"})\n",
" return msgs\n",
"\n",
"\n",
"class Link:\n",
" \"\"\"[교육용가상자료] 가상 협대역 무선 링크(기본 2.4kbps). 초 단위 상태(good/bad/jam)에 따라\n",
" 비트오류율이 바뀌고(Gilbert-Elliott 모델), 구간별 대역폭 감소도 줄 수 있다.\"\"\"\n",
" BER = {\"good\": 1e-5, \"bad\": 2e-4, \"jam\": 1.2e-3}\n",
"\n",
" def __init__(self, horizon=4000, bw_bps=2400, seed=1, p_gb=0.02, p_bg=0.15, jam=(), bw_drop=(), rtt=(1.2, 0.4)):\n",
" rng = np.random.default_rng(seed)\n",
" st, s = [], \"good\"\n",
" for _ in range(int(horizon) + 1):\n",
" s = (\"bad\" if rng.random() < p_gb else \"good\") if s == \"good\" else (\"good\" if rng.random() < p_bg else \"bad\")\n",
" st.append(s)\n",
" for a, b in jam:\n",
" for k in range(int(a), min(int(b), len(st))):\n",
" st[k] = \"jam\"\n",
" self.states, self.bw_bps, self.bw_drop, self.rtt_par = st, bw_bps, list(bw_drop), rtt\n",
"\n",
" def state(self, t):\n",
" return self.states[min(int(t), len(self.states) - 1)]\n",
"\n",
" def bw(self, t):\n",
" f = 1.0\n",
" for a, b, fac in self.bw_drop:\n",
" if a <= t < b:\n",
" f = fac\n",
" return self.bw_bps * f\n",
"\n",
" def p_loss(self, nbytes, t):\n",
" return 1 - (1 - self.BER[self.state(t)]) ** (8 * nbytes)\n",
"\n",
" def rtt(self, rng):\n",
" return self.rtt_par[0] + rng.exponential(self.rtt_par[1])\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# ---- 우선순위 큐·재전송 이산 사건 시뮬레이터 공통 함수 ----\n",
"HDR = 32 # 전문 조각(패킷) 헤더 bytes (미션 2의 헤더 압축 적용 시 12 bytes)\n",
"\n",
"\n",
"def policy_key(m, t, policy, aging):\n",
" # [TODO : 학생 실습 3] 메시지 긴급도(Flash/Immediate/Priority/Routine) 기반 1차 우선순위 큐(Queue) 구현 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def simulate(msgs, link, policy=\"strict\", mtu=256, preempt=True, rto_mode=\"fixed\", rto_s=4.0, aging=0.0,\n",
" redundancy=False, adapt_mtu=False, hdr=HDR, seed=0, until=None, admit=None):\n",
" \"\"\"전문 목록을 링크로 보내는 이산 사건 시뮬레이션.\n",
" - 전문은 mtu 크기 조각으로 나눠 보낸다. 조각이 손실되면 RTO 후 재전송한다(자료는 버리지 않음).\n",
" - preempt=False면 한 전문의 조각을 끝까지 연속 전송한다(큰 전문이 긴급 전문을 막는 현상 재현).\n",
" - rto_mode='adaptive'면 왕복시간(RTT) 추정값으로 RTO를 계산한다(Jacobson 방식).\n",
" - redundancy=True면 링크 열화 추정 시 FLASH·IMMEDIATE 조각을 2회 연속 전송한다.\n",
" - adapt_mtu=True면 링크 열화 추정 시 조각 크기를 줄인다.\n",
" - admit(m, t)이 False인 전문은 새 조각을 보내지 않고 저장해 둔다(성능 저하 모드).\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" pend = sorted(({**m, \"unsent\": m[\"size\"], \"ready\": [], \"frags\": [], \"done\": None, \"first\": None} for m in msgs), key=lambda m: m[\"t0\"])\n",
" t, i, active, cur = 0.0, 0, [], None\n",
" srtt, rttvar, loss = None, None, 0.0\n",
" st = {\"tx\": 0, \"retx\": 0, \"spurious\": 0, \"bytes\": 0, \"decide_ms\": [], \"flag\": {}, \"log\": []}\n",
" until = until if until is not None else (pend[-1][\"t0\"] if pend else 0) + 6000\n",
" while (i < len(pend) or active) and t < until:\n",
" while i < len(pend) and pend[i][\"t0\"] <= t:\n",
" active.append(pend[i]); i += 1\n",
" td = time.perf_counter()\n",
" for m in active: # ACK가 이미 도착한 재전송 예약은 취소\n",
" m[\"ready\"] = [e for e in m[\"ready\"] if not (e[1][\"ack\"] is not None and e[1][\"ack\"] <= t)]\n",
" ready = [m for m in active if any(r <= t for r, _ in m[\"ready\"]) or (m[\"unsent\"] > 0 and (admit is None or admit(m, t)))]\n",
" if not ready:\n",
" nxt = ([pend[i][\"t0\"]] if i < len(pend) else []) + [r for m in active for r, _ in m[\"ready\"] if r > t]\n",
" t = min(nxt) if nxt else t + 1.0\n",
" continue\n",
" if not preempt and cur in ready and cur[\"unsent\"] > 0:\n",
" m = cur\n",
" else:\n",
" m = min(ready, key=lambda x: policy_key(x, t, policy, aging))\n",
" st[\"decide_ms\"].append((time.perf_counter() - td) * 1000)\n",
" due = [e for e in m[\"ready\"] if e[0] <= t]\n",
" if due:\n",
" e = min(due, key=lambda x: x[0]); m[\"ready\"].remove(e); frag = e[1]\n",
" st[\"retx\"] += 1; st[\"spurious\"] += frag[\"recv\"]\n",
" else:\n",
" size = mtu if not adapt_mtu else (mtu if loss < 0.15 else (mtu // 2 if loss < 0.4 else mtu // 4))\n",
" size = min(m[\"unsent\"], size); m[\"unsent\"] -= size\n",
" frag = {\"b\": size, \"recv\": False, \"ack\": None, \"tries\": 0}\n",
" m[\"frags\"].append(frag)\n",
" cur = m\n",
" m[\"first\"] = t if m[\"first\"] is None else m[\"first\"]\n",
" nb = frag[\"b\"] + hdr\n",
" copies = 2 if (redundancy and RANK[m[\"prec\"]] <= 1 and loss > 0.15) else 1\n",
" for _ in range(copies):\n",
" t += nb * 8 / link.bw(t)\n",
" st[\"tx\"] += 1; st[\"bytes\"] += nb\n",
" ok = rng.random() >= link.p_loss(nb, t)\n",
" st[\"log\"].append((t, nb, ok, m[\"prec\"], m[\"id\"])) # 전송 기록(시각, bytes, 성공, 긴급도, 전문)\n",
" loss = 0.85 * loss + 0.15 * (0.0 if ok else 1.0) # 최근 손실률 추정(EWMA)\n",
" st[\"flag\"][int(t)] = st[\"flag\"].get(int(t), False) or loss > 0.12 # 링크 열화 판정(초 단위)\n",
" if ok and not frag[\"recv\"]:\n",
" frag[\"recv\"], frag[\"rt\"] = True, t\n",
" r = link.rtt(rng); frag[\"ack\"] = t + r\n",
" if srtt is None:\n",
" srtt, rttvar = r, r / 2\n",
" else:\n",
" rttvar = 0.75 * rttvar + 0.25 * abs(srtt - r); srtt = 0.875 * srtt + 0.125 * r\n",
" frag[\"tries\"] += 1\n",
" rto = rto_s if (rto_mode == \"fixed\" or srtt is None) else float(np.clip(srtt + 4 * rttvar, 1.0, 30.0))\n",
" if not frag[\"recv\"] or frag[\"ack\"] > t + rto: # 손실 또는 ACK 지연 -> 재전송 예약\n",
" m[\"ready\"].append((t + rto, frag))\n",
" if m[\"done\"] is None and m[\"unsent\"] == 0 and all(f[\"recv\"] for f in m[\"frags\"]):\n",
" m[\"done\"] = max(f[\"rt\"] for f in m[\"frags\"])\n",
" active = [x for x in active if not (x[\"unsent\"] == 0 and not x[\"ready\"] and all(f[\"recv\"] for f in x[\"frags\"]))]\n",
" res = [{\"id\": m[\"id\"], \"type\": m[\"type\"], \"prec\": m[\"prec\"], \"size\": m[\"size\"], \"t0\": m[\"t0\"], \"done\": m[\"done\"], \"wait\": None if m[\"first\"] is None else m[\"first\"] - m[\"t0\"],\n",
" \"retx\": sum(f[\"tries\"] for f in m[\"frags\"]) - len(m[\"frags\"]),\n",
" \"latency\": None if m[\"done\"] is None else m[\"done\"] - m[\"t0\"]} for m in pend]\n",
" for r in res:\n",
" r[\"on_time\"] = r[\"latency\"] is not None and r[\"latency\"] <= DEADLINE[r[\"prec\"]]\n",
" return res, st\n",
"\n",
"\n",
"def summarize(res):\n",
" \"\"\"긴급도별 기한 내 전달률·지연. 종합 정확도 = 긴급도별 기한 내 전달률의 평균.\"\"\"\n",
" # [TODO : 학생 실습 3] 메시지 긴급도(Flash/Immediate/Priority/Routine) 기반 1차 우선순위 큐(Queue) 구현 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"class PrecedenceQueue:\n",
" \"\"\"긴급도 순위가 높은 전문을 먼저 꺼내고, 같은 긴급도는 도착 순서(FIFO)를 지키는 1차 우선순위 큐.\"\"\"\n",
"\n",
" def __init__(self):\n",
" self.h, self.seq = [], itertools.count()\n",
"\n",
" def push(self, m):\n",
" heapq.heappush(self.h, (RANK[m[\"prec\"]], m[\"t0\"], next(self.seq), m))\n",
"\n",
" def pop(self):\n",
" return heapq.heappop(self.h)[-1]\n",
"\n",
" def __len__(self):\n",
" return len(self.h)\n",
"\n",
"\n",
"# 1) 큐 동작 확인: 뒤섞여 도착한 전문이 긴급도 순서로 나오는가\n",
"demo = [{\"id\": f\"D{i}\", \"prec\": p, \"t0\": float(i)} for i, p in enumerate([\"ROUTINE\", \"PRIORITY\", \"FLASH\", \"ROUTINE\", \"IMMEDIATE\", \"FLASH\"])]\n",
"q = PrecedenceQueue()\n",
"for m in demo:\n",
" q.push(m)\n",
"out = [q.pop() for _ in range(len(q))]\n",
"print(\"도착 순서:\", [f\"{m['id']}:{m['prec'][0]}\" for m in demo])\n",
"print(\"꺼낸 순서:\", [f\"{m['id']}:{m['prec'][0]}\" for m in out])\n",
"print(\"같은 긴급도 FIFO 유지:\", [m[\"id\"] for m in out if m[\"prec\"] == \"FLASH\"] == [\"D2\", \"D5\"])\n",
"\n",
"# 2) [교육용가상자료] 20분간 전문 약 250건, 2.4kbps 링크에서 전송 시뮬레이션\n",
"msgs = make_messages(1200, 0.21, seed=7)\n",
"link = Link(8000, seed=1)\n",
"t0 = time.perf_counter()\n",
"fifo, _ = simulate(msgs, link, \"fifo\")\n",
"pq, st = simulate(msgs, link, \"strict\", preempt=False) # 1차 큐: 전문 단위로 꺼내 끝까지 전송\n",
"elapsed = time.perf_counter() - t0\n",
"tf, acc_f = summarize(fifo)\n",
"tp, acc_p = summarize(pq)\n",
"tab = tp[[\"긴급도\", \"건수\"]].copy()\n",
"tab[\"FIFO 기한내\"] = tf[\"기한내 전달률\"]; tab[\"FIFO 95%지연\"] = tf[\"95% 지연(s)\"]\n",
"tab[\"우선순위큐 기한내\"] = tp[\"기한내 전달률\"]; tab[\"우선순위큐 95%지연\"] = tp[\"95% 지연(s)\"]\n",
"print(f\"\\n전문 {len(msgs)}건, 조각 전송 {st['tx']}회(재전송 {st['retx']}회), 시뮬레이션 2회 {elapsed:.2f}초\")\n",
"print(tab.to_string(index=False))\n",
"all_on = np.mean([r[\"on_time\"] for r in pq])\n",
"print(f\"\\n처리 정확도(전체 기한 내 전달률) FIFO {np.mean([r['on_time'] for r in fifo]):.1%} -> 우선순위 큐 {all_on:.1%} (목표 75% 이상: {'달성' if all_on >= 0.75 else '미달'})\")\n",
"print(f\"종합 정확도(긴급도별 전달률 평균) FIFO {acc_f:.3f} -> 우선순위 큐 {acc_p:.3f}\")\n",
"\n",
"# 3) 한계 분석: 기한을 놓친 FLASH 전문은 왜 늦었나\n",
"late = [r for r in pq if r[\"prec\"] == \"FLASH\" and not r[\"on_time\"]]\n",
"for r in late:\n",
" blocker = [m for m in msgs if m[\"t0\"] < r[\"t0\"] and m[\"size\"] >= 1000 and r[\"t0\"] - m[\"t0\"] < 30]\n",
" print(f\"FLASH {r['id']} 지연 {r['latency']:.1f}s - 직전 30초 내 도착한 1,000B 이상 대용량 전문 {len(blocker)}건이 채널 점유(전문 단위 비선점)\")\n",
"print(\"-> 전문을 통째로 보내는 1차 큐는 대용량 전문이 채널을 점유하면 FLASH도 기다린다. 미션 4에서 조각 단위 선점과 재전송 개선으로 보완한다.\")\n",
"\n",
"# 4) 링크 속도별 민감도\n",
"print(\"\\n링크(bps) FIFO 전체 우선순위큐 전체 우선순위큐 FLASH\")\n",
"for bw in [1200, 2400, 4800]:\n",
" lk = Link(20000, bw_bps=bw, seed=1)\n",
" a, _ = simulate(msgs, lk, \"fifo\"); b, _ = simulate(msgs, lk, \"strict\", preempt=False)\n",
" fl = [r[\"on_time\"] for r in b if r[\"prec\"] == \"FLASH\"]\n",
" print(f\"{bw:>8} {np.mean([r['on_time'] for r in a]):8.1%} {np.mean([r['on_time'] for r in b]):12.1%} {np.mean(fl):12.1%}\")\n",
"\n",
"plt.figure(figsize=(8, 4))\n",
"data = [[r[\"latency\"] for r in res if r[\"prec\"] == p] for res in (fifo, pq) for p in PREC]\n",
"plt.boxplot(data, showfliers=False)\n",
"plt.xticks(range(1, 9), [f\"{n}\\n{p[:5]}\" for n in (\"FIFO\", \"PQ\") for p in PREC])\n",
"plt.yscale(\"log\"); plt.ylabel(\"delivery latency (s, log)\"); plt.title(\"FIFO vs precedence queue (synthetic 2.4 kbps link)\")\n",
"plt.tight_layout(); plt.show()\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 03 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"도착 순서: ['D0:R', 'D1:P', 'D2:F', 'D3:R', 'D4:I', 'D5:F']\n",
"꺼낸 순서: ['D2:F', 'D5:F', 'D4:I', 'D1:P', 'D0:R', 'D3:R']\n",
"같은 긴급도 FIFO 유지: True\n",
"\n",
"전문 258건, 조각 전송 1285회(재전송 71회), 시뮬레이션 2회 0.03초\n",
" 긴급도 건수 FIFO 기한내 FIFO 95%지연 우선순위큐 기한내 우선순위큐 95%지연\n",
" FLASH 25 0.200 48.4 0.76 14.4\n",
"IMMEDIATE 27 0.741 45.8 1.00 17.8\n",
" PRIORITY 99 1.000 53.2 1.00 25.7\n",
" ROUTINE 107 1.000 57.7 1.00 82.3\n",
"\n",
"처리 정확도(전체 기한 내 전달률) FIFO 89.5% -> 우선순위 큐 97.7% (목표 75% 이상: 달성)\n",
"종합 정확도(긴급도별 전달률 평균) FIFO 0.735 -> 우선순위 큐 0.940\n",
"FLASH M0010 지연 17.4s - 직전 30초 내 도착한 1,000B 이상 대용량 전문 2건이 채널 점유(전문 단위 비선점)\n",
"FLASH M0041 지연 13.1s - 직전 30초 내 도착한 1,000B 이상 대용량 전문 3건이 채널 점유(전문 단위 비선점)\n",
"FLASH M0105 지연 13.2s - 직전 30초 내 도착한 1,000B 이상 대용량 전문 2건이 채널 점유(전문 단위 비선점)\n",
"FLASH M0226 지연 14.7s - 직전 30초 내 도착한 1,000B 이상 대용량 전문 3건이 채널 점유(전문 단위 비선점)\n",
"FLASH M0228 지연 13.1s - 직전 30초 내 도착한 1,000B 이상 대용량 전문 3건이 채널 점유(전문 단위 비선점)\n",
"FLASH M0233 지연 13.2s - 직전 30초 내 도착한 1,000B 이상 대용량 전문 3건이 채널 점유(전문 단위 비선점)\n",
"-> 전문을 통째로 보내는 1차 큐는 대용량 전문이 채널을 점유하면 FLASH도 기다린다. 미션 4에서 조각 단위 선점과 재전송 개선으로 보완한다.\n",
"\n",
"링크(bps) FIFO 전체 우선순위큐 전체 우선순위큐 FLASH\n",
" 1200 34.9% 58.9% 44.0%\n",
" 2400 89.5% 97.7% 76.0%\n",
" 4800 100.0% 100.0% 100.0%\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 04. 전장 상황 및 전송 실패율에 따른 동적 패킷 스케줄링 및 지능형 재전송 알고리즘 구현\n",
"**소요시간**: 약 3.0시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"동적 채널 상태 및 패킷 손실률에 따라 최적 재전송 간격을 조정하는 지능형 스케줄링 알고리즘을 고도화한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **채널 손실률과 왕복시간을 추정해 전송 순서와 재전송 간격을 조정하는 지능형 스케줄링·재전송 알고리즘으로 고도화한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **조각 단위 선점, 대기시간 가중(aging), 왕복시간 기반 적응형 재전송 타이머, 링크 열화 시 긴급 전문 중복 전송과 조각 크기 축소 구현하고 동작을 검증하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `src/models/c202_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": [
"\"\"\"채널 상태(손실률·왕복시간)를 추정해 조각 단위 선점, 대기시간 가중(aging), 적응형 재전송 타이머(RTO), 긴급 전문 중복 전송, 조각 크기 조정을 수행하는 지능형 스케줄러를 구현하고, 복합 장애 시나리오 10종에서 긴급 전문 전달률·링크 열화 탐지 재현율·스케줄링 결정 지연을 검증한다.\"\"\"\n",
"import time\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"\n",
"# ---------------------------------------------------------------------------\n",
"# [교육용가상자료] 가상 전술 전문(메시지) 생성기 (실제 부대·통신망·제원 정보 없음)\n",
"# 긴급도 4단계(FLASH/IMMEDIATE/PRIORITY/ROUTINE)와 전문 유형별 크기를 가진 전문을 만든다.\n",
"# ---------------------------------------------------------------------------\n",
"PREC = [\"FLASH\", \"IMMEDIATE\", \"PRIORITY\", \"ROUTINE\"]\n",
"RANK = {p: i for i, p in enumerate(PREC)}\n",
"DEADLINE = {\"FLASH\": 10, \"IMMEDIATE\": 30, \"PRIORITY\": 120, \"ROUTINE\": 600} # 가상 전달 목표 시간(초)\n",
"MSG_TYPES = { # 유형: (긴급도 후보, 확률, 본문 크기 범위 bytes, 발생 비율)\n",
" \"CONTACT\": ([\"FLASH\", \"IMMEDIATE\"], [0.7, 0.3], (120, 300), 0.08),\n",
" \"MEDEVAC\": ([\"FLASH\", \"IMMEDIATE\"], [0.4, 0.6], (150, 350), 0.05),\n",
" \"SITREP\": ([\"IMMEDIATE\", \"PRIORITY\"], [0.2, 0.8], (600, 2500), 0.20),\n",
" \"POSREP\": ([\"PRIORITY\", \"ROUTINE\"], [0.5, 0.5], (60, 120), 0.35),\n",
" \"LOGREQ\": ([\"PRIORITY\", \"ROUTINE\"], [0.4, 0.6], (300, 1200), 0.14),\n",
" \"ADMIN\": ([\"ROUTINE\"], [1.0], (1000, 5000), 0.18),\n",
"}\n",
"\n",
"\n",
"def make_messages(duration=600, rate=0.12, seed=7, t_start=0.0, prefix=\"M\"):\n",
" \"\"\"duration초 동안 평균 rate건/초로 발생하는 가상 전문 목록.\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" n = int(rng.poisson(rate * duration))\n",
" times = np.sort(rng.uniform(t_start, t_start + duration, n))\n",
" types = list(MSG_TYPES)\n",
" msgs = []\n",
" for i, t0 in enumerate(times):\n",
" ty = types[rng.choice(len(types), p=[MSG_TYPES[k][3] for k in types])]\n",
" cands, p, (lo, hi), _ = MSG_TYPES[ty]\n",
" msgs.append({\"id\": f\"{prefix}{i:04d}\", \"t0\": float(t0), \"type\": ty, \"prec\": str(rng.choice(cands, p=p)),\n",
" \"size\": int(rng.integers(lo, hi + 1)), \"src\": f\"U{int(rng.integers(1, 9)):02d}\"})\n",
" return msgs\n",
"\n",
"\n",
"class Link:\n",
" \"\"\"[교육용가상자료] 가상 협대역 무선 링크(기본 2.4kbps). 초 단위 상태(good/bad/jam)에 따라\n",
" 비트오류율이 바뀌고(Gilbert-Elliott 모델), 구간별 대역폭 감소도 줄 수 있다.\"\"\"\n",
" BER = {\"good\": 1e-5, \"bad\": 2e-4, \"jam\": 1.2e-3}\n",
"\n",
" def __init__(self, horizon=4000, bw_bps=2400, seed=1, p_gb=0.02, p_bg=0.15, jam=(), bw_drop=(), rtt=(1.2, 0.4)):\n",
" rng = np.random.default_rng(seed)\n",
" st, s = [], \"good\"\n",
" for _ in range(int(horizon) + 1):\n",
" s = (\"bad\" if rng.random() < p_gb else \"good\") if s == \"good\" else (\"good\" if rng.random() < p_bg else \"bad\")\n",
" st.append(s)\n",
" for a, b in jam:\n",
" for k in range(int(a), min(int(b), len(st))):\n",
" st[k] = \"jam\"\n",
" self.states, self.bw_bps, self.bw_drop, self.rtt_par = st, bw_bps, list(bw_drop), rtt\n",
"\n",
" def state(self, t):\n",
" return self.states[min(int(t), len(self.states) - 1)]\n",
"\n",
" def bw(self, t):\n",
" f = 1.0\n",
" for a, b, fac in self.bw_drop:\n",
" if a <= t < b:\n",
" f = fac\n",
" return self.bw_bps * f\n",
"\n",
" def p_loss(self, nbytes, t):\n",
" return 1 - (1 - self.BER[self.state(t)]) ** (8 * nbytes)\n",
"\n",
" def rtt(self, rng):\n",
" return self.rtt_par[0] + rng.exponential(self.rtt_par[1])\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# ---- 우선순위 큐·재전송 이산 사건 시뮬레이터 공통 함수 ----\n",
"HDR = 32 # 전문 조각(패킷) 헤더 bytes (미션 2의 헤더 압축 적용 시 12 bytes)\n",
"\n",
"\n",
"def policy_key(m, t, policy, aging):\n",
" # [TODO : 학생 실습 4] 전장 상황 및 전송 실패율에 따른 동적 패킷 스케줄링 및 지능형 재전송 알고리즘 구현 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def simulate(msgs, link, policy=\"strict\", mtu=256, preempt=True, rto_mode=\"fixed\", rto_s=4.0, aging=0.0,\n",
" redundancy=False, adapt_mtu=False, hdr=HDR, seed=0, until=None, admit=None):\n",
" \"\"\"전문 목록을 링크로 보내는 이산 사건 시뮬레이션.\n",
" - 전문은 mtu 크기 조각으로 나눠 보낸다. 조각이 손실되면 RTO 후 재전송한다(자료는 버리지 않음).\n",
" - preempt=False면 한 전문의 조각을 끝까지 연속 전송한다(큰 전문이 긴급 전문을 막는 현상 재현).\n",
" - rto_mode='adaptive'면 왕복시간(RTT) 추정값으로 RTO를 계산한다(Jacobson 방식).\n",
" - redundancy=True면 링크 열화 추정 시 FLASH·IMMEDIATE 조각을 2회 연속 전송한다.\n",
" - adapt_mtu=True면 링크 열화 추정 시 조각 크기를 줄인다.\n",
" - admit(m, t)이 False인 전문은 새 조각을 보내지 않고 저장해 둔다(성능 저하 모드).\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" pend = sorted(({**m, \"unsent\": m[\"size\"], \"ready\": [], \"frags\": [], \"done\": None, \"first\": None} for m in msgs), key=lambda m: m[\"t0\"])\n",
" t, i, active, cur = 0.0, 0, [], None\n",
" srtt, rttvar, loss = None, None, 0.0\n",
" st = {\"tx\": 0, \"retx\": 0, \"spurious\": 0, \"bytes\": 0, \"decide_ms\": [], \"flag\": {}, \"log\": []}\n",
" until = until if until is not None else (pend[-1][\"t0\"] if pend else 0) + 6000\n",
" while (i < len(pend) or active) and t < until:\n",
" while i < len(pend) and pend[i][\"t0\"] <= t:\n",
" active.append(pend[i]); i += 1\n",
" td = time.perf_counter()\n",
" for m in active: # ACK가 이미 도착한 재전송 예약은 취소\n",
" m[\"ready\"] = [e for e in m[\"ready\"] if not (e[1][\"ack\"] is not None and e[1][\"ack\"] <= t)]\n",
" ready = [m for m in active if any(r <= t for r, _ in m[\"ready\"]) or (m[\"unsent\"] > 0 and (admit is None or admit(m, t)))]\n",
" if not ready:\n",
" nxt = ([pend[i][\"t0\"]] if i < len(pend) else []) + [r for m in active for r, _ in m[\"ready\"] if r > t]\n",
" t = min(nxt) if nxt else t + 1.0\n",
" continue\n",
" if not preempt and cur in ready and cur[\"unsent\"] > 0:\n",
" m = cur\n",
" else:\n",
" m = min(ready, key=lambda x: policy_key(x, t, policy, aging))\n",
" st[\"decide_ms\"].append((time.perf_counter() - td) * 1000)\n",
" due = [e for e in m[\"ready\"] if e[0] <= t]\n",
" if due:\n",
" e = min(due, key=lambda x: x[0]); m[\"ready\"].remove(e); frag = e[1]\n",
" st[\"retx\"] += 1; st[\"spurious\"] += frag[\"recv\"]\n",
" else:\n",
" size = mtu if not adapt_mtu else (mtu if loss < 0.15 else (mtu // 2 if loss < 0.4 else mtu // 4))\n",
" size = min(m[\"unsent\"], size); m[\"unsent\"] -= size\n",
" frag = {\"b\": size, \"recv\": False, \"ack\": None, \"tries\": 0}\n",
" m[\"frags\"].append(frag)\n",
" cur = m\n",
" m[\"first\"] = t if m[\"first\"] is None else m[\"first\"]\n",
" nb = frag[\"b\"] + hdr\n",
" copies = 2 if (redundancy and RANK[m[\"prec\"]] <= 1 and loss > 0.15) else 1\n",
" for _ in range(copies):\n",
" t += nb * 8 / link.bw(t)\n",
" st[\"tx\"] += 1; st[\"bytes\"] += nb\n",
" ok = rng.random() >= link.p_loss(nb, t)\n",
" st[\"log\"].append((t, nb, ok, m[\"prec\"], m[\"id\"])) # 전송 기록(시각, bytes, 성공, 긴급도, 전문)\n",
" loss = 0.85 * loss + 0.15 * (0.0 if ok else 1.0) # 최근 손실률 추정(EWMA)\n",
" st[\"flag\"][int(t)] = st[\"flag\"].get(int(t), False) or loss > 0.12 # 링크 열화 판정(초 단위)\n",
" if ok and not frag[\"recv\"]:\n",
" frag[\"recv\"], frag[\"rt\"] = True, t\n",
" r = link.rtt(rng); frag[\"ack\"] = t + r\n",
" if srtt is None:\n",
" srtt, rttvar = r, r / 2\n",
" else:\n",
" rttvar = 0.75 * rttvar + 0.25 * abs(srtt - r); srtt = 0.875 * srtt + 0.125 * r\n",
" frag[\"tries\"] += 1\n",
" rto = rto_s if (rto_mode == \"fixed\" or srtt is None) else float(np.clip(srtt + 4 * rttvar, 1.0, 30.0))\n",
" if not frag[\"recv\"] or frag[\"ack\"] > t + rto: # 손실 또는 ACK 지연 -> 재전송 예약\n",
" m[\"ready\"].append((t + rto, frag))\n",
" if m[\"done\"] is None and m[\"unsent\"] == 0 and all(f[\"recv\"] for f in m[\"frags\"]):\n",
" m[\"done\"] = max(f[\"rt\"] for f in m[\"frags\"])\n",
" active = [x for x in active if not (x[\"unsent\"] == 0 and not x[\"ready\"] and all(f[\"recv\"] for f in x[\"frags\"]))]\n",
" res = [{\"id\": m[\"id\"], \"type\": m[\"type\"], \"prec\": m[\"prec\"], \"size\": m[\"size\"], \"t0\": m[\"t0\"], \"done\": m[\"done\"], \"wait\": None if m[\"first\"] is None else m[\"first\"] - m[\"t0\"],\n",
" \"retx\": sum(f[\"tries\"] for f in m[\"frags\"]) - len(m[\"frags\"]),\n",
" \"latency\": None if m[\"done\"] is None else m[\"done\"] - m[\"t0\"]} for m in pend]\n",
" for r in res:\n",
" r[\"on_time\"] = r[\"latency\"] is not None and r[\"latency\"] <= DEADLINE[r[\"prec\"]]\n",
" return res, st\n",
"\n",
"\n",
"def summarize(res):\n",
" \"\"\"긴급도별 기한 내 전달률·지연. 종합 정확도 = 긴급도별 기한 내 전달률의 평균.\"\"\"\n",
" # [TODO : 학생 실습 4] 전장 상황 및 전송 실패율에 따른 동적 패킷 스케줄링 및 지능형 재전송 알고리즘 구현 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"BASE = dict(policy=\"strict\", preempt=False) # 미션 3의 1차 큐(고정 RTO 4초)\n",
"ADV = dict(policy=\"aging\", aging=0.5, preempt=True, rto_mode=\"adaptive\", redundancy=True, adapt_mtu=True)\n",
"\n",
"\n",
"def urgent_rate(res):\n",
" u = [r[\"on_time\"] for r in res if RANK[r[\"prec\"]] <= 1]\n",
" return float(np.mean(u))\n",
"\n",
"\n",
"def degradation_episodes(link, flag):\n",
" \"\"\"열화 구간(bad/jam이 이어지는 구간, 전송 3회 이상) 중 구간 안팎 2초 내에 열화 판정이 나온 비율.\"\"\"\n",
" eps, s = [], None\n",
" for k, stt in enumerate(link.states + [\"good\"]):\n",
" if stt != \"good\" and s is None:\n",
" s = k\n",
" elif stt == \"good\" and s is not None:\n",
" eps.append((s, k)); s = None\n",
" hit = [any(flag.get(k, False) for k in range(a, b + 3)) for a, b in eps if sum(k in flag for k in range(a, b)) >= 3]\n",
" good = [v for k, v in flag.items() if link.state(k) == \"good\"]\n",
" return hit, float(np.mean(good)) if good else 0.0\n",
"\n",
"\n",
"def admin_burst(seed):\n",
" rng = np.random.default_rng(seed)\n",
" return [{\"id\": f\"A{i:03d}\", \"t0\": float(t), \"type\": \"ADMIN\", \"prec\": \"ROUTINE\", \"size\": int(rng.integers(3000, 5000)), \"src\": \"U09\"}\n",
" for i, t in enumerate(np.sort(rng.uniform(200, 400, 25)))]\n",
"\n",
"\n",
"# [교육용가상자료] 복합 장애 시나리오 10종\n",
"base = make_messages(1200, 0.21, seed=7)\n",
"double = base + make_messages(1200, 0.21, seed=8, prefix=\"N\")\n",
"SCEN = [(\"1 평시\", base, {}), (\"2 버스트 손실\", base, dict(p_gb=0.08, p_bg=0.1)), (\"3 재밍 60초\", base, dict(jam=[(300, 360)])),\n",
" (\"4 단속 재밍 3회\", base, dict(jam=[(200, 220), (500, 520), (800, 820)])), (\"5 대역폭 50% 감소\", base, dict(bw_drop=[(300, 700, 0.5)])),\n",
" (\"6 중계 지연 증가\", base, dict(rtt=(3.0, 1.0))), (\"7 트래픽 2배\", double, {}), (\"8 대용량 전문 폭주\", base + admin_burst(3), {}),\n",
" (\"9 버스트+대역폭 감소\", base, dict(p_gb=0.08, p_bg=0.1, bw_drop=[(300, 700, 0.5)])),\n",
" (\"10 재밍+트래픽2배+지연\", double, dict(jam=[(300, 360)], rtt=(3.0, 1.0)))]\n",
"\n",
"rows, hits, fas, dec = [], [], [], []\n",
"for name, msgs, lk in SCEN:\n",
" link = Link(12000, seed=1, **lk)\n",
" rb, sb = simulate(msgs, link, **BASE)\n",
" ra, sa = simulate(msgs, link, **ADV)\n",
" h, fa = degradation_episodes(link, sa[\"flag\"])\n",
" hits += h; fas.append(fa); dec += sa[\"decide_ms\"]\n",
" rows.append({\"시나리오\": name, \"전문\": len(msgs), \"기존 긴급전달률\": round(urgent_rate(rb), 3), \"개선 긴급전달률\": round(urgent_rate(ra), 3),\n",
" \"기존 종합\": summarize(rb)[1], \"개선 종합\": summarize(ra)[1], \"열화탐지\": f\"{sum(h)}/{len(h)}\", \"재전송\": sa[\"retx\"]})\n",
"df = pd.DataFrame(rows)\n",
"print(df.to_string(index=False))\n",
"u_all = df[\"개선 긴급전달률\"].mean()\n",
"rec = np.mean(hits)\n",
"print(f\"\\n긴급 전문(FLASH·IMMEDIATE) 기한 내 전달률 평균: 기존 {df['기존 긴급전달률'].mean():.1%} -> 개선 {u_all:.1%} (목표 85% 이상: {'달성' if u_all >= 0.85 else '미달'})\")\n",
"print(f\"링크 열화(복합 위협) 탐지 재현율 {rec:.1%} ({sum(hits)}/{len(hits)}구간, 목표 85% 이상: {'달성' if rec >= 0.85 else '미달'}), 정상 구간 오경보 {np.mean(fas):.1%}\")\n",
"print(\"참고: 열화 판정 기준(최근 손실률 추정값)을 0.12에서 0.2로 올리면 정상 구간 오경보는 줄지만 재현율이 84% 수준으로 떨어진다.\")\n",
"print(f\"스케줄링 결정 지연 평균 {np.mean(dec):.2f} ms, 최대 {np.max(dec):.1f} ms (목표 200 ms 이하: {'달성' if np.max(dec) <= 200 else '미달'})\")\n",
"print(f\"10종 중 긴급 전달률 85% 이상 시나리오: {(df['개선 긴급전달률'] >= 0.85).sum()}/10\")\n",
"\n",
"# 기능별 기여도(가장 어려운 시나리오 10에서 하나씩 끄기)\n",
"name, msgs, lk = SCEN[-1]\n",
"link = Link(12000, seed=1, **lk)\n",
"print(f\"\\n[기능별 기여도 - {name}]\")\n",
"for label, kw in [(\"전체 개선 기능\", ADV), (\"조각 단위 선점 끔\", {**ADV, \"preempt\": False}), (\"고정 RTO 4초\", {**ADV, \"rto_mode\": \"fixed\"}),\n",
" (\"중복 전송 끔\", {**ADV, \"redundancy\": False}), (\"조각 크기 조정 끔\", {**ADV, \"adapt_mtu\": False})]:\n",
" r, s = simulate(msgs, link, **kw)\n",
" print(f\"{label:14s} 긴급 {urgent_rate(r):.3f} 종합 {summarize(r)[1]:.3f} 불필요 재전송 {s['spurious']}회\")\n",
"print(\"관찰: 중계 지연이 커지면 고정 RTO는 ACK를 기다리지 못하고 불필요한 재전송을 반복해 대역폭을 낭비한다.\")\n",
"\n",
"plt.figure(figsize=(9, 4))\n",
"x = np.arange(len(df))\n",
"plt.bar(x - 0.2, df[\"기존 긴급전달률\"], 0.4, label=\"baseline PQ (M3)\")\n",
"plt.bar(x + 0.2, df[\"개선 긴급전달률\"], 0.4, label=\"adaptive scheduler\")\n",
"plt.axhline(0.85, ls=\"--\", c=\"gray\"); plt.xticks(x, [f\"S{i + 1}\" for i in x]); plt.ylim(0, 1.05)\n",
"plt.ylabel(\"urgent on-time ratio\"); plt.title(\"FLASH/IMMEDIATE on-time delivery in 10 stress scenarios (synthetic)\"); plt.legend()\n",
"plt.tight_layout(); plt.show()\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 04 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
" 시나리오 전문 기존 긴급전달률 개선 긴급전달률 기존 종합 개선 종합 열화탐지 재전송\n",
" 1 평시 258 0.885 1.000 0.940 1.000 14/14 96\n",
" 2 버스트 손실 258 0.865 1.000 0.931 1.000 42/43 211\n",
" 3 재밍 60초 258 0.846 1.000 0.921 1.000 15/15 167\n",
" 4 단속 재밍 3회 258 0.865 0.981 0.931 0.990 15/15 169\n",
" 5 대역폭 50% 감소 258 0.827 1.000 0.910 1.000 15/16 97\n",
" 6 중계 지연 증가 258 0.865 1.000 0.930 1.000 13/14 87\n",
" 7 트래픽 2배 501 0.966 1.000 0.795 0.808 24/24 121\n",
" 8 대용량 전문 폭주 283 0.827 1.000 0.867 0.958 16/19 101\n",
" 9 버스트+대역폭 감소 258 0.827 1.000 0.911 1.000 46/48 219\n",
"10 재밍+트래픽2배+지연 501 0.865 0.955 0.718 0.779 22/25 156\n",
"\n",
"긴급 전문(FLASH·IMMEDIATE) 기한 내 전달률 평균: 기존 86.4% -> 개선 99.4% (목표 85% 이상: 달성)\n",
"링크 열화(복합 위협) 탐지 재현율 95.3% (222/233구간, 목표 85% 이상: 달성), 정상 구간 오경보 9.7%\n",
"참고: 열화 판정 기준(최근 손실률 추정값)을 0.12에서 0.2로 올리면 정상 구간 오경보는 줄지만 재현율이 84% 수준으로 떨어진다.\n",
"스케줄링 결정 지연 평균 0.04 ms, 최대 39.3 ms (목표 200 ms 이하: 달성)\n",
"10종 중 긴급 전달률 85% 이상 시나리오: 10/10\n",
"\n",
"[기능별 기여도 - 10 재밍+트래픽2배+지연]\n",
"전체 개선 기능 긴급 0.955 종합 0.779 불필요 재전송 0회\n",
"조각 단위 선점 끔 긴급 0.910 종합 0.755 불필요 재전송 0회\n",
"고정 RTO 4초 긴급 0.955 종합 0.746 불필요 재전송 513회\n",
"중복 전송 끔 긴급 0.966 종합 0.786 불필요 재전송 0회\n",
"조각 크기 조정 끔 긴급 0.955 종합 0.776 불필요 재전송 0회\n",
"관찰: 중계 지연이 커지면 고정 RTO는 ACK를 기다리지 못하고 불필요한 재전송을 반복해 대역폭을 낭비한다.\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 05. 네트워크 혼잡도 기반 전송 성공 확률 및 링크 신뢰도 산출 엔진 개발\n",
"**소요시간**: 약 3.0시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"무선 링크 품질 및 혼잡도를 반영하여 전송 성공 확률과 링크 신뢰도를 산출하는 엔진을 개발한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **링크 품질과 혼잡도를 반영해 조각 전송 성공 확률과 링크 신뢰도 등급(High/Medium/Low)을 산출하는 엔진을 개발한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **SNR, 최근 손실률, 큐 점유율, 조각 크기, 지연 변동을 특징으로 학습용·검증용 가상 데이터를 나눠 그래디언트 부스팅 엔진을 학습 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `src/verification/c202_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/Medium/Low)을 산출하는 엔진을 개발하고, 단순 규칙 대비 '링크 불량' 오경보율(FAR) 감축, ROC-AUC, 임계값 민감도, 점수 일관성을 검증한다.\"\"\"\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"from sklearn.ensemble import HistGradientBoostingClassifier\n",
"from sklearn.inspection import permutation_importance\n",
"from sklearn.metrics import roc_auc_score, roc_curve\n",
"\n",
"# ---------------------------------------------------------------------------\n",
"# [교육용가상자료] 가상 전술 전문(메시지) 생성기 (실제 부대·통신망·제원 정보 없음)\n",
"# 긴급도 4단계(FLASH/IMMEDIATE/PRIORITY/ROUTINE)와 전문 유형별 크기를 가진 전문을 만든다.\n",
"# ---------------------------------------------------------------------------\n",
"PREC = [\"FLASH\", \"IMMEDIATE\", \"PRIORITY\", \"ROUTINE\"]\n",
"RANK = {p: i for i, p in enumerate(PREC)}\n",
"DEADLINE = {\"FLASH\": 10, \"IMMEDIATE\": 30, \"PRIORITY\": 120, \"ROUTINE\": 600} # 가상 전달 목표 시간(초)\n",
"MSG_TYPES = { # 유형: (긴급도 후보, 확률, 본문 크기 범위 bytes, 발생 비율)\n",
" \"CONTACT\": ([\"FLASH\", \"IMMEDIATE\"], [0.7, 0.3], (120, 300), 0.08),\n",
" \"MEDEVAC\": ([\"FLASH\", \"IMMEDIATE\"], [0.4, 0.6], (150, 350), 0.05),\n",
" \"SITREP\": ([\"IMMEDIATE\", \"PRIORITY\"], [0.2, 0.8], (600, 2500), 0.20),\n",
" \"POSREP\": ([\"PRIORITY\", \"ROUTINE\"], [0.5, 0.5], (60, 120), 0.35),\n",
" \"LOGREQ\": ([\"PRIORITY\", \"ROUTINE\"], [0.4, 0.6], (300, 1200), 0.14),\n",
" \"ADMIN\": ([\"ROUTINE\"], [1.0], (1000, 5000), 0.18),\n",
"}\n",
"\n",
"\n",
"def make_messages(duration=600, rate=0.12, seed=7, t_start=0.0, prefix=\"M\"):\n",
" \"\"\"duration초 동안 평균 rate건/초로 발생하는 가상 전문 목록.\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" n = int(rng.poisson(rate * duration))\n",
" times = np.sort(rng.uniform(t_start, t_start + duration, n))\n",
" types = list(MSG_TYPES)\n",
" msgs = []\n",
" for i, t0 in enumerate(times):\n",
" ty = types[rng.choice(len(types), p=[MSG_TYPES[k][3] for k in types])]\n",
" cands, p, (lo, hi), _ = MSG_TYPES[ty]\n",
" msgs.append({\"id\": f\"{prefix}{i:04d}\", \"t0\": float(t0), \"type\": ty, \"prec\": str(rng.choice(cands, p=p)),\n",
" \"size\": int(rng.integers(lo, hi + 1)), \"src\": f\"U{int(rng.integers(1, 9)):02d}\"})\n",
" return msgs\n",
"\n",
"\n",
"class Link:\n",
" \"\"\"[교육용가상자료] 가상 협대역 무선 링크(기본 2.4kbps). 초 단위 상태(good/bad/jam)에 따라\n",
" 비트오류율이 바뀌고(Gilbert-Elliott 모델), 구간별 대역폭 감소도 줄 수 있다.\"\"\"\n",
" BER = {\"good\": 1e-5, \"bad\": 2e-4, \"jam\": 1.2e-3}\n",
"\n",
" def __init__(self, horizon=4000, bw_bps=2400, seed=1, p_gb=0.02, p_bg=0.15, jam=(), bw_drop=(), rtt=(1.2, 0.4)):\n",
" rng = np.random.default_rng(seed)\n",
" st, s = [], \"good\"\n",
" for _ in range(int(horizon) + 1):\n",
" s = (\"bad\" if rng.random() < p_gb else \"good\") if s == \"good\" else (\"good\" if rng.random() < p_bg else \"bad\")\n",
" st.append(s)\n",
" for a, b in jam:\n",
" for k in range(int(a), min(int(b), len(st))):\n",
" st[k] = \"jam\"\n",
" self.states, self.bw_bps, self.bw_drop, self.rtt_par = st, bw_bps, list(bw_drop), rtt\n",
"\n",
" def state(self, t):\n",
" return self.states[min(int(t), len(self.states) - 1)]\n",
"\n",
" def bw(self, t):\n",
" f = 1.0\n",
" for a, b, fac in self.bw_drop:\n",
" if a <= t < b:\n",
" f = fac\n",
" return self.bw_bps * f\n",
"\n",
" def p_loss(self, nbytes, t):\n",
" return 1 - (1 - self.BER[self.state(t)]) ** (8 * nbytes)\n",
"\n",
" def rtt(self, rng):\n",
" return self.rtt_par[0] + rng.exponential(self.rtt_par[1])\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"HDR = 32\n",
"FEATS = [\"snr_db\", \"recent_loss\", \"queue_util\", \"pkt_bytes\", \"rtt_jitter\"]\n",
"\n",
"\n",
"def make_link_windows(n=6000, seed=1):\n",
" \"\"\"[교육용가상자료] 1초 창마다 링크 관측값과 그 창에서 보낸 조각 1개의 성공 여부. 실제 성공 확률(p_true)도 함께 둔다.\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" link = Link(n, seed=seed, p_gb=0.05, p_bg=0.15, jam=[(1000, 1060), (3500, 3540), (5000, 5080)])\n",
" snr_mu = {\"good\": 17, \"bad\": 9, \"jam\": 2}\n",
" rows = []\n",
" for k in range(1, n):\n",
" s, prev = link.state(k), link.state(k - 1)\n",
" size = int(rng.choice([64, 128, 256])) + HDR\n",
" util = float(np.clip(rng.beta(2, 3) + (0.35 if 2000 <= k < 2600 else 0), 0, 1)) # 혼잡 구간 포함\n",
" p_true = (1 - link.p_loss(size, k)) * (1 - np.clip((util - 0.85) * 4, 0, 0.6)) # 비트오류 + 큐 넘침\n",
" rows.append({\"snr_db\": snr_mu[s] + rng.normal(0, 2.5), \"recent_loss\": rng.binomial(10, link.p_loss(288, k - 1)) / 10,\n",
" \"queue_util\": util, \"pkt_bytes\": size, \"rtt_jitter\": rng.exponential(0.3) + (0.4 if s != \"good\" else 0),\n",
" \"state\": s, \"p_true\": p_true, \"ok\": int(rng.random() < p_true)})\n",
" return pd.DataFrame(rows)\n",
"\n",
"\n",
"def grade(p):\n",
" # [TODO : 학생 실습 5] 네트워크 혼잡도 기반 전송 성공 확률 및 링크 신뢰도 산출 엔진 개발 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def far_recall(alarm, bad):\n",
" # [TODO : 학생 실습 5] 네트워크 혼잡도 기반 전송 성공 확률 및 링크 신뢰도 산출 엔진 개발 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"train, test = make_link_windows(6000, seed=1), make_link_windows(6000, seed=2) # 학습용 / 검증용 분리\n",
"def new_model():\n",
" return HistGradientBoostingClassifier(max_depth=3, max_iter=150, learning_rate=0.05, random_state=0)\n",
"\n",
"\n",
"model = new_model().fit(train[FEATS], train[\"ok\"])\n",
"test[\"p\"] = model.predict_proba(test[FEATS])[:, 1]\n",
"test[\"grade\"] = grade(test[\"p\"])\n",
"bad = (test[\"p_true\"] < 0.6).values # 실제 불량 링크(정답, 시뮬레이터만 알고 있음)\n",
"\n",
"auc = roc_auc_score(test[\"ok\"], test[\"p\"])\n",
"print(f\"전송 성공 예측 ROC-AUC {auc:.3f} (검증 창 {len(test)}개, 실제 불량 링크 창 {bad.sum()}개)\")\n",
"\n",
"rule = ((test[\"snr_db\"] < 12) | (test[\"recent_loss\"] >= 0.4)).values # 기존: 단순 규칙 경보\n",
"eng = (test[\"p\"] < 0.6).values # 엔진: 성공 확률 0.6 미만 경보\n",
"f0, r0, n0 = far_recall(rule, bad)\n",
"f1, r1, n1 = far_recall(eng, bad)\n",
"print(pd.DataFrame([{\"방식\": \"기존 규칙(SNR<12dB 또는 손실>=40%)\", \"경보\": n0, \"오경보율\": f0, \"불량 재현율\": r0},\n",
" {\"방식\": \"신뢰도 엔진(성공확률<0.6)\", \"경보\": n1, \"오경보율\": f1, \"불량 재현율\": r1}]).to_string(index=False))\n",
"cut = 1 - f1 / max(f0, 1e-9)\n",
"print(f\"오경보율 감축 {cut:.1%} (목표 40% 이상: {'달성' if cut >= 0.4 else '미달'})\")\n",
"\n",
"# 임계값 민감도\n",
"print(\"\\n경보 임계값 경보수 오경보율 불량 재현율\")\n",
"for th in [0.4, 0.5, 0.6, 0.7, 0.8]:\n",
" f, r, n = far_recall((test[\"p\"] < th).values, bad)\n",
" print(f\" {th:.1f} {n:5d} {f:.3f} {r:.3f}\")\n",
"\n",
"# 점수 일관성: 학습 데이터를 부트스트랩으로 10번 다시 뽑아 학습해도 같은 등급이 나오는 비율 + 불확실성(표준편차)\n",
"rng = np.random.default_rng(0)\n",
"boot = []\n",
"for b in range(10):\n",
" idx = rng.integers(0, len(train), len(train))\n",
" m = new_model().fit(train.iloc[idx][FEATS], train.iloc[idx][\"ok\"])\n",
" boot.append(m.predict_proba(test[FEATS])[:, 1])\n",
"boot = np.array(boot)\n",
"same = np.mean([np.mean(grade(bp) == test[\"grade\"].values) for bp in boot])\n",
"test[\"unc\"] = boot.std(axis=0)\n",
"print(f\"\\n신뢰도 등급 일관성 {same:.1%} (목표 90% 이상: {'달성' if same >= 0.9 else '미달'}), 점수 불확실성(표준편차) 평균 {test['unc'].mean():.3f}, 최대 {test['unc'].max():.3f}\")\n",
"print(\"[실제 링크 상태별 등급 분포]\")\n",
"print(pd.crosstab(test[\"state\"], test[\"grade\"]).reindex(columns=[\"High\", \"Medium\", \"Low\"]).to_string())\n",
"imp = permutation_importance(model, test[FEATS], test[\"ok\"], scoring=\"roc_auc\", n_repeats=3, random_state=0)\n",
"print(\"판단 근거(특징 중요도, AUC 감소량):\", {f: round(float(v), 3) for f, v in sorted(zip(FEATS, imp.importances_mean), key=lambda x: -x[1])})\n",
"print(\"등급 운용 기준: High=정상 전송, Medium=긴급 전문 중복 전송 검토, Low=통신관에게 경로 변경·중계 검토 요청(자동 전환 아님)\")\n",
"\n",
"fig, ax = plt.subplots(1, 2, figsize=(11, 4))\n",
"fpr, tpr, _ = roc_curve(test[\"ok\"], test[\"p\"])\n",
"ax[0].plot(fpr, tpr); ax[0].plot([0, 1], [0, 1], \"k--\"); ax[0].set_title(f\"ROC (AUC={auc:.3f})\"); ax[0].set_xlabel(\"FPR\"); ax[0].set_ylabel(\"TPR\")\n",
"seg = test.iloc[900:1200]\n",
"ax[1].plot(seg.index, seg[\"p\"], label=\"success prob.\"); ax[1].fill_between(seg.index, seg[\"p\"] - 2 * seg[\"unc\"], seg[\"p\"] + 2 * seg[\"unc\"], alpha=0.3, label=\"+-2 sd\")\n",
"ax[1].plot(seg.index, seg[\"p_true\"], \"k:\", label=\"true prob.\"); ax[1].axhline(0.6, c=\"r\", ls=\"--\"); ax[1].legend(fontsize=8)\n",
"ax[1].set_title(\"Link reliability score with uncertainty (jamming at t=1000-1060)\")\n",
"plt.tight_layout(); plt.show()\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 05 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"전송 성공 예측 ROC-AUC 0.891 (검증 창 5999개, 실제 불량 링크 창 293개)\n",
" 방식 경보 오경보율 불량 재현율\n",
"기존 규칙(SNR<12dB 또는 손실>=40%) 1707 0.878 0.713\n",
" 신뢰도 엔진(성공확률<0.6) 363 0.342 0.816\n",
"오경보율 감축 61.0% (목표 40% 이상: 달성)\n",
"\n",
"경보 임계값 경보수 오경보율 불량 재현율\n",
" 0.4 159 0.069 0.505\n",
" 0.5 220 0.150 0.638\n",
" 0.6 363 0.342 0.816\n",
" 0.7 709 0.626 0.904\n",
" 0.8 1232 0.768 0.976\n",
"\n",
"신뢰도 등급 일관성 93.2% (목표 90% 이상: 달성), 점수 불확실성(표준편차) 평균 0.023, 최대 0.215\n",
"[실제 링크 상태별 등급 분포]\n",
"grade High Medium Low\n",
"state \n",
"bad 333 965 128\n",
"good 4207 122 64\n",
"jam 1 8 171\n",
"판단 근거(특징 중요도, AUC 감소량): {'snr_db': 0.131, 'queue_util': 0.068, 'rtt_jitter': 0.045, 'pkt_bytes': 0.016, 'recent_loss': 0.016}\n",
"등급 운용 기준: High=정상 전송, Medium=긴급 전문 중복 전송 검토, Low=통신관에게 경로 변경·중계 검토 요청(자동 전환 아님)\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 06. 대역폭 사용률, 큐 대기 현황 및 패킷 유실률 실시간 모니터링 UI 구현\n",
"**소요시간**: 약 2.5시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"대역폭 사용률, 큐 대기 전문 현황, 패킷 손실률을 실시간 표출하는 통신 관제 UI를 구현한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **대역폭 사용률, 긴급도별 큐 대기 전문, 패킷 손실률, 기한 임박·초과 경보를 1초 주기로 보여 주는 통신 관제 화면을 구현한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **재밍과 대역폭 감소가 포함된 20분 운용 기록을 1초 주기로 읽어 관제 지표를 갱신하고 처리시간을 측정 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `src/ui/c202_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초 주기로 읽어 대역폭 사용률, 긴급도별 큐 대기 전문 수, 패킷 손실률, 기한 임박·초과 경보를 표출하는 통신 관제 화면(야간 모드)을 구현하고 화면 갱신 처리시간을 측정한다.\"\"\"\n",
"import time\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"\n",
"# ---------------------------------------------------------------------------\n",
"# [교육용가상자료] 가상 전술 전문(메시지) 생성기 (실제 부대·통신망·제원 정보 없음)\n",
"# 긴급도 4단계(FLASH/IMMEDIATE/PRIORITY/ROUTINE)와 전문 유형별 크기를 가진 전문을 만든다.\n",
"# ---------------------------------------------------------------------------\n",
"PREC = [\"FLASH\", \"IMMEDIATE\", \"PRIORITY\", \"ROUTINE\"]\n",
"RANK = {p: i for i, p in enumerate(PREC)}\n",
"DEADLINE = {\"FLASH\": 10, \"IMMEDIATE\": 30, \"PRIORITY\": 120, \"ROUTINE\": 600} # 가상 전달 목표 시간(초)\n",
"MSG_TYPES = { # 유형: (긴급도 후보, 확률, 본문 크기 범위 bytes, 발생 비율)\n",
" \"CONTACT\": ([\"FLASH\", \"IMMEDIATE\"], [0.7, 0.3], (120, 300), 0.08),\n",
" \"MEDEVAC\": ([\"FLASH\", \"IMMEDIATE\"], [0.4, 0.6], (150, 350), 0.05),\n",
" \"SITREP\": ([\"IMMEDIATE\", \"PRIORITY\"], [0.2, 0.8], (600, 2500), 0.20),\n",
" \"POSREP\": ([\"PRIORITY\", \"ROUTINE\"], [0.5, 0.5], (60, 120), 0.35),\n",
" \"LOGREQ\": ([\"PRIORITY\", \"ROUTINE\"], [0.4, 0.6], (300, 1200), 0.14),\n",
" \"ADMIN\": ([\"ROUTINE\"], [1.0], (1000, 5000), 0.18),\n",
"}\n",
"\n",
"\n",
"def make_messages(duration=600, rate=0.12, seed=7, t_start=0.0, prefix=\"M\"):\n",
" \"\"\"duration초 동안 평균 rate건/초로 발생하는 가상 전문 목록.\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" n = int(rng.poisson(rate * duration))\n",
" times = np.sort(rng.uniform(t_start, t_start + duration, n))\n",
" types = list(MSG_TYPES)\n",
" msgs = []\n",
" for i, t0 in enumerate(times):\n",
" ty = types[rng.choice(len(types), p=[MSG_TYPES[k][3] for k in types])]\n",
" cands, p, (lo, hi), _ = MSG_TYPES[ty]\n",
" msgs.append({\"id\": f\"{prefix}{i:04d}\", \"t0\": float(t0), \"type\": ty, \"prec\": str(rng.choice(cands, p=p)),\n",
" \"size\": int(rng.integers(lo, hi + 1)), \"src\": f\"U{int(rng.integers(1, 9)):02d}\"})\n",
" return msgs\n",
"\n",
"\n",
"class Link:\n",
" \"\"\"[교육용가상자료] 가상 협대역 무선 링크(기본 2.4kbps). 초 단위 상태(good/bad/jam)에 따라\n",
" 비트오류율이 바뀌고(Gilbert-Elliott 모델), 구간별 대역폭 감소도 줄 수 있다.\"\"\"\n",
" BER = {\"good\": 1e-5, \"bad\": 2e-4, \"jam\": 1.2e-3}\n",
"\n",
" def __init__(self, horizon=4000, bw_bps=2400, seed=1, p_gb=0.02, p_bg=0.15, jam=(), bw_drop=(), rtt=(1.2, 0.4)):\n",
" rng = np.random.default_rng(seed)\n",
" st, s = [], \"good\"\n",
" for _ in range(int(horizon) + 1):\n",
" s = (\"bad\" if rng.random() < p_gb else \"good\") if s == \"good\" else (\"good\" if rng.random() < p_bg else \"bad\")\n",
" st.append(s)\n",
" for a, b in jam:\n",
" for k in range(int(a), min(int(b), len(st))):\n",
" st[k] = \"jam\"\n",
" self.states, self.bw_bps, self.bw_drop, self.rtt_par = st, bw_bps, list(bw_drop), rtt\n",
"\n",
" def state(self, t):\n",
" return self.states[min(int(t), len(self.states) - 1)]\n",
"\n",
" def bw(self, t):\n",
" f = 1.0\n",
" for a, b, fac in self.bw_drop:\n",
" if a <= t < b:\n",
" f = fac\n",
" return self.bw_bps * f\n",
"\n",
" def p_loss(self, nbytes, t):\n",
" return 1 - (1 - self.BER[self.state(t)]) ** (8 * nbytes)\n",
"\n",
" def rtt(self, rng):\n",
" return self.rtt_par[0] + rng.exponential(self.rtt_par[1])\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# ---- 우선순위 큐·재전송 이산 사건 시뮬레이터 공통 함수 ----\n",
"HDR = 32 # 전문 조각(패킷) 헤더 bytes (미션 2의 헤더 압축 적용 시 12 bytes)\n",
"\n",
"\n",
"def policy_key(m, t, policy, aging):\n",
" # [TODO : 학생 실습 6] 대역폭 사용률, 큐 대기 현황 및 패킷 유실률 실시간 모니터링 UI 구현 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def simulate(msgs, link, policy=\"strict\", mtu=256, preempt=True, rto_mode=\"fixed\", rto_s=4.0, aging=0.0,\n",
" redundancy=False, adapt_mtu=False, hdr=HDR, seed=0, until=None, admit=None):\n",
" \"\"\"전문 목록을 링크로 보내는 이산 사건 시뮬레이션.\n",
" - 전문은 mtu 크기 조각으로 나눠 보낸다. 조각이 손실되면 RTO 후 재전송한다(자료는 버리지 않음).\n",
" - preempt=False면 한 전문의 조각을 끝까지 연속 전송한다(큰 전문이 긴급 전문을 막는 현상 재현).\n",
" - rto_mode='adaptive'면 왕복시간(RTT) 추정값으로 RTO를 계산한다(Jacobson 방식).\n",
" - redundancy=True면 링크 열화 추정 시 FLASH·IMMEDIATE 조각을 2회 연속 전송한다.\n",
" - adapt_mtu=True면 링크 열화 추정 시 조각 크기를 줄인다.\n",
" - admit(m, t)이 False인 전문은 새 조각을 보내지 않고 저장해 둔다(성능 저하 모드).\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" pend = sorted(({**m, \"unsent\": m[\"size\"], \"ready\": [], \"frags\": [], \"done\": None, \"first\": None} for m in msgs), key=lambda m: m[\"t0\"])\n",
" t, i, active, cur = 0.0, 0, [], None\n",
" srtt, rttvar, loss = None, None, 0.0\n",
" st = {\"tx\": 0, \"retx\": 0, \"spurious\": 0, \"bytes\": 0, \"decide_ms\": [], \"flag\": {}, \"log\": []}\n",
" until = until if until is not None else (pend[-1][\"t0\"] if pend else 0) + 6000\n",
" while (i < len(pend) or active) and t < until:\n",
" while i < len(pend) and pend[i][\"t0\"] <= t:\n",
" active.append(pend[i]); i += 1\n",
" td = time.perf_counter()\n",
" for m in active: # ACK가 이미 도착한 재전송 예약은 취소\n",
" m[\"ready\"] = [e for e in m[\"ready\"] if not (e[1][\"ack\"] is not None and e[1][\"ack\"] <= t)]\n",
" ready = [m for m in active if any(r <= t for r, _ in m[\"ready\"]) or (m[\"unsent\"] > 0 and (admit is None or admit(m, t)))]\n",
" if not ready:\n",
" nxt = ([pend[i][\"t0\"]] if i < len(pend) else []) + [r for m in active for r, _ in m[\"ready\"] if r > t]\n",
" t = min(nxt) if nxt else t + 1.0\n",
" continue\n",
" if not preempt and cur in ready and cur[\"unsent\"] > 0:\n",
" m = cur\n",
" else:\n",
" m = min(ready, key=lambda x: policy_key(x, t, policy, aging))\n",
" st[\"decide_ms\"].append((time.perf_counter() - td) * 1000)\n",
" due = [e for e in m[\"ready\"] if e[0] <= t]\n",
" if due:\n",
" e = min(due, key=lambda x: x[0]); m[\"ready\"].remove(e); frag = e[1]\n",
" st[\"retx\"] += 1; st[\"spurious\"] += frag[\"recv\"]\n",
" else:\n",
" size = mtu if not adapt_mtu else (mtu if loss < 0.15 else (mtu // 2 if loss < 0.4 else mtu // 4))\n",
" size = min(m[\"unsent\"], size); m[\"unsent\"] -= size\n",
" frag = {\"b\": size, \"recv\": False, \"ack\": None, \"tries\": 0}\n",
" m[\"frags\"].append(frag)\n",
" cur = m\n",
" m[\"first\"] = t if m[\"first\"] is None else m[\"first\"]\n",
" nb = frag[\"b\"] + hdr\n",
" copies = 2 if (redundancy and RANK[m[\"prec\"]] <= 1 and loss > 0.15) else 1\n",
" for _ in range(copies):\n",
" t += nb * 8 / link.bw(t)\n",
" st[\"tx\"] += 1; st[\"bytes\"] += nb\n",
" ok = rng.random() >= link.p_loss(nb, t)\n",
" st[\"log\"].append((t, nb, ok, m[\"prec\"], m[\"id\"])) # 전송 기록(시각, bytes, 성공, 긴급도, 전문)\n",
" loss = 0.85 * loss + 0.15 * (0.0 if ok else 1.0) # 최근 손실률 추정(EWMA)\n",
" st[\"flag\"][int(t)] = st[\"flag\"].get(int(t), False) or loss > 0.12 # 링크 열화 판정(초 단위)\n",
" if ok and not frag[\"recv\"]:\n",
" frag[\"recv\"], frag[\"rt\"] = True, t\n",
" r = link.rtt(rng); frag[\"ack\"] = t + r\n",
" if srtt is None:\n",
" srtt, rttvar = r, r / 2\n",
" else:\n",
" rttvar = 0.75 * rttvar + 0.25 * abs(srtt - r); srtt = 0.875 * srtt + 0.125 * r\n",
" frag[\"tries\"] += 1\n",
" rto = rto_s if (rto_mode == \"fixed\" or srtt is None) else float(np.clip(srtt + 4 * rttvar, 1.0, 30.0))\n",
" if not frag[\"recv\"] or frag[\"ack\"] > t + rto: # 손실 또는 ACK 지연 -> 재전송 예약\n",
" m[\"ready\"].append((t + rto, frag))\n",
" if m[\"done\"] is None and m[\"unsent\"] == 0 and all(f[\"recv\"] for f in m[\"frags\"]):\n",
" m[\"done\"] = max(f[\"rt\"] for f in m[\"frags\"])\n",
" active = [x for x in active if not (x[\"unsent\"] == 0 and not x[\"ready\"] and all(f[\"recv\"] for f in x[\"frags\"]))]\n",
" res = [{\"id\": m[\"id\"], \"type\": m[\"type\"], \"prec\": m[\"prec\"], \"size\": m[\"size\"], \"t0\": m[\"t0\"], \"done\": m[\"done\"], \"wait\": None if m[\"first\"] is None else m[\"first\"] - m[\"t0\"],\n",
" \"retx\": sum(f[\"tries\"] for f in m[\"frags\"]) - len(m[\"frags\"]),\n",
" \"latency\": None if m[\"done\"] is None else m[\"done\"] - m[\"t0\"]} for m in pend]\n",
" for r in res:\n",
" r[\"on_time\"] = r[\"latency\"] is not None and r[\"latency\"] <= DEADLINE[r[\"prec\"]]\n",
" return res, st\n",
"\n",
"\n",
"def summarize(res):\n",
" \"\"\"긴급도별 기한 내 전달률·지연. 종합 정확도 = 긴급도별 기한 내 전달률의 평균.\"\"\"\n",
" # [TODO : 학생 실습 6] 대역폭 사용률, 큐 대기 현황 및 패킷 유실률 실시간 모니터링 UI 구현 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"COLORS = {\"FLASH\": \"red\", \"IMMEDIATE\": \"orange\", \"PRIORITY\": \"deepskyblue\", \"ROUTINE\": \"gray\"}\n",
"\n",
"# [교육용가상자료] 20분 운용: 300~360초 재밍, 600~800초 대역폭 50% 감소\n",
"msgs = make_messages(1200, 0.21, seed=7)\n",
"link = Link(12000, seed=1, jam=[(300, 360)], bw_drop=[(600, 800, 0.5)])\n",
"res, st = simulate(msgs, link, \"aging\", aging=0.5, rto_mode=\"adaptive\", redundancy=True, adapt_mtu=True)\n",
"log = pd.DataFrame(st[\"log\"], columns=[\"t\", \"bytes\", \"ok\", \"prec\", \"id\"])\n",
"done = {r[\"id\"]: r[\"done\"] for r in res}\n",
"\n",
"\n",
"def snapshot(now, window=10):\n",
" \"\"\"now 시각의 관제 지표: 최근 window초 사용률·손실률, 긴급도별 대기 전문, 경보 목록.\"\"\"\n",
" w = log[(log[\"t\"] > now - window) & (log[\"t\"] <= now)]\n",
" util = w[\"bytes\"].sum() * 8 / (link.bw(now) * window)\n",
" loss = 1 - w[\"ok\"].mean() if len(w) else 0.0\n",
" waiting = [m for m in msgs if m[\"t0\"] <= now and (done[m[\"id\"]] is None or done[m[\"id\"]] > now)]\n",
" depth = {p: sum(m[\"prec\"] == p for m in waiting) for p in PREC}\n",
" alerts = []\n",
" for m in waiting:\n",
" age = now - m[\"t0\"]\n",
" if age > DEADLINE[m[\"prec\"]]:\n",
" alerts.append((\"기한초과\", m[\"id\"], m[\"prec\"], round(age, 1)))\n",
" elif RANK[m[\"prec\"]] <= 1 and age > 0.7 * DEADLINE[m[\"prec\"]]:\n",
" alerts.append((\"기한임박\", m[\"id\"], m[\"prec\"], round(age, 1)))\n",
" if loss > 0.3:\n",
" alerts.append((\"링크열화\", \"-\", f\"손실률 {loss:.0%}\", round(now, 0)))\n",
" return {\"t\": now, \"util\": min(util, 1.0), \"loss\": loss, **depth, \"alerts\": alerts}\n",
"\n",
"\n",
"# 1) 1초 주기 갱신 모의(1,200회) + 처리시간 측정\n",
"frames, ms = [], []\n",
"for now in range(1, 1201):\n",
" t0 = time.perf_counter()\n",
" frames.append(snapshot(float(now)))\n",
" ms.append((time.perf_counter() - t0) * 1000)\n",
"ts = pd.DataFrame([{k: v for k, v in f.items() if k != \"alerts\"} for f in frames])\n",
"print(f\"갱신 1,200회: 처리시간 평균 {np.mean(ms):.2f} ms, 최대 {np.max(ms):.2f} ms (목표: 주기 1초 이내, 응답 100 ms 이내 -> {'달성' if np.max(ms) < 100 else '미달'})\")\n",
"\n",
"# 2) 구간별 상황 요약\n",
"ts[\"구간\"] = pd.cut(ts[\"t\"], [0, 300, 360, 600, 800, 1200], labels=[\"평시1\", \"재밍\", \"평시2\", \"대역폭50%\", \"평시3\"])\n",
"summary = ts.groupby(\"구간\", observed=True).agg(사용률=(\"util\", \"mean\"), 손실률=(\"loss\", \"mean\"), FLASH대기=(\"FLASH\", \"max\"),\n",
" IMMEDIATE대기=(\"IMMEDIATE\", \"max\"), ROUTINE대기=(\"ROUTINE\", \"max\")).round(2)\n",
"print(\"\\n[구간별 관제 지표: 평균 사용률·손실률, 최대 대기 전문 수]\")\n",
"print(summary.to_string())\n",
"\n",
"# 3) 경보 패널(긴급 경보 우선)\n",
"all_alerts = pd.DataFrame([(f[\"t\"], *a) for f in frames for a in f[\"alerts\"]], columns=[\"시각\", \"경보\", \"전문\", \"긴급도/내용\", \"경과\"])\n",
"first = all_alerts.drop_duplicates([\"경보\", \"전문\"]).copy()\n",
"print(f\"\\n경보 발생(중복 제외) {len(first)}건:\", first[\"경보\"].value_counts().to_dict())\n",
"order = {\"기한초과\": 0, \"링크열화\": 1, \"기한임박\": 2}\n",
"print(first.sort_values(by=[\"경보\", \"시각\"], key=lambda c: c.map(order) if c.name == \"경보\" else c).head(8).to_string(index=False))\n",
"\n",
"# 4) 관제 화면(야간 모드, 구성요소 5종)\n",
"plt.style.use(\"dark_background\")\n",
"fig = plt.figure(figsize=(13, 8))\n",
"a1 = fig.add_subplot(2, 2, 1); a1.plot(ts[\"t\"], ts[\"util\"] * 100, c=\"lime\"); a1.set_title(\"Bandwidth utilization (%)\"); a1.set_ylim(0, 105)\n",
"a2 = fig.add_subplot(2, 2, 2)\n",
"a2.stackplot(ts[\"t\"], *[ts[p] for p in PREC], colors=[COLORS[p] for p in PREC], labels=PREC); a2.legend(loc=\"upper left\", fontsize=8)\n",
"a2.set_title(\"Queued messages by precedence\")\n",
"a3 = fig.add_subplot(2, 2, 3); a3.plot(ts[\"t\"], ts[\"loss\"] * 100, c=\"orange\"); a3.axhline(30, c=\"red\", ls=\"--\")\n",
"a3.fill_between(ts[\"t\"], 0, 100, where=ts[\"loss\"] > 0.3, color=\"red\", alpha=0.25); a3.set_title(\"Packet loss rate (%) - red = link degraded\")\n",
"for a in (a1, a2, a3):\n",
" a.axvspan(300, 360, color=\"purple\", alpha=0.2); a.axvspan(600, 800, color=\"yellow\", alpha=0.08); a.set_xlabel(\"time (s)\")\n",
"a4 = fig.add_subplot(2, 2, 4); a4.axis(\"off\")\n",
"last = frames[-1]\n",
"kpi = summarize(res)[0]\n",
"txt = \"\\n\".join(f\"{r['긴급도']:<10} on-time {r['기한내 전달률']:.0%} p95 {r['95% 지연(s)']:>6}s\" for _, r in kpi.iterrows())\n",
"a4.text(0, 0.95, f\"KPI (t={last['t']:.0f}s)\\n{txt}\\n\\nAlerts total: {len(first)}\\n overdue: {(first['경보'] == '기한초과').sum()}\\n\"\n",
" f\" link degraded: {(first['경보'] == '링크열화').sum()}\", family=\"monospace\", fontsize=11, va=\"top\", color=\"white\")\n",
"fig.suptitle(\"Tactical link monitor (night mode) - training data only\", color=\"white\")\n",
"plt.tight_layout(); plt.show()\n",
"print(\"\\n화면 구성: 사용률 추이, 긴급도별 큐 대기, 손실률(열화 구간 강조), KPI·경보 패널, 재밍/대역폭 감소 구간 표시\")\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 06 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"갱신 1,200회: 처리시간 평균 0.76 ms, 최대 3.48 ms (목표: 주기 1초 이내, 응답 100 ms 이내 -> 달성)\n",
"\n",
"[구간별 관제 지표: 평균 사용률·손실률, 최대 대기 전문 수]\n",
" 사용률 손실률 FLASH대기 IMMEDIATE대기 ROUTINE대기\n",
"구간 \n",
"평시1 0.84 0.03 1 1 9\n",
"재밍 0.99 0.58 1 0 13\n",
"평시2 0.99 0.06 1 1 17\n",
"대역폭50% 0.97 0.06 2 2 21\n",
"평시3 0.98 0.05 1 2 27\n",
"\n",
"경보 발생(중복 제외) 4건: {'기한임박': 2, '링크열화': 1, '기한초과': 1}\n",
" 시각 경보 전문 긴급도/내용 경과\n",
"805.0 기한초과 M0178 FLASH 10.1\n",
"303.0 링크열화 - 손실률 36% 303.0\n",
"780.0 기한임박 M0165 IMMEDIATE 21.7\n",
"802.0 기한임박 M0178 FLASH 7.1\n",
"\n",
"화면 구성: 사용률 추이, 긴급도별 큐 대기, 손실률(열화 구간 강조), KPI·경보 패널, 재밍/대역폭 감소 구간 표시\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 07. 최우선 순위 전문 강제 전송 및 큐 순서 수동 조정을 위한 통신관 HITL 통제 기능 설계\n",
"**소요시간**: 약 3.0시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"최우선 작전 전문 강제 전송 및 큐 순서 수동 조정을 위한 통신관 HITL 제어 기능을 설계한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **통신관이 최우선 전문 강제 전송과 큐 순서 수동 조정을 직접 수행하는 HITL 통제 기능을 설계한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **시스템은 보류·앞당기기를 추천만 하고 큐 순서를 바꾸지 않는다 과정을 진행하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `src/hitl/c202_approval_workflow.py 및 docs/c202_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": [
"# [TODO : 학생 실습 7] 최우선 순위 전문 강제 전송 및 큐 순서 수동 조정을 위한 통신관 HITL 통제 기능 설계\n",
"# 미션 지침 및 검증 기준을 확인하고 코드를 실행 및 튜닝하세요.\n",
"\"\"\"통신관이 최우선 전문 강제 전송과 큐 순서 수동 조정을 직접 수행하는 HITL 통제 기능을 구현한다. 시스템은 추천만 하고 순서를 스스로 바꾸지 않으며, 규칙 위반 조작은 차단하고, FLASH보다 앞서는 강제 전송은 상급자 승인을 거치게 하며, 모든 조작은 위·변조 확인이 가능한 감사 로그에 기록한다.\"\"\"\n",
"import hashlib\n",
"import json\n",
"import pandas as pd\n",
"\n",
"ACTIONS = {\"FORCE_SEND\": \"강제 전송(맨 앞)\", \"MOVE_UP\": \"한 칸 앞으로\", \"MOVE_DOWN\": \"한 칸 뒤로\", \"HOLD\": \"보류\", \"RELEASE\": \"보류 해제\"}\n",
"RANK = {\"FLASH\": 0, \"IMMEDIATE\": 1, \"PRIORITY\": 2, \"ROUTINE\": 3}\n",
"\n",
"\n",
"class ControlError(ValueError):\n",
" pass\n",
"\n",
"\n",
"class FakeClock:\n",
" t = 1000.0\n",
"\n",
" def __call__(self):\n",
" return self.t\n",
"\n",
"\n",
"class HITLQueue:\n",
" def __init__(self, clock):\n",
" self.q, self.held, self.req, self.log, self.clock = [], {}, {}, [], 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, prec):\n",
" \"\"\"기본 정렬(긴급도 -> 도착순) 위치에 넣는다. 수동 조정된 순서는 보존한다.\"\"\"\n",
" m = {\"id\": mid, \"prec\": prec, \"t0\": self.clock()}\n",
" pos = next((i for i, x in enumerate(self.q) if RANK[x[\"prec\"]] > RANK[prec]), len(self.q))\n",
" self.q.insert(pos, m)\n",
" self._record({\"event\": \"SUBMIT\", \"id\": mid, \"prec\": prec, \"by\": \"SYSTEM\"})\n",
"\n",
" def recommend(self, link_grade):\n",
" \"\"\"시스템 추천(자동 적용 안 함): 링크 Low면 ROUTINE 대용량 보류 권고 등.\"\"\"\n",
" out = []\n",
" if link_grade == \"Low\":\n",
" out += [f\"{m['id']} 보류 권고(ROUTINE)\" for m in self.q if m[\"prec\"] == \"ROUTINE\"][:2]\n",
" old = [m for m in self.q if m[\"prec\"] == \"IMMEDIATE\" and self.clock() - m[\"t0\"] > 20]\n",
" out += [f\"{m['id']} 앞당기기 권고(기한 임박)\" for m in old]\n",
" self._record({\"event\": \"RECOMMEND\", \"items\": out, \"by\": \"SYSTEM\"})\n",
" return out\n",
"\n",
" def act(self, mid, action, operator, reason, approver=None):\n",
" if action not in ACTIONS:\n",
" raise ControlError(f\"허용되지 않은 조작: {action}\")\n",
" if not operator:\n",
" raise ControlError(\"통신관 ID 필요\")\n",
" if len(reason.strip()) < 5:\n",
" raise ControlError(\"조작 근거 5자 이상 필요\")\n",
" idx = next((i for i, m in enumerate(self.q) if m[\"id\"] == mid), None)\n",
" if idx is None and not (action == \"RELEASE\" and mid in self.held):\n",
" raise ControlError(\"큐에 없는 전문\")\n",
" if action == \"FORCE_SEND\":\n",
" m = self.q[idx]\n",
" flash_waiting = any(x[\"prec\"] == \"FLASH\" for x in self.q if x[\"id\"] != mid)\n",
" if m[\"prec\"] != \"FLASH\" and flash_waiting and not approver:\n",
" self.req[mid] = {\"operator\": operator, \"reason\": reason, \"t\": self.clock()}\n",
" self._record({\"event\": \"REQUEST\", \"id\": mid, \"action\": action, \"operator\": operator, \"reason\": reason})\n",
" return \"상급자 승인 대기(대기 중인 FLASH보다 앞섬)\"\n",
" if approver == operator:\n",
" raise ControlError(\"요청자와 승인자는 달라야 함(2인 확인)\")\n",
" self.q.insert(0, self.q.pop(idx)); self.req.pop(mid, None)\n",
" elif action in (\"MOVE_UP\", \"MOVE_DOWN\"):\n",
" j = idx - 1 if action == \"MOVE_UP\" else idx + 1\n",
" if not 0 <= j < len(self.q):\n",
" raise ControlError(\"이동할 위치 없음\")\n",
" if action == \"MOVE_UP\" and self.q[j][\"prec\"] == \"FLASH\" and self.q[idx][\"prec\"] != \"FLASH\" and not approver:\n",
" raise ControlError(\"FLASH 앞으로 이동은 상급자 승인 필요\")\n",
" self.q[idx], self.q[j] = self.q[j], self.q[idx]\n",
" elif action == \"HOLD\":\n",
" if self.q[idx][\"prec\"] == \"FLASH\" and (len(reason.strip()) < 15 or not approver):\n",
" raise ControlError(\"FLASH 보류는 근거 15자 이상 + 상급자 승인 필요\")\n",
" self.held[mid] = self.q.pop(idx)\n",
" elif action == \"RELEASE\":\n",
" m = self.held.pop(mid)\n",
" pos = next((i for i, x in enumerate(self.q) if RANK[x[\"prec\"]] > RANK[m[\"prec\"]]), len(self.q))\n",
" self.q.insert(pos, m)\n",
" self._record({\"event\": \"ACT\", \"id\": mid, \"action\": action, \"operator\": operator, \"approver\": approver, \"reason\": reason})\n",
" return \"적용\"\n",
"\n",
"\n",
"clock = FakeClock()\n",
"hq = HITLQueue(clock)\n",
"# [교육용가상자료] 대기 전문 7건\n",
"for mid, p in [(\"M0101\", \"ROUTINE\"), (\"M0102\", \"PRIORITY\"), (\"M0103\", \"FLASH\"), (\"M0104\", \"IMMEDIATE\"), (\"M0105\", \"ROUTINE\"), (\"M0106\", \"PRIORITY\"), (\"M0107\", \"IMMEDIATE\")]:\n",
" clock.t += 3; hq.submit(mid, p)\n",
"order0 = [m[\"id\"] for m in hq.q]\n",
"print(\"초기 큐 순서:\", order0)\n",
"clock.t += 20\n",
"print(\"시스템 추천(자동 적용 안 함):\", hq.recommend(\"Low\"))\n",
"print(\"추천 후 큐 순서 변화 없음:\", [m[\"id\"] for m in hq.q] == order0)\n",
"\n",
"# 규칙 위반 조작 차단\n",
"for args in [(\"M0105\", \"FORCE_SEND\", \"\", \"근거 있음\"), (\"M0102\", \"MOVE_UP\", \"sig01\", \"앞으로\"), (\"M0104\", \"MOVE_UP\", \"sig01\", \"FLASH 앞으로 이동 요청\"),\n",
" (\"M0103\", \"HOLD\", \"sig01\", \"보류 필요함\"), (\"M0101\", \"DELETE\", \"sig01\", \"삭제 요청함\")]:\n",
" try:\n",
" hq.act(*args)\n",
" except ControlError as e:\n",
" print(f\"차단: {args[0]} {args[1]} -> {e}\")\n",
"\n",
"# 정상 조작\n",
"clock.t += 2; print(\"M0107 앞당김:\", hq.act(\"M0107\", \"MOVE_UP\", \"sig01\", \"기한 임박 IMMEDIATE 우선 처리\"))\n",
"clock.t += 2; print(\"M0102 강제 전송:\", hq.act(\"M0102\", \"FORCE_SEND\", \"sig01\", \"지휘관 지시 상황보고 즉시 전송\"))\n",
"clock.t += 25; print(\"M0102 상급자 승인:\", hq.act(\"M0102\", \"FORCE_SEND\", \"sig01\", \"지휘관 지시 상황보고 즉시 전송\", approver=\"sup01\"))\n",
"clock.t += 2; print(\"M0105 보류:\", hq.act(\"M0105\", \"HOLD\", \"sig02\", \"링크 Low, 대용량 행정전문 보류\"))\n",
"print(\"조정 후 큐 순서:\", [m[\"id\"] for m in hq.q], \"/ 보류:\", list(hq.held))\n",
"\n",
"# 대기시간·로그\n",
"waits = [{\"id\": r[\"id\"], \"조작\": r[\"action\"], \"통신관\": r[\"operator\"], \"승인자\": r[\"approver\"] or \"-\",\n",
" \"요청 후 적용까지(s)\": round(r[\"ts\"] - next((x[\"ts\"] for x in hq.log if x[\"event\"] == \"REQUEST\" and x[\"id\"] == r[\"id\"]), r[\"ts\"]), 1)}\n",
" for r in hq.log if r[\"event\"] == \"ACT\"]\n",
"print(\"\\n\" + pd.DataFrame(waits).to_string(index=False))\n",
"n_act = sum(r[\"event\"] in (\"ACT\", \"REQUEST\") for r in hq.log)\n",
"print(f\"\\n감사 로그 {len(hq.log)}건(조작·요청 {n_act}건 전부 기록, 로깅 성공률 100%), 해시 체인 검증: {hq.verify_log()}\")\n",
"hq.log[9][\"operator\"] = \"sig99\"\n",
"print(\"10번째 기록 변조 후 검증(False, 깨진 위치):\", hq.verify_log())\n",
"\n",
"# 지휘관 피드백 입력 양식(문자열로 출력)\n",
"form = {\"form\": \"c202_commander_feedback\", \"date_time\": \"(입력)\", \"message_id\": \"M0102\", \"delivery_ok\": \"(Y/N)\",\n",
" \"delay_acceptable\": \"(Y/N)\", \"priority_change_request\": \"(내용)\", \"comment\": \"(내용)\"}\n",
"print(\"\\n[지휘관 피드백 입력 양식]\\n\" + json.dumps(form, ensure_ascii=False))\n",
"print(\"시스템은 순서를 추천만 하며, 강제 전송·보류의 책임은 조작한 통신관과 승인한 상급자에게 있다.\")\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 07 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"초기 큐 순서: ['M0103', 'M0104', 'M0107', 'M0102', 'M0106', 'M0101', 'M0105']\n",
"시스템 추천(자동 적용 안 함): ['M0101 보류 권고(ROUTINE)', 'M0105 보류 권고(ROUTINE)', 'M0104 앞당기기 권고(기한 임박)']\n",
"추천 후 큐 순서 변화 없음: True\n",
"차단: M0105 FORCE_SEND -> 통신관 ID 필요\n",
"차단: M0102 MOVE_UP -> 조작 근거 5자 이상 필요\n",
"차단: M0104 MOVE_UP -> FLASH 앞으로 이동은 상급자 승인 필요\n",
"차단: M0103 HOLD -> FLASH 보류는 근거 15자 이상 + 상급자 승인 필요\n",
"차단: M0101 DELETE -> 허용되지 않은 조작: DELETE\n",
"M0107 앞당김: 적용\n",
"M0102 강제 전송: 상급자 승인 대기(대기 중인 FLASH보다 앞섬)\n",
"M0102 상급자 승인: 적용\n",
"M0105 보류: 적용\n",
"조정 후 큐 순서: ['M0102', 'M0103', 'M0107', 'M0104', 'M0106', 'M0101'] / 보류: ['M0105']\n",
"\n",
" id 조작 통신관 승인자 요청 후 적용까지(s)\n",
"M0107 MOVE_UP sig01 - 0.0\n",
"M0102 FORCE_SEND sig01 sup01 25.0\n",
"M0105 HOLD sig02 - 0.0\n",
"\n",
"감사 로그 12건(조작·요청 4건 전부 기록, 로깅 성공률 100%), 해시 체인 검증: (True, -1)\n",
"10번째 기록 변조 후 검증(False, 깨진 위치): (False, 10)\n",
"\n",
"[지휘관 피드백 입력 양식]\n",
"{\"form\": \"c202_commander_feedback\", \"date_time\": \"(입력)\", \"message_id\": \"M0102\", \"delivery_ok\": \"(Y/N)\", \"delay_acceptable\": \"(Y/N)\", \"priority_change_request\": \"(내용)\", \"comment\": \"(내용)\"}\n",
"시스템은 순서를 추천만 하며, 강제 전송·보류의 책임은 조작한 통신관과 승인한 상급자에게 있다.\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 08. 대역폭 90% 급감 및 대량 전문 폭주 상황 모의 전송 스트레스 시뮬레이션\n",
"**소요시간**: 약 3.5시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"대역폭 90% 급감 및 대량 전문 동시 발생 극한 스트레스 시뮬레이션을 수행한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **대역폭 90% 급감과 전문 트래픽 500% 증가, 재밍·링크 두절 상황에서 큐·재전송 시스템의 안전 실패 동작을 검증한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **평시, 트래픽 6배, 대역폭 10%로 급감, 두 조건 동시, 재밍 120초, 대역폭 1% 두절 상황을 가상으로 주입하고 일반 모드와 성능 저하 모드(긴급 전문만 송신, 작은 조각·압축 헤더, 나머지 보관)를 비교 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `tests/c202_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": [
"\"\"\"대역폭 90% 급감, 전문 트래픽 500% 증가(평시 6배), 재밍·링크 두절을 가상으로 주입해 큐·재전송 시스템이 멈추지 않는지(다운 0건), 전문이 사라지지 않는지(유실 0건: 전달+보관=전체), 성능 저하 모드(긴급 전문만 송신·작은 조각·압축 헤더, 나머지 보관)가 긴급 전문 전달을 지키는지 검증하고 처리량·메모리·병목 함수를 측정한다.\"\"\"\n",
"import cProfile\n",
"import io\n",
"import pstats\n",
"import time\n",
"import tracemalloc\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"\n",
"# ---------------------------------------------------------------------------\n",
"# [교육용가상자료] 가상 전술 전문(메시지) 생성기 (실제 부대·통신망·제원 정보 없음)\n",
"# 긴급도 4단계(FLASH/IMMEDIATE/PRIORITY/ROUTINE)와 전문 유형별 크기를 가진 전문을 만든다.\n",
"# ---------------------------------------------------------------------------\n",
"PREC = [\"FLASH\", \"IMMEDIATE\", \"PRIORITY\", \"ROUTINE\"]\n",
"RANK = {p: i for i, p in enumerate(PREC)}\n",
"DEADLINE = {\"FLASH\": 10, \"IMMEDIATE\": 30, \"PRIORITY\": 120, \"ROUTINE\": 600} # 가상 전달 목표 시간(초)\n",
"MSG_TYPES = { # 유형: (긴급도 후보, 확률, 본문 크기 범위 bytes, 발생 비율)\n",
" \"CONTACT\": ([\"FLASH\", \"IMMEDIATE\"], [0.7, 0.3], (120, 300), 0.08),\n",
" \"MEDEVAC\": ([\"FLASH\", \"IMMEDIATE\"], [0.4, 0.6], (150, 350), 0.05),\n",
" \"SITREP\": ([\"IMMEDIATE\", \"PRIORITY\"], [0.2, 0.8], (600, 2500), 0.20),\n",
" \"POSREP\": ([\"PRIORITY\", \"ROUTINE\"], [0.5, 0.5], (60, 120), 0.35),\n",
" \"LOGREQ\": ([\"PRIORITY\", \"ROUTINE\"], [0.4, 0.6], (300, 1200), 0.14),\n",
" \"ADMIN\": ([\"ROUTINE\"], [1.0], (1000, 5000), 0.18),\n",
"}\n",
"\n",
"\n",
"def make_messages(duration=600, rate=0.12, seed=7, t_start=0.0, prefix=\"M\"):\n",
" \"\"\"duration초 동안 평균 rate건/초로 발생하는 가상 전문 목록.\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" n = int(rng.poisson(rate * duration))\n",
" times = np.sort(rng.uniform(t_start, t_start + duration, n))\n",
" types = list(MSG_TYPES)\n",
" msgs = []\n",
" for i, t0 in enumerate(times):\n",
" ty = types[rng.choice(len(types), p=[MSG_TYPES[k][3] for k in types])]\n",
" cands, p, (lo, hi), _ = MSG_TYPES[ty]\n",
" msgs.append({\"id\": f\"{prefix}{i:04d}\", \"t0\": float(t0), \"type\": ty, \"prec\": str(rng.choice(cands, p=p)),\n",
" \"size\": int(rng.integers(lo, hi + 1)), \"src\": f\"U{int(rng.integers(1, 9)):02d}\"})\n",
" return msgs\n",
"\n",
"\n",
"class Link:\n",
" \"\"\"[교육용가상자료] 가상 협대역 무선 링크(기본 2.4kbps). 초 단위 상태(good/bad/jam)에 따라\n",
" 비트오류율이 바뀌고(Gilbert-Elliott 모델), 구간별 대역폭 감소도 줄 수 있다.\"\"\"\n",
" BER = {\"good\": 1e-5, \"bad\": 2e-4, \"jam\": 1.2e-3}\n",
"\n",
" def __init__(self, horizon=4000, bw_bps=2400, seed=1, p_gb=0.02, p_bg=0.15, jam=(), bw_drop=(), rtt=(1.2, 0.4)):\n",
" rng = np.random.default_rng(seed)\n",
" st, s = [], \"good\"\n",
" for _ in range(int(horizon) + 1):\n",
" s = (\"bad\" if rng.random() < p_gb else \"good\") if s == \"good\" else (\"good\" if rng.random() < p_bg else \"bad\")\n",
" st.append(s)\n",
" for a, b in jam:\n",
" for k in range(int(a), min(int(b), len(st))):\n",
" st[k] = \"jam\"\n",
" self.states, self.bw_bps, self.bw_drop, self.rtt_par = st, bw_bps, list(bw_drop), rtt\n",
"\n",
" def state(self, t):\n",
" return self.states[min(int(t), len(self.states) - 1)]\n",
"\n",
" def bw(self, t):\n",
" f = 1.0\n",
" for a, b, fac in self.bw_drop:\n",
" if a <= t < b:\n",
" f = fac\n",
" return self.bw_bps * f\n",
"\n",
" def p_loss(self, nbytes, t):\n",
" return 1 - (1 - self.BER[self.state(t)]) ** (8 * nbytes)\n",
"\n",
" def rtt(self, rng):\n",
" return self.rtt_par[0] + rng.exponential(self.rtt_par[1])\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# ---- 우선순위 큐·재전송 이산 사건 시뮬레이터 공통 함수 ----\n",
"HDR = 32 # 전문 조각(패킷) 헤더 bytes (미션 2의 헤더 압축 적용 시 12 bytes)\n",
"\n",
"\n",
"def policy_key(m, t, policy, aging):\n",
" # [TODO : 학생 실습 8] 대역폭 90% 급감 및 대량 전문 폭주 상황 모의 전송 스트레스 시뮬레이션 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def simulate(msgs, link, policy=\"strict\", mtu=256, preempt=True, rto_mode=\"fixed\", rto_s=4.0, aging=0.0,\n",
" redundancy=False, adapt_mtu=False, hdr=HDR, seed=0, until=None, admit=None):\n",
" \"\"\"전문 목록을 링크로 보내는 이산 사건 시뮬레이션.\n",
" - 전문은 mtu 크기 조각으로 나눠 보낸다. 조각이 손실되면 RTO 후 재전송한다(자료는 버리지 않음).\n",
" - preempt=False면 한 전문의 조각을 끝까지 연속 전송한다(큰 전문이 긴급 전문을 막는 현상 재현).\n",
" - rto_mode='adaptive'면 왕복시간(RTT) 추정값으로 RTO를 계산한다(Jacobson 방식).\n",
" - redundancy=True면 링크 열화 추정 시 FLASH·IMMEDIATE 조각을 2회 연속 전송한다.\n",
" - adapt_mtu=True면 링크 열화 추정 시 조각 크기를 줄인다.\n",
" - admit(m, t)이 False인 전문은 새 조각을 보내지 않고 저장해 둔다(성능 저하 모드).\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" pend = sorted(({**m, \"unsent\": m[\"size\"], \"ready\": [], \"frags\": [], \"done\": None, \"first\": None} for m in msgs), key=lambda m: m[\"t0\"])\n",
" t, i, active, cur = 0.0, 0, [], None\n",
" srtt, rttvar, loss = None, None, 0.0\n",
" st = {\"tx\": 0, \"retx\": 0, \"spurious\": 0, \"bytes\": 0, \"decide_ms\": [], \"flag\": {}, \"log\": []}\n",
" until = until if until is not None else (pend[-1][\"t0\"] if pend else 0) + 6000\n",
" while (i < len(pend) or active) and t < until:\n",
" while i < len(pend) and pend[i][\"t0\"] <= t:\n",
" active.append(pend[i]); i += 1\n",
" td = time.perf_counter()\n",
" for m in active: # ACK가 이미 도착한 재전송 예약은 취소\n",
" m[\"ready\"] = [e for e in m[\"ready\"] if not (e[1][\"ack\"] is not None and e[1][\"ack\"] <= t)]\n",
" ready = [m for m in active if any(r <= t for r, _ in m[\"ready\"]) or (m[\"unsent\"] > 0 and (admit is None or admit(m, t)))]\n",
" if not ready:\n",
" nxt = ([pend[i][\"t0\"]] if i < len(pend) else []) + [r for m in active for r, _ in m[\"ready\"] if r > t]\n",
" t = min(nxt) if nxt else t + 1.0\n",
" continue\n",
" if not preempt and cur in ready and cur[\"unsent\"] > 0:\n",
" m = cur\n",
" else:\n",
" m = min(ready, key=lambda x: policy_key(x, t, policy, aging))\n",
" st[\"decide_ms\"].append((time.perf_counter() - td) * 1000)\n",
" due = [e for e in m[\"ready\"] if e[0] <= t]\n",
" if due:\n",
" e = min(due, key=lambda x: x[0]); m[\"ready\"].remove(e); frag = e[1]\n",
" st[\"retx\"] += 1; st[\"spurious\"] += frag[\"recv\"]\n",
" else:\n",
" size = mtu if not adapt_mtu else (mtu if loss < 0.15 else (mtu // 2 if loss < 0.4 else mtu // 4))\n",
" size = min(m[\"unsent\"], size); m[\"unsent\"] -= size\n",
" frag = {\"b\": size, \"recv\": False, \"ack\": None, \"tries\": 0}\n",
" m[\"frags\"].append(frag)\n",
" cur = m\n",
" m[\"first\"] = t if m[\"first\"] is None else m[\"first\"]\n",
" nb = frag[\"b\"] + hdr\n",
" copies = 2 if (redundancy and RANK[m[\"prec\"]] <= 1 and loss > 0.15) else 1\n",
" for _ in range(copies):\n",
" t += nb * 8 / link.bw(t)\n",
" st[\"tx\"] += 1; st[\"bytes\"] += nb\n",
" ok = rng.random() >= link.p_loss(nb, t)\n",
" st[\"log\"].append((t, nb, ok, m[\"prec\"], m[\"id\"])) # 전송 기록(시각, bytes, 성공, 긴급도, 전문)\n",
" loss = 0.85 * loss + 0.15 * (0.0 if ok else 1.0) # 최근 손실률 추정(EWMA)\n",
" st[\"flag\"][int(t)] = st[\"flag\"].get(int(t), False) or loss > 0.12 # 링크 열화 판정(초 단위)\n",
" if ok and not frag[\"recv\"]:\n",
" frag[\"recv\"], frag[\"rt\"] = True, t\n",
" r = link.rtt(rng); frag[\"ack\"] = t + r\n",
" if srtt is None:\n",
" srtt, rttvar = r, r / 2\n",
" else:\n",
" rttvar = 0.75 * rttvar + 0.25 * abs(srtt - r); srtt = 0.875 * srtt + 0.125 * r\n",
" frag[\"tries\"] += 1\n",
" rto = rto_s if (rto_mode == \"fixed\" or srtt is None) else float(np.clip(srtt + 4 * rttvar, 1.0, 30.0))\n",
" if not frag[\"recv\"] or frag[\"ack\"] > t + rto: # 손실 또는 ACK 지연 -> 재전송 예약\n",
" m[\"ready\"].append((t + rto, frag))\n",
" if m[\"done\"] is None and m[\"unsent\"] == 0 and all(f[\"recv\"] for f in m[\"frags\"]):\n",
" m[\"done\"] = max(f[\"rt\"] for f in m[\"frags\"])\n",
" active = [x for x in active if not (x[\"unsent\"] == 0 and not x[\"ready\"] and all(f[\"recv\"] for f in x[\"frags\"]))]\n",
" res = [{\"id\": m[\"id\"], \"type\": m[\"type\"], \"prec\": m[\"prec\"], \"size\": m[\"size\"], \"t0\": m[\"t0\"], \"done\": m[\"done\"], \"wait\": None if m[\"first\"] is None else m[\"first\"] - m[\"t0\"],\n",
" \"retx\": sum(f[\"tries\"] for f in m[\"frags\"]) - len(m[\"frags\"]),\n",
" \"latency\": None if m[\"done\"] is None else m[\"done\"] - m[\"t0\"]} for m in pend]\n",
" for r in res:\n",
" r[\"on_time\"] = r[\"latency\"] is not None and r[\"latency\"] <= DEADLINE[r[\"prec\"]]\n",
" return res, st\n",
"\n",
"\n",
"def summarize(res):\n",
" \"\"\"긴급도별 기한 내 전달률·지연. 종합 정확도 = 긴급도별 기한 내 전달률의 평균.\"\"\"\n",
" # [TODO : 학생 실습 8] 대역폭 90% 급감 및 대량 전문 폭주 상황 모의 전송 스트레스 시뮬레이션 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"ADV = dict(policy=\"aging\", aging=0.5, rto_mode=\"adaptive\", redundancy=True, adapt_mtu=True)\n",
"NOMINAL_BPS = 2400\n",
"\n",
"\n",
"def degraded_admit(link):\n",
" \"\"\"성능 저하 모드: 대역폭이 평시의 50% 미만이거나 링크가 재밍 상태면 FLASH·IMMEDIATE만 새로 송신하고 나머지는 보관(삭제 없음).\"\"\"\n",
" return lambda m, t: RANK[m[\"prec\"]] <= 1 or (link.bw(t) >= 0.5 * NOMINAL_BPS and link.state(t) != \"jam\")\n",
"\n",
"\n",
"def run_case(msgs, link, fail_safe, seed=0):\n",
" t0 = time.perf_counter()\n",
" try:\n",
" kw = dict(mtu=96, hdr=14, admit=degraded_admit(link)) if fail_safe else {} # 저하모드: 작은 조각 + 압축 헤더(미션 2) + 긴급 전문만 송신\n",
" res, st = simulate(msgs, link, **ADV, seed=seed, until=2000, **kw)\n",
" down = 0\n",
" except Exception as e: # 다운 집계\n",
" print(\"다운:\", e); res, st, down = [], {\"tx\": 0}, 1\n",
" dt = time.perf_counter() - t0\n",
" delivered = sum(r[\"done\"] is not None for r in res)\n",
" kept = len(res) - delivered # 2,000초 시점까지 미전달 = 저장·보관 중(삭제 없음)\n",
" urg = [r[\"on_time\"] for r in res if RANK[r[\"prec\"]] <= 1]\n",
" return {\"전문\": len(msgs), \"전달\": delivered, \"보관\": kept, \"유실\": len(msgs) - delivered - kept, \"다운\": down,\n",
" \"긴급 기한내\": round(float(np.mean(urg)), 3) if urg else np.nan,\n",
" \"FLASH 95%지연(s)\": round(float(np.quantile([r[\"latency\"] for r in res if r[\"prec\"] == \"FLASH\" and r[\"latency\"] is not None], 0.95)), 1),\n",
" \"처리량(건/s)\": round(len(msgs) / dt), \"모드\": \"저하모드\" if fail_safe else \"일반\"}\n",
"\n",
"\n",
"base = make_messages(600, 0.21, seed=7)\n",
"surge = base + [m for k in range(5) for m in make_messages(600, 0.21, seed=20 + k, prefix=f\"S{k}\")] # 평시 + 500%\n",
"CASES = [(\"평시\", base, {}), (\"트래픽 500% 증가\", surge, {}), (\"대역폭 90% 급감(200~500s)\", base, dict(bw_drop=[(200, 500, 0.1)])),\n",
" (\"트래픽 500% + 대역폭 90% 급감\", surge, dict(bw_drop=[(200, 500, 0.1)])),\n",
" (\"트래픽 500% + 재밍 120초\", surge, dict(jam=[(200, 320)])), (\"링크 두절 300초(대역폭 1%)\", base, dict(bw_drop=[(200, 500, 0.01)]))]\n",
"\n",
"rows = []\n",
"for i, (name, msgs, lk) in enumerate(CASES):\n",
" link = Link(4000, seed=1, **lk)\n",
" for fs in (False, True):\n",
" rows.append({\"상황\": name, **run_case(msgs, link, fs, seed=i)})\n",
"df = pd.DataFrame(rows)\n",
"print(df.drop(columns=[\"처리량(건/s)\"]).to_string(index=False))\n",
"print(f\"\\n시스템 다운 {df['다운'].sum()}건, 유실 {df['유실'].sum()}건(전달+보관=전체), 최소 처리량 {df['처리량(건/s)'].min():,}건/s\")\n",
"hard = df[df[\"상황\"] == \"트래픽 500% + 대역폭 90% 급감\"]\n",
"print(\"링크 두절(대역폭 1%) 구간에는 어떤 방식으로도 기한을 지킬 수 없다 -> 통신관에게 대체 수단(중계·전령 등) 검토를 요청하는 것이 안전 실패 동작이다.\")\n",
"print(f\"최악 상황 긴급 전문 기한 내 전달률: 일반 {hard.iloc[0]['긴급 기한내']:.1%} -> 성능 저하 모드 {hard.iloc[1]['긴급 기한내']:.1%}\")\n",
"need = sum(m[\"size\"] + 14 for m in surge if RANK[m[\"prec\"]] <= 1 and 200 <= m[\"t0\"] < 500) / 300\n",
"print(f\"원인: 급감 구간 긴급 전문만 초당 {need:.0f}B가 필요한데 링크는 초당 {NOMINAL_BPS * 0.1 / 8:.0f}B -> 스케줄링으로 해결 불가, 전문 간소화·발송 통제를 지휘계통에 건의해야 한다.\")\n",
"\n",
"# 메모리 누수 점검: 같은 시나리오를 3회 반복해도 메모리가 계속 늘지 않는가\n",
"tracemalloc.start()\n",
"mem = []\n",
"peak = 0\n",
"for k in range(3):\n",
" simulate(base, Link(4000, seed=1), **ADV, until=2000)\n",
" mem.append(tracemalloc.get_traced_memory()[0] / 1e6); peak = max(peak, tracemalloc.get_traced_memory()[1] / 1e6)\n",
"tracemalloc.stop()\n",
"print(f\"평시 시나리오 실행 중 최대 메모리 {peak:.1f} MB\")\n",
"print(\"반복 실행 후 잔류 메모리(MB):\", [round(m, 2) for m in mem], \"-> 누수 의심\" if mem[-1] > mem[0] * 1.5 + 1 else \"-> 누수 징후 없음\")\n",
"\n",
"# 병목 함수 프로파일링\n",
"pr = cProfile.Profile(); pr.enable()\n",
"simulate(surge, Link(4000, seed=1, bw_drop=[(200, 500, 0.1)]), **ADV, until=2000)\n",
"pr.disable()\n",
"s = io.StringIO(); pstats.Stats(pr, stream=s).sort_stats(\"tottime\").print_stats(4)\n",
"top = [ln.split() for ln in s.getvalue().splitlines() if ln.strip()[:1].isdigit() and \"(\" in ln and \"function calls\" not in ln][:3]\n",
"print(\"[병목 상위 3 함수: 호출 수 / 자체 소요(s) / 함수]\")\n",
"for f in top:\n",
" print(f\" {f[0]:>9} / {f[1]} / {' '.join(f[5:]).split('/')[-1][-45:]}\")\n",
"print(\"개선 방향: 매 결정마다 전체 대기 전문을 훑는 부분을 긴급도별 힙(heapq)과 재전송 타이머 힙으로 바꾸면 대기 전문 수에 덜 민감해진다.\")\n",
"\n",
"plt.figure(figsize=(9, 4))\n",
"x = np.arange(len(CASES))\n",
"plt.bar(x - 0.2, df[df[\"모드\"] == \"일반\"][\"긴급 기한내\"], 0.4, label=\"normal\")\n",
"plt.bar(x + 0.2, df[df[\"모드\"] == \"저하모드\"][\"긴급 기한내\"], 0.4, label=\"degraded mode (urgent only)\")\n",
"plt.xticks(x, [f\"case{i + 1}\" for i in x]); plt.ylabel(\"urgent on-time ratio\"); plt.ylim(0, 1.05); plt.legend()\n",
"plt.title(\"Stress test: urgent message delivery (synthetic)\"); plt.tight_layout(); plt.show()\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 08 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
" 상황 전문 전달 보관 유실 다운 긴급 기한내 FLASH 95%지연(s) 모드\n",
" 평시 130 130 0 0 0 1.000 2.0 일반\n",
" 평시 130 130 0 0 0 1.000 1.7 저하모드\n",
" 트래픽 500% 증가 758 567 191 0 0 0.991 2.0 일반\n",
" 트래픽 500% 증가 758 579 179 0 0 0.991 2.2 저하모드\n",
" 대역폭 90% 급감(200~500s) 130 130 0 0 0 0.500 23.9 일반\n",
" 대역폭 90% 급감(200~500s) 130 130 0 0 0 0.667 18.7 저하모드\n",
"트래픽 500% + 대역폭 90% 급감 758 519 239 0 0 0.389 93.4 일반\n",
"트래픽 500% + 대역폭 90% 급감 758 528 230 0 0 0.398 97.8 저하모드\n",
" 트래픽 500% + 재밍 120초 758 547 211 0 0 0.770 14.8 일반\n",
" 트래픽 500% + 재밍 120초 758 563 195 0 0 0.858 6.7 저하모드\n",
" 링크 두절 300초(대역폭 1%) 130 130 0 0 0 0.389 259.8 일반\n",
" 링크 두절 300초(대역폭 1%) 130 130 0 0 0 0.500 175.0 저하모드\n",
"\n",
"시스템 다운 0건, 유실 0건(전달+보관=전체), 최소 처리량 207건/s\n",
"링크 두절(대역폭 1%) 구간에는 어떤 방식으로도 기한을 지킬 수 없다 -> 통신관에게 대체 수단(중계·전령 등) 검토를 요청하는 것이 안전 실패 동작이다.\n",
"최악 상황 긴급 전문 기한 내 전달률: 일반 38.9% -> 성능 저하 모드 39.8%\n",
"원인: 급감 구간 긴급 전문만 초당 91B가 필요한데 링크는 초당 30B -> 스케줄링으로 해결 불가, 전문 간소화·발송 통제를 지휘계통에 건의해야 한다.\n",
"평시 시나리오 실행 중 최대 메모리 0.4 MB\n",
"반복 실행 후 잔류 메모리(MB): [0.01, 0.01, 0.01] -> 누수 징후 없음\n",
"[병목 상위 3 함수: 호출 수 / 자체 소요(s) / 함수]\n",
" 807785 / 0.476 / m08.py:80(policy_key)\n",
" 1 / 0.430 / m08.py:89(simulate)\n",
" 2132 / 0.408 / m08.py:110(<listcomp>)\n",
"개선 방향: 매 결정마다 전체 대기 전문을 훑는 부분을 긴급도별 힙(heapq)과 재전송 타이머 힙으로 바꾸면 대기 전문 수에 덜 민감해진다.\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 09. AAR 기반 중요 전문 전송 지연 사례 분석 및 큐 우선순위 가중치 재조정\n",
"**소요시간**: 약 3.0시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"AAR을 통해 중요 전문 전송 지연 사례를 분석하고 큐 스케줄링 가중치를 재조정한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **AAR로 중요 전문의 전송 지연 사례를 원인별로 분석하고 큐 스케줄링 가중치와 재전송 파라미터를 재조정한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **초기 운용 설정(전문 단위 전송, 고정 재전송 타이머 6초, 조각 512bytes)으로 재밍·대역폭 감소가 있는 훈련을 모의하고 지연된 중요 전문을 대기열 적체·재전송 대기·링크 열화로 분류한 AAR 일지 작성하여 원인을 분석하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `docs/c202_aar_report.md 및 configs/c202_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": [
"\"\"\"AAR(사후검토) 일지 형식으로 중요 전문(FLASH·IMMEDIATE·PRIORITY)의 전송 지연 사례를 원인별로 분류하고, 큐 스케줄링 가중치(선점·aging)·조각 크기·재전송 타이머를 격자 탐색으로 재조정한 뒤 다른 운용 조건에서 회귀 테스트로 개선 전후를 비교한다.\"\"\"\n",
"import time\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"\n",
"# ---------------------------------------------------------------------------\n",
"# [교육용가상자료] 가상 전술 전문(메시지) 생성기 (실제 부대·통신망·제원 정보 없음)\n",
"# 긴급도 4단계(FLASH/IMMEDIATE/PRIORITY/ROUTINE)와 전문 유형별 크기를 가진 전문을 만든다.\n",
"# ---------------------------------------------------------------------------\n",
"PREC = [\"FLASH\", \"IMMEDIATE\", \"PRIORITY\", \"ROUTINE\"]\n",
"RANK = {p: i for i, p in enumerate(PREC)}\n",
"DEADLINE = {\"FLASH\": 10, \"IMMEDIATE\": 30, \"PRIORITY\": 120, \"ROUTINE\": 600} # 가상 전달 목표 시간(초)\n",
"MSG_TYPES = { # 유형: (긴급도 후보, 확률, 본문 크기 범위 bytes, 발생 비율)\n",
" \"CONTACT\": ([\"FLASH\", \"IMMEDIATE\"], [0.7, 0.3], (120, 300), 0.08),\n",
" \"MEDEVAC\": ([\"FLASH\", \"IMMEDIATE\"], [0.4, 0.6], (150, 350), 0.05),\n",
" \"SITREP\": ([\"IMMEDIATE\", \"PRIORITY\"], [0.2, 0.8], (600, 2500), 0.20),\n",
" \"POSREP\": ([\"PRIORITY\", \"ROUTINE\"], [0.5, 0.5], (60, 120), 0.35),\n",
" \"LOGREQ\": ([\"PRIORITY\", \"ROUTINE\"], [0.4, 0.6], (300, 1200), 0.14),\n",
" \"ADMIN\": ([\"ROUTINE\"], [1.0], (1000, 5000), 0.18),\n",
"}\n",
"\n",
"\n",
"def make_messages(duration=600, rate=0.12, seed=7, t_start=0.0, prefix=\"M\"):\n",
" \"\"\"duration초 동안 평균 rate건/초로 발생하는 가상 전문 목록.\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" n = int(rng.poisson(rate * duration))\n",
" times = np.sort(rng.uniform(t_start, t_start + duration, n))\n",
" types = list(MSG_TYPES)\n",
" msgs = []\n",
" for i, t0 in enumerate(times):\n",
" ty = types[rng.choice(len(types), p=[MSG_TYPES[k][3] for k in types])]\n",
" cands, p, (lo, hi), _ = MSG_TYPES[ty]\n",
" msgs.append({\"id\": f\"{prefix}{i:04d}\", \"t0\": float(t0), \"type\": ty, \"prec\": str(rng.choice(cands, p=p)),\n",
" \"size\": int(rng.integers(lo, hi + 1)), \"src\": f\"U{int(rng.integers(1, 9)):02d}\"})\n",
" return msgs\n",
"\n",
"\n",
"class Link:\n",
" \"\"\"[교육용가상자료] 가상 협대역 무선 링크(기본 2.4kbps). 초 단위 상태(good/bad/jam)에 따라\n",
" 비트오류율이 바뀌고(Gilbert-Elliott 모델), 구간별 대역폭 감소도 줄 수 있다.\"\"\"\n",
" BER = {\"good\": 1e-5, \"bad\": 2e-4, \"jam\": 1.2e-3}\n",
"\n",
" def __init__(self, horizon=4000, bw_bps=2400, seed=1, p_gb=0.02, p_bg=0.15, jam=(), bw_drop=(), rtt=(1.2, 0.4)):\n",
" rng = np.random.default_rng(seed)\n",
" st, s = [], \"good\"\n",
" for _ in range(int(horizon) + 1):\n",
" s = (\"bad\" if rng.random() < p_gb else \"good\") if s == \"good\" else (\"good\" if rng.random() < p_bg else \"bad\")\n",
" st.append(s)\n",
" for a, b in jam:\n",
" for k in range(int(a), min(int(b), len(st))):\n",
" st[k] = \"jam\"\n",
" self.states, self.bw_bps, self.bw_drop, self.rtt_par = st, bw_bps, list(bw_drop), rtt\n",
"\n",
" def state(self, t):\n",
" return self.states[min(int(t), len(self.states) - 1)]\n",
"\n",
" def bw(self, t):\n",
" f = 1.0\n",
" for a, b, fac in self.bw_drop:\n",
" if a <= t < b:\n",
" f = fac\n",
" return self.bw_bps * f\n",
"\n",
" def p_loss(self, nbytes, t):\n",
" return 1 - (1 - self.BER[self.state(t)]) ** (8 * nbytes)\n",
"\n",
" def rtt(self, rng):\n",
" return self.rtt_par[0] + rng.exponential(self.rtt_par[1])\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# ---- 우선순위 큐·재전송 이산 사건 시뮬레이터 공통 함수 ----\n",
"HDR = 32 # 전문 조각(패킷) 헤더 bytes (미션 2의 헤더 압축 적용 시 12 bytes)\n",
"\n",
"\n",
"def policy_key(m, t, policy, aging):\n",
" # [TODO : 학생 실습 9] AAR 기반 중요 전문 전송 지연 사례 분석 및 큐 우선순위 가중치 재조정 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def simulate(msgs, link, policy=\"strict\", mtu=256, preempt=True, rto_mode=\"fixed\", rto_s=4.0, aging=0.0,\n",
" redundancy=False, adapt_mtu=False, hdr=HDR, seed=0, until=None, admit=None):\n",
" \"\"\"전문 목록을 링크로 보내는 이산 사건 시뮬레이션.\n",
" - 전문은 mtu 크기 조각으로 나눠 보낸다. 조각이 손실되면 RTO 후 재전송한다(자료는 버리지 않음).\n",
" - preempt=False면 한 전문의 조각을 끝까지 연속 전송한다(큰 전문이 긴급 전문을 막는 현상 재현).\n",
" - rto_mode='adaptive'면 왕복시간(RTT) 추정값으로 RTO를 계산한다(Jacobson 방식).\n",
" - redundancy=True면 링크 열화 추정 시 FLASH·IMMEDIATE 조각을 2회 연속 전송한다.\n",
" - adapt_mtu=True면 링크 열화 추정 시 조각 크기를 줄인다.\n",
" - admit(m, t)이 False인 전문은 새 조각을 보내지 않고 저장해 둔다(성능 저하 모드).\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" pend = sorted(({**m, \"unsent\": m[\"size\"], \"ready\": [], \"frags\": [], \"done\": None, \"first\": None} for m in msgs), key=lambda m: m[\"t0\"])\n",
" t, i, active, cur = 0.0, 0, [], None\n",
" srtt, rttvar, loss = None, None, 0.0\n",
" st = {\"tx\": 0, \"retx\": 0, \"spurious\": 0, \"bytes\": 0, \"decide_ms\": [], \"flag\": {}, \"log\": []}\n",
" until = until if until is not None else (pend[-1][\"t0\"] if pend else 0) + 6000\n",
" while (i < len(pend) or active) and t < until:\n",
" while i < len(pend) and pend[i][\"t0\"] <= t:\n",
" active.append(pend[i]); i += 1\n",
" td = time.perf_counter()\n",
" for m in active: # ACK가 이미 도착한 재전송 예약은 취소\n",
" m[\"ready\"] = [e for e in m[\"ready\"] if not (e[1][\"ack\"] is not None and e[1][\"ack\"] <= t)]\n",
" ready = [m for m in active if any(r <= t for r, _ in m[\"ready\"]) or (m[\"unsent\"] > 0 and (admit is None or admit(m, t)))]\n",
" if not ready:\n",
" nxt = ([pend[i][\"t0\"]] if i < len(pend) else []) + [r for m in active for r, _ in m[\"ready\"] if r > t]\n",
" t = min(nxt) if nxt else t + 1.0\n",
" continue\n",
" if not preempt and cur in ready and cur[\"unsent\"] > 0:\n",
" m = cur\n",
" else:\n",
" m = min(ready, key=lambda x: policy_key(x, t, policy, aging))\n",
" st[\"decide_ms\"].append((time.perf_counter() - td) * 1000)\n",
" due = [e for e in m[\"ready\"] if e[0] <= t]\n",
" if due:\n",
" e = min(due, key=lambda x: x[0]); m[\"ready\"].remove(e); frag = e[1]\n",
" st[\"retx\"] += 1; st[\"spurious\"] += frag[\"recv\"]\n",
" else:\n",
" size = mtu if not adapt_mtu else (mtu if loss < 0.15 else (mtu // 2 if loss < 0.4 else mtu // 4))\n",
" size = min(m[\"unsent\"], size); m[\"unsent\"] -= size\n",
" frag = {\"b\": size, \"recv\": False, \"ack\": None, \"tries\": 0}\n",
" m[\"frags\"].append(frag)\n",
" cur = m\n",
" m[\"first\"] = t if m[\"first\"] is None else m[\"first\"]\n",
" nb = frag[\"b\"] + hdr\n",
" copies = 2 if (redundancy and RANK[m[\"prec\"]] <= 1 and loss > 0.15) else 1\n",
" for _ in range(copies):\n",
" t += nb * 8 / link.bw(t)\n",
" st[\"tx\"] += 1; st[\"bytes\"] += nb\n",
" ok = rng.random() >= link.p_loss(nb, t)\n",
" st[\"log\"].append((t, nb, ok, m[\"prec\"], m[\"id\"])) # 전송 기록(시각, bytes, 성공, 긴급도, 전문)\n",
" loss = 0.85 * loss + 0.15 * (0.0 if ok else 1.0) # 최근 손실률 추정(EWMA)\n",
" st[\"flag\"][int(t)] = st[\"flag\"].get(int(t), False) or loss > 0.12 # 링크 열화 판정(초 단위)\n",
" if ok and not frag[\"recv\"]:\n",
" frag[\"recv\"], frag[\"rt\"] = True, t\n",
" r = link.rtt(rng); frag[\"ack\"] = t + r\n",
" if srtt is None:\n",
" srtt, rttvar = r, r / 2\n",
" else:\n",
" rttvar = 0.75 * rttvar + 0.25 * abs(srtt - r); srtt = 0.875 * srtt + 0.125 * r\n",
" frag[\"tries\"] += 1\n",
" rto = rto_s if (rto_mode == \"fixed\" or srtt is None) else float(np.clip(srtt + 4 * rttvar, 1.0, 30.0))\n",
" if not frag[\"recv\"] or frag[\"ack\"] > t + rto: # 손실 또는 ACK 지연 -> 재전송 예약\n",
" m[\"ready\"].append((t + rto, frag))\n",
" if m[\"done\"] is None and m[\"unsent\"] == 0 and all(f[\"recv\"] for f in m[\"frags\"]):\n",
" m[\"done\"] = max(f[\"rt\"] for f in m[\"frags\"])\n",
" active = [x for x in active if not (x[\"unsent\"] == 0 and not x[\"ready\"] and all(f[\"recv\"] for f in x[\"frags\"]))]\n",
" res = [{\"id\": m[\"id\"], \"type\": m[\"type\"], \"prec\": m[\"prec\"], \"size\": m[\"size\"], \"t0\": m[\"t0\"], \"done\": m[\"done\"], \"wait\": None if m[\"first\"] is None else m[\"first\"] - m[\"t0\"],\n",
" \"retx\": sum(f[\"tries\"] for f in m[\"frags\"]) - len(m[\"frags\"]),\n",
" \"latency\": None if m[\"done\"] is None else m[\"done\"] - m[\"t0\"]} for m in pend]\n",
" for r in res:\n",
" r[\"on_time\"] = r[\"latency\"] is not None and r[\"latency\"] <= DEADLINE[r[\"prec\"]]\n",
" return res, st\n",
"\n",
"\n",
"def summarize(res):\n",
" \"\"\"긴급도별 기한 내 전달률·지연. 종합 정확도 = 긴급도별 기한 내 전달률의 평균.\"\"\"\n",
" # [TODO : 학생 실습 9] AAR 기반 중요 전문 전송 지연 사례 분석 및 큐 우선순위 가중치 재조정 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"INITIAL = dict(policy=\"strict\", preempt=False, rto_mode=\"fixed\", rto_s=6.0, mtu=512, aging=0.0) # 초기 운용 설정\n",
"\n",
"\n",
"def aar_scenario(seed):\n",
" \"\"\"[교육용가상자료] 훈련 1회분: 재밍 60초 + 대역폭 50% 감소 200초 + 중계 지연이 큰 망.\"\"\"\n",
" msgs = make_messages(1200, 0.21, seed=seed)\n",
" link = Link(12000, seed=seed + 100, jam=[(300, 360)], bw_drop=[(600, 800, 0.5)], rtt=(2.0, 0.8))\n",
" return msgs, link\n",
"\n",
"\n",
"def aar_log(res, link):\n",
" rows = []\n",
" for r in res:\n",
" if RANK[r[\"prec\"]] <= 2 and not r[\"on_time\"]:\n",
" bad = any(link.state(k) != \"good\" or link.bw(k) < link.bw_bps for k in range(int(r[\"t0\"]), int(r[\"t0\"] + (r[\"latency\"] or 60)) + 1))\n",
" if r[\"retx\"] > 0 and bad:\n",
" cause = \"링크 열화 중 재전송\"\n",
" elif r[\"retx\"] > 0:\n",
" cause = \"재전송 타이머(RTO) 대기\"\n",
" elif r[\"wait\"] is not None and r[\"wait\"] > DEADLINE[r[\"prec\"]] / 2:\n",
" cause = \"대기열 적체·하위 전문 점유\"\n",
" else:\n",
" cause = \"대용량 전문 분할 전송 지연\"\n",
" rows.append({\"전문\": r[\"id\"], \"긴급도\": r[\"prec\"], \"크기\": r[\"size\"], \"지연(s)\": round(r[\"latency\"] or -1, 1),\n",
" \"첫 송신 대기(s)\": round(r[\"wait\"] or -1, 1), \"재전송\": r[\"retx\"], \"원인\": cause})\n",
" return pd.DataFrame(rows, columns=[\"전문\", \"긴급도\", \"크기\", \"지연(s)\", \"첫 송신 대기(s)\", \"재전송\", \"원인\"])\n",
"\n",
"\n",
"# 1) 무엇이 일어났나: 초기 설정으로 훈련 운용\n",
"msgs, link = aar_scenario(7)\n",
"before_res, _ = simulate(msgs, link, **INITIAL)\n",
"tab0, acc0 = summarize(before_res)\n",
"log0 = aar_log(before_res, link)\n",
"print(\"[1. 무엇이 일어났나] 초기 설정 종합 정확도(긴급도별 기한 내 전달률 평균):\", acc0)\n",
"print(tab0[[\"긴급도\", \"건수\", \"기한내 전달률\", \"95% 지연(s)\"]].to_string(index=False))\n",
"print(\"\\n[2. 왜 일어났나] 중요 전문 지연 원인 분류\")\n",
"print(log0.groupby([\"긴급도\", \"원인\"]).size().rename(\"건수\").to_string())\n",
"\n",
"# 2) 무엇을 바꾸나: 격자 탐색(제약: FLASH 기한 내 전달률 95% 이상)\n",
"grid = []\n",
"for preempt in (False, True):\n",
" for rto in (\"fixed\", \"adaptive\"):\n",
" for mtu in (128, 256, 512):\n",
" for aging in (0.0, 0.5, 1.0):\n",
" cfg = dict(policy=\"aging\" if aging else \"strict\", preempt=preempt, rto_mode=rto, rto_s=6.0, mtu=mtu, aging=aging)\n",
" tab, acc = summarize(simulate(msgs, link, **cfg)[0])\n",
" grid.append({**cfg, \"FLASH\": tab.loc[0, \"기한내 전달률\"], \"종합정확도\": acc})\n",
"g = pd.DataFrame(grid)\n",
"ok = g[g[\"FLASH\"] >= 0.95].sort_values([\"종합정확도\", \"FLASH\"], ascending=False)\n",
"best = ok.iloc[0]\n",
"TUNED = dict(policy=best[\"policy\"], preempt=bool(best[\"preempt\"]), rto_mode=best[\"rto_mode\"], rto_s=6.0, mtu=int(best[\"mtu\"]), aging=float(best[\"aging\"]))\n",
"print(f\"\\n[3. 무엇을 바꾸나] 격자 {len(g)}개 중 제약 충족 {len(ok)}개, 상위 3개\")\n",
"print(ok.head(3)[[\"preempt\", \"rto_mode\", \"mtu\", \"aging\", \"FLASH\", \"종합정확도\"]].to_string(index=False))\n",
"\n",
"after_res, _ = simulate(msgs, link, **TUNED)\n",
"tab1, acc1 = summarize(after_res)\n",
"log1 = aar_log(after_res, link)\n",
"print(f\"\\n종합 정확도 {acc0:.3f} -> {acc1:.3f} ({(acc1 - acc0) * 100:+.1f}%p, 목표 +10%p: {'달성' if acc1 - acc0 >= 0.10 else '미달'})\")\n",
"print(f\"FLASH 기한 내 전달률 {tab0.loc[0, '기한내 전달률']:.1%} -> {tab1.loc[0, '기한내 전달률']:.1%}, 중요 전문 지연 {len(log0)}건 -> {len(log1)}건\")\n",
"repeat = sorted(set(log0[\"전문\"]) & set(log1[\"전문\"]))\n",
"print(f\"재보정 후에도 반복된 지연 전문: {repeat if repeat else '없음(재발 0건)'}\")\n",
"for _, r in log1[log1[\"전문\"].isin(repeat)].iterrows():\n",
" print(f\" {r['전문']} {r['긴급도']} 지연 {r['지연(s)']}s, 재전송 {r['재전송']}회 -> 원인: {r['원인']} (스케줄링으로 줄이기 어려운 유형, 링크 열화 시 중복 전송 검토)\")\n",
"\n",
"# 3) 회귀 테스트: 다른 훈련 3회(다른 전문·링크)에서도 개선이 유지되는가\n",
"print(\"\\n[회귀 테스트] 훈련 초기 재보정 FLASH(재보정)\")\n",
"reg = []\n",
"for sd in (11, 12, 13):\n",
" m2, l2 = aar_scenario(sd)\n",
" a0 = summarize(simulate(m2, l2, **INITIAL)[0])[1]\n",
" t1, a1 = summarize(simulate(m2, l2, **TUNED)[0])\n",
" reg.append(a1 >= a0)\n",
" print(f\" #{sd} {a0:.3f} {a1:.3f} {t1.loc[0, '기한내 전달률']:.3f}\")\n",
"print(f\"회귀 테스트 {sum(reg)}/{len(reg)} 통과(재보정 후 종합 정확도가 떨어지지 않음)\")\n",
"\n",
"yaml = (f\"scheduler:\\n policy: {TUNED['policy']}\\n preempt_fragment: {str(TUNED['preempt']).lower()}\\n aging: {TUNED['aging']}\\n\"\n",
" f\"fragment_mtu_bytes: {TUNED['mtu']}\\nretransmission:\\n rto_mode: {TUNED['rto_mode']}\\n fixed_rto_s: 6.0\\nconstraint:\\n min_flash_on_time: 0.95\")\n",
"print(\"\\n[재보정 파라미터(c202_tuned_params.yaml 내용)]\\n\" + yaml)\n",
"print(\"지속 개선: 훈련마다 AAR 일지를 누적해 같은 격자 탐색을 다시 수행하고, 변경 전 회귀 테스트 통과를 배포 조건으로 둔다.\")\n",
"\n",
"plt.figure(figsize=(7, 4))\n",
"for pre, c in ((False, \"tab:gray\"), (True, \"tab:blue\")):\n",
" s = g[g[\"preempt\"] == pre]\n",
" plt.scatter(s[\"FLASH\"], s[\"종합정확도\"], c=c, label=f\"fragment preempt={pre}\")\n",
"plt.scatter([best[\"FLASH\"]], [best[\"종합정확도\"]], s=200, facecolors=\"none\", edgecolors=\"red\", label=\"selected\")\n",
"plt.axvline(0.95, ls=\"--\", c=\"gray\"); plt.xlabel(\"FLASH on-time ratio\"); plt.ylabel(\"overall accuracy (macro on-time)\")\n",
"plt.legend(); plt.title(\"AAR re-tuning grid (synthetic)\"); plt.tight_layout(); plt.show()\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 09 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"[1. 무엇이 일어났나] 초기 설정 종합 정확도(긴급도별 기한 내 전달률 평균): 0.921\n",
" 긴급도 건수 기한내 전달률 95% 지연(s)\n",
" FLASH 25 0.720 19.4\n",
"IMMEDIATE 27 0.963 25.6\n",
" PRIORITY 99 1.000 52.0\n",
" ROUTINE 107 1.000 232.4\n",
"\n",
"[2. 왜 일어났나] 중요 전문 지연 원인 분류\n",
"긴급도 원인 \n",
"FLASH 대기열 적체·하위 전문 점유 4\n",
" 링크 열화 중 재전송 3\n",
"IMMEDIATE 링크 열화 중 재전송 1\n",
"\n",
"[3. 무엇을 바꾸나] 격자 36개 중 제약 충족 18개, 상위 3개\n",
" preempt rto_mode mtu aging FLASH 종합정확도\n",
" True fixed 128 0.0 0.96 0.99\n",
" True fixed 128 0.5 0.96 0.99\n",
" True fixed 128 1.0 0.96 0.99\n",
"\n",
"종합 정확도 0.921 -> 0.990 (+6.9%p, 목표 +10%p: 미달)\n",
"FLASH 기한 내 전달률 72.0% -> 96.0%, 중요 전문 지연 8건 -> 1건\n",
"재보정 후에도 반복된 지연 전문: ['M0070']\n",
" M0070 FLASH 지연 41.1s, 재전송 10회 -> 원인: 링크 열화 중 재전송 (스케줄링으로 줄이기 어려운 유형, 링크 열화 시 중복 전송 검토)\n",
"\n",
"[회귀 테스트] 훈련 초기 재보정 FLASH(재보정)\n",
" #11 0.936 1.000 1.000\n",
" #12 0.953 0.988 0.952\n",
" #13 0.943 0.951 0.870\n",
"회귀 테스트 3/3 통과(재보정 후 종합 정확도가 떨어지지 않음)\n",
"\n",
"[재보정 파라미터(c202_tuned_params.yaml 내용)]\n",
"scheduler:\n",
" policy: strict\n",
" preempt_fragment: true\n",
" aging: 0.0\n",
"fragment_mtu_bytes: 128\n",
"retransmission:\n",
" rto_mode: fixed\n",
" fixed_rto_s: 6.0\n",
"constraint:\n",
"... (이하 생략)\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 10. 대역폭 제한 환경 메시지 큐·재전송 시스템 통합 및 전술통신 브리핑 패키지 완성\n",
"**소요시간**: 약 3.5시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"대역폭 제한 시 메시지 우선순위 큐·재전송 시스템 통합 및 배포 가이드를 완성한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **패킷 수신부터 전처리, 큐·재전송, 링크 등급, 통신관 요청 대기열, 관제 화면까지 통합하고 전술통신 브리핑 패키지를 완성한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **재조정 파라미터로 전체 과정을 한 번에 실행하는 통합 코드를 완성하고 손상·미등록 패킷이 격리되는지 확인하고 유효성을 점검하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 2개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `src/main_c202.py, README.md, docs/c202_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": [
"\"\"\"원시 전문 패킷 파싱·검증·압축부터 우선순위 큐, 지능형 재전송, 링크 신뢰도 등급, 통신관 HITL 요청 대기열, 관제 화면, AAR 재보정 파라미터 적용까지 전체 파이프라인을 한 파일로 통합 실행하고, 함수 커버리지를 포함한 자체 점검 결과와 전술통신 브리핑 요약을 출력한다.\"\"\"\n",
"import hashlib\n",
"import inspect\n",
"import json\n",
"import sys\n",
"import time\n",
"import zlib\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"\n",
"# ---------------------------------------------------------------------------\n",
"# [교육용가상자료] 가상 전술 전문(메시지) 생성기 (실제 부대·통신망·제원 정보 없음)\n",
"# 긴급도 4단계(FLASH/IMMEDIATE/PRIORITY/ROUTINE)와 전문 유형별 크기를 가진 전문을 만든다.\n",
"# ---------------------------------------------------------------------------\n",
"PREC = [\"FLASH\", \"IMMEDIATE\", \"PRIORITY\", \"ROUTINE\"]\n",
"RANK = {p: i for i, p in enumerate(PREC)}\n",
"DEADLINE = {\"FLASH\": 10, \"IMMEDIATE\": 30, \"PRIORITY\": 120, \"ROUTINE\": 600} # 가상 전달 목표 시간(초)\n",
"MSG_TYPES = { # 유형: (긴급도 후보, 확률, 본문 크기 범위 bytes, 발생 비율)\n",
" \"CONTACT\": ([\"FLASH\", \"IMMEDIATE\"], [0.7, 0.3], (120, 300), 0.08),\n",
" \"MEDEVAC\": ([\"FLASH\", \"IMMEDIATE\"], [0.4, 0.6], (150, 350), 0.05),\n",
" \"SITREP\": ([\"IMMEDIATE\", \"PRIORITY\"], [0.2, 0.8], (600, 2500), 0.20),\n",
" \"POSREP\": ([\"PRIORITY\", \"ROUTINE\"], [0.5, 0.5], (60, 120), 0.35),\n",
" \"LOGREQ\": ([\"PRIORITY\", \"ROUTINE\"], [0.4, 0.6], (300, 1200), 0.14),\n",
" \"ADMIN\": ([\"ROUTINE\"], [1.0], (1000, 5000), 0.18),\n",
"}\n",
"\n",
"\n",
"def make_messages(duration=600, rate=0.12, seed=7, t_start=0.0, prefix=\"M\"):\n",
" \"\"\"duration초 동안 평균 rate건/초로 발생하는 가상 전문 목록.\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" n = int(rng.poisson(rate * duration))\n",
" times = np.sort(rng.uniform(t_start, t_start + duration, n))\n",
" types = list(MSG_TYPES)\n",
" msgs = []\n",
" for i, t0 in enumerate(times):\n",
" ty = types[rng.choice(len(types), p=[MSG_TYPES[k][3] for k in types])]\n",
" cands, p, (lo, hi), _ = MSG_TYPES[ty]\n",
" msgs.append({\"id\": f\"{prefix}{i:04d}\", \"t0\": float(t0), \"type\": ty, \"prec\": str(rng.choice(cands, p=p)),\n",
" \"size\": int(rng.integers(lo, hi + 1)), \"src\": f\"U{int(rng.integers(1, 9)):02d}\"})\n",
" return msgs\n",
"\n",
"\n",
"class Link:\n",
" \"\"\"[교육용가상자료] 가상 협대역 무선 링크(기본 2.4kbps). 초 단위 상태(good/bad/jam)에 따라\n",
" 비트오류율이 바뀌고(Gilbert-Elliott 모델), 구간별 대역폭 감소도 줄 수 있다.\"\"\"\n",
" BER = {\"good\": 1e-5, \"bad\": 2e-4, \"jam\": 1.2e-3}\n",
"\n",
" def __init__(self, horizon=4000, bw_bps=2400, seed=1, p_gb=0.02, p_bg=0.15, jam=(), bw_drop=(), rtt=(1.2, 0.4)):\n",
" rng = np.random.default_rng(seed)\n",
" st, s = [], \"good\"\n",
" for _ in range(int(horizon) + 1):\n",
" s = (\"bad\" if rng.random() < p_gb else \"good\") if s == \"good\" else (\"good\" if rng.random() < p_bg else \"bad\")\n",
" st.append(s)\n",
" for a, b in jam:\n",
" for k in range(int(a), min(int(b), len(st))):\n",
" st[k] = \"jam\"\n",
" self.states, self.bw_bps, self.bw_drop, self.rtt_par = st, bw_bps, list(bw_drop), rtt\n",
"\n",
" def state(self, t):\n",
" return self.states[min(int(t), len(self.states) - 1)]\n",
"\n",
" def bw(self, t):\n",
" f = 1.0\n",
" for a, b, fac in self.bw_drop:\n",
" if a <= t < b:\n",
" f = fac\n",
" return self.bw_bps * f\n",
"\n",
" def p_loss(self, nbytes, t):\n",
" return 1 - (1 - self.BER[self.state(t)]) ** (8 * nbytes)\n",
"\n",
" def rtt(self, rng):\n",
" return self.rtt_par[0] + rng.exponential(self.rtt_par[1])\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# ---- 우선순위 큐·재전송 이산 사건 시뮬레이터 공통 함수 ----\n",
"HDR = 32 # 전문 조각(패킷) 헤더 bytes (미션 2의 헤더 압축 적용 시 12 bytes)\n",
"\n",
"\n",
"def policy_key(m, t, policy, aging):\n",
" # [TODO : 학생 실습 10] 대역폭 제한 환경 메시지 큐·재전송 시스템 통합 및 전술통신 브리핑 패키지 완성 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def simulate(msgs, link, policy=\"strict\", mtu=256, preempt=True, rto_mode=\"fixed\", rto_s=4.0, aging=0.0,\n",
" redundancy=False, adapt_mtu=False, hdr=HDR, seed=0, until=None, admit=None):\n",
" \"\"\"전문 목록을 링크로 보내는 이산 사건 시뮬레이션.\n",
" - 전문은 mtu 크기 조각으로 나눠 보낸다. 조각이 손실되면 RTO 후 재전송한다(자료는 버리지 않음).\n",
" - preempt=False면 한 전문의 조각을 끝까지 연속 전송한다(큰 전문이 긴급 전문을 막는 현상 재현).\n",
" - rto_mode='adaptive'면 왕복시간(RTT) 추정값으로 RTO를 계산한다(Jacobson 방식).\n",
" - redundancy=True면 링크 열화 추정 시 FLASH·IMMEDIATE 조각을 2회 연속 전송한다.\n",
" - adapt_mtu=True면 링크 열화 추정 시 조각 크기를 줄인다.\n",
" - admit(m, t)이 False인 전문은 새 조각을 보내지 않고 저장해 둔다(성능 저하 모드).\"\"\"\n",
" rng = np.random.default_rng(seed)\n",
" pend = sorted(({**m, \"unsent\": m[\"size\"], \"ready\": [], \"frags\": [], \"done\": None, \"first\": None} for m in msgs), key=lambda m: m[\"t0\"])\n",
" t, i, active, cur = 0.0, 0, [], None\n",
" srtt, rttvar, loss = None, None, 0.0\n",
" st = {\"tx\": 0, \"retx\": 0, \"spurious\": 0, \"bytes\": 0, \"decide_ms\": [], \"flag\": {}, \"log\": []}\n",
" until = until if until is not None else (pend[-1][\"t0\"] if pend else 0) + 6000\n",
" while (i < len(pend) or active) and t < until:\n",
" while i < len(pend) and pend[i][\"t0\"] <= t:\n",
" active.append(pend[i]); i += 1\n",
" td = time.perf_counter()\n",
" for m in active: # ACK가 이미 도착한 재전송 예약은 취소\n",
" m[\"ready\"] = [e for e in m[\"ready\"] if not (e[1][\"ack\"] is not None and e[1][\"ack\"] <= t)]\n",
" ready = [m for m in active if any(r <= t for r, _ in m[\"ready\"]) or (m[\"unsent\"] > 0 and (admit is None or admit(m, t)))]\n",
" if not ready:\n",
" nxt = ([pend[i][\"t0\"]] if i < len(pend) else []) + [r for m in active for r, _ in m[\"ready\"] if r > t]\n",
" t = min(nxt) if nxt else t + 1.0\n",
" continue\n",
" if not preempt and cur in ready and cur[\"unsent\"] > 0:\n",
" m = cur\n",
" else:\n",
" m = min(ready, key=lambda x: policy_key(x, t, policy, aging))\n",
" st[\"decide_ms\"].append((time.perf_counter() - td) * 1000)\n",
" due = [e for e in m[\"ready\"] if e[0] <= t]\n",
" if due:\n",
" e = min(due, key=lambda x: x[0]); m[\"ready\"].remove(e); frag = e[1]\n",
" st[\"retx\"] += 1; st[\"spurious\"] += frag[\"recv\"]\n",
" else:\n",
" size = mtu if not adapt_mtu else (mtu if loss < 0.15 else (mtu // 2 if loss < 0.4 else mtu // 4))\n",
" size = min(m[\"unsent\"], size); m[\"unsent\"] -= size\n",
" frag = {\"b\": size, \"recv\": False, \"ack\": None, \"tries\": 0}\n",
" m[\"frags\"].append(frag)\n",
" cur = m\n",
" m[\"first\"] = t if m[\"first\"] is None else m[\"first\"]\n",
" nb = frag[\"b\"] + hdr\n",
" copies = 2 if (redundancy and RANK[m[\"prec\"]] <= 1 and loss > 0.15) else 1\n",
" for _ in range(copies):\n",
" t += nb * 8 / link.bw(t)\n",
" st[\"tx\"] += 1; st[\"bytes\"] += nb\n",
" ok = rng.random() >= link.p_loss(nb, t)\n",
" st[\"log\"].append((t, nb, ok, m[\"prec\"], m[\"id\"])) # 전송 기록(시각, bytes, 성공, 긴급도, 전문)\n",
" loss = 0.85 * loss + 0.15 * (0.0 if ok else 1.0) # 최근 손실률 추정(EWMA)\n",
" st[\"flag\"][int(t)] = st[\"flag\"].get(int(t), False) or loss > 0.12 # 링크 열화 판정(초 단위)\n",
" if ok and not frag[\"recv\"]:\n",
" frag[\"recv\"], frag[\"rt\"] = True, t\n",
" r = link.rtt(rng); frag[\"ack\"] = t + r\n",
" if srtt is None:\n",
" srtt, rttvar = r, r / 2\n",
" else:\n",
" rttvar = 0.75 * rttvar + 0.25 * abs(srtt - r); srtt = 0.875 * srtt + 0.125 * r\n",
" frag[\"tries\"] += 1\n",
" rto = rto_s if (rto_mode == \"fixed\" or srtt is None) else float(np.clip(srtt + 4 * rttvar, 1.0, 30.0))\n",
" if not frag[\"recv\"] or frag[\"ack\"] > t + rto: # 손실 또는 ACK 지연 -> 재전송 예약\n",
" m[\"ready\"].append((t + rto, frag))\n",
" if m[\"done\"] is None and m[\"unsent\"] == 0 and all(f[\"recv\"] for f in m[\"frags\"]):\n",
" m[\"done\"] = max(f[\"rt\"] for f in m[\"frags\"])\n",
" active = [x for x in active if not (x[\"unsent\"] == 0 and not x[\"ready\"] and all(f[\"recv\"] for f in x[\"frags\"]))]\n",
" res = [{\"id\": m[\"id\"], \"type\": m[\"type\"], \"prec\": m[\"prec\"], \"size\": m[\"size\"], \"t0\": m[\"t0\"], \"done\": m[\"done\"], \"wait\": None if m[\"first\"] is None else m[\"first\"] - m[\"t0\"],\n",
" \"retx\": sum(f[\"tries\"] for f in m[\"frags\"]) - len(m[\"frags\"]),\n",
" \"latency\": None if m[\"done\"] is None else m[\"done\"] - m[\"t0\"]} for m in pend]\n",
" for r in res:\n",
" r[\"on_time\"] = r[\"latency\"] is not None and r[\"latency\"] <= DEADLINE[r[\"prec\"]]\n",
" return res, st\n",
"\n",
"\n",
"def summarize(res):\n",
" \"\"\"긴급도별 기한 내 전달률·지연. 종합 정확도 = 긴급도별 기한 내 전달률의 평균.\"\"\"\n",
" # [TODO : 학생 실습 10] 대역폭 제한 환경 메시지 큐·재전송 시스템 통합 및 전술통신 브리핑 패키지 완성 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"CONFIG = {\"policy\": \"strict\", \"preempt\": True, \"rto_mode\": \"fixed\", \"rto_s\": 6.0, \"mtu\": 128, \"aging\": 0.0, # 미션 9 재보정값\n",
" \"redundancy\": True, \"hdr\": 14} # 미션 4 중복 전송, 미션 2 헤더 압축\n",
"ALIAS = {\"Z\": \"FLASH\", \"O\": \"IMMEDIATE\", \"P\": \"PRIORITY\", \"R\": \"ROUTINE\"}\n",
"ROSTER = {f\"U{i:02d}\" for i in range(1, 9)}\n",
"\n",
"\n",
"def to_packet(m):\n",
" body = \"\".join(f\"{m['type']} {m['src']} I{(k * 7919 + int(m['t0'] * 10)) % 997:03d} Q{(k * 31 + m['size']) % 89:02d};\" for k in range(m[\"size\"] // 20 + 1))[:m[\"size\"]]\n",
" return f\"ZCZC|ID={m['id']}|PREC={'ZOPR'[RANK[m['prec']]]}|TYPE={m['type']}|SRC={m['src']}|T={m['t0']:.1f}|CRC={zlib.crc32(body.encode()):08x}|BODY={body}\"\n",
"\n",
"\n",
"def parse(line):\n",
" kv = dict(p.split(\"=\", 1) for p in line.split(\"|\")[1:] if \"=\" in p)\n",
" if set(kv) != {\"ID\", \"PREC\", \"TYPE\", \"SRC\", \"T\", \"CRC\", \"BODY\"} or kv[\"PREC\"] not in ALIAS or kv[\"SRC\"] not in ROSTER:\n",
" raise ValueError(\"형식·긴급도·송신국 오류\")\n",
" if f\"{zlib.crc32(kv['BODY'].encode()):08x}\" != kv[\"CRC\"]:\n",
" raise ValueError(\"CRC 불일치\")\n",
" body_z = zlib.compress(kv[\"BODY\"].encode(), 9)\n",
" return {\"id\": kv[\"ID\"], \"prec\": ALIAS[kv[\"PREC\"]], \"type\": kv[\"TYPE\"], \"src\": kv[\"SRC\"], \"t0\": float(kv[\"T\"]), \"size\": len(body_z)}\n",
"\n",
"\n",
"def link_grade(loss):\n",
" return \"High\" if loss < 0.1 else (\"Medium\" if loss < 0.3 else \"Low\")\n",
"\n",
"\n",
"class RequestQueue:\n",
" \"\"\"통신관 조작 요청 대기열: 시스템은 요청만 만들고 적용은 사람이 한다.\"\"\"\n",
"\n",
" def __init__(self):\n",
" self.items, self.log = [], []\n",
"\n",
" def submit(self, kind, target, why):\n",
" prev = self.log[-1][\"hash\"] if self.log else \"GENESIS\"\n",
" body = {\"event\": \"REQUEST\", \"kind\": kind, \"target\": target, \"why\": why}\n",
" self.log.append({**body, \"prev\": prev, \"hash\": hashlib.sha256((prev + json.dumps(body, ensure_ascii=False, sort_keys=True)).encode()).hexdigest()})\n",
" self.items.append({**body, \"status\": \"PENDING\"})\n",
"\n",
"\n",
"# 함수 커버리지 측정(표준 라이브러리 sys.setprofile): 이 파일에서 정의한 함수 중 실제 호출된 비율\n",
"called = set()\n",
"sys.setprofile(lambda f, e, a: called.add(f.f_code.co_name) if e == \"call\" and f.f_globals is globals() else None)\n",
"t_start = time.perf_counter()\n",
"\n",
"# ---------------- 1. 수신·전처리 ----------------\n",
"msgs = make_messages(1200, 0.5, seed=21) # 평시의 약 2.4배 트래픽\n",
"lines = [to_packet(m) for m in msgs]\n",
"lines[5] = lines[5].replace(\"SRC=U\", \"SRC=X\") # 손상·미등록 송신국 패킷 주입\n",
"lines[9] = lines[9][:-3] + \"###\"\n",
"parsed, rejected = [], {}\n",
"for ln in lines:\n",
" try:\n",
" parsed.append(parse(ln))\n",
" except ValueError as e:\n",
" rejected[ln.split(\"|\")[1][3:]] = str(e)\n",
"\n",
"# ---------------- 2. 큐·재전송 ----------------\n",
"link = Link(12000, seed=31, jam=[(300, 360)], bw_drop=[(600, 800, 0.5)], rtt=(2.0, 0.8))\n",
"res, st = simulate(parsed, link, **CONFIG)\n",
"tab, acc = summarize(res)\n",
"lg = pd.DataFrame(st[\"log\"], columns=[\"t\", \"bytes\", \"ok\", \"prec\", \"id\"])\n",
"lg[\"sec\"] = (lg[\"t\"] // 10 * 10).astype(int)\n",
"win = lg.groupby(\"sec\")[\"ok\"].apply(lambda s: 1 - s.mean()).rename(\"loss\").reset_index()\n",
"win[\"grade\"] = win[\"loss\"].map(link_grade)\n",
"\n",
"# ---------------- 3. HITL 요청(자동 적용 없음) ----------------\n",
"rq = RequestQueue()\n",
"for _, w in win[win[\"grade\"] == \"Low\"].iterrows():\n",
" rq.submit(\"ROUTE_REVIEW\", f\"t={w['sec']}s\", f\"손실률 {w['loss']:.0%}, 경로·중계 검토\")\n",
"for r in res:\n",
" if r[\"prec\"] == \"FLASH\" and not r[\"on_time\"]:\n",
" rq.submit(\"FLASH_LATE\", r[\"id\"], f\"지연 {r['latency']:.1f}s, 통신관 확인\")\n",
"sys.setprofile(None)\n",
"\n",
"# ---------------- 4. 자체 점검 ----------------\n",
"mine = [n for n, o in list(globals().items()) if inspect.isfunction(o) and o.__module__ == \"__main__\"]\n",
"mine += [\"submit\", \"state\", \"bw\", \"p_loss\", \"rtt\", \"__init__\"]\n",
"cov = len(set(mine) & called) / len(set(mine))\n",
"urg = np.mean([r[\"on_time\"] for r in res if RANK[r[\"prec\"]] <= 1])\n",
"elapsed = time.perf_counter() - t_start\n",
"chk = {\n",
" \"손상 패킷 2건 격리\": len(rejected) == 2,\n",
" \"정상 전문 전부 큐 등록\": len(parsed) == len(msgs) - 2,\n",
" \"전문 유실 0건(전달+보관=전체)\": sum(r[\"done\"] is not None for r in res) + sum(r[\"done\"] is None for r in res) == len(parsed),\n",
" \"FLASH 기한 내 전달률 90% 이상\": tab.loc[0, \"기한내 전달률\"] >= 0.9,\n",
" \"긴급 전문 기한 내 85% 이상\": urg >= 0.85,\n",
" \"종합 정확도 0.9 이상\": acc >= 0.9,\n",
" \"자동 적용 0건(요청 전부 PENDING)\": all(i[\"status\"] == \"PENDING\" for i in rq.items),\n",
" \"감사 로그 건수 일치\": len(rq.log) == len(rq.items),\n",
" \"스케줄링 결정 200ms 이내\": max(st[\"decide_ms\"]) < 200,\n",
" \"함수 커버리지 85% 이상\": cov >= 0.85,\n",
"}\n",
"passed = sum(chk.values())\n",
"for k, v in chk.items():\n",
" print(f\"[{'PASS' if v else 'FAIL'}] {k}\")\n",
"print(f\"자체 점검 {passed}/{len(chk)} 통과, 함수 커버리지 {cov:.0%}, 파이프라인 실행 성공 {'예' if passed == len(chk) else '아니오'}\")\n",
"print(tab[[\"긴급도\", \"건수\", \"기한내 전달률\", \"95% 지연(s)\", \"미전달\"]].to_string(index=False))\n",
"\n",
"# ---------------- 5. 관제 화면 ----------------\n",
"fig, ax = plt.subplots(2, 2, figsize=(12, 7))\n",
"ax[0, 0].bar(tab[\"긴급도\"], tab[\"기한내 전달률\"], color=[\"red\", \"orange\", \"deepskyblue\", \"gray\"]); ax[0, 0].set_ylim(0, 1.05); ax[0, 0].set_title(\"On-time ratio by precedence\")\n",
"ax[0, 1].plot(win[\"sec\"], win[\"loss\"] * 100); ax[0, 1].axhline(30, c=\"r\", ls=\"--\"); ax[0, 1].set_title(\"Loss rate per 10 s (%)\")\n",
"for a, b, c in [(300, 360, \"purple\"), (600, 800, \"gold\")]:\n",
" ax[0, 1].axvspan(a, b, color=c, alpha=0.2)\n",
"lat = pd.DataFrame(res)\n",
"for p, c in zip(PREC, [\"red\", \"orange\", \"deepskyblue\", \"gray\"]):\n",
" d = lat[lat[\"prec\"] == p]; ax[1, 0].scatter(d[\"t0\"], d[\"latency\"], s=8, c=c, label=p)\n",
"ax[1, 0].set_yscale(\"log\"); ax[1, 0].legend(fontsize=7); ax[1, 0].set_title(\"Delivery latency (s, log)\")\n",
"ax[1, 1].axis(\"off\")\n",
"ax[1, 1].text(0, 0.9, f\"Messages {len(parsed)} (+{len(rejected)} rejected)\\nOverall acc {acc:.3f}\\nUrgent ok {urg:.1%}\\n\"\n",
" f\"HITL pending {len(rq.items)}\\nSelf-check {passed}/{len(chk)}\", family=\"monospace\", fontsize=13, va=\"top\")\n",
"plt.suptitle(\"C202 integrated message queue - training data only\"); plt.tight_layout(); plt.show()\n",
"\n",
"# ---------------- 6. 전술통신 브리핑 요약 ----------------\n",
"print(f\"\"\"\n",
"[전술통신 브리핑 요약 - 교육용가상자료]\n",
"1. 처리: 수신 {len(lines)}건 중 {len(parsed)}건 전송, 손상·미등록 {len(rejected)}건 격리, 전체 소요 {elapsed:.1f}s\n",
"2. 성능: 종합 정확도 {acc:.3f}, FLASH 기한 내 {tab.loc[0, '기한내 전달률']:.1%}, 긴급 전문 기한 내 {urg:.1%}, 재전송 {st['retx']}회\n",
"3. 통신관 검토 요청: {len(rq.items)}건(경로 검토 {sum(i['kind'] == 'ROUTE_REVIEW' for i in rq.items)}, FLASH 지연 {sum(i['kind'] == 'FLASH_LATE' for i in rq.items)}), 자동 적용 0건\n",
"4. 운용 유의: 강제 전송·보류는 통신관이 근거를 기록해 수행하고, FLASH보다 앞서는 조작은 상급자 승인을 받는다.\n",
"5. 한계·확장: 가상 링크 모델 기준 결과로 실제 장비·망에서는 재측정이 필요하다. 인접 체계 연동은 전문 입력·전달 결과 조회 API로 확장한다.\"\"\")\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 10 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"[PASS] 손상 패킷 2건 격리\n",
"[PASS] 정상 전문 전부 큐 등록\n",
"[PASS] 전문 유실 0건(전달+보관=전체)\n",
"[PASS] FLASH 기한 내 전달률 90% 이상\n",
"[PASS] 긴급 전문 기한 내 85% 이상\n",
"[PASS] 종합 정확도 0.9 이상\n",
"[PASS] 자동 적용 0건(요청 전부 PENDING)\n",
"[PASS] 감사 로그 건수 일치\n",
"[PASS] 스케줄링 결정 200ms 이내\n",
"[PASS] 함수 커버리지 85% 이상\n",
"자체 점검 10/10 통과, 함수 커버리지 100%, 파이프라인 실행 성공 예\n",
" 긴급도 건수 기한내 전달률 95% 지연(s) 미전달\n",
" FLASH 45 0.978 7.3 0\n",
"IMMEDIATE 55 1.000 7.3 0\n",
" PRIORITY 248 1.000 8.3 0\n",
" ROUTINE 271 1.000 46.8 0\n",
"\n",
"[전술통신 브리핑 요약 - 교육용가상자료]\n",
"1. 처리: 수신 621건 중 619건 전송, 손상·미등록 2건 격리, 전체 소요 0.2s\n",
"2. 성능: 종합 정확도 0.994, FLASH 기한 내 97.8%, 긴급 전문 기한 내 99.0%, 재전송 111회\n",
"3. 통신관 검토 요청: 7건(경로 검토 6, FLASH 지연 1), 자동 적용 0건\n",
"4. 운용 유의: 강제 전송·보류는 통신관이 근거를 기록해 수행하고, FLASH보다 앞서는 조작은 상급자 승인을 받는다.\n",
"5. 한계·확장: 가상 링크 모델 기준 결과로 실제 장비·망에서는 재측정이 필요하다. 인접 체계 연동은 전문 입력·전달 결과 조회 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