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

2791 lines
190 KiB
Plaintext
Raw Permalink Blame History

{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# 🎯 [WEL01] 더미 생체신호 기반 열·한랭 스트레스 경보 프로토타입\n",
"### Google Colab 실습 환경 (장병 복지·지원 / 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/WEL01/WEL01_solution.ipynb)\n",
"\n",
"---\n",
"\n",
"## 📌 1. 프로젝트 개요 및 배경\n",
"\n",
"| 항목 | 내용 |\n",
"| :--- | :--- |\n",
"| **과정 분류** | 장병 복지·지원 (Level 2) |\n",
"| **학습 소요 시간** | 10개 미션 (5일 / 총 27.5시간) |\n",
"| **환경 사양** | Python 3.10+, numpy, pandas, matplotlib, scipy, scikit-learn |\n",
"\n",
"### 📖 프로젝트 목표\n",
"러시아-우크라이나 전쟁에서는 혹한과 폭염 속에서 장시간 기동하는 전투원의 열·한랭 손상이 작전 지속력을 좌우하는 문제로 드러났고, 웨어러블 센서 신호로 위험 징후를 일찍 알아차리되 오경보와 센서 오류를 가려내는 일이 중요해졌다. 본 프로젝트에서는 교육용 가상 병사의 더미 생체신호(심박수, 체온, 피부전도도)와 기상을 코드로 직접 만들어, 데이터 스키마 정의, 결측·센서 튐 전처리, 체온-심박 상관 기반 Baseline, 기상·개인 특성을 결합한 복합 위험도 모델, 개인화 신뢰도 엔진, 소대장·의무관용 상황판, 군의관 원격 검토 중심의 HITL 절차, 폭염·혹한 대량 데이터 유입 시험, AAR 재보정, 최종 통합까지 열·한랭 스트레스 경보 프로토타입을 단계별로 구현한다. 최종 목표는 위험 징후를 놓치지 않으면서 오경보를 줄이고, 모든 조치 판단은 진단이 아닌 보조 정보로서 군의관·의무대와 연계되도록 설계하는 것이다.\n",
"\n",
"> **⚠️ [교육용 가상 데이터셋 원칙]**\n",
"> - 본 노트북의 모든 데이터(제원, 로그, 시나리오 등)는 보안 규정을 준수하여 **코드 내 난수 기반 가상 데이터로 생성**됩니다.\n",
"> - 실제 부대 위치, 작전 인가, 실제 전장 정보, 군사 기밀 데이터는 일체 포함되지 않습니다.\n",
"\n",
"---\n",
"\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# ============================================================================\n",
"# [Google Colab 환경 설정] 필수 라이브러리 설치 및 임포트\n",
"# ============================================================================\n",
"!pip install -q numpy pandas matplotlib scipy scikit-learn\n",
"\n",
"import os\n",
"import sys\n",
"import time\n",
"import math\n",
"import json\n",
"import re\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"import warnings\n",
"warnings.filterwarnings('ignore')\n",
"\n",
"# 그래프 영문 폰트 기본 설정 (환경 독립성 보장)\n",
"plt.rcParams['font.sans-serif'] = 'DejaVu Sans'\n",
"plt.rcParams['axes.unicode_minus'] = False\n",
"\n",
"print(f\"실행 환경 준비 완료 (Python {sys.version.split()[0]})\")\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 01. 전투원 착용 웨어러블 더미 생체신호(심박수, 체온, 피부전도도) 데이터 규격 정의\n",
"**소요시간**: 약 1.5시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"전투원 웨어러블 더미 생체신호(심박수, 체온, 피부전도도) 데이터셋을 정의한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **웨어러블 더미 생체신호(심박수, 체온, 피부전도도)와 가상 병사 특성·기상·기기 상태의 데이터 규격을 정의하고, 잘못된 데이터를 걸러내는 검증기를 만든다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **신호 3종의 허용 범위, 샘플링 간격, 기기 종류와 접촉 품질, 병사 특성(안정시 심박, 연령, 체격, 더위 순응 여부), 기상(기온, 습도, 풍속) 필드를 담은 스키마 설계하고 작성하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `data/schema/wel01_input_schema.json 및 docs/wel01_data_spec.md` 형태의 결과물을 완성한다.\n",
"\n",
"### 🔍 평가 및 검증 기준 (PASS 조건)\n",
"#### 미션 판정\n",
"\n",
"**평가 구분:** 필수\n",
"\n",
"**핵심 평가영역:** 장병 복지·지원 실무 및 러시아-우크라이나 전쟁 전장 환경과의 군사적 부합성\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": [
"\"\"\"전투원 웨어러블 더미 생체신호(심박수, 체온, 피부전도도)와 가상 병사 특성·기상·기기 상태의 데이터 스키마를 정의하고, 가상 샘플 20건을 검증한다. 진단 목적이 아닌 교육용 가상 자료이다.\"\"\"\n",
"import json\n",
"import re\n",
"import numpy as np\n",
"import pandas as pd\n",
"\n",
"# ---------------- [교육용가상자료] 더미 생체신호 생성기 (실제 장병·부대·의무 기록 없음) ----------------\n",
"# 가상 병사(가명 ID)가 60분간 걷기·휴식·고강도 활동을 하는 동안의 심박수·체온(추정 심부)·피부전도도를 1분 간격으로 만든다.\n",
"# 진단 도구가 아니며, 상태 라벨은 가상 생리 모델의 '교육용 단계'일 뿐이다.\n",
"N_MIN = 60\n",
"STAGE = {0: \"정상\", 1: \"주의\", 2: \"경보\"}\n",
"TYPES = [\"normal\", \"heat\", \"cold\"]\n",
"\n",
"\n",
"def make_person(i, rng):\n",
" return {\"id\": f\"S{i:03d}\", \"rest_hr\": float(rng.normal(64, 7)), \"age\": int(rng.integers(19, 27)), \"bmi\": float(rng.normal(23.5, 2.5)),\n",
" \"acclim\": int(rng.random() < 0.4), \"fit\": float(np.clip(rng.normal(0.5, 0.2), 0.05, 0.95))}\n",
"\n",
"\n",
"def make_env(kind, rng):\n",
" if kind == \"heat\":\n",
" return {\"temp_c\": float(rng.uniform(30, 40)), \"rh\": float(rng.uniform(45, 85)), \"wind\": float(rng.uniform(0.3, 3))}\n",
" if kind == \"cold\":\n",
" return {\"temp_c\": float(rng.uniform(-22, 2)), \"rh\": float(rng.uniform(40, 80)), \"wind\": float(rng.uniform(2, 11))}\n",
" return {\"temp_c\": float(rng.uniform(8, 24)), \"rh\": float(rng.uniform(35, 70)), \"wind\": float(rng.uniform(0.3, 5))}\n",
"\n",
"\n",
"def make_session(sid, kind, rng, p=None, glitch=True):\n",
" \"\"\"반환: dict(person, env, kind, df[hr,temp,eda,act + 정답 core·stage·rtype]). 센서 결측·튐·고착 포함.\"\"\"\n",
" p = p or make_person(sid, rng)\n",
" env = make_env(kind, rng)\n",
" act = np.zeros(N_MIN); t = 0\n",
" while t < N_MIN: # 10~20분 단위 활동 블록: 휴식 1.5 / 행군 4.5 / 고강도 7 MET\n",
" d = int(rng.integers(10, 21)); act[t:t + d] = rng.choice([1.5, 4.5, 7.0], p=[0.25, 0.5, 0.25]); t += d\n",
" hi = env[\"temp_c\"] + (0.12 * (env[\"rh\"] - 40) if env[\"temp_c\"] > 25 else 0)\n",
" wc = env[\"temp_c\"] - 1.6 * env[\"wind\"]\n",
" core, cores = 36.9, []\n",
" for m in range(N_MIN):\n",
" heat = 0.085 * max(0, hi - 24) * (1 - 0.25 * p[\"acclim\"]) * (1 + 0.5 * m / N_MIN) * (1 + 0.012 * (p[\"bmi\"] - 23.5))\n",
" cold = -0.075 * max(0, 8 - wc) * (1 - 0.09 * act[m])\n",
" ss = 36.9 + 0.09 * act[m] + heat + cold\n",
" core += (ss - core) / 18 + rng.normal(0, 0.03)\n",
" cores.append(core)\n",
" core = np.array(cores)\n",
" hr = p[\"rest_hr\"] + 9.5 * (act - 1.5) * (1.15 - 0.3 * p[\"fit\"]) + 11 * (core - 36.9) + rng.normal(0, 3.2, N_MIN)\n",
" eda = 2.0 + 0.25 * act + 3.2 * np.maximum(core - 37.2, 0) - 0.04 * np.maximum(8 - wc, 0) + rng.normal(0, 0.25, N_MIN)\n",
" bias = rng.normal(0, 0.1)\n",
" temp = core + bias + rng.normal(0, 0.1, N_MIN) - np.where((kind == \"cold\") & (env[\"wind\"] > 7), 0.25, 0.0) # 센서 접촉 부위 바람 냉각 오차\n",
" df = pd.DataFrame({\"min\": np.arange(N_MIN), \"hr\": hr, \"temp\": temp, \"eda\": np.maximum(eda, 0.2), \"act\": act})\n",
" if glitch:\n",
" for col in (\"hr\", \"temp\", \"eda\"):\n",
" df.loc[rng.random(N_MIN) < 0.03, col] = np.nan # 결측\n",
" s = int(rng.integers(5, N_MIN))\n",
" if rng.random() < 0.12:\n",
" df.loc[s, \"hr\"] = 230.0 # 센서 튐\n",
" if rng.random() < 0.12:\n",
" df.loc[s, \"temp\"] = 45.0\n",
" ht = np.where(core >= 39.0, 2, np.where(core >= 38.3, 1, 0)); cd = np.where(core <= 35.4, 2, np.where(core <= 36.2, 1, 0))\n",
" df[\"core\"] = core\n",
" df[\"stage\"] = np.maximum(ht, cd); df[\"rtype\"] = np.where(ht > 0, \"heat\", np.where(cd > 0, \"cold\", \"normal\"))\n",
" return {\"id\": p[\"id\"], \"person\": p, \"env\": env, \"kind\": kind, \"df\": df}\n",
"\n",
"\n",
"def make_dataset(n, seed):\n",
" rng = np.random.default_rng(seed)\n",
" kinds = [\"normal\", \"heat\", \"cold\"]\n",
" return [make_session(i, kinds[i % 3], rng) for i in range(n)]\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# 1) 스키마 정의: 신호 규격, 가상 병사 특성, 기상(온·습도·풍속), 기기 상태\n",
"SCHEMA = {\n",
" \"required\": [\"record_id\", \"soldier_id\", \"n_samples\", \"sampling_s\", \"device\", \"person\", \"weather\", \"samples\"],\n",
" \"fields\": {\n",
" \"record_id\": {\"type\": str, \"pattern\": r\"^REC-\\d{4}$\"},\n",
" \"soldier_id\": {\"type\": str, \"pattern\": r\"^S\\d{3}$\"},\n",
" \"n_samples\": {\"type\": int, \"min\": 1, \"max\": 100000},\n",
" \"sampling_s\": {\"type\": (int, float), \"min\": 1, \"max\": 600},\n",
" },\n",
" \"device\": {\n",
" \"type\": {\"type\": str, \"enum\": [\"wrist\", \"chest\"]},\n",
" \"battery_pct\": {\"type\": (int, float), \"min\": 0, \"max\": 100},\n",
" \"contact_quality\": {\"type\": (int, float), \"min\": 0, \"max\": 1},\n",
" },\n",
" \"person\": {\n",
" \"rest_hr\": {\"type\": (int, float), \"min\": 40, \"max\": 100},\n",
" \"age\": {\"type\": int, \"min\": 18, \"max\": 45},\n",
" \"bmi\": {\"type\": (int, float), \"min\": 15, \"max\": 40},\n",
" \"acclim\": {\"type\": int, \"min\": 0, \"max\": 1},\n",
" },\n",
" \"weather\": {\n",
" \"temp_c\": {\"type\": (int, float), \"min\": -40, \"max\": 55},\n",
" \"rh\": {\"type\": (int, float), \"min\": 0, \"max\": 100},\n",
" \"wind\": {\"type\": (int, float), \"min\": 0, \"max\": 30},\n",
" },\n",
" \"sample\": {\n",
" \"hr\": {\"type\": (int, float), \"min\": 30, \"max\": 220},\n",
" \"temp\": {\"type\": (int, float), \"min\": 30, \"max\": 43},\n",
" \"eda\": {\"type\": (int, float), \"min\": 0, \"max\": 40},\n",
" },\n",
"}\n",
"\n",
"\n",
"def check(name, value, rule):\n",
" \"\"\"값 하나를 규칙(type, enum, min, max, pattern)으로 검사해 오류 목록을 돌려준다.\"\"\"\n",
" # [TODO : 학생 실습 1] 전투원 착용 웨어러블 더미 생체신호(심박수, 체온, 피부전도도) 데이터 규격 정의 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def validate(rec):\n",
" # [TODO : 학생 실습 1] 전투원 착용 웨어러블 더미 생체신호(심박수, 체온, 피부전도도) 데이터 규격 정의 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"rng = np.random.default_rng(1)\n",
"records = []\n",
"for i in range(20):\n",
" s = make_session(i, TYPES[i % 3], rng, glitch=False)\n",
" smp = [{\"hr\": round(float(r.hr), 1), \"temp\": round(float(r.temp), 2), \"eda\": round(float(r.eda), 2)} for r in s[\"df\"].itertuples()]\n",
" p, e = s[\"person\"], s[\"env\"]\n",
" records.append({\"record_id\": f\"REC-{i:04d}\", \"soldier_id\": s[\"id\"], \"n_samples\": len(smp), \"sampling_s\": 60,\n",
" \"device\": {\"type\": str(rng.choice([\"wrist\", \"chest\"])), \"battery_pct\": int(rng.integers(30, 100)), \"contact_quality\": round(float(rng.uniform(0.7, 1)), 2)},\n",
" \"person\": {\"rest_hr\": round(p[\"rest_hr\"], 1), \"age\": p[\"age\"], \"bmi\": round(p[\"bmi\"], 1), \"acclim\": p[\"acclim\"]},\n",
" \"weather\": {k: round(v, 1) for k, v in e.items()}, \"samples\": smp})\n",
"\n",
"# 3) 정상 데이터 검증\n",
"ok = sum(1 for r in records if not validate(r))\n",
"print(f\"정상 샘플 {len(records)}건 중 검증 통과 {ok}건, 오류 {len(records) - ok}건 -> 목표(검증 오류 0건): {'달성' if ok == len(records) else '미달'}\")\n",
"\n",
"# 4) 오류 주입 시험\n",
"bad = json.loads(json.dumps(records[0]))\n",
"bad[\"soldier_id\"] = \"SOLDIER-1\" # 형식\n",
"bad[\"samples\"][0][\"temp\"] = 45.0 # 범위(센서 튐)\n",
"del bad[\"weather\"][\"wind\"] # 누락(풍속)\n",
"bad[\"device\"][\"type\"] = \"ankle\" # 허용값\n",
"bad[\"n_samples\"] = 99 # 논리(개수 불일치)\n",
"errs = validate(bad)\n",
"print(f\"\\n[오류 주입 결과] 주입 5종 -> 검출 {len(errs)}건\")\n",
"for e in errs:\n",
" print(\" -\", e)\n",
"print(f\"주입 오류 전부 검출: {'예' if len(errs) >= 5 else '아니오'}\")\n",
"\n",
"# 5) 스키마 필드 정의서(명세서 초안)\n",
"rows = [(\"기본\", k, r[\"type\"].__name__ if isinstance(r[\"type\"], type) else \"number\", str({x: y for x, y in r.items() if x != \"type\"})) for k, r in SCHEMA[\"fields\"].items()]\n",
"for g in (\"device\", \"person\", \"weather\", \"sample\"):\n",
" for k, r in SCHEMA[g].items():\n",
" t = r[\"type\"].__name__ if isinstance(r[\"type\"], type) else \"number\"\n",
" rows.append((g, k, t, str({x: y for x, y in r.items() if x != \"type\"})))\n",
"print(\"\\n[데이터 명세 요약] 필드 수\", len(rows))\n",
"print(pd.DataFrame(rows, columns=[\"그룹\", \"필드\", \"자료형\", \"규칙\"]).to_string(index=False))\n",
"print(\"\\n상태 단계(교육용 가상 라벨):\", STAGE, \"/ 위험 유형:\", TYPES)\n",
"print(\"본 데이터는 코드로 만든 더미 생체신호이며 진단·처방 자료가 아니다. 실제 건강 이상은 의무대·군의관에게 연계한다.\")\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 01 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"정상 샘플 20건 중 검증 통과 20건, 오류 0건 -> 목표(검증 오류 0건): 달성\n",
"\n",
"[오류 주입 결과] 주입 5종 -> 검출 5건\n",
" - soldier_id: 형식 불일치 'SOLDIER-1'\n",
" - device.type: 허용되지 않은 값 'ankle'\n",
" - weather.wind: 필수 필드 누락\n",
" - samples[0].temp: 최대값 43 초과\n",
" - n_samples: 샘플 개수와 불일치\n",
"주입 오류 전부 검출: 예\n",
"\n",
"[데이터 명세 요약] 필드 수 17\n",
" 그룹 필드 자료형 규칙\n",
" 기본 record_id str {'pattern': '^REC-\\\\d{4}$'}\n",
" 기본 soldier_id str {'pattern': '^S\\\\d{3}$'}\n",
" 기본 n_samples int {'min': 1, 'max': 100000}\n",
" 기본 sampling_s number {'min': 1, 'max': 600}\n",
" device type str {'enum': ['wrist', 'chest']}\n",
" device battery_pct number {'min': 0, 'max': 100}\n",
" device contact_quality number {'min': 0, 'max': 1}\n",
" person rest_hr number {'min': 40, 'max': 100}\n",
" person age int {'min': 18, 'max': 45}\n",
" person bmi number {'min': 15, 'max': 40}\n",
" person acclim int {'min': 0, 'max': 1}\n",
"weather temp_c number {'min': -40, 'max': 55}\n",
"weather rh number {'min': 0, 'max': 100}\n",
"weather wind number {'min': 0, 'max': 30}\n",
" sample hr number {'min': 30, 'max': 220}\n",
" sample temp number {'min': 30, 'max': 43}\n",
" sample eda number {'min': 0, 'max': 40}\n",
"\n",
"상태 단계(교육용 가상 라벨): {0: '정상', 1: '주의', 2: '경보'} / 위험 유형: ['normal', 'heat', 'cold']\n",
"본 데이터는 코드로 만든 더미 생체신호이며 진단·처방 자료가 아니다. 실제 건강 이상은 의무대·군의관에게 연계한다.\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 02. 야전 활동 중 발생하는 생체신호 결측치 보정 및 센서 노이즈 필터링 전처리 구현\n",
"**소요시간**: 약 2.0시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"야전 활동 중 생체신호 결측치 보정 및 센서 노이즈 필터링 전처리 모듈을 구현한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **야전 활동 중 발생하는 결측, 센서 튐, 센서 고착(단절)을 탐지·보정하는 전처리 모듈을 구현한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **생리적으로 불가능한 범위 밖 값을 결측으로 바꾸고 선형 보간과 3분 이동 중앙값으로 정리하며, 같은 값이 8분 이상 이어지는 센서 고착을 찾아 결측 처리 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `src/preprocess/wel01_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": [
"\"\"\"야전 활동 중 생체신호의 결측·센서 튐·고착(센서 단절)을 탐지·보정하는 전처리 모듈을 구현하고, 처리 속도·단위 테스트 5종·보정 전후 체온 오차와 경보 분류 성능을 측정한다. 진단 목적이 아닌 교육용 가상 자료이다.\"\"\"\n",
"import time\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"\n",
"# ---------------- [교육용가상자료] 더미 생체신호 생성기 (실제 장병·부대·의무 기록 없음) ----------------\n",
"# 가상 병사(가명 ID)가 60분간 걷기·휴식·고강도 활동을 하는 동안의 심박수·체온(추정 심부)·피부전도도를 1분 간격으로 만든다.\n",
"# 진단 도구가 아니며, 상태 라벨은 가상 생리 모델의 '교육용 단계'일 뿐이다.\n",
"N_MIN = 60\n",
"STAGE = {0: \"정상\", 1: \"주의\", 2: \"경보\"}\n",
"TYPES = [\"normal\", \"heat\", \"cold\"]\n",
"\n",
"\n",
"def make_person(i, rng):\n",
" return {\"id\": f\"S{i:03d}\", \"rest_hr\": float(rng.normal(64, 7)), \"age\": int(rng.integers(19, 27)), \"bmi\": float(rng.normal(23.5, 2.5)),\n",
" \"acclim\": int(rng.random() < 0.4), \"fit\": float(np.clip(rng.normal(0.5, 0.2), 0.05, 0.95))}\n",
"\n",
"\n",
"def make_env(kind, rng):\n",
" if kind == \"heat\":\n",
" return {\"temp_c\": float(rng.uniform(30, 40)), \"rh\": float(rng.uniform(45, 85)), \"wind\": float(rng.uniform(0.3, 3))}\n",
" if kind == \"cold\":\n",
" return {\"temp_c\": float(rng.uniform(-22, 2)), \"rh\": float(rng.uniform(40, 80)), \"wind\": float(rng.uniform(2, 11))}\n",
" return {\"temp_c\": float(rng.uniform(8, 24)), \"rh\": float(rng.uniform(35, 70)), \"wind\": float(rng.uniform(0.3, 5))}\n",
"\n",
"\n",
"def make_session(sid, kind, rng, p=None, glitch=True):\n",
" \"\"\"반환: dict(person, env, kind, df[hr,temp,eda,act + 정답 core·stage·rtype]). 센서 결측·튐·고착 포함.\"\"\"\n",
" p = p or make_person(sid, rng)\n",
" env = make_env(kind, rng)\n",
" act = np.zeros(N_MIN); t = 0\n",
" while t < N_MIN: # 10~20분 단위 활동 블록: 휴식 1.5 / 행군 4.5 / 고강도 7 MET\n",
" d = int(rng.integers(10, 21)); act[t:t + d] = rng.choice([1.5, 4.5, 7.0], p=[0.25, 0.5, 0.25]); t += d\n",
" hi = env[\"temp_c\"] + (0.12 * (env[\"rh\"] - 40) if env[\"temp_c\"] > 25 else 0)\n",
" wc = env[\"temp_c\"] - 1.6 * env[\"wind\"]\n",
" core, cores = 36.9, []\n",
" for m in range(N_MIN):\n",
" heat = 0.085 * max(0, hi - 24) * (1 - 0.25 * p[\"acclim\"]) * (1 + 0.5 * m / N_MIN) * (1 + 0.012 * (p[\"bmi\"] - 23.5))\n",
" cold = -0.075 * max(0, 8 - wc) * (1 - 0.09 * act[m])\n",
" ss = 36.9 + 0.09 * act[m] + heat + cold\n",
" core += (ss - core) / 18 + rng.normal(0, 0.03)\n",
" cores.append(core)\n",
" core = np.array(cores)\n",
" hr = p[\"rest_hr\"] + 9.5 * (act - 1.5) * (1.15 - 0.3 * p[\"fit\"]) + 11 * (core - 36.9) + rng.normal(0, 3.2, N_MIN)\n",
" eda = 2.0 + 0.25 * act + 3.2 * np.maximum(core - 37.2, 0) - 0.04 * np.maximum(8 - wc, 0) + rng.normal(0, 0.25, N_MIN)\n",
" bias = rng.normal(0, 0.1)\n",
" temp = core + bias + rng.normal(0, 0.1, N_MIN) - np.where((kind == \"cold\") & (env[\"wind\"] > 7), 0.25, 0.0) # 센서 접촉 부위 바람 냉각 오차\n",
" df = pd.DataFrame({\"min\": np.arange(N_MIN), \"hr\": hr, \"temp\": temp, \"eda\": np.maximum(eda, 0.2), \"act\": act})\n",
" if glitch:\n",
" for col in (\"hr\", \"temp\", \"eda\"):\n",
" df.loc[rng.random(N_MIN) < 0.03, col] = np.nan # 결측\n",
" s = int(rng.integers(5, N_MIN))\n",
" if rng.random() < 0.12:\n",
" df.loc[s, \"hr\"] = 230.0 # 센서 튐\n",
" if rng.random() < 0.12:\n",
" df.loc[s, \"temp\"] = 45.0\n",
" ht = np.where(core >= 39.0, 2, np.where(core >= 38.3, 1, 0)); cd = np.where(core <= 35.4, 2, np.where(core <= 36.2, 1, 0))\n",
" df[\"core\"] = core\n",
" df[\"stage\"] = np.maximum(ht, cd); df[\"rtype\"] = np.where(ht > 0, \"heat\", np.where(cd > 0, \"cold\", \"normal\"))\n",
" return {\"id\": p[\"id\"], \"person\": p, \"env\": env, \"kind\": kind, \"df\": df}\n",
"\n",
"\n",
"def make_dataset(n, seed):\n",
" rng = np.random.default_rng(seed)\n",
" kinds = [\"normal\", \"heat\", \"cold\"]\n",
" return [make_session(i, kinds[i % 3], rng) for i in range(n)]\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# ---------------- 전처리·특징 추출 공통 함수 ----------------\n",
"LIMITS = {\"hr\": (30, 220), \"temp\": (30.0, 43.0), \"eda\": (0.0, 40.0)} # 생리적으로 가능한 범위(범위 밖 = 센서 오류)\n",
"\n",
"\n",
"def clean(df):\n",
" \"\"\"범위 밖 값은 결측 처리 -> 선형 보간(양 끝은 가까운 값) -> 3분 이동 중앙값으로 튐 제거. 반환: (정제 df, 보정 건수)\"\"\"\n",
" # [TODO : 학생 실습 2] 야전 활동 중 발생하는 생체신호 결측치 보정 및 센서 노이즈 필터링 전처리 구현 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def env_index(env):\n",
" # [TODO : 학생 실습 2] 야전 활동 중 발생하는 생체신호 결측치 보정 및 센서 노이즈 필터링 전처리 구현 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def features(s, use_env=True, personal=True):\n",
" \"\"\"세션 1개 -> 분 단위 특징표(최근 5분 평균·기울기 포함). 정답 stage/rtype 포함.\"\"\"\n",
" df, _ = clean(s[\"df\"]); p, e = s[\"person\"], s[\"env\"]\n",
" f = pd.DataFrame({\"sid\": s[\"id\"], \"min\": df[\"min\"]})\n",
" f[\"temp5\"] = df[\"temp\"].rolling(5, min_periods=1).mean()\n",
" f[\"hr5\"] = df[\"hr\"].rolling(5, min_periods=1).mean()\n",
" f[\"eda5\"] = df[\"eda\"].rolling(5, min_periods=1).mean()\n",
" f[\"dtemp\"] = (f[\"temp5\"] - f[\"temp5\"].shift(4)).fillna(0)\n",
" f[\"act\"] = df[\"act\"]\n",
" if personal:\n",
" f[\"hr_rel\"] = f[\"hr5\"] - p[\"rest_hr\"]\n",
" f[\"acclim\"], f[\"bmi\"], f[\"fit\"] = p[\"acclim\"], p[\"bmi\"], p[\"fit\"]\n",
" if use_env:\n",
" f[\"hi\"], f[\"wc\"] = env_index(e); f[\"wind\"] = e[\"wind\"]\n",
" f[\"stage\"], f[\"rtype\"], f[\"kind\"] = df[\"stage\"], df[\"rtype\"], s[\"kind\"]\n",
" return f[f[\"min\"] >= 5].reset_index(drop=True)\n",
"\n",
"\n",
"def table(sessions, **kw):\n",
" return pd.concat([features(s, **kw) for s in sessions], ignore_index=True)\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"\n",
"def flatline(series, run=8):\n",
" \"\"\"같은 값이 run분 이상 이어지면 센서 고착(단절)으로 보고 해당 구간을 True로 표시한다.\"\"\"\n",
" same = series.diff().abs().fillna(1) < 1e-9\n",
" grp = (~same).cumsum()\n",
" return same.groupby(grp).transform(\"sum\").add(1).where(same | same.shift(-1, fill_value=False), 1) >= run\n",
"\n",
"\n",
"def preprocess(df):\n",
" \"\"\"고착 구간을 결측으로 바꾼 뒤 clean()으로 범위 검사·보간·튐 제거. 반환: (정제 df, 보정 건수, 고착 건수)\"\"\"\n",
" d = df.copy(); stuck = 0\n",
" for c in (\"hr\", \"temp\"):\n",
" fl = flatline(d[c].round(3))\n",
" stuck += int(fl.sum()); d.loc[fl, c] = np.nan\n",
" out, fixed = clean(d)\n",
" return out, fixed, stuck\n",
"\n",
"\n",
"def inject_faults(df, rng):\n",
" \"\"\"센서 단절 모의: 체온이 가장 크게 변하는 시점 직전부터 센서가 12분간 같은 값으로 고착된다(실제 변화를 따라가지 못함).\"\"\"\n",
" d = df.copy(); s = int(np.clip(np.argmax(np.abs(d[\"core\"].values - 36.9)) - 12 + rng.integers(0, 4), 3, N_MIN - 13))\n",
" d.loc[s:s + 11, \"temp\"] = d.loc[s, \"temp\"]\n",
" return d\n",
"\n",
"\n",
"rng = np.random.default_rng(5)\n",
"data = make_dataset(120, seed=21)\n",
"\n",
"# 1) 처리 속도와 오류율\n",
"t0 = time.perf_counter(); n_rows = 0; errors = 0; fixed_all = 0\n",
"for s in data:\n",
" try:\n",
" out, fx, st = preprocess(s[\"df\"]); n_rows += len(out) * 3; fixed_all += fx\n",
" errors += int(out[[\"hr\", \"temp\", \"eda\"]].isna().any().any() or (out[\"temp\"] > 43).any())\n",
" except Exception:\n",
" errors += 1\n",
"sec = time.perf_counter() - t0\n",
"print(f\"전처리 {n_rows:,}건(신호값) / {sec:.2f}초 = {n_rows / sec:,.0f}건/초 (목표 1,000 이상: {'달성' if n_rows / sec >= 1000 else '미달'})\")\n",
"print(f\"전처리 오류 {errors}건 -> 오류율 {errors / len(data):.0%} (목표 0%), 범위 밖·결측 보정 {fixed_all}건\")\n",
"\n",
"\n",
"# 2) 단위 테스트 5종\n",
"def t1():\n",
" d = pd.DataFrame({\"min\": range(6), \"hr\": [80, np.nan, 82, 83, 84, 85.], \"temp\": 37.0, \"eda\": 2.0, \"act\": 1.5, \"stage\": 0, \"rtype\": \"normal\"})\n",
" out, fx = clean(d); assert out[\"hr\"].notna().all() and fx == 1\n",
"def t2():\n",
" d = pd.DataFrame({\"min\": range(6), \"hr\": [80, 81, 230, 82, 83, 84.], \"temp\": 37.0, \"eda\": 2.0, \"act\": 1.5, \"stage\": 0, \"rtype\": \"normal\"})\n",
" out, fx = clean(d); assert out[\"hr\"].max() < 100 and fx == 1\n",
"def t3():\n",
" s = pd.Series([36.9, 37.0, 37.0, 37.0, 37.0, 37.0, 37.0, 37.0, 37.0, 37.0, 37.1]); assert flatline(s).sum() >= 8\n",
"def t4():\n",
" s = pd.Series(np.linspace(36.5, 37.5, 20)); assert not flatline(s).any()\n",
"def t5():\n",
" out, fx, st = preprocess(inject_faults(make_session(1, \"normal\", np.random.default_rng(3))[\"df\"], np.random.default_rng(4))); assert st >= 8 and len(out) == N_MIN\n",
"tests = {\"결측 보간\": t1, \"센서 튐 제거\": t2, \"고착 구간 탐지\": t3, \"정상 변화 오탐 없음\": t4, \"고착 주입 후 보정\": t5}\n",
"passed = 0\n",
"for name, fn in tests.items():\n",
" try:\n",
" fn(); passed += 1; print(f\"[PASS] {name}\")\n",
" except AssertionError:\n",
" print(f\"[FAIL] {name}\")\n",
"print(f\"단위 테스트 {passed}/5 통과\")\n",
"\n",
"# 3) 보정 전후 체온 오차(정답 core 대비 RMSE)와 경보 분류 성능\n",
"def rmse(a, b):\n",
" m = ~np.isnan(a)\n",
" return float(np.sqrt(np.mean((a[m] - b[m]) ** 2)))\n",
"\n",
"\n",
"def hot_cold_f1(df_use, stage_src):\n",
" \"\"\"체온 38.0 이상 또는 36.3 이하를 위험으로 본 경보와 정답 stage>=1의 F1.\"\"\"\n",
" pred = (df_use[\"temp\"] >= 38.0) | (df_use[\"temp\"] <= 36.3)\n",
" truth = stage_src[\"stage\"] >= 1\n",
" tp = (pred & truth).sum(); p = tp / max(pred.sum(), 1); r = tp / max(truth.sum(), 1)\n",
" return round(2 * p * r / max(p + r, 1e-9), 3), int((pred & ~truth).sum())\n",
"\n",
"\n",
"rows = []\n",
"for kind, label in ((\"normal\", \"평시\"), (\"heat\", \"폭염\"), (\"cold\", \"혹한\")):\n",
" raw, cln, tr_ = [], [], []\n",
" for s in [x for x in data if x[\"kind\"] == kind]:\n",
" bad = inject_faults(s[\"df\"], rng)\n",
" out, _, _ = preprocess(bad)\n",
" raw.append(bad); cln.append(out); tr_.append(s[\"df\"])\n",
" raw, cln, tr_ = pd.concat(raw), pd.concat(cln), pd.concat(tr_)\n",
" f_raw, fp_raw = hot_cold_f1(raw, tr_)\n",
" f_cln, fp_cln = hot_cold_f1(cln, tr_)\n",
" rows.append({\"상황\": label, \"체온 RMSE 전\": round(rmse(raw[\"temp\"].where(raw[\"temp\"] <= 43).values, tr_[\"core\"].values), 3),\n",
" \"체온 RMSE 후\": round(rmse(cln[\"temp\"].values, tr_[\"core\"].values), 3), \"경보 F1 전\": f_raw, \"경보 F1 후\": f_cln, \"오경보 전\": fp_raw, \"오경보 후\": fp_cln})\n",
"print(\"\\n[보정 전후 비교] (전: 센서 튐·결측·고착 그대로 / 후: 전처리 적용)\")\n",
"print(pd.DataFrame(rows).to_string(index=False))\n",
"print(\"전술 노이즈 대응: 체온 센서 고착(단절)은 8분 이상 같은 값이면 결측 처리 후 보간하고, 범위 밖 값은 센서 오류로 본다. 스푸핑 판별은 본 과제 범위 밖이다.\")\n",
"\n",
"s = data[1]; bad = inject_faults(s[\"df\"], rng); out, _, _ = preprocess(bad)\n",
"plt.figure(figsize=(9, 4))\n",
"plt.plot(bad[\"min\"], bad[\"temp\"], \".\", color=\"tab:red\", label=\"raw (faults)\"); plt.plot(out[\"min\"], out[\"temp\"], color=\"tab:blue\", label=\"cleaned\")\n",
"plt.plot(s[\"df\"][\"min\"], s[\"df\"][\"core\"], \"k:\", label=\"true (simulated)\"); plt.ylim(34, 41); plt.xlabel(\"minute\"); plt.ylabel(\"temp (C)\")\n",
"plt.title(f\"{s['id']} temperature before/after preprocessing (synthetic)\"); plt.legend(); plt.show()\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 02 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"전처리 21,600건(신호값) / 1.20초 = 18,030건/초 (목표 1,000 이상: 달성)\n",
"전처리 오류 0건 -> 오류율 0% (목표 0%), 범위 밖·결측 보정 663건\n",
"[PASS] 결측 보간\n",
"[PASS] 센서 튐 제거\n",
"[PASS] 고착 구간 탐지\n",
"[PASS] 정상 변화 오탐 없음\n",
"[PASS] 고착 주입 후 보정\n",
"단위 테스트 5/5 통과\n",
"\n",
"[보정 전후 비교] (전: 센서 튐·결측·고착 그대로 / 후: 전처리 적용)\n",
"상황 체온 RMSE 전 체온 RMSE 후 경보 F1 전 경보 F1 후 오경보 전 오경보 후\n",
"평시 0.141 0.121 0.000 0.000 4 0\n",
"폭염 0.146 0.104 0.818 0.820 378 417\n",
"혹한 0.219 0.203 0.814 0.828 397 419\n",
"전술 노이즈 대응: 체온 센서 고착(단절)은 8분 이상 같은 값이면 결측 처리 후 보간하고, 범위 밖 값은 센서 오류로 본다. 스푸핑 판별은 본 과제 범위 밖이다.\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 03. 체온-심박수 상관관계 기반 1차 열사병/저체온증 경보 Baseline 모델 구현\n",
"**소요시간**: 약 2.5시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"체온-심박수 상관관계 기반 1차 온열/한랭 스트레스 경보 Baseline 모델을 구현한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **체온과 심박수의 상관관계를 이용해 온열·한랭 스트레스를 1차로 경보하는 규칙 기반 Baseline을 구현하고 판단 기준을 투명하게 기록한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **최근 5분 평균 체온이 38.0도 이상이면서 심박수가 95 이상이면 온열 스트레스 의심, 36.3도 이하이면 한랭 스트레스 의심으로 분류 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `src/models/wel01_baseline_model.py 및 기본 추론 결과 보고서` 형태의 결과물을 완성한다.\n",
"\n",
"### 🔍 평가 및 검증 기준 (PASS 조건)\n",
"#### 미션 판정\n",
"\n",
"**평가 구분:** 필수\n",
"\n",
"**핵심 평가영역:** 알고리즘의 논리적 흐름 및 군사적 판단 기준의 투명성\n",
"\n",
"#### 미션 PASS 조건\n",
"\n",
"- 정량 기준: 기본 데이터셋 대상 탐지/처리 정확도 75% 이상 달성\n",
"\n",
"\n",
"- 진행 기준: 단위 기능 동작 확인 및 기본 시나리오 실행 로그 생성\n",
"\n",
"\n",
"- 검증 방법: 기준 벤치마크 데이터셋 대상 추론 결과 오차율 비교\n",
"\n",
"\n",
"#### 채점자 판정 메모\n",
"\n",
"> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
"\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"\"\"\"체온과 심박수의 상관관계(체온 상승 시 심박수 동반 상승)를 이용한 규칙 기반 1차 온열·한랭 스트레스 경보 Baseline을 구현하고, 가상 병사 데이터에서 정확도와 위험 구간 재현율을 측정한다. 진단 아님·의무대 연계 원칙을 따른다.\"\"\"\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"\n",
"# ---------------- [교육용가상자료] 더미 생체신호 생성기 (실제 장병·부대·의무 기록 없음) ----------------\n",
"# 가상 병사(가명 ID)가 60분간 걷기·휴식·고강도 활동을 하는 동안의 심박수·체온(추정 심부)·피부전도도를 1분 간격으로 만든다.\n",
"# 진단 도구가 아니며, 상태 라벨은 가상 생리 모델의 '교육용 단계'일 뿐이다.\n",
"N_MIN = 60\n",
"STAGE = {0: \"정상\", 1: \"주의\", 2: \"경보\"}\n",
"TYPES = [\"normal\", \"heat\", \"cold\"]\n",
"\n",
"\n",
"def make_person(i, rng):\n",
" return {\"id\": f\"S{i:03d}\", \"rest_hr\": float(rng.normal(64, 7)), \"age\": int(rng.integers(19, 27)), \"bmi\": float(rng.normal(23.5, 2.5)),\n",
" \"acclim\": int(rng.random() < 0.4), \"fit\": float(np.clip(rng.normal(0.5, 0.2), 0.05, 0.95))}\n",
"\n",
"\n",
"def make_env(kind, rng):\n",
" if kind == \"heat\":\n",
" return {\"temp_c\": float(rng.uniform(30, 40)), \"rh\": float(rng.uniform(45, 85)), \"wind\": float(rng.uniform(0.3, 3))}\n",
" if kind == \"cold\":\n",
" return {\"temp_c\": float(rng.uniform(-22, 2)), \"rh\": float(rng.uniform(40, 80)), \"wind\": float(rng.uniform(2, 11))}\n",
" return {\"temp_c\": float(rng.uniform(8, 24)), \"rh\": float(rng.uniform(35, 70)), \"wind\": float(rng.uniform(0.3, 5))}\n",
"\n",
"\n",
"def make_session(sid, kind, rng, p=None, glitch=True):\n",
" \"\"\"반환: dict(person, env, kind, df[hr,temp,eda,act + 정답 core·stage·rtype]). 센서 결측·튐·고착 포함.\"\"\"\n",
" p = p or make_person(sid, rng)\n",
" env = make_env(kind, rng)\n",
" act = np.zeros(N_MIN); t = 0\n",
" while t < N_MIN: # 10~20분 단위 활동 블록: 휴식 1.5 / 행군 4.5 / 고강도 7 MET\n",
" d = int(rng.integers(10, 21)); act[t:t + d] = rng.choice([1.5, 4.5, 7.0], p=[0.25, 0.5, 0.25]); t += d\n",
" hi = env[\"temp_c\"] + (0.12 * (env[\"rh\"] - 40) if env[\"temp_c\"] > 25 else 0)\n",
" wc = env[\"temp_c\"] - 1.6 * env[\"wind\"]\n",
" core, cores = 36.9, []\n",
" for m in range(N_MIN):\n",
" heat = 0.085 * max(0, hi - 24) * (1 - 0.25 * p[\"acclim\"]) * (1 + 0.5 * m / N_MIN) * (1 + 0.012 * (p[\"bmi\"] - 23.5))\n",
" cold = -0.075 * max(0, 8 - wc) * (1 - 0.09 * act[m])\n",
" ss = 36.9 + 0.09 * act[m] + heat + cold\n",
" core += (ss - core) / 18 + rng.normal(0, 0.03)\n",
" cores.append(core)\n",
" core = np.array(cores)\n",
" hr = p[\"rest_hr\"] + 9.5 * (act - 1.5) * (1.15 - 0.3 * p[\"fit\"]) + 11 * (core - 36.9) + rng.normal(0, 3.2, N_MIN)\n",
" eda = 2.0 + 0.25 * act + 3.2 * np.maximum(core - 37.2, 0) - 0.04 * np.maximum(8 - wc, 0) + rng.normal(0, 0.25, N_MIN)\n",
" bias = rng.normal(0, 0.1)\n",
" temp = core + bias + rng.normal(0, 0.1, N_MIN) - np.where((kind == \"cold\") & (env[\"wind\"] > 7), 0.25, 0.0) # 센서 접촉 부위 바람 냉각 오차\n",
" df = pd.DataFrame({\"min\": np.arange(N_MIN), \"hr\": hr, \"temp\": temp, \"eda\": np.maximum(eda, 0.2), \"act\": act})\n",
" if glitch:\n",
" for col in (\"hr\", \"temp\", \"eda\"):\n",
" df.loc[rng.random(N_MIN) < 0.03, col] = np.nan # 결측\n",
" s = int(rng.integers(5, N_MIN))\n",
" if rng.random() < 0.12:\n",
" df.loc[s, \"hr\"] = 230.0 # 센서 튐\n",
" if rng.random() < 0.12:\n",
" df.loc[s, \"temp\"] = 45.0\n",
" ht = np.where(core >= 39.0, 2, np.where(core >= 38.3, 1, 0)); cd = np.where(core <= 35.4, 2, np.where(core <= 36.2, 1, 0))\n",
" df[\"core\"] = core\n",
" df[\"stage\"] = np.maximum(ht, cd); df[\"rtype\"] = np.where(ht > 0, \"heat\", np.where(cd > 0, \"cold\", \"normal\"))\n",
" return {\"id\": p[\"id\"], \"person\": p, \"env\": env, \"kind\": kind, \"df\": df}\n",
"\n",
"\n",
"def make_dataset(n, seed):\n",
" rng = np.random.default_rng(seed)\n",
" kinds = [\"normal\", \"heat\", \"cold\"]\n",
" return [make_session(i, kinds[i % 3], rng) for i in range(n)]\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# ---------------- 전처리·특징 추출 공통 함수 ----------------\n",
"LIMITS = {\"hr\": (30, 220), \"temp\": (30.0, 43.0), \"eda\": (0.0, 40.0)} # 생리적으로 가능한 범위(범위 밖 = 센서 오류)\n",
"\n",
"\n",
"def clean(df):\n",
" \"\"\"범위 밖 값은 결측 처리 -> 선형 보간(양 끝은 가까운 값) -> 3분 이동 중앙값으로 튐 제거. 반환: (정제 df, 보정 건수)\"\"\"\n",
" # [TODO : 학생 실습 3] 체온-심박수 상관관계 기반 1차 열사병/저체온증 경보 Baseline 모델 구현 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def env_index(env):\n",
" # [TODO : 학생 실습 3] 체온-심박수 상관관계 기반 1차 열사병/저체온증 경보 Baseline 모델 구현 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def features(s, use_env=True, personal=True):\n",
" \"\"\"세션 1개 -> 분 단위 특징표(최근 5분 평균·기울기 포함). 정답 stage/rtype 포함.\"\"\"\n",
" df, _ = clean(s[\"df\"]); p, e = s[\"person\"], s[\"env\"]\n",
" f = pd.DataFrame({\"sid\": s[\"id\"], \"min\": df[\"min\"]})\n",
" f[\"temp5\"] = df[\"temp\"].rolling(5, min_periods=1).mean()\n",
" f[\"hr5\"] = df[\"hr\"].rolling(5, min_periods=1).mean()\n",
" f[\"eda5\"] = df[\"eda\"].rolling(5, min_periods=1).mean()\n",
" f[\"dtemp\"] = (f[\"temp5\"] - f[\"temp5\"].shift(4)).fillna(0)\n",
" f[\"act\"] = df[\"act\"]\n",
" if personal:\n",
" f[\"hr_rel\"] = f[\"hr5\"] - p[\"rest_hr\"]\n",
" f[\"acclim\"], f[\"bmi\"], f[\"fit\"] = p[\"acclim\"], p[\"bmi\"], p[\"fit\"]\n",
" if use_env:\n",
" f[\"hi\"], f[\"wc\"] = env_index(e); f[\"wind\"] = e[\"wind\"]\n",
" f[\"stage\"], f[\"rtype\"], f[\"kind\"] = df[\"stage\"], df[\"rtype\"], s[\"kind\"]\n",
" return f[f[\"min\"] >= 5].reset_index(drop=True)\n",
"\n",
"\n",
"def table(sessions, **kw):\n",
" return pd.concat([features(s, **kw) for s in sessions], ignore_index=True)\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# 판단 기준(투명성): ① 최근 5분 평균 체온이 38.0도 이상이고 심박수가 95 이상이면 '온열 스트레스 의심'\n",
"# ② 최근 5분 평균 체온이 36.3도 이하이면 '한랭 스트레스 의심' ③ 그 외 정상\n",
"TH_HEAT, TH_COLD, HR_GATE = 38.0, 36.3, 95\n",
"\n",
"\n",
"def rule(r, th=TH_HEAT, tc=TH_COLD, hg=HR_GATE):\n",
" if r[\"temp5\"] >= th and r[\"hr5\"] >= hg:\n",
" return \"heat\"\n",
" if r[\"temp5\"] <= tc:\n",
" return \"cold\"\n",
" return \"normal\"\n",
"\n",
"\n",
"data = make_dataset(90, seed=7)\n",
"tab = table(data, use_env=False, personal=False)\n",
"tab[\"pred\"] = tab.apply(rule, axis=1)\n",
"acc = (tab[\"pred\"] == tab[\"rtype\"]).mean()\n",
"print(f\"[Baseline] 가상 병사 90명 x 55분 = {len(tab):,}개 구간, 정답 분포 {tab['rtype'].value_counts().to_dict()}\")\n",
"print(f\"경보 유형 분류 정확도 {acc:.1%} (목표 75% 이상) -> {'달성' if acc >= 0.75 else '미달'}\")\n",
"\n",
"print(\"\\n[유형별 판정] (행: 정답, 열: 판정)\")\n",
"cm = pd.crosstab(tab[\"rtype\"], tab[\"pred\"]).reindex(index=TYPES, columns=TYPES, fill_value=0)\n",
"print(cm.to_string())\n",
"for c in (\"heat\", \"cold\"):\n",
" rec = cm.loc[c, c] / cm.loc[c].sum(); prec = cm.loc[c, c] / max(cm[c].sum(), 1)\n",
" print(f\"{c}: 재현율 {rec:.1%}, 정밀도 {prec:.1%}\")\n",
"\n",
"print(\"\\n[단계별 위험 구간 재현율] (위험 유형을 위험으로 맞춘 비율)\")\n",
"risk = tab[tab[\"stage\"] >= 1]\n",
"print(risk.groupby(\"stage\").apply(lambda g: f\"{(g['pred'] == g['rtype']).mean():.1%} ({len(g)}구간)\").to_string())\n",
"\n",
"print(\"\\n[체온-심박수 상관] 전체 상관계수 %.2f / 폭염 세션 %.2f / 혹한 세션 %.2f\" % (\n",
" tab[[\"temp5\", \"hr5\"]].corr().iloc[0, 1], tab[tab[\"kind\"] == \"heat\"][[\"temp5\", \"hr5\"]].corr().iloc[0, 1], tab[tab[\"kind\"] == \"cold\"][[\"temp5\", \"hr5\"]].corr().iloc[0, 1]))\n",
"\n",
"print(\"\\n[임계값 민감도] 온열 체온 기준 변화\")\n",
"for th in (37.7, 38.0, 38.3, 38.6):\n",
" p = tab.apply(lambda r: rule(r, th=th), axis=1)\n",
" r = ((p == \"heat\") & (tab[\"rtype\"] == \"heat\")).sum() / (tab[\"rtype\"] == \"heat\").sum(); f = ((p == \"heat\") & (tab[\"rtype\"] != \"heat\")).sum()\n",
" print(f\"기준 {th}: 온열 재현율 {r:.1%}, 오경보 {f}구간\")\n",
"print(\"한계: 개인별 안정시 심박·기상·활동 강도를 반영하지 않아 운동 직후 심박 상승과 바람에 의한 센서 냉각을 구분하지 못한다 -> 미션 4·5에서 보완한다.\")\n",
"print(\"본 결과는 교육용 가상 자료의 경보 단계이며 진단이 아니다. 실제 이상 징후는 의무대·군의관에게 연계한다.\")\n",
"\n",
"colr = {\"normal\": \"tab:gray\", \"heat\": \"tab:red\", \"cold\": \"tab:blue\"}\n",
"plt.figure(figsize=(7, 5))\n",
"for k, g in tab.groupby(\"rtype\"):\n",
" plt.scatter(g[\"temp5\"], g[\"hr5\"], s=6, alpha=0.4, color=colr[k], label=k)\n",
"plt.axvline(TH_HEAT, ls=\"--\", color=\"k\"); plt.axvline(TH_COLD, ls=\"--\", color=\"k\"); plt.axhline(HR_GATE, ls=\":\", color=\"k\")\n",
"plt.xlabel(\"temperature (5-min mean, C)\"); plt.ylabel(\"heart rate (5-min mean, bpm)\"); plt.legend(); plt.title(\"Baseline rule on dummy signals\")\n",
"plt.show()\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 03 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"[Baseline] 가상 병사 90명 x 55분 = 4,950개 구간, 정답 분포 {'normal': 3536, 'cold': 761, 'heat': 653}\n",
"경보 유형 분류 정확도 86.9% (목표 75% 이상) -> 달성\n",
"\n",
"[유형별 판정] (행: 정답, 열: 판정)\n",
"pred normal heat cold\n",
"rtype \n",
"normal 3056 246 234\n",
"heat 160 493 0\n",
"cold 7 0 754\n",
"heat: 재현율 75.5%, 정밀도 66.7%\n",
"cold: 재현율 99.1%, 정밀도 76.3%\n",
"\n",
"[단계별 위험 구간 재현율] (위험 유형을 위험으로 맞춘 비율)\n",
"stage\n",
"1 86.8% (1221구간)\n",
"2 96.9% (193구간)\n",
"\n",
"[체온-심박수 상관] 전체 상관계수 0.48 / 폭염 세션 0.32 / 혹한 세션 0.41\n",
"\n",
"[임계값 민감도] 온열 체온 기준 변화\n",
"기준 37.7: 온열 재현율 75.7%, 오경보 462구간\n",
"기준 38.0: 온열 재현율 75.5%, 오경보 246구간\n",
"기준 38.3: 온열 재현율 67.7%, 오경보 38구간\n",
"기준 38.6: 온열 재현율 31.1%, 오경보 0구간\n",
"한계: 개인별 안정시 심박·기상·활동 강도를 반영하지 않아 운동 직후 심박 상승과 바람에 의한 센서 냉각을 구분하지 못한다 -> 미션 4·5에서 보완한다.\n",
"본 결과는 교육용 가상 자료의 경보 단계이며 진단이 아니다. 실제 이상 징후는 의무대·군의관에게 연계한다.\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 04. 기상 데이터(온·습도, 풍속) 결합 온열/한랭 스트레스 복합 위험도 예측 모델 고도화\n",
"**소요시간**: 약 3.0시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"기상 데이터와 결합한 복합 환경 위험도 예측 모델을 고도화한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **기상(온도·습도·풍속)과 개인 특성을 결합한 복합 환경 위험도 분류 모델로 고도화하고, 센서 이상·교란 신호를 점검하는 예외 탐지를 추가한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **체감 더위·추위 지수, 안정시 심박 대비 심박 상승, 체온 변화율, 더위 순응 여부 등을 특징으로 그래디언트 부스팅 모델을 학습 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `src/models/wel01_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": [
"\"\"\"체온·심박수·피부전도도에 기상(온도·습도·풍속)과 개인 특성(안정시 심박, 순응 여부, 체격)을 결합한 복합 환경 위험도 분류 모델(그래디언트 부스팅)로 고도화하고, 위험 구간 재현율·처리 지연을 Baseline과 비교한다. 센서 이상 예외 탐지 10종도 시험한다. 진단 아님·의무대 연계.\"\"\"\n",
"import time\n",
"import numpy as np\n",
"import pandas as pd\n",
"from sklearn.ensemble import GradientBoostingClassifier\n",
"\n",
"# ---------------- [교육용가상자료] 더미 생체신호 생성기 (실제 장병·부대·의무 기록 없음) ----------------\n",
"# 가상 병사(가명 ID)가 60분간 걷기·휴식·고강도 활동을 하는 동안의 심박수·체온(추정 심부)·피부전도도를 1분 간격으로 만든다.\n",
"# 진단 도구가 아니며, 상태 라벨은 가상 생리 모델의 '교육용 단계'일 뿐이다.\n",
"N_MIN = 60\n",
"STAGE = {0: \"정상\", 1: \"주의\", 2: \"경보\"}\n",
"TYPES = [\"normal\", \"heat\", \"cold\"]\n",
"\n",
"\n",
"def make_person(i, rng):\n",
" return {\"id\": f\"S{i:03d}\", \"rest_hr\": float(rng.normal(64, 7)), \"age\": int(rng.integers(19, 27)), \"bmi\": float(rng.normal(23.5, 2.5)),\n",
" \"acclim\": int(rng.random() < 0.4), \"fit\": float(np.clip(rng.normal(0.5, 0.2), 0.05, 0.95))}\n",
"\n",
"\n",
"def make_env(kind, rng):\n",
" if kind == \"heat\":\n",
" return {\"temp_c\": float(rng.uniform(30, 40)), \"rh\": float(rng.uniform(45, 85)), \"wind\": float(rng.uniform(0.3, 3))}\n",
" if kind == \"cold\":\n",
" return {\"temp_c\": float(rng.uniform(-22, 2)), \"rh\": float(rng.uniform(40, 80)), \"wind\": float(rng.uniform(2, 11))}\n",
" return {\"temp_c\": float(rng.uniform(8, 24)), \"rh\": float(rng.uniform(35, 70)), \"wind\": float(rng.uniform(0.3, 5))}\n",
"\n",
"\n",
"def make_session(sid, kind, rng, p=None, glitch=True):\n",
" \"\"\"반환: dict(person, env, kind, df[hr,temp,eda,act + 정답 core·stage·rtype]). 센서 결측·튐·고착 포함.\"\"\"\n",
" p = p or make_person(sid, rng)\n",
" env = make_env(kind, rng)\n",
" act = np.zeros(N_MIN); t = 0\n",
" while t < N_MIN: # 10~20분 단위 활동 블록: 휴식 1.5 / 행군 4.5 / 고강도 7 MET\n",
" d = int(rng.integers(10, 21)); act[t:t + d] = rng.choice([1.5, 4.5, 7.0], p=[0.25, 0.5, 0.25]); t += d\n",
" hi = env[\"temp_c\"] + (0.12 * (env[\"rh\"] - 40) if env[\"temp_c\"] > 25 else 0)\n",
" wc = env[\"temp_c\"] - 1.6 * env[\"wind\"]\n",
" core, cores = 36.9, []\n",
" for m in range(N_MIN):\n",
" heat = 0.085 * max(0, hi - 24) * (1 - 0.25 * p[\"acclim\"]) * (1 + 0.5 * m / N_MIN) * (1 + 0.012 * (p[\"bmi\"] - 23.5))\n",
" cold = -0.075 * max(0, 8 - wc) * (1 - 0.09 * act[m])\n",
" ss = 36.9 + 0.09 * act[m] + heat + cold\n",
" core += (ss - core) / 18 + rng.normal(0, 0.03)\n",
" cores.append(core)\n",
" core = np.array(cores)\n",
" hr = p[\"rest_hr\"] + 9.5 * (act - 1.5) * (1.15 - 0.3 * p[\"fit\"]) + 11 * (core - 36.9) + rng.normal(0, 3.2, N_MIN)\n",
" eda = 2.0 + 0.25 * act + 3.2 * np.maximum(core - 37.2, 0) - 0.04 * np.maximum(8 - wc, 0) + rng.normal(0, 0.25, N_MIN)\n",
" bias = rng.normal(0, 0.1)\n",
" temp = core + bias + rng.normal(0, 0.1, N_MIN) - np.where((kind == \"cold\") & (env[\"wind\"] > 7), 0.25, 0.0) # 센서 접촉 부위 바람 냉각 오차\n",
" df = pd.DataFrame({\"min\": np.arange(N_MIN), \"hr\": hr, \"temp\": temp, \"eda\": np.maximum(eda, 0.2), \"act\": act})\n",
" if glitch:\n",
" for col in (\"hr\", \"temp\", \"eda\"):\n",
" df.loc[rng.random(N_MIN) < 0.03, col] = np.nan # 결측\n",
" s = int(rng.integers(5, N_MIN))\n",
" if rng.random() < 0.12:\n",
" df.loc[s, \"hr\"] = 230.0 # 센서 튐\n",
" if rng.random() < 0.12:\n",
" df.loc[s, \"temp\"] = 45.0\n",
" ht = np.where(core >= 39.0, 2, np.where(core >= 38.3, 1, 0)); cd = np.where(core <= 35.4, 2, np.where(core <= 36.2, 1, 0))\n",
" df[\"core\"] = core\n",
" df[\"stage\"] = np.maximum(ht, cd); df[\"rtype\"] = np.where(ht > 0, \"heat\", np.where(cd > 0, \"cold\", \"normal\"))\n",
" return {\"id\": p[\"id\"], \"person\": p, \"env\": env, \"kind\": kind, \"df\": df}\n",
"\n",
"\n",
"def make_dataset(n, seed):\n",
" rng = np.random.default_rng(seed)\n",
" kinds = [\"normal\", \"heat\", \"cold\"]\n",
" return [make_session(i, kinds[i % 3], rng) for i in range(n)]\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# ---------------- 전처리·특징 추출 공통 함수 ----------------\n",
"LIMITS = {\"hr\": (30, 220), \"temp\": (30.0, 43.0), \"eda\": (0.0, 40.0)} # 생리적으로 가능한 범위(범위 밖 = 센서 오류)\n",
"\n",
"\n",
"def clean(df):\n",
" \"\"\"범위 밖 값은 결측 처리 -> 선형 보간(양 끝은 가까운 값) -> 3분 이동 중앙값으로 튐 제거. 반환: (정제 df, 보정 건수)\"\"\"\n",
" # [TODO : 학생 실습 4] 기상 데이터(온·습도, 풍속) 결합 온열/한랭 스트레스 복합 위험도 예측 모델 고도화 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def env_index(env):\n",
" # [TODO : 학생 실습 4] 기상 데이터(온·습도, 풍속) 결합 온열/한랭 스트레스 복합 위험도 예측 모델 고도화 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def features(s, use_env=True, personal=True):\n",
" \"\"\"세션 1개 -> 분 단위 특징표(최근 5분 평균·기울기 포함). 정답 stage/rtype 포함.\"\"\"\n",
" df, _ = clean(s[\"df\"]); p, e = s[\"person\"], s[\"env\"]\n",
" f = pd.DataFrame({\"sid\": s[\"id\"], \"min\": df[\"min\"]})\n",
" f[\"temp5\"] = df[\"temp\"].rolling(5, min_periods=1).mean()\n",
" f[\"hr5\"] = df[\"hr\"].rolling(5, min_periods=1).mean()\n",
" f[\"eda5\"] = df[\"eda\"].rolling(5, min_periods=1).mean()\n",
" f[\"dtemp\"] = (f[\"temp5\"] - f[\"temp5\"].shift(4)).fillna(0)\n",
" f[\"act\"] = df[\"act\"]\n",
" if personal:\n",
" f[\"hr_rel\"] = f[\"hr5\"] - p[\"rest_hr\"]\n",
" f[\"acclim\"], f[\"bmi\"], f[\"fit\"] = p[\"acclim\"], p[\"bmi\"], p[\"fit\"]\n",
" if use_env:\n",
" f[\"hi\"], f[\"wc\"] = env_index(e); f[\"wind\"] = e[\"wind\"]\n",
" f[\"stage\"], f[\"rtype\"], f[\"kind\"] = df[\"stage\"], df[\"rtype\"], s[\"kind\"]\n",
" return f[f[\"min\"] >= 5].reset_index(drop=True)\n",
"\n",
"\n",
"def table(sessions, **kw):\n",
" return pd.concat([features(s, **kw) for s in sessions], ignore_index=True)\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"ADV_FEATS = [\"temp5\", \"hr5\", \"eda5\", \"dtemp\", \"act\", \"hr_rel\", \"acclim\", \"bmi\", \"fit\", \"hi\", \"wc\", \"wind\"]\n",
"LABEL = {\"normal\": 0, \"heat\": 1, \"cold\": 2}\n",
"\n",
"\n",
"def baseline_pred(t):\n",
" return np.where((t[\"temp5\"] >= 38.0) & (t[\"hr5\"] >= 95), \"heat\", np.where(t[\"temp5\"] <= 36.3, \"cold\", \"normal\"))\n",
"\n",
"\n",
"train = table(make_dataset(150, seed=31))\n",
"clf = GradientBoostingClassifier(n_estimators=80, max_depth=3, learning_rate=0.1, random_state=0).fit(train[ADV_FEATS], train[\"rtype\"].map(LABEL))\n",
"inv = {v: k for k, v in LABEL.items()}\n",
"\n",
"# 시험셋 2종: 일반 / 복합 교란(강풍 혹한 + 고강도 활동 + 센서 튐 많음)\n",
"std = make_dataset(90, seed=77)\n",
"rng = np.random.default_rng(9)\n",
"hard = []\n",
"for i in range(60):\n",
" s = make_session(i, [\"heat\", \"cold\"][i % 2], rng)\n",
" if s[\"kind\"] == \"cold\":\n",
" s[\"env\"][\"wind\"] = float(rng.uniform(7, 11))\n",
" for col, val in ((\"hr\", 230.0), (\"temp\", 45.0)):\n",
" idx = rng.choice(np.arange(5, N_MIN), 3, replace=False); s[\"df\"].loc[idx, col] = val\n",
" hard.append(s)\n",
"\n",
"rows = []\n",
"for name, sess in ((\"일반\", std), (\"복합 교란\", hard)):\n",
" t = table(sess)\n",
" pb = baseline_pred(t)\n",
" t0 = time.perf_counter(); pa = np.array([inv[v] for v in clf.predict(t[ADV_FEATS])]); ms = (time.perf_counter() - t0) * 1000 / len(t)\n",
" risk = t[\"stage\"] >= 1; danger = t[\"stage\"] == 2\n",
" rows.append({\"시험셋\": name, \"구간 수\": len(t), \"Baseline 위험 재현율\": round(float((pb[risk] == t[\"rtype\"][risk]).mean()), 3),\n",
" \"고도화 위험 재현율\": round(float((pa[risk] == t[\"rtype\"][risk]).mean()), 3),\n",
" \"고도화 경보(2단계) 재현율\": round(float((pa[danger] == t[\"rtype\"][danger]).mean()), 3),\n",
" \"Baseline 오경보\": int(((pb != \"normal\") & ~risk).sum()), \"고도화 오경보\": int(((pa != \"normal\") & ~risk).sum()), \"지연(ms/구간)\": round(ms, 3)})\n",
"out = pd.DataFrame(rows)\n",
"print(out.T.to_string(header=False))\n",
"rec = out[\"고도화 위험 재현율\"].min(); lat = out[\"지연(ms/구간)\"].max()\n",
"imp = pd.Series(clf.feature_importances_, index=ADV_FEATS).sort_values(ascending=False)\n",
"print(\"\\n특징 중요도 상위:\", \", \".join(f\"{k} {v:.2f}\" for k, v in imp.head(5).items()))\n",
"print(f\"\\n복합 위협 탐지 재현율(최저 시험셋) {rec:.1%} (목표 85% 이상) -> {'달성' if rec >= 0.85 else '미달'}\")\n",
"print(f\"처리 지연 최대 {lat:.3f} ms/구간 (목표 200 ms 이하) -> {'달성' if lat <= 200 else '미달'}\")\n",
"\n",
"\n",
"# ---- 예외(센서 이상·기만) 탐지 10종: 모델에 넣기 전 신호 자체의 신뢰성을 점검한다 ----\n",
"def anomaly(w):\n",
" \"\"\"최근 10분 원시 신호 창(hr, temp, eda 배열 dict)의 이상 유형 목록\"\"\"\n",
" flags = []\n",
" for c in (\"hr\", \"temp\", \"eda\"):\n",
" v = w[c]\n",
" if np.isnan(v).all():\n",
" flags.append(f\"{c}_missing_all\")\n",
" else:\n",
" lo, hi = LIMITS[c]\n",
" if ((v < lo) | (v > hi)).any():\n",
" flags.append(f\"{c}_out_of_range\")\n",
" if np.isnan(v).mean() > 0.3:\n",
" flags.append(f\"{c}_many_missing\")\n",
" ok = v[~np.isnan(v)]\n",
" if len(ok) >= 8 and np.ptp(ok) < 1e-6:\n",
" flags.append(f\"{c}_flatline\")\n",
" if len(ok) >= 3 and np.abs(np.diff(ok)).max() > {\"hr\": 60, \"temp\": 1.5, \"eda\": 10}[c]:\n",
" flags.append(f\"{c}_jump\")\n",
" if \"temp_out_of_range\" not in flags and \"hr_out_of_range\" not in flags and not np.isnan(w[\"temp\"]).all() and not np.isnan(w[\"hr\"]).all():\n",
" if np.nanmean(w[\"temp\"]) >= 38.5 and np.nanmean(w[\"hr\"]) < 60:\n",
" flags.append(\"temp_hr_inconsistent\")\n",
" return flags\n",
"\n",
"\n",
"def win(hr=80, temp=37.0, eda=2.0, n=10):\n",
" return {\"hr\": np.full(n, float(hr)) + np.arange(n) % 3, \"temp\": np.full(n, float(temp)) + np.arange(n) * 0.01, \"eda\": np.full(n, float(eda)) + np.arange(n) % 2 * 0.1}\n",
"\n",
"\n",
"cases = []\n",
"w = win(); cases.append((\"1 정상\", w, []))\n",
"w = win(); w[\"hr\"][4] = 230; cases.append((\"2 심박 튐\", w, [\"hr_out_of_range\", \"hr_jump\"]))\n",
"w = win(); w[\"temp\"][:] = 37.0; cases.append((\"3 체온 고착\", w, [\"temp_flatline\"]))\n",
"w = win(); w[\"temp\"][:6] = np.nan; cases.append((\"4 체온 결측 다수\", w, [\"temp_many_missing\"]))\n",
"w = win(); w[\"hr\"][:] = np.nan; cases.append((\"5 심박 전체 두절\", w, [\"hr_missing_all\"]))\n",
"w = win(); w[\"temp\"][3] = 45.0; cases.append((\"6 체온 범위 이탈\", w, [\"temp_out_of_range\", \"temp_jump\"]))\n",
"w = win(temp=39.0, hr=55); cases.append((\"7 체온-심박 불일치\", w, [\"temp_hr_inconsistent\"]))\n",
"w = win(); w[\"eda\"][:] = 2.0; cases.append((\"8 피부전도 고착\", w, [\"eda_flatline\"]))\n",
"w = win(); w[\"temp\"][5:] += 2.0; cases.append((\"9 체온 급변\", w, [\"temp_jump\"]))\n",
"w = win(); w[\"hr\"][:] = 80.0; w[\"temp\"][:] = 37.0; cases.append((\"10 심박·체온 동시 고착\", w, [\"hr_flatline\", \"temp_flatline\"]))\n",
"hit = 0\n",
"print(\"\\n[센서 이상 탐지 시험 10종]\")\n",
"for name, w, expect in cases:\n",
" got = anomaly(w); good = all(e in got for e in expect) and (bool(got) == bool(expect)); hit += good\n",
" print(f\"[{'PASS' if good else 'FAIL'}] {name}: {', '.join(got) or '이상 없음'}\")\n",
"print(f\"통과 {hit}/10. 이상 플래그가 있으면 모델 경보보다 먼저 '센서 점검 필요'를 표시하고, 판단은 소대장·의무관 확인으로 넘긴다(기만·교란 신호로 인한 오판 방지).\")\n",
"print(\"본 결과는 교육용 가상 자료 기준이며 진단이 아니다. 실제 이상 징후는 의무대·군의관에게 연계한다.\")\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 04 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"시험셋 일반 복합 교란\n",
"구간 수 4950 3300\n",
"Baseline 위험 재현율 0.879 0.916\n",
"고도화 위험 재현율 0.929 0.934\n",
"고도화 경보(2단계) 재현율 1.0 1.0\n",
"Baseline 오경보 446 422\n",
"고도화 오경보 108 98\n",
"지연(ms/구간) 0.004 0.004\n",
"\n",
"특징 중요도 상위: temp5 0.48, eda5 0.42, hr_rel 0.05, hr5 0.01, fit 0.01\n",
"\n",
"복합 위협 탐지 재현율(최저 시험셋) 92.9% (목표 85% 이상) -> 달성\n",
"처리 지연 최대 0.004 ms/구간 (목표 200 ms 이하) -> 달성\n",
"\n",
"[센서 이상 탐지 시험 10종]\n",
"[PASS] 1 정상: 이상 없음\n",
"[PASS] 2 심박 튐: hr_out_of_range, hr_jump\n",
"[PASS] 3 체온 고착: temp_flatline\n",
"[PASS] 4 체온 결측 다수: temp_many_missing\n",
"[PASS] 5 심박 전체 두절: hr_missing_all\n",
"[PASS] 6 체온 범위 이탈: temp_out_of_range, temp_jump\n",
"[PASS] 7 체온-심박 불일치: temp_hr_inconsistent\n",
"[PASS] 8 피부전도 고착: eda_flatline\n",
"[PASS] 9 체온 급변: temp_jump\n",
"[PASS] 10 심박·체온 동시 고착: hr_flatline, temp_flatline\n",
"통과 10/10. 이상 플래그가 있으면 모델 경보보다 먼저 '센서 점검 필요'를 표시하고, 판단은 소대장·의무관 확인으로 넘긴다(기만·교란 신호로 인한 오판 방지).\n",
"본 결과는 교육용 가상 자료 기준이며 진단이 아니다. 실제 이상 징후는 의무대·군의관에게 연계한다.\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 05. 전투원 신체 조건별 개인화 위험 지수 산출 및 오경보 방지 신뢰도 엔진 개발\n",
"**소요시간**: 약 3.0시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"개인화 위험 지수 산출 및 오경보 방지 신뢰도 엔진을 개발한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **병사 개인의 체력·순응 여부를 반영한 개인화 위험 지수를 산출하고 오경보를 줄이는 신뢰도 엔진을 개발한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **활동 강도와 개인 체력으로 예상되는 심박을 빼 '설명되지 않는 심박 상승'과 체온 편차를 만들고, 로지스틱 회귀로 위험 확률을 계산해 0.8 이상 High, 0.5 이상 Medium, 그 미만 Low 등급을 부여 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `src/verification/wel01_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": [
"\"\"\"개인별 안정시 심박·체격·순응 여부를 반영해 활동 강도로 설명되는 심박 상승을 보정한 개인화 위험 지수를 계산하고, 오경보를 줄이는 신뢰도 점수와 등급(High/Medium/Low)을 산출해 고정 임계값 방식 대비 오경보율(FAR) 감축과 점수 일관성을 검증한다. 진단 아님·의무대 연계.\"\"\"\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"from sklearn.linear_model import LogisticRegression\n",
"from sklearn.metrics import roc_auc_score, roc_curve\n",
"from sklearn.pipeline import make_pipeline\n",
"from sklearn.preprocessing import StandardScaler\n",
"\n",
"# ---------------- [교육용가상자료] 더미 생체신호 생성기 (실제 장병·부대·의무 기록 없음) ----------------\n",
"# 가상 병사(가명 ID)가 60분간 걷기·휴식·고강도 활동을 하는 동안의 심박수·체온(추정 심부)·피부전도도를 1분 간격으로 만든다.\n",
"# 진단 도구가 아니며, 상태 라벨은 가상 생리 모델의 '교육용 단계'일 뿐이다.\n",
"N_MIN = 60\n",
"STAGE = {0: \"정상\", 1: \"주의\", 2: \"경보\"}\n",
"TYPES = [\"normal\", \"heat\", \"cold\"]\n",
"\n",
"\n",
"def make_person(i, rng):\n",
" return {\"id\": f\"S{i:03d}\", \"rest_hr\": float(rng.normal(64, 7)), \"age\": int(rng.integers(19, 27)), \"bmi\": float(rng.normal(23.5, 2.5)),\n",
" \"acclim\": int(rng.random() < 0.4), \"fit\": float(np.clip(rng.normal(0.5, 0.2), 0.05, 0.95))}\n",
"\n",
"\n",
"def make_env(kind, rng):\n",
" if kind == \"heat\":\n",
" return {\"temp_c\": float(rng.uniform(30, 40)), \"rh\": float(rng.uniform(45, 85)), \"wind\": float(rng.uniform(0.3, 3))}\n",
" if kind == \"cold\":\n",
" return {\"temp_c\": float(rng.uniform(-22, 2)), \"rh\": float(rng.uniform(40, 80)), \"wind\": float(rng.uniform(2, 11))}\n",
" return {\"temp_c\": float(rng.uniform(8, 24)), \"rh\": float(rng.uniform(35, 70)), \"wind\": float(rng.uniform(0.3, 5))}\n",
"\n",
"\n",
"def make_session(sid, kind, rng, p=None, glitch=True):\n",
" \"\"\"반환: dict(person, env, kind, df[hr,temp,eda,act + 정답 core·stage·rtype]). 센서 결측·튐·고착 포함.\"\"\"\n",
" p = p or make_person(sid, rng)\n",
" env = make_env(kind, rng)\n",
" act = np.zeros(N_MIN); t = 0\n",
" while t < N_MIN: # 10~20분 단위 활동 블록: 휴식 1.5 / 행군 4.5 / 고강도 7 MET\n",
" d = int(rng.integers(10, 21)); act[t:t + d] = rng.choice([1.5, 4.5, 7.0], p=[0.25, 0.5, 0.25]); t += d\n",
" hi = env[\"temp_c\"] + (0.12 * (env[\"rh\"] - 40) if env[\"temp_c\"] > 25 else 0)\n",
" wc = env[\"temp_c\"] - 1.6 * env[\"wind\"]\n",
" core, cores = 36.9, []\n",
" for m in range(N_MIN):\n",
" heat = 0.085 * max(0, hi - 24) * (1 - 0.25 * p[\"acclim\"]) * (1 + 0.5 * m / N_MIN) * (1 + 0.012 * (p[\"bmi\"] - 23.5))\n",
" cold = -0.075 * max(0, 8 - wc) * (1 - 0.09 * act[m])\n",
" ss = 36.9 + 0.09 * act[m] + heat + cold\n",
" core += (ss - core) / 18 + rng.normal(0, 0.03)\n",
" cores.append(core)\n",
" core = np.array(cores)\n",
" hr = p[\"rest_hr\"] + 9.5 * (act - 1.5) * (1.15 - 0.3 * p[\"fit\"]) + 11 * (core - 36.9) + rng.normal(0, 3.2, N_MIN)\n",
" eda = 2.0 + 0.25 * act + 3.2 * np.maximum(core - 37.2, 0) - 0.04 * np.maximum(8 - wc, 0) + rng.normal(0, 0.25, N_MIN)\n",
" bias = rng.normal(0, 0.1)\n",
" temp = core + bias + rng.normal(0, 0.1, N_MIN) - np.where((kind == \"cold\") & (env[\"wind\"] > 7), 0.25, 0.0) # 센서 접촉 부위 바람 냉각 오차\n",
" df = pd.DataFrame({\"min\": np.arange(N_MIN), \"hr\": hr, \"temp\": temp, \"eda\": np.maximum(eda, 0.2), \"act\": act})\n",
" if glitch:\n",
" for col in (\"hr\", \"temp\", \"eda\"):\n",
" df.loc[rng.random(N_MIN) < 0.03, col] = np.nan # 결측\n",
" s = int(rng.integers(5, N_MIN))\n",
" if rng.random() < 0.12:\n",
" df.loc[s, \"hr\"] = 230.0 # 센서 튐\n",
" if rng.random() < 0.12:\n",
" df.loc[s, \"temp\"] = 45.0\n",
" ht = np.where(core >= 39.0, 2, np.where(core >= 38.3, 1, 0)); cd = np.where(core <= 35.4, 2, np.where(core <= 36.2, 1, 0))\n",
" df[\"core\"] = core\n",
" df[\"stage\"] = np.maximum(ht, cd); df[\"rtype\"] = np.where(ht > 0, \"heat\", np.where(cd > 0, \"cold\", \"normal\"))\n",
" return {\"id\": p[\"id\"], \"person\": p, \"env\": env, \"kind\": kind, \"df\": df}\n",
"\n",
"\n",
"def make_dataset(n, seed):\n",
" rng = np.random.default_rng(seed)\n",
" kinds = [\"normal\", \"heat\", \"cold\"]\n",
" return [make_session(i, kinds[i % 3], rng) for i in range(n)]\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# ---------------- 전처리·특징 추출 공통 함수 ----------------\n",
"LIMITS = {\"hr\": (30, 220), \"temp\": (30.0, 43.0), \"eda\": (0.0, 40.0)} # 생리적으로 가능한 범위(범위 밖 = 센서 오류)\n",
"\n",
"\n",
"def clean(df):\n",
" \"\"\"범위 밖 값은 결측 처리 -> 선형 보간(양 끝은 가까운 값) -> 3분 이동 중앙값으로 튐 제거. 반환: (정제 df, 보정 건수)\"\"\"\n",
" # [TODO : 학생 실습 5] 전투원 신체 조건별 개인화 위험 지수 산출 및 오경보 방지 신뢰도 엔진 개발 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def env_index(env):\n",
" # [TODO : 학생 실습 5] 전투원 신체 조건별 개인화 위험 지수 산출 및 오경보 방지 신뢰도 엔진 개발 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def features(s, use_env=True, personal=True):\n",
" \"\"\"세션 1개 -> 분 단위 특징표(최근 5분 평균·기울기 포함). 정답 stage/rtype 포함.\"\"\"\n",
" df, _ = clean(s[\"df\"]); p, e = s[\"person\"], s[\"env\"]\n",
" f = pd.DataFrame({\"sid\": s[\"id\"], \"min\": df[\"min\"]})\n",
" f[\"temp5\"] = df[\"temp\"].rolling(5, min_periods=1).mean()\n",
" f[\"hr5\"] = df[\"hr\"].rolling(5, min_periods=1).mean()\n",
" f[\"eda5\"] = df[\"eda\"].rolling(5, min_periods=1).mean()\n",
" f[\"dtemp\"] = (f[\"temp5\"] - f[\"temp5\"].shift(4)).fillna(0)\n",
" f[\"act\"] = df[\"act\"]\n",
" if personal:\n",
" f[\"hr_rel\"] = f[\"hr5\"] - p[\"rest_hr\"]\n",
" f[\"acclim\"], f[\"bmi\"], f[\"fit\"] = p[\"acclim\"], p[\"bmi\"], p[\"fit\"]\n",
" if use_env:\n",
" f[\"hi\"], f[\"wc\"] = env_index(e); f[\"wind\"] = e[\"wind\"]\n",
" f[\"stage\"], f[\"rtype\"], f[\"kind\"] = df[\"stage\"], df[\"rtype\"], s[\"kind\"]\n",
" return f[f[\"min\"] >= 5].reset_index(drop=True)\n",
"\n",
"\n",
"def table(sessions, **kw):\n",
" return pd.concat([features(s, **kw) for s in sessions], ignore_index=True)\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"PF = [\"temp5\", \"dtemp\", \"hr_excess\", \"eda5\", \"hi\", \"wc\", \"acclim\", \"bmi\", \"tdev\"]\n",
"\n",
"\n",
"def personalize(t):\n",
" \"\"\"활동 강도·개인 체력으로 예상되는 심박을 빼 '설명되지 않는 심박 상승(hr_excess)'과 체온 편차(tdev)를 만든다.\"\"\"\n",
" t = t.copy()\n",
" expected = t[\"hr5\"] - t[\"hr_rel\"] + 9.5 * (t[\"act\"].rolling(5, min_periods=1).mean() - 1.5) * (1.15 - 0.3 * t[\"fit\"])\n",
" t[\"hr_excess\"] = t[\"hr5\"] - expected\n",
" t[\"tdev\"] = (t[\"temp5\"] - 36.9).abs()\n",
" return t\n",
"\n",
"\n",
"def grade(p):\n",
" return \"High\" if p >= 0.8 else (\"Medium\" if p >= 0.5 else \"Low\")\n",
"\n",
"\n",
"train = personalize(table(make_dataset(180, seed=41)))\n",
"test = personalize(table(make_dataset(90, seed=43)))\n",
"model = make_pipeline(StandardScaler(), LogisticRegression(class_weight=\"balanced\", C=1.0, max_iter=3000, random_state=0))\n",
"model.fit(train[PF], train[\"stage\"] >= 1)\n",
"test[\"score\"] = model.predict_proba(test[PF])[:, 1]\n",
"test[\"grade\"] = test[\"score\"].map(grade)\n",
"true = (test[\"stage\"] >= 1).values\n",
"\n",
"fixed = ((test[\"temp5\"] >= 37.8) | (test[\"temp5\"] <= 36.5) | (test[\"hr5\"] >= 130)).values # 기존: 고정 임계값 경보\n",
"eng = (test[\"score\"] >= 0.5).values\n",
"\n",
"\n",
"def far_rec(alert):\n",
" return (alert & ~true).sum() / max(alert.sum(), 1), (alert & true).sum() / true.sum()\n",
"\n",
"\n",
"f0, r0 = far_rec(fixed); f1, r1 = far_rec(eng)\n",
"print(pd.DataFrame([{\"방식\": \"고정 임계값(체온 37.8 이상·36.5 이하·심박 130 이상)\", \"경보 구간\": int(fixed.sum()), \"오경보율\": round(f0, 3), \"위험 재현율\": round(r0, 3)},\n",
" {\"방식\": \"개인화 신뢰도 엔진(0.5 이상)\", \"경보 구간\": int(eng.sum()), \"오경보율\": round(f1, 3), \"위험 재현율\": round(r1, 3)}]).to_string(index=False))\n",
"cut = 1 - f1 / f0\n",
"print(f\"\\n오경보율 감축 {cut:.1%} (목표 40% 이상: {'달성' if cut >= 0.4 else '미달'}), ROC-AUC {roc_auc_score(true, test['score']):.3f}\")\n",
"\n",
"# 점수 일관성: 센서 값에 +-2% 측정 흔들림을 주어도 같은 등급이 유지되는 비율\n",
"rng = np.random.default_rng(0)\n",
"X = test[PF].values; g0 = test[\"grade\"].values\n",
"same = [np.mean(np.array([grade(v) for v in model.predict_proba(pd.DataFrame(X * (1 + rng.normal(0, 0.02, X.shape)), columns=PF))[:, 1]]) == g0) for _ in range(10)]\n",
"print(f\"신뢰도 등급 일관성 {np.mean(same):.1%} (같은 입력 재실행 100%, 목표 90% 이상: {'달성' if np.mean(same) >= 0.9 else '미달'})\")\n",
"\n",
"print(\"\\n[단계별 등급 분포] (행: 정답 단계)\")\n",
"print(pd.crosstab(test[\"stage\"].map(STAGE).rename(\"정답 단계\"), test[\"grade\"].rename(\"등급\")).to_string())\n",
"\n",
"# 개인화 효과: 같은 심박·체온이라도 체력·순응 여부에 따라 지수가 달라지는 예\n",
"ex = test.iloc[[0]].copy()\n",
"rows = []\n",
"for name, fit, acc in ((\"체력 낮음·비순응\", 0.2, 0), (\"체력 높음·순응\", 0.8, 1)):\n",
" e = ex.copy(); e[\"fit\"], e[\"acclim\"] = fit, acc\n",
" e[\"hr_rel\"] = e[\"hr5\"] - 64; e[\"hr_excess\"] = 0; e = personalize(e); e[\"act\"] = 4.5\n",
" e[\"hr5\"] = 128.0; e[\"temp5\"] = 38.3; e[\"hi\"], e[\"wc\"] = 41.0, 30.0; e[\"tdev\"] = 1.1; e[\"dtemp\"] = 0.2; e[\"eda5\"] = 3.5; e = personalize(e)\n",
" rows.append({\"가상 병사\": name, \"초과 심박\": round(float(e[\"hr_excess\"].iloc[0]), 1), \"위험 점수\": round(float(model.predict_proba(e[PF])[:, 1][0]), 3)})\n",
"print(\"\\n[개인화 예시: 같은 심박 128·체온 38.3]\")\n",
"print(pd.DataFrame(rows).to_string(index=False))\n",
"print(\"같은 측정값이라도 체력·더위 순응 여부에 따라 위험 점수가 달라진다(순응 병사는 같은 체온에서 점수가 낮아짐).\")\n",
"\n",
"sc, lr = model[0], model[-1]\n",
"i = int(np.argmax(test[\"score\"].values)); c = (X[i] - sc.mean_) / sc.scale_ * lr.coef_[0]\n",
"print(f\"\\n최고 점수 구간 근거: \", \", \".join(f\"{PF[j]}({c[j]:+.2f})\" for j in np.argsort(-np.abs(c))[:3]))\n",
"print(\"등급 기준: High(0.8 이상)는 즉시 소대장 확인·의무대 연락 검토, Medium(0.5~0.8)은 재측정·휴식 확인, Low(0.5 미만)는 모니터링 유지. 점수는 확인 순서를 정하는 보조 정보이며 진단이 아니다.\")\n",
"\n",
"fig, ax = plt.subplots(1, 2, figsize=(11, 4))\n",
"for st, col in ((0, \"tab:gray\"), (1, \"tab:orange\"), (2, \"tab:red\")):\n",
" ax[0].hist(test.loc[test[\"stage\"] == st, \"score\"], bins=10, range=(0, 1), alpha=0.6, color=col, label=f\"stage {st}\")\n",
"ax[0].axvline(0.5, color=\"k\", ls=\"--\"); ax[0].set_xlabel(\"confidence score\"); ax[0].legend(); ax[0].set_title(\"Score by true stage\")\n",
"fpr, tpr, _ = roc_curve(true, test[\"score\"]); ax[1].plot(fpr, tpr); ax[1].plot([0, 1], [0, 1], \"k:\"); ax[1].set_title(\"ROC\"); ax[1].set_xlabel(\"FPR\"); ax[1].set_ylabel(\"TPR\")\n",
"plt.tight_layout(); plt.show()\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 05 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
" 방식 경보 구간 오경보율 위험 재현율\n",
"고정 임계값(체온 37.8 이상·36.5 이하·심박 130 이상) 2614 0.392 0.999\n",
" 개인화 신뢰도 엔진(0.5 이상) 1797 0.165 0.943\n",
"\n",
"오경보율 감축 57.9% (목표 40% 이상: 달성), ROC-AUC 0.986\n",
"신뢰도 등급 일관성 92.9% (같은 입력 재실행 100%, 목표 90% 이상: 달성)\n",
"\n",
"[단계별 등급 분포] (행: 정답 단계)\n",
"등급 High Low Medium\n",
"정답 단계 \n",
"경보 183 0 0\n",
"정상 96 3063 201\n",
"주의 1153 90 164\n",
"\n",
"[개인화 예시: 같은 심박 128·체온 38.3]\n",
" 가상 병사 초과 심박 위험 점수\n",
"체력 낮음·비순응 1.9 0.288\n",
" 체력 높음·순응 7.0 0.170\n",
"같은 측정값이라도 체력·더위 순응 여부에 따라 위험 점수가 달라진다(순응 병사는 같은 체온에서 점수가 낮아짐).\n",
"\n",
"최고 점수 구간 근거: tdev(+21.02), hi(+8.18), wc(-7.12)\n",
"등급 기준: High(0.8 이상)는 즉시 소대장 확인·의무대 연락 검토, Medium(0.5~0.8)은 재측정·휴식 확인, Low(0.5 미만)는 모니터링 유지. 점수는 확인 순서를 정하는 보조 정보이며 진단이 아니다.\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 06. 소대장/의무관용 장병 생체 위험 신호 모니터링 및 긴급 경보 대시보드 구현\n",
"**소요시간**: 약 2.5시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"지휘관/의무관용 장병 생체 위험 신호 모니터링 대시보드를 구현한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **소대장과 의무관이 가상 소대 병사들의 생체 위험 신호를 한눈에 보고 확인 대상을 정하는 모니터링 상황판을 구현한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **가상 병사 24명의 상태를 정상·주의·경보·신호두절로 나누고 체온·심박·추세·권고 문구를 표로 작성하여 원인을 분석하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `src/ui/wel01_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": [
"\"\"\"소대장·의무관용 가상 소대 24명의 생체 위험 신호 모니터링 상황판을 구현한다. 병사별 상태(정상·주의·경보·신호두절), 체온·심박 추세, 조치 권고 문구를 1초 주기로 갱신하고 처리시간을 측정한다. 진단 아님·의무대 연계.\"\"\"\n",
"import time\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"from sklearn.linear_model import LogisticRegression\n",
"from sklearn.pipeline import make_pipeline\n",
"from sklearn.preprocessing import StandardScaler\n",
"\n",
"# ---------------- [교육용가상자료] 더미 생체신호 생성기 (실제 장병·부대·의무 기록 없음) ----------------\n",
"# 가상 병사(가명 ID)가 60분간 걷기·휴식·고강도 활동을 하는 동안의 심박수·체온(추정 심부)·피부전도도를 1분 간격으로 만든다.\n",
"# 진단 도구가 아니며, 상태 라벨은 가상 생리 모델의 '교육용 단계'일 뿐이다.\n",
"N_MIN = 60\n",
"STAGE = {0: \"정상\", 1: \"주의\", 2: \"경보\"}\n",
"TYPES = [\"normal\", \"heat\", \"cold\"]\n",
"\n",
"\n",
"def make_person(i, rng):\n",
" return {\"id\": f\"S{i:03d}\", \"rest_hr\": float(rng.normal(64, 7)), \"age\": int(rng.integers(19, 27)), \"bmi\": float(rng.normal(23.5, 2.5)),\n",
" \"acclim\": int(rng.random() < 0.4), \"fit\": float(np.clip(rng.normal(0.5, 0.2), 0.05, 0.95))}\n",
"\n",
"\n",
"def make_env(kind, rng):\n",
" if kind == \"heat\":\n",
" return {\"temp_c\": float(rng.uniform(30, 40)), \"rh\": float(rng.uniform(45, 85)), \"wind\": float(rng.uniform(0.3, 3))}\n",
" if kind == \"cold\":\n",
" return {\"temp_c\": float(rng.uniform(-22, 2)), \"rh\": float(rng.uniform(40, 80)), \"wind\": float(rng.uniform(2, 11))}\n",
" return {\"temp_c\": float(rng.uniform(8, 24)), \"rh\": float(rng.uniform(35, 70)), \"wind\": float(rng.uniform(0.3, 5))}\n",
"\n",
"\n",
"def make_session(sid, kind, rng, p=None, glitch=True):\n",
" \"\"\"반환: dict(person, env, kind, df[hr,temp,eda,act + 정답 core·stage·rtype]). 센서 결측·튐·고착 포함.\"\"\"\n",
" p = p or make_person(sid, rng)\n",
" env = make_env(kind, rng)\n",
" act = np.zeros(N_MIN); t = 0\n",
" while t < N_MIN: # 10~20분 단위 활동 블록: 휴식 1.5 / 행군 4.5 / 고강도 7 MET\n",
" d = int(rng.integers(10, 21)); act[t:t + d] = rng.choice([1.5, 4.5, 7.0], p=[0.25, 0.5, 0.25]); t += d\n",
" hi = env[\"temp_c\"] + (0.12 * (env[\"rh\"] - 40) if env[\"temp_c\"] > 25 else 0)\n",
" wc = env[\"temp_c\"] - 1.6 * env[\"wind\"]\n",
" core, cores = 36.9, []\n",
" for m in range(N_MIN):\n",
" heat = 0.085 * max(0, hi - 24) * (1 - 0.25 * p[\"acclim\"]) * (1 + 0.5 * m / N_MIN) * (1 + 0.012 * (p[\"bmi\"] - 23.5))\n",
" cold = -0.075 * max(0, 8 - wc) * (1 - 0.09 * act[m])\n",
" ss = 36.9 + 0.09 * act[m] + heat + cold\n",
" core += (ss - core) / 18 + rng.normal(0, 0.03)\n",
" cores.append(core)\n",
" core = np.array(cores)\n",
" hr = p[\"rest_hr\"] + 9.5 * (act - 1.5) * (1.15 - 0.3 * p[\"fit\"]) + 11 * (core - 36.9) + rng.normal(0, 3.2, N_MIN)\n",
" eda = 2.0 + 0.25 * act + 3.2 * np.maximum(core - 37.2, 0) - 0.04 * np.maximum(8 - wc, 0) + rng.normal(0, 0.25, N_MIN)\n",
" bias = rng.normal(0, 0.1)\n",
" temp = core + bias + rng.normal(0, 0.1, N_MIN) - np.where((kind == \"cold\") & (env[\"wind\"] > 7), 0.25, 0.0) # 센서 접촉 부위 바람 냉각 오차\n",
" df = pd.DataFrame({\"min\": np.arange(N_MIN), \"hr\": hr, \"temp\": temp, \"eda\": np.maximum(eda, 0.2), \"act\": act})\n",
" if glitch:\n",
" for col in (\"hr\", \"temp\", \"eda\"):\n",
" df.loc[rng.random(N_MIN) < 0.03, col] = np.nan # 결측\n",
" s = int(rng.integers(5, N_MIN))\n",
" if rng.random() < 0.12:\n",
" df.loc[s, \"hr\"] = 230.0 # 센서 튐\n",
" if rng.random() < 0.12:\n",
" df.loc[s, \"temp\"] = 45.0\n",
" ht = np.where(core >= 39.0, 2, np.where(core >= 38.3, 1, 0)); cd = np.where(core <= 35.4, 2, np.where(core <= 36.2, 1, 0))\n",
" df[\"core\"] = core\n",
" df[\"stage\"] = np.maximum(ht, cd); df[\"rtype\"] = np.where(ht > 0, \"heat\", np.where(cd > 0, \"cold\", \"normal\"))\n",
" return {\"id\": p[\"id\"], \"person\": p, \"env\": env, \"kind\": kind, \"df\": df}\n",
"\n",
"\n",
"def make_dataset(n, seed):\n",
" rng = np.random.default_rng(seed)\n",
" kinds = [\"normal\", \"heat\", \"cold\"]\n",
" return [make_session(i, kinds[i % 3], rng) for i in range(n)]\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# ---------------- 전처리·특징 추출 공통 함수 ----------------\n",
"LIMITS = {\"hr\": (30, 220), \"temp\": (30.0, 43.0), \"eda\": (0.0, 40.0)} # 생리적으로 가능한 범위(범위 밖 = 센서 오류)\n",
"\n",
"\n",
"def clean(df):\n",
" \"\"\"범위 밖 값은 결측 처리 -> 선형 보간(양 끝은 가까운 값) -> 3분 이동 중앙값으로 튐 제거. 반환: (정제 df, 보정 건수)\"\"\"\n",
" # [TODO : 학생 실습 6] 소대장/의무관용 장병 생체 위험 신호 모니터링 및 긴급 경보 대시보드 구현 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def env_index(env):\n",
" # [TODO : 학생 실습 6] 소대장/의무관용 장병 생체 위험 신호 모니터링 및 긴급 경보 대시보드 구현 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def features(s, use_env=True, personal=True):\n",
" \"\"\"세션 1개 -> 분 단위 특징표(최근 5분 평균·기울기 포함). 정답 stage/rtype 포함.\"\"\"\n",
" df, _ = clean(s[\"df\"]); p, e = s[\"person\"], s[\"env\"]\n",
" f = pd.DataFrame({\"sid\": s[\"id\"], \"min\": df[\"min\"]})\n",
" f[\"temp5\"] = df[\"temp\"].rolling(5, min_periods=1).mean()\n",
" f[\"hr5\"] = df[\"hr\"].rolling(5, min_periods=1).mean()\n",
" f[\"eda5\"] = df[\"eda\"].rolling(5, min_periods=1).mean()\n",
" f[\"dtemp\"] = (f[\"temp5\"] - f[\"temp5\"].shift(4)).fillna(0)\n",
" f[\"act\"] = df[\"act\"]\n",
" if personal:\n",
" f[\"hr_rel\"] = f[\"hr5\"] - p[\"rest_hr\"]\n",
" f[\"acclim\"], f[\"bmi\"], f[\"fit\"] = p[\"acclim\"], p[\"bmi\"], p[\"fit\"]\n",
" if use_env:\n",
" f[\"hi\"], f[\"wc\"] = env_index(e); f[\"wind\"] = e[\"wind\"]\n",
" f[\"stage\"], f[\"rtype\"], f[\"kind\"] = df[\"stage\"], df[\"rtype\"], s[\"kind\"]\n",
" return f[f[\"min\"] >= 5].reset_index(drop=True)\n",
"\n",
"\n",
"def table(sessions, **kw):\n",
" return pd.concat([features(s, **kw) for s in sessions], ignore_index=True)\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"FE = [\"temp5\", \"dtemp\", \"hr5\", \"eda5\", \"hr_rel\", \"acclim\", \"bmi\", \"fit\", \"hi\", \"wc\", \"wind\"]\n",
"tr = table(make_dataset(120, seed=51))\n",
"model = make_pipeline(StandardScaler(), LogisticRegression(class_weight=\"balanced\", max_iter=3000, random_state=0)).fit(tr[FE], tr[\"stage\"] >= 1)\n",
"\n",
"# [교육용가상자료] 가상 소대 24명(가명 ID): 폭염 8, 혹한 8, 평시 8 구역에서 활동 중인 병사들의 화면\n",
"rng = np.random.default_rng(3)\n",
"kinds = [\"heat\"] * 8 + [\"cold\"] * 8 + [\"normal\"] * 8\n",
"squad = [make_session(i, k, rng) for i, k in enumerate(kinds)]\n",
"for s in squad[:2]:\n",
" s[\"df\"].loc[40:, [\"hr\", \"temp\", \"eda\"]] = np.nan # 신호 두절 2명(기기 이탈·배터리 방전 모의)\n",
"FEAT = {s[\"id\"]: features(s) for s in squad} # 병사별 분 단위 특징표(재생용)\n",
"RAW = {s[\"id\"]: s[\"df\"] for s in squad}\n",
"\n",
"\n",
"def snapshot(minute):\n",
" \"\"\"분 단위 1틱: 병사별 상태·점수·권고를 계산해 상황판 표를 만든다.\"\"\"\n",
" rows = []\n",
" for s in squad:\n",
" f = FEAT[s[\"id\"]]; r = f[f[\"min\"] == minute]\n",
" raw = RAW[s[\"id\"]].loc[max(minute - 4, 0):minute, [\"hr\", \"temp\"]]\n",
" if raw.isna().all().all():\n",
" rows.append({\"병사\": s[\"id\"], \"상태\": \"신호두절\", \"점수\": np.nan, \"체온\": np.nan, \"심박\": np.nan, \"추세\": \"-\", \"권고\": \"기기 착용·배터리 확인, 소대장 대면 확인\"}); continue\n",
" sc = float(model.predict_proba(r[FE])[:, 1][0])\n",
" st = \"경보\" if sc >= 0.8 else (\"주의\" if sc >= 0.5 else \"정상\")\n",
" rec = {\"경보\": \"소대장 즉시 확인, 의무대 연락 검토\", \"주의\": \"재측정·휴식 상태 확인\", \"정상\": \"-\"}[st]\n",
" rows.append({\"병사\": s[\"id\"], \"상태\": st, \"점수\": round(sc, 2), \"체온\": round(float(r[\"temp5\"].iloc[0]), 1), \"심박\": int(r[\"hr5\"].iloc[0]),\n",
" \"추세\": f\"{float(r['dtemp'].iloc[0]):+.1f}\", \"권고\": rec})\n",
" return pd.DataFrame(rows)\n",
"\n",
"\n",
"# 1) 1초 주기 갱신 모의: 5틱 갱신 시간 측정\n",
"times = []\n",
"for m in range(45, 50):\n",
" t0 = time.perf_counter(); board = snapshot(m); times.append((time.perf_counter() - t0) * 1000)\n",
"print(f\"상황판 갱신 처리시간 평균 {np.mean(times):.1f} ms, 최대 {max(times):.1f} ms (목표: 갱신 주기 1초 이내 -> {'달성' if max(times) < 1000 else '미달'})\")\n",
"\n",
"# 2) 병사 24명 -> 확인 대상 요약(인지 부하 감소)\n",
"order = {\"신호두절\": 0, \"경보\": 1, \"주의\": 2, \"정상\": 3}\n",
"board = board.sort_values([\"상태\", \"점수\"], key=lambda s: s.map(order) if s.name == \"상태\" else -s.fillna(2), kind=\"stable\")\n",
"cnt = board[\"상태\"].value_counts().reindex([\"경보\", \"주의\", \"신호두절\", \"정상\"]).fillna(0).astype(int)\n",
"print(f\"\\n[소대 현황] 병사 {len(board)}명 -> 확인 대상 {len(board) - cnt['정상']}명으로 압축:\", cnt.to_dict())\n",
"print(\"\\n[우선 확인 목록: 신호두절 > 경보 > 주의 순]\")\n",
"print(board[board[\"상태\"] != \"정상\"].head(12).fillna(\"-\").to_string(index=False))\n",
"\n",
"# 3) 렌더링 시간(UI 응답성)과 상황판 그림\n",
"t0 = time.perf_counter()\n",
"col = {\"경보\": \"tab:red\", \"주의\": \"orange\", \"정상\": \"tab:green\", \"신호두절\": \"dimgray\"}; EN = {\"경보\": \"ALERT\", \"주의\": \"CAUTION\", \"정상\": \"NORMAL\", \"신호두절\": \"NO SIGNAL\"}\n",
"fig, ax = plt.subplots(2, 3, figsize=(13, 6))\n",
"focus = board[board[\"상태\"].isin([\"경보\", \"주의\"])].head(5)[\"병사\"].tolist()\n",
"for a, sid in zip(ax.ravel()[:5], focus):\n",
" d = RAW[sid]; st = board[board[\"병사\"] == sid][\"상태\"].iloc[0]\n",
" a.plot(d[\"min\"], d[\"temp\"].interpolate(), color=col[st]); a.axvline(49, color=\"k\", ls=\":\"); a.set_ylim(34, 41)\n",
" a.set_title(f\"{sid} [{EN[st]}] temperature trend\", fontsize=9); a.set_xlabel(\"minute\")\n",
"ax.ravel()[5].bar([EN[k] for k in cnt.index], cnt.values, color=[col[k] for k in cnt.index]); ax.ravel()[5].set_title(\"Platoon status\")\n",
"plt.suptitle(\"Soldier bio-signal board (dummy data, not diagnosis)\"); plt.tight_layout(); fig.canvas.draw()\n",
"render = (time.perf_counter() - t0) * 1000\n",
"print(f\"\\n한 화면 렌더링 {render:.0f} ms (UI 응답 목표 100 ms: 데이터 갱신 {np.mean(times):.1f} ms는 달성, 렌더링은 실행 환경에 따라 달라짐)\")\n",
"print(\"상황판의 상태와 권고는 소대장·의무관의 확인을 돕는 보조 정보이며 진단이 아니다. 의학적 판단과 조치는 군의관·의무대가 하며, 조치 명령은 미션 7의 HITL 절차를 따른다.\")\n",
"plt.show()\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 06 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"상황판 갱신 처리시간 평균 74.0 ms, 최대 77.3 ms (목표: 갱신 주기 1초 이내 -> 달성)\n",
"\n",
"[소대 현황] 병사 24명 -> 확인 대상 12명으로 압축: {'경보': 7, '주의': 3, '신호두절': 2, '정상': 12}\n",
"\n",
"[우선 확인 목록: 신호두절 > 경보 > 주의 순]\n",
" 병사 상태 점수 체온 심박 추세 권고\n",
"S000 신호두절 - - - - 기기 착용·배터리 확인, 소대장 대면 확인\n",
"S001 신호두절 - - - - 기기 착용·배터리 확인, 소대장 대면 확인\n",
"S003 경보 1.0 38.7 119.0 +0.1 소대장 즉시 확인, 의무대 연락 검토\n",
"S010 경보 1.0 35.4 65.0 -0.1 소대장 즉시 확인, 의무대 연락 검토\n",
"S009 경보 0.99 35.3 93.0 -0.1 소대장 즉시 확인, 의무대 연락 검토\n",
"S006 경보 0.96 38.5 110.0 +0.1 소대장 즉시 확인, 의무대 연락 검토\n",
"S014 경보 0.94 35.9 85.0 -0.1 소대장 즉시 확인, 의무대 연락 검토\n",
"S004 경보 0.93 38.3 81.0 -0.0 소대장 즉시 확인, 의무대 연락 검토\n",
"S005 경보 0.86 38.1 125.0 +0.0 소대장 즉시 확인, 의무대 연락 검토\n",
"S007 주의 0.67 38.5 110.0 +0.0 재측정·휴식 상태 확인\n",
"S002 주의 0.62 38.3 91.0 +0.1 재측정·휴식 상태 확인\n",
"S012 주의 0.52 36.2 92.0 -0.0 재측정·휴식 상태 확인\n",
"\n",
"한 화면 렌더링 565 ms (UI 응답 목표 100 ms: 데이터 갱신 74.0 ms는 달성, 렌더링은 실행 환경에 따라 달라짐)\n",
"상황판의 상태와 권고는 소대장·의무관의 확인을 돕는 보조 정보이며 진단이 아니다. 의학적 판단과 조치는 군의관·의무대가 하며, 조치 명령은 미션 7의 HITL 절차를 따른다.\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",
"이 문제를 해결하기 위해 본 미션에서는 **AI 위험 신호는 점수와 무관하게 검토 대기로 등록하고 등급·점수 순으로 검토 순서를 정 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `src/hitl/wel01_approval_workflow.py 및 docs/wel01_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) 절차를 구현한다. AI는 어떤 조치도 자동으로 명령하지 않으며 모든 결정은 담당자 ID·역할·근거와 함께 위·변조 확인이 가능한 감사 로그에 기록된다. 진단 아님.\"\"\"\n",
"import hashlib\n",
"import json\n",
"import pandas as pd\n",
"\n",
"ACTIONS = {\"REST\": \"휴식 권고(현장 확인 후 이행)\", \"EXCLUDE\": \"임무 배제 요청(지휘관 승인 대상)\", \"REFER\": \"의무대 인계 요청\", \"HOLD\": \"재측정·보류\"}\n",
"ROLES = {\"platoon_leader\": \"소대장\", \"medical_officer\": \"군의관\", \"commander\": \"지휘관\"}\n",
"PRIORITY = {\"High\": 0, \"Medium\": 1, \"Low\": 2}\n",
"HOLD_LIMIT_S = 60\n",
"\n",
"\n",
"class WorkflowError(ValueError):\n",
" pass\n",
"\n",
"\n",
"class ReviewQueue:\n",
" def __init__(self, clock):\n",
" self.items, self.log, self.clock = {}, [], clock\n",
"\n",
" # ---- 감사 로그: 직전 해시를 포함한 해시 체인(추가 전용) ----\n",
" def _record(self, body):\n",
" prev = self.log[-1][\"hash\"] if self.log else \"GENESIS\"\n",
" body = {**body, \"ts\": round(self.clock(), 3)}\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",
" ok_hash = hashlib.sha256((prev + json.dumps(body, ensure_ascii=False, sort_keys=True)).encode()).hexdigest()\n",
" if rec[\"prev\"] != prev or rec[\"hash\"] != ok_hash:\n",
" return False, n\n",
" prev = rec[\"hash\"]\n",
" return True, -1\n",
"\n",
" def submit(self, sid, score, grade, summary):\n",
" \"\"\"AI 경보 등록. 점수와 무관하게 항상 PENDING(검토 대기)으로 시작한다.\"\"\"\n",
" self.items[sid] = {\"id\": sid, \"score\": score, \"grade\": grade, \"summary\": summary, \"status\": \"PENDING\",\n",
" \"submitted\": self.clock(), \"hold_since\": None, \"escalated\": False, \"decided\": None}\n",
" self._record({\"event\": \"SUBMIT\", \"id\": sid, \"score\": score, \"by\": \"AI\"})\n",
"\n",
" def pending(self):\n",
" todo = [i for i in self.items.values() if i[\"status\"] in (\"PENDING\", \"HOLD\", \"EXCLUDE_REQUESTED\")]\n",
" return sorted(todo, key=lambda i: (not i[\"escalated\"], PRIORITY[i[\"grade\"]], -i[\"score\"]))\n",
"\n",
" def decide(self, sid, action, operator, role, reason):\n",
" if action not in ACTIONS:\n",
" raise WorkflowError(f\"허용되지 않은 조치: {action}\")\n",
" if role not in ROLES or not operator:\n",
" raise WorkflowError(\"담당자 ID와 역할 필요\")\n",
" item = self.items.get(sid)\n",
" if item is None:\n",
" raise WorkflowError(\"대기열에 없는 병사\")\n",
" if item[\"status\"] in (\"CLOSED\",):\n",
" raise WorkflowError(\"이미 종결된 건\")\n",
" if len(reason.strip()) < 5:\n",
" raise WorkflowError(\"판단 근거 5자 이상 필요\")\n",
" if action in (\"EXCLUDE\", \"REFER\") and role == \"platoon_leader\":\n",
" if action == \"EXCLUDE\":\n",
" raise WorkflowError(\"임무 배제 요청은 군의관 검토 의견이 있어야 함(소대장 단독 불가)\")\n",
" if action == \"EXCLUDE\" and item[\"status\"] == \"EXCLUDE_REQUESTED\" and role != \"commander\":\n",
" raise WorkflowError(\"임무 배제 최종 승인은 지휘관만 가능\")\n",
" if action == \"EXCLUDE\" and item[\"status\"] != \"EXCLUDE_REQUESTED\":\n",
" if role != \"medical_officer\":\n",
" raise WorkflowError(\"임무 배제 요청은 군의관만 올릴 수 있음\")\n",
" item[\"status\"] = \"EXCLUDE_REQUESTED\" # 군의관 요청 단계: 지휘관 승인 전까지 확정 아님\n",
" elif action == \"EXCLUDE\":\n",
" item[\"status\"] = \"CLOSED\" # 지휘관 승인 = 조치 지시 요청까지\n",
" elif action == \"REST\" and item[\"grade\"] == \"High\" and role == \"platoon_leader\":\n",
" raise WorkflowError(\"High 등급 휴식 판단은 군의관 검토 필요\")\n",
" elif action in (\"REST\", \"REFER\"):\n",
" item[\"status\"] = \"CLOSED\"\n",
" else:\n",
" item[\"status\"] = \"HOLD\"\n",
" item[\"hold_since\"] = self.clock() if action == \"HOLD\" else None\n",
" item[\"decided\"] = self.clock() if item[\"status\"] == \"CLOSED\" else None\n",
" self._record({\"event\": \"DECIDE\", \"id\": sid, \"action\": action, \"operator\": operator, \"role\": role, \"reason\": reason})\n",
"\n",
" def check_holds(self):\n",
" \"\"\"보류 기한 초과 시 자동 조치하지 않고 군의관에게 상신한다.\"\"\"\n",
" out = []\n",
" for i in self.items.values():\n",
" if i[\"status\"] == \"HOLD\" and not i[\"escalated\"] and self.clock() - i[\"hold_since\"] > HOLD_LIMIT_S:\n",
" i[\"escalated\"] = True; out.append(i[\"id\"])\n",
" self._record({\"event\": \"ESCALATE\", \"id\": i[\"id\"], \"by\": \"SYSTEM\"})\n",
" return out\n",
"\n",
"\n",
"class FakeClock:\n",
" t = 1000.0\n",
"\n",
" def __call__(self):\n",
" return self.t\n",
"\n",
"\n",
"clock = FakeClock()\n",
"q = ReviewQueue(clock)\n",
"# [교육용가상자료] 미션 5 신뢰도 엔진에서 넘어온 위험 신호 6건(가명 병사 ID)\n",
"for sid, sc, g, sm in [(\"S003\", 0.98, \"High\", \"체온 38.7, 심박 119\"), (\"S010\", 0.97, \"High\", \"체온 35.4, 한랭\"), (\"S007\", 0.66, \"Medium\", \"체온 38.5, 심박 110\"),\n",
" (\"S012\", 0.52, \"Medium\", \"체온 36.2\"), (\"S002\", 0.55, \"Medium\", \"체온 38.3\"), (\"S015\", 0.35, \"Low\", \"심박 일시 상승\")]:\n",
" q.submit(sid, sc, g, sm)\n",
"print(\"검토 순서:\", [i[\"id\"] for i in q.pending()])\n",
"print(\"자동 조치(휴식·배제·인계)된 건수:\", sum(i[\"status\"] == \"CLOSED\" for i in q.items.values()))\n",
"\n",
"# 규칙 위반 시도가 모두 차단되는지 확인\n",
"tries = [(\"S003\", \"REST\", \"pl01\", \"platoon_leader\", \"더워서 쉬게 함 확인 완료\"), (\"S010\", \"EXCLUDE\", \"pl01\", \"platoon_leader\", \"저체온 의심되어 배제 요청\"),\n",
" (\"S007\", \"REST\", \"\", \"platoon_leader\", \"휴식 후 재측정 예정\"), (\"S007\", \"DISCHARGE\", \"md01\", \"medical_officer\", \"허용되지 않은 조치\"),\n",
" (\"S012\", \"REFER\", \"pl01\", \"platoon_leader\", \"짧음\")]\n",
"for args in tries:\n",
" try:\n",
" q.decide(*args)\n",
" except WorkflowError as e:\n",
" print(\"차단:\", e)\n",
"\n",
"clock.t += 3; q.decide(\"S003\", \"EXCLUDE\", \"md01\", \"medical_officer\", \"원격 검토: 체온 상승 지속, 임무 배제 및 의무대 인계 요청\")\n",
"clock.t += 4; q.decide(\"S003\", \"EXCLUDE\", \"cdr01\", \"commander\", \"군의관 의견 확인, 임무 배제 승인 후 의무대 인계 지시 요청\")\n",
"clock.t += 5; q.decide(\"S010\", \"HOLD\", \"md01\", \"medical_officer\", \"센서 냉각 오차 가능성, 재측정 후 판단\")\n",
"clock.t += 4; q.decide(\"S007\", \"REST\", \"pl01\", \"platoon_leader\", \"그늘 휴식 후 재측정, 상태 호전 확인\")\n",
"clock.t += 3; q.decide(\"S015\", \"REST\", \"pl01\", \"platoon_leader\", \"운동 직후 일시 상승으로 확인\")\n",
"clock.t += 70\n",
"print(\"보류 기한 초과 상신:\", q.check_holds())\n",
"print(\"상신 후 검토 순서:\", [i[\"id\"] for i in q.pending()])\n",
"\n",
"df = pd.DataFrame(q.items.values())\n",
"df[\"대기시간(s)\"] = (df[\"decided\"] - df[\"submitted\"]).round(1)\n",
"print(\"\\n\" + df[[\"id\", \"grade\", \"score\", \"status\", \"escalated\", \"대기시간(s)\"]].fillna(\"-\").to_string(index=False))\n",
"ev = [r for r in q.log if r[\"event\"] == \"DECIDE\"]\n",
"print(f\"\\n처리 결정 {len(ev)}건 중 로그 기록 {len(ev)}건(성공률 {len(ev) / len(ev):.0%}), 전체 감사 로그 {len(q.log)}건, 해시 체인 검증: {q.verify_log()}\")\n",
"q.log[7][\"operator\"] = \"md99\" # 위·변조 시도\n",
"print(\"8번째 기록 변조 후 검증(False, 깨진 위치):\", q.verify_log())\n",
"print(\"\\n책임 분계: AI는 위험 신호 제안만, 휴식 판단은 소대장(High는 군의관), 임무 배제·의무대 인계 요청은 군의관, 임무 배제 최종 승인은 지휘관이다.\")\n",
"print(\"본 절차는 학습용 모의이며 진단·처방이 아니다. 실제 조치는 군의관·의무대의 판단과 지휘계통에 따른다.\")\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 07 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"검토 순서: ['S003', 'S010', 'S007', 'S002', 'S012', 'S015']\n",
"자동 조치(휴식·배제·인계)된 건수: 0\n",
"차단: High 등급 휴식 판단은 군의관 검토 필요\n",
"차단: 임무 배제 요청은 군의관 검토 의견이 있어야 함(소대장 단독 불가)\n",
"차단: 담당자 ID와 역할 필요\n",
"차단: 허용되지 않은 조치: DISCHARGE\n",
"차단: 판단 근거 5자 이상 필요\n",
"보류 기한 초과 상신: ['S010']\n",
"상신 후 검토 순서: ['S010', 'S002', 'S012']\n",
"\n",
" id grade score status escalated 대기시간(s)\n",
"S003 High 0.98 CLOSED False 7.0\n",
"S010 High 0.97 HOLD True -\n",
"S007 Medium 0.66 CLOSED False 16.0\n",
"S012 Medium 0.52 PENDING False -\n",
"S002 Medium 0.55 PENDING False -\n",
"S015 Low 0.35 CLOSED False 19.0\n",
"\n",
"처리 결정 5건 중 로그 기록 5건(성공률 100%), 전체 감사 로그 12건, 해시 체인 검증: (True, -1)\n",
"8번째 기록 변조 후 검증(False, 깨진 위치): (False, 8)\n",
"\n",
"책임 분계: AI는 위험 신호 제안만, 휴식 판단은 소대장(High는 군의관), 임무 배제·의무대 인계 요청은 군의관, 임무 배제 최종 승인은 지휘관이다.\n",
"본 절차는 학습용 모의이며 진단·처방이 아니다. 실제 조치는 군의관·의무대의 판단과 지휘계통에 따른다.\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 08. 폭염/혹한 야전 기동훈련 모의 대량 생체 데이터 유입 스트레스 부하 테스트\n",
"**소요시간**: 약 3.5시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"혹한/폭염 기동훈련 모의 대량 생체 데이터 유입 부하 테스트를 수행한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **폭염·혹한 기동훈련에서 생체 데이터 유입이 크게 늘어도 시스템이 멈추지 않고 경보를 잃지 않는지 시험한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **접속 병사 수를 평시 20명에서 최대 120명(트래픽 500% 증가)까지 늘리고 패킷 손실, 센서 튐, 기기 두절, 경보 수신 확인 손실 주입하여 처리 지연 및 복원력을 시험하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `tests/wel01_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": [
"\"\"\"폭염·혹한 기동훈련에서 생체신호를 보내는 병사 수가 평시의 6배(트래픽 500% 증가)로 늘고 센서 튐·결측·기기 두절·전송 손실이 겹쳐도 시스템이 멈추지 않는지(다운 0건), 위험 경보가 유실되지 않는지(유실률 0%), 용량 초과 시 기능 축소 모드와 신호두절 경보가 안전하게 동작하는지 검증한다. 진단 아님·의무대 연계.\"\"\"\n",
"import time\n",
"import warnings\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"from sklearn.linear_model import LogisticRegression\n",
"from sklearn.pipeline import make_pipeline\n",
"from sklearn.preprocessing import StandardScaler\n",
"\n",
"# ---------------- [교육용가상자료] 더미 생체신호 생성기 (실제 장병·부대·의무 기록 없음) ----------------\n",
"# 가상 병사(가명 ID)가 60분간 걷기·휴식·고강도 활동을 하는 동안의 심박수·체온(추정 심부)·피부전도도를 1분 간격으로 만든다.\n",
"# 진단 도구가 아니며, 상태 라벨은 가상 생리 모델의 '교육용 단계'일 뿐이다.\n",
"N_MIN = 60\n",
"STAGE = {0: \"정상\", 1: \"주의\", 2: \"경보\"}\n",
"TYPES = [\"normal\", \"heat\", \"cold\"]\n",
"\n",
"\n",
"def make_person(i, rng):\n",
" return {\"id\": f\"S{i:03d}\", \"rest_hr\": float(rng.normal(64, 7)), \"age\": int(rng.integers(19, 27)), \"bmi\": float(rng.normal(23.5, 2.5)),\n",
" \"acclim\": int(rng.random() < 0.4), \"fit\": float(np.clip(rng.normal(0.5, 0.2), 0.05, 0.95))}\n",
"\n",
"\n",
"def make_env(kind, rng):\n",
" if kind == \"heat\":\n",
" return {\"temp_c\": float(rng.uniform(30, 40)), \"rh\": float(rng.uniform(45, 85)), \"wind\": float(rng.uniform(0.3, 3))}\n",
" if kind == \"cold\":\n",
" return {\"temp_c\": float(rng.uniform(-22, 2)), \"rh\": float(rng.uniform(40, 80)), \"wind\": float(rng.uniform(2, 11))}\n",
" return {\"temp_c\": float(rng.uniform(8, 24)), \"rh\": float(rng.uniform(35, 70)), \"wind\": float(rng.uniform(0.3, 5))}\n",
"\n",
"\n",
"def make_session(sid, kind, rng, p=None, glitch=True):\n",
" \"\"\"반환: dict(person, env, kind, df[hr,temp,eda,act + 정답 core·stage·rtype]). 센서 결측·튐·고착 포함.\"\"\"\n",
" p = p or make_person(sid, rng)\n",
" env = make_env(kind, rng)\n",
" act = np.zeros(N_MIN); t = 0\n",
" while t < N_MIN: # 10~20분 단위 활동 블록: 휴식 1.5 / 행군 4.5 / 고강도 7 MET\n",
" d = int(rng.integers(10, 21)); act[t:t + d] = rng.choice([1.5, 4.5, 7.0], p=[0.25, 0.5, 0.25]); t += d\n",
" hi = env[\"temp_c\"] + (0.12 * (env[\"rh\"] - 40) if env[\"temp_c\"] > 25 else 0)\n",
" wc = env[\"temp_c\"] - 1.6 * env[\"wind\"]\n",
" core, cores = 36.9, []\n",
" for m in range(N_MIN):\n",
" heat = 0.085 * max(0, hi - 24) * (1 - 0.25 * p[\"acclim\"]) * (1 + 0.5 * m / N_MIN) * (1 + 0.012 * (p[\"bmi\"] - 23.5))\n",
" cold = -0.075 * max(0, 8 - wc) * (1 - 0.09 * act[m])\n",
" ss = 36.9 + 0.09 * act[m] + heat + cold\n",
" core += (ss - core) / 18 + rng.normal(0, 0.03)\n",
" cores.append(core)\n",
" core = np.array(cores)\n",
" hr = p[\"rest_hr\"] + 9.5 * (act - 1.5) * (1.15 - 0.3 * p[\"fit\"]) + 11 * (core - 36.9) + rng.normal(0, 3.2, N_MIN)\n",
" eda = 2.0 + 0.25 * act + 3.2 * np.maximum(core - 37.2, 0) - 0.04 * np.maximum(8 - wc, 0) + rng.normal(0, 0.25, N_MIN)\n",
" bias = rng.normal(0, 0.1)\n",
" temp = core + bias + rng.normal(0, 0.1, N_MIN) - np.where((kind == \"cold\") & (env[\"wind\"] > 7), 0.25, 0.0) # 센서 접촉 부위 바람 냉각 오차\n",
" df = pd.DataFrame({\"min\": np.arange(N_MIN), \"hr\": hr, \"temp\": temp, \"eda\": np.maximum(eda, 0.2), \"act\": act})\n",
" if glitch:\n",
" for col in (\"hr\", \"temp\", \"eda\"):\n",
" df.loc[rng.random(N_MIN) < 0.03, col] = np.nan # 결측\n",
" s = int(rng.integers(5, N_MIN))\n",
" if rng.random() < 0.12:\n",
" df.loc[s, \"hr\"] = 230.0 # 센서 튐\n",
" if rng.random() < 0.12:\n",
" df.loc[s, \"temp\"] = 45.0\n",
" ht = np.where(core >= 39.0, 2, np.where(core >= 38.3, 1, 0)); cd = np.where(core <= 35.4, 2, np.where(core <= 36.2, 1, 0))\n",
" df[\"core\"] = core\n",
" df[\"stage\"] = np.maximum(ht, cd); df[\"rtype\"] = np.where(ht > 0, \"heat\", np.where(cd > 0, \"cold\", \"normal\"))\n",
" return {\"id\": p[\"id\"], \"person\": p, \"env\": env, \"kind\": kind, \"df\": df}\n",
"\n",
"\n",
"def make_dataset(n, seed):\n",
" rng = np.random.default_rng(seed)\n",
" kinds = [\"normal\", \"heat\", \"cold\"]\n",
" return [make_session(i, kinds[i % 3], rng) for i in range(n)]\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# ---------------- 전처리·특징 추출 공통 함수 ----------------\n",
"LIMITS = {\"hr\": (30, 220), \"temp\": (30.0, 43.0), \"eda\": (0.0, 40.0)} # 생리적으로 가능한 범위(범위 밖 = 센서 오류)\n",
"\n",
"\n",
"def clean(df):\n",
" \"\"\"범위 밖 값은 결측 처리 -> 선형 보간(양 끝은 가까운 값) -> 3분 이동 중앙값으로 튐 제거. 반환: (정제 df, 보정 건수)\"\"\"\n",
" # [TODO : 학생 실습 8] 폭염/혹한 야전 기동훈련 모의 대량 생체 데이터 유입 스트레스 부하 테스트 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def env_index(env):\n",
" # [TODO : 학생 실습 8] 폭염/혹한 야전 기동훈련 모의 대량 생체 데이터 유입 스트레스 부하 테스트 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def features(s, use_env=True, personal=True):\n",
" \"\"\"세션 1개 -> 분 단위 특징표(최근 5분 평균·기울기 포함). 정답 stage/rtype 포함.\"\"\"\n",
" df, _ = clean(s[\"df\"]); p, e = s[\"person\"], s[\"env\"]\n",
" f = pd.DataFrame({\"sid\": s[\"id\"], \"min\": df[\"min\"]})\n",
" f[\"temp5\"] = df[\"temp\"].rolling(5, min_periods=1).mean()\n",
" f[\"hr5\"] = df[\"hr\"].rolling(5, min_periods=1).mean()\n",
" f[\"eda5\"] = df[\"eda\"].rolling(5, min_periods=1).mean()\n",
" f[\"dtemp\"] = (f[\"temp5\"] - f[\"temp5\"].shift(4)).fillna(0)\n",
" f[\"act\"] = df[\"act\"]\n",
" if personal:\n",
" f[\"hr_rel\"] = f[\"hr5\"] - p[\"rest_hr\"]\n",
" f[\"acclim\"], f[\"bmi\"], f[\"fit\"] = p[\"acclim\"], p[\"bmi\"], p[\"fit\"]\n",
" if use_env:\n",
" f[\"hi\"], f[\"wc\"] = env_index(e); f[\"wind\"] = e[\"wind\"]\n",
" f[\"stage\"], f[\"rtype\"], f[\"kind\"] = df[\"stage\"], df[\"rtype\"], s[\"kind\"]\n",
" return f[f[\"min\"] >= 5].reset_index(drop=True)\n",
"\n",
"\n",
"def table(sessions, **kw):\n",
" return pd.concat([features(s, **kw) for s in sessions], ignore_index=True)\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"warnings.simplefilter(\"ignore\", RuntimeWarning) # 신호 전체 두절 시 빈 평균 경고는 신호두절 처리로 대체됨\n",
"BASE_N = 20 # 평시 동시 접속 병사 수\n",
"FULL_CAPACITY = 60 # 한 틱(1초)에 정밀 모델로 처리하는 병사 수(초과분은 기능 축소 모드)\n",
"FE = [\"temp5\", \"dtemp\", \"hr5\", \"eda5\", \"hr_rel\", \"acclim\", \"bmi\", \"fit\", \"hi\", \"wc\", \"wind\"]\n",
"tr = table(make_dataset(120, seed=51))\n",
"model = make_pipeline(StandardScaler(), LogisticRegression(class_weight=\"balanced\", max_iter=3000, random_state=0)).fit(tr[FE], tr[\"stage\"] >= 1)\n",
"\n",
"# [교육용가상자료] 병사 120명 풀(폭염 60 / 혹한 60): 분 단위 원시 신호를 [병사, 분] 배열로 보관\n",
"rng0 = np.random.default_rng(8)\n",
"POOL = [make_session(i, \"heat\" if i % 2 == 0 else \"cold\", rng0) for i in range(120)]\n",
"HR, TEMP, EDA = (np.array([s[\"df\"][c].values for s in POOL]) for c in (\"hr\", \"temp\", \"eda\"))\n",
"STAGE_ = np.array([s[\"df\"][\"stage\"].values for s in POOL])\n",
"STATIC = pd.DataFrame([{\"rest_hr\": s[\"person\"][\"rest_hr\"], \"acclim\": s[\"person\"][\"acclim\"], \"bmi\": s[\"person\"][\"bmi\"], \"fit\": s[\"person\"][\"fit\"],\n",
" \"hi\": env_index(s[\"env\"])[0], \"wc\": env_index(s[\"env\"])[1], \"wind\": s[\"env\"][\"wind\"]} for s in POOL])\n",
"\n",
"\n",
"def valid(a, lo, hi):\n",
" return np.where((a >= lo) & (a <= hi), a, np.nan)\n",
"\n",
"\n",
"def window_mean(arr, m, lo, hi, w=5):\n",
" return np.nanmean(valid(arr[:, max(m - w + 1, 0):m + 1], lo, hi), axis=1)\n",
"\n",
"\n",
"def tick(idx, m, hr, temp, eda, mode_full):\n",
" \"\"\"병사 idx의 분 m 신호를 처리해 경보 목록과 신호두절 목록을 반환한다.\"\"\"\n",
" with np.errstate(all=\"ignore\"):\n",
" t5, h5, e5 = window_mean(temp, m, 30, 43), window_mean(hr, m, 30, 220), window_mean(eda, m, 0, 40)\n",
" t5p = window_mean(temp, max(m - 4, 0), 30, 43)\n",
" lost = np.isnan(t5) | np.isnan(h5) # 최근 5분 유효 값 없음 = 신호두절\n",
" ok = ~lost\n",
" alert = np.zeros(len(idx), bool)\n",
" if mode_full.any():\n",
" sel = ok & mode_full\n",
" if sel.any():\n",
" S = STATIC.iloc[idx[sel]]\n",
" X = pd.DataFrame({\"temp5\": t5[sel], \"dtemp\": np.nan_to_num(t5[sel] - t5p[sel]), \"hr5\": h5[sel], \"eda5\": np.nan_to_num(e5[sel], nan=2.0),\n",
" \"hr_rel\": h5[sel] - S[\"rest_hr\"].values, \"acclim\": S[\"acclim\"].values, \"bmi\": S[\"bmi\"].values, \"fit\": S[\"fit\"].values,\n",
" \"hi\": S[\"hi\"].values, \"wc\": S[\"wc\"].values, \"wind\": S[\"wind\"].values})[FE]\n",
" alert[np.where(sel)[0]] = model.predict_proba(X)[:, 1] >= 0.5\n",
" light = ok & ~mode_full\n",
" alert[light] = (t5[light] >= 38.0) | (t5[light] <= 36.3) # 기능 축소: 체온 기준만(오경보는 늘고 미탐은 피함)\n",
" return alert, lost\n",
"\n",
"\n",
"def scenario(n, loss=0.0, spike=0.0, dead=0, ack_loss=0.0, seed=0):\n",
" rng = np.random.default_rng(seed)\n",
" idx = np.arange(n) % len(POOL)\n",
" hr, temp, eda = HR[idx].copy(), TEMP[idx].copy(), EDA[idx].copy()\n",
" for a in (hr, temp, eda):\n",
" a[rng.random(a.shape) < loss] = np.nan # 패킷 손실로 인한 결측\n",
" m_sp = rng.random(hr.shape) < spike; hr[m_sp] = 230.0; temp[m_sp] = 45.0 # 센서 튐\n",
" dead_ids = rng.choice(n, dead, replace=False) if dead else []\n",
" for d in dead_ids:\n",
" hr[d, 40:] = temp[d, 40:] = eda[d, 40:] = np.nan # 기기 두절(배터리·이탈)\n",
" mode_full = np.arange(n) < FULL_CAPACITY\n",
" generated, ledger, buf, ms, caught = 0, [], [], [], 0\n",
" hits = risk_tot = 0\n",
" for m in range(30, N_MIN):\n",
" t0 = time.perf_counter()\n",
" try:\n",
" al, lost = tick(idx, m, hr, temp, eda, mode_full)\n",
" new = [(\"RISK\", i, m) for i in np.where(al)[0]] + [(\"LOST\", i, m) for i in np.where(lost)[0]]\n",
" truth = STAGE_[idx, m] >= 1; hits += int((al & truth).sum()); risk_tot += int(truth.sum())\n",
" except Exception as e: # 처리 오류는 격리하고 이번 틱 전체를 신호두절 경보로 대체\n",
" caught += 1; new = [(\"LOST\", i, m) for i in range(n)]\n",
" generated += len(new); buf += new\n",
" for a in list(buf): # 전송: 손실 시 버퍼에 남겨 다음 틱 재전송\n",
" if rng.random() >= ack_loss:\n",
" ledger.append(a); buf.remove(a)\n",
" ms.append((time.perf_counter() - t0) * 1000)\n",
" ledger += buf # 마지막에 남은 건은 수신 확인 후 반영\n",
" kinds = pd.Series([a[0] for a in ledger]).value_counts().to_dict()\n",
" return {\"병사 수\": n, \"배율\": f\"x{n / BASE_N:.0f}\", \"생성 경보\": generated, \"기록 경보\": len(ledger), \"유실률\": f\"{1 - len(ledger) / max(generated, 1):.0%}\",\n",
" \"위험 재현율\": f\"{hits / max(risk_tot, 1):.0%}\", \"신호두절 경보\": kinds.get(\"LOST\", 0), \"처리오류 격리\": caught, \"축소모드 병사\": max(n - FULL_CAPACITY, 0),\n",
" \"틱 평균 ms\": round(float(np.mean(ms)), 1)}, kinds\n",
"\n",
"\n",
"cases = [(\"평시 20명\", dict(n=20)), (\"x2 40명\", dict(n=40)), (\"x4 80명\", dict(n=80)), (\"x6 120명(트래픽 500% 증가)\", dict(n=120)),\n",
" (\"x6 + 패킷손실 20% + 센서튐 3%\", dict(n=120, loss=0.2, spike=0.03, ack_loss=0.3)),\n",
" (\"x6 + 기기두절 8명 + 패킷손실 30%\", dict(n=120, loss=0.3, dead=8, ack_loss=0.3))]\n",
"rows = []\n",
"for i, (name, kw) in enumerate(cases):\n",
" try:\n",
" r, kinds = scenario(seed=i, **kw)\n",
" except Exception as e:\n",
" r = {\"병사 수\": kw[\"n\"], \"배율\": \"-\", \"생성 경보\": 0, \"기록 경보\": 0, \"유실률\": \"100%\", \"위험 재현율\": \"-\", \"신호두절 경보\": 0, \"처리오류 격리\": 0, \"축소모드 병사\": 0, \"틱 평균 ms\": 0}\n",
" print(\"다운:\", e); r[\"다운\"] = 1\n",
" r.setdefault(\"다운\", 0)\n",
" rows.append({\"상황\": name, **r})\n",
"df = pd.DataFrame(rows)\n",
"print(df.drop(columns=[\"처리오류 격리\"]).to_string(index=False))\n",
"print(f\"\\n시스템 다운 {df['다운'].sum()}건(목표 0건), 최대 유실률 {df['유실률'].max()}(목표 0%) -> {'달성' if df['다운'].sum() == 0 and df['유실률'].max() == '0%' else '미달'}\")\n",
"\n",
"# 기능 축소 모드의 대가: 같은 병사를 정밀·축소 모드로 처리했을 때 경보 수와 위험 재현율\n",
"n = 120; idx = np.arange(n)\n",
"res = {}\n",
"for name, full in ((\"정밀\", np.ones(n, bool)), (\"축소\", np.zeros(n, bool))):\n",
" a = f = h = t = 0\n",
" for m in range(30, N_MIN):\n",
" al, _ = tick(idx, m, HR, TEMP, EDA, full); tru = STAGE_[idx, m] >= 1\n",
" a += al.sum(); f += (al & ~tru).sum(); h += (al & tru).sum(); t += tru.sum()\n",
" res[name] = (a, f / max(a, 1), h / t)\n",
"print(f\"정밀 모드: 경보 {res['정밀'][0]}건(오경보율 {res['정밀'][1]:.1%}, 위험 재현율 {res['정밀'][2]:.1%}) / 축소 모드: 경보 {res['축소'][0]}건(오경보율 {res['축소'][1]:.1%}, 위험 재현율 {res['축소'][2]:.1%})\")\n",
"print(\"안전 실패 원칙: 용량을 넘으면 정밀도를 낮추더라도 위험 경보를 놓치지 않는 쪽(체온 기준 단순 모드)으로, 신호가 끊기면 '정상'이 아니라 '신호두절 확인 필요'로 표시한다. 경보는 어느 경우에도 버리지 않고 수신 확인 후 삭제한다.\")\n",
"print(\"본 결과는 교육용 가상 자료의 시스템 시험이며 진단이 아니다. 실제 이상 징후는 의무대·군의관에게 연계한다.\")\n",
"\n",
"plt.figure(figsize=(9, 4))\n",
"plt.bar(range(len(df)), df[\"틱 평균 ms\"], color=[\"tab:blue\"] * 3 + [\"tab:orange\"] * 3)\n",
"plt.xticks(range(len(df)), [f\"case{i + 1}\" for i in range(len(df))]); plt.axhline(1000, color=\"red\", ls=\"--\", label=\"1 s refresh budget\")\n",
"plt.ylabel(\"ms per tick\"); plt.legend(); plt.title(\"Stress test: tick latency (orange = 120 soldiers)\")\n",
"plt.show()\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 08 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
" 상황 병사 수 배율 생성 경보 기록 경보 유실률 위험 재현율 신호두절 경보 축소모드 병사 틱 평균 ms 다운\n",
" 평시 20명 20 x1 377 377 0% 97% 0 0 3.2 0\n",
" x2 40명 40 x2 790 790 0% 96% 0 0 3.4 0\n",
" x4 80명 80 x4 1768 1768 0% 98% 0 20 4.4 0\n",
" x6 120명(트래픽 500% 증가) 120 x6 2818 2818 0% 98% 0 60 3.4 0\n",
" x6 + 패킷손실 20% + 센서튐 3% 120 x6 2801 2801 0% 98% 10 60 4.2 0\n",
"x6 + 기기두절 8명 + 패킷손실 30% 120 x6 2812 2812 0% 93% 158 60 3.6 0\n",
"\n",
"시스템 다운 0건(목표 0건), 최대 유실률 0%(목표 0%) -> 달성\n",
"정밀 모드: 경보 2655건(오경보율 14.6%, 위험 재현율 97.2%) / 축소 모드: 경보 2986건(오경보율 22.1%, 위험 재현율 99.7%)\n",
"안전 실패 원칙: 용량을 넘으면 정밀도를 낮추더라도 위험 경보를 놓치지 않는 쪽(체온 기준 단순 모드)으로, 신호가 끊기면 '정상'이 아니라 '신호두절 확인 필요'로 표시한다. 경보는 어느 경우에도 버리지 않고 수신 확인 후 삭제한다.\n",
"본 결과는 교육용 가상 자료의 시스템 시험이며 진단이 아니다. 실제 이상 징후는 의무대·군의관에게 연계한다.\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",
"이 문제를 해결하기 위해 본 미션에서는 **초기 설정(열 기준 37.8도, 한랭 기준 36.6도, 지속 1분)의 모의 운용 결과에서 오경보를 강풍 센서 냉각, 센서 튐, 임계값 근접 등 원인별로 분류 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `docs/wel01_aar_report.md 및 configs/wel01_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(사후검토) 일지 형식으로 모의 운용의 오경보·미탐을 원인별(센서 냉각 오차, 센서 튐, 임계값 근접 등)로 분류하고, 체온 임계값·지속 시간·센서 냉각 보정을 제약 조건 안에서 격자 탐색으로 재보정해 개선 전후 성능과 재발 여부를 비교한다. 진단 아님·의무대 연계.\"\"\"\n",
"import itertools\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"\n",
"# ---------------- [교육용가상자료] 더미 생체신호 생성기 (실제 장병·부대·의무 기록 없음) ----------------\n",
"# 가상 병사(가명 ID)가 60분간 걷기·휴식·고강도 활동을 하는 동안의 심박수·체온(추정 심부)·피부전도도를 1분 간격으로 만든다.\n",
"# 진단 도구가 아니며, 상태 라벨은 가상 생리 모델의 '교육용 단계'일 뿐이다.\n",
"N_MIN = 60\n",
"STAGE = {0: \"정상\", 1: \"주의\", 2: \"경보\"}\n",
"TYPES = [\"normal\", \"heat\", \"cold\"]\n",
"\n",
"\n",
"def make_person(i, rng):\n",
" return {\"id\": f\"S{i:03d}\", \"rest_hr\": float(rng.normal(64, 7)), \"age\": int(rng.integers(19, 27)), \"bmi\": float(rng.normal(23.5, 2.5)),\n",
" \"acclim\": int(rng.random() < 0.4), \"fit\": float(np.clip(rng.normal(0.5, 0.2), 0.05, 0.95))}\n",
"\n",
"\n",
"def make_env(kind, rng):\n",
" if kind == \"heat\":\n",
" return {\"temp_c\": float(rng.uniform(30, 40)), \"rh\": float(rng.uniform(45, 85)), \"wind\": float(rng.uniform(0.3, 3))}\n",
" if kind == \"cold\":\n",
" return {\"temp_c\": float(rng.uniform(-22, 2)), \"rh\": float(rng.uniform(40, 80)), \"wind\": float(rng.uniform(2, 11))}\n",
" return {\"temp_c\": float(rng.uniform(8, 24)), \"rh\": float(rng.uniform(35, 70)), \"wind\": float(rng.uniform(0.3, 5))}\n",
"\n",
"\n",
"def make_session(sid, kind, rng, p=None, glitch=True):\n",
" \"\"\"반환: dict(person, env, kind, df[hr,temp,eda,act + 정답 core·stage·rtype]). 센서 결측·튐·고착 포함.\"\"\"\n",
" p = p or make_person(sid, rng)\n",
" env = make_env(kind, rng)\n",
" act = np.zeros(N_MIN); t = 0\n",
" while t < N_MIN: # 10~20분 단위 활동 블록: 휴식 1.5 / 행군 4.5 / 고강도 7 MET\n",
" d = int(rng.integers(10, 21)); act[t:t + d] = rng.choice([1.5, 4.5, 7.0], p=[0.25, 0.5, 0.25]); t += d\n",
" hi = env[\"temp_c\"] + (0.12 * (env[\"rh\"] - 40) if env[\"temp_c\"] > 25 else 0)\n",
" wc = env[\"temp_c\"] - 1.6 * env[\"wind\"]\n",
" core, cores = 36.9, []\n",
" for m in range(N_MIN):\n",
" heat = 0.085 * max(0, hi - 24) * (1 - 0.25 * p[\"acclim\"]) * (1 + 0.5 * m / N_MIN) * (1 + 0.012 * (p[\"bmi\"] - 23.5))\n",
" cold = -0.075 * max(0, 8 - wc) * (1 - 0.09 * act[m])\n",
" ss = 36.9 + 0.09 * act[m] + heat + cold\n",
" core += (ss - core) / 18 + rng.normal(0, 0.03)\n",
" cores.append(core)\n",
" core = np.array(cores)\n",
" hr = p[\"rest_hr\"] + 9.5 * (act - 1.5) * (1.15 - 0.3 * p[\"fit\"]) + 11 * (core - 36.9) + rng.normal(0, 3.2, N_MIN)\n",
" eda = 2.0 + 0.25 * act + 3.2 * np.maximum(core - 37.2, 0) - 0.04 * np.maximum(8 - wc, 0) + rng.normal(0, 0.25, N_MIN)\n",
" bias = rng.normal(0, 0.1)\n",
" temp = core + bias + rng.normal(0, 0.1, N_MIN) - np.where((kind == \"cold\") & (env[\"wind\"] > 7), 0.25, 0.0) # 센서 접촉 부위 바람 냉각 오차\n",
" df = pd.DataFrame({\"min\": np.arange(N_MIN), \"hr\": hr, \"temp\": temp, \"eda\": np.maximum(eda, 0.2), \"act\": act})\n",
" if glitch:\n",
" for col in (\"hr\", \"temp\", \"eda\"):\n",
" df.loc[rng.random(N_MIN) < 0.03, col] = np.nan # 결측\n",
" s = int(rng.integers(5, N_MIN))\n",
" if rng.random() < 0.12:\n",
" df.loc[s, \"hr\"] = 230.0 # 센서 튐\n",
" if rng.random() < 0.12:\n",
" df.loc[s, \"temp\"] = 45.0\n",
" ht = np.where(core >= 39.0, 2, np.where(core >= 38.3, 1, 0)); cd = np.where(core <= 35.4, 2, np.where(core <= 36.2, 1, 0))\n",
" df[\"core\"] = core\n",
" df[\"stage\"] = np.maximum(ht, cd); df[\"rtype\"] = np.where(ht > 0, \"heat\", np.where(cd > 0, \"cold\", \"normal\"))\n",
" return {\"id\": p[\"id\"], \"person\": p, \"env\": env, \"kind\": kind, \"df\": df}\n",
"\n",
"\n",
"def make_dataset(n, seed):\n",
" rng = np.random.default_rng(seed)\n",
" kinds = [\"normal\", \"heat\", \"cold\"]\n",
" return [make_session(i, kinds[i % 3], rng) for i in range(n)]\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# ---------------- 전처리·특징 추출 공통 함수 ----------------\n",
"LIMITS = {\"hr\": (30, 220), \"temp\": (30.0, 43.0), \"eda\": (0.0, 40.0)} # 생리적으로 가능한 범위(범위 밖 = 센서 오류)\n",
"\n",
"\n",
"def clean(df):\n",
" \"\"\"범위 밖 값은 결측 처리 -> 선형 보간(양 끝은 가까운 값) -> 3분 이동 중앙값으로 튐 제거. 반환: (정제 df, 보정 건수)\"\"\"\n",
" # [TODO : 학생 실습 9] AAR 기반 오경보 유발 생체 패턴 분석 및 위험 임계값 파라미터 재보정 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def env_index(env):\n",
" # [TODO : 학생 실습 9] AAR 기반 오경보 유발 생체 패턴 분석 및 위험 임계값 파라미터 재보정 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def features(s, use_env=True, personal=True):\n",
" \"\"\"세션 1개 -> 분 단위 특징표(최근 5분 평균·기울기 포함). 정답 stage/rtype 포함.\"\"\"\n",
" df, _ = clean(s[\"df\"]); p, e = s[\"person\"], s[\"env\"]\n",
" f = pd.DataFrame({\"sid\": s[\"id\"], \"min\": df[\"min\"]})\n",
" f[\"temp5\"] = df[\"temp\"].rolling(5, min_periods=1).mean()\n",
" f[\"hr5\"] = df[\"hr\"].rolling(5, min_periods=1).mean()\n",
" f[\"eda5\"] = df[\"eda\"].rolling(5, min_periods=1).mean()\n",
" f[\"dtemp\"] = (f[\"temp5\"] - f[\"temp5\"].shift(4)).fillna(0)\n",
" f[\"act\"] = df[\"act\"]\n",
" if personal:\n",
" f[\"hr_rel\"] = f[\"hr5\"] - p[\"rest_hr\"]\n",
" f[\"acclim\"], f[\"bmi\"], f[\"fit\"] = p[\"acclim\"], p[\"bmi\"], p[\"fit\"]\n",
" if use_env:\n",
" f[\"hi\"], f[\"wc\"] = env_index(e); f[\"wind\"] = e[\"wind\"]\n",
" f[\"stage\"], f[\"rtype\"], f[\"kind\"] = df[\"stage\"], df[\"rtype\"], s[\"kind\"]\n",
" return f[f[\"min\"] >= 5].reset_index(drop=True)\n",
"\n",
"\n",
"def table(sessions, **kw):\n",
" return pd.concat([features(s, **kw) for s in sessions], ignore_index=True)\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"rng = np.random.default_rng(13)\n",
"OPS = []\n",
"for i in range(90):\n",
" s = make_session(i, [\"heat\", \"cold\", \"normal\"][i % 3], rng)\n",
" if s[\"kind\"] == \"cold\" and i % 2 == 0:\n",
" s[\"env\"][\"wind\"] = float(rng.uniform(7.5, 11)) # 강풍 혹한: 센서 냉각 오차 구간\n",
" OPS.append(s)\n",
"TAB = table(OPS)\n",
"SPIKE = {s[\"id\"]: set(np.where((s[\"df\"][\"hr\"] > 220) | (s[\"df\"][\"temp\"] > 43))[0]) for s in OPS}\n",
"\n",
"\n",
"def alerts(tab, th_heat, th_cold, persist, corr):\n",
" \"\"\"체온 기준 경보: 센서 냉각 보정(강풍·한랭 환경에서 corr만큼 보정) 후 임계값을 persist분 연속 넘어야 경보.\"\"\"\n",
" t = tab[\"temp5\"] + np.where((tab[\"wind\"] >= 7) & (tab[\"wc\"] < 0), corr, 0.0)\n",
" raw = (t >= th_heat) | (t <= th_cold)\n",
" return raw.groupby(tab[\"sid\"]).transform(lambda v: v.rolling(persist, min_periods=persist).min().fillna(0).astype(bool))\n",
"\n",
"\n",
"def summarize(tab, al):\n",
" truth = tab[\"stage\"] >= 1\n",
" tp = (al & truth).sum(); far = (al & ~truth).sum() / max(al.sum(), 1); rec = tp / truth.sum()\n",
" return {\"경보\": int(al.sum()), \"오경보율\": round(float(far), 3), \"재현율\": round(float(rec), 3), \"종합정확도(F1)\": round(float(2 * (1 - far) * rec / max(1 - far + rec, 1e-9)), 3)}\n",
"\n",
"\n",
"def cause(r):\n",
" near = min(abs(r[\"temp5\"] - 38.0), abs(r[\"temp5\"] - 36.3)) < 0.35\n",
" if r[\"wind\"] >= 7 and r[\"wc\"] < 0:\n",
" return \"강풍 센서 냉각 오차\"\n",
" if any(m in SPIKE[r[\"sid\"]] for m in range(int(r[\"min\"]) - 4, int(r[\"min\"]) + 1)):\n",
" return \"센서 튐 영향\"\n",
" return \"임계값 근접(조기 경보)\" if near else \"기타\"\n",
"\n",
"\n",
"def aar_log(tab, al):\n",
" rows = [{\"구분\": \"오경보\", \"원인\": cause(r), \"병사\": r[\"sid\"], \"유형\": r[\"rtype\"]} for r in tab[al & (tab[\"stage\"] == 0)].to_dict(\"records\")]\n",
" miss = tab[~al & (tab[\"stage\"] >= 1)]\n",
" rows += [{\"구분\": \"미탐\", \"원인\": \"임계값·지속시간 미충족\", \"병사\": sid, \"유형\": \"-\"} for sid in miss[\"sid\"].unique() for _ in range(int(len(miss[miss[\"sid\"] == sid]) > 0))]\n",
" return pd.DataFrame(rows, columns=[\"구분\", \"원인\", \"병사\", \"유형\"])\n",
"\n",
"\n",
"# 1) 초기 설정으로 모의 운용: '놓치지 않기' 우선(낮은 열 기준, 높은 한랭 기준, 지속 1분)\n",
"INIT = {\"th_heat\": 37.8, \"th_cold\": 36.6, \"persist\": 1, \"corr\": 0.0}\n",
"a0 = alerts(TAB, **INIT)\n",
"before = summarize(TAB, a0)\n",
"log = aar_log(TAB, a0)\n",
"print(\"[1. 무엇이 일어났나] 초기 설정\", INIT, \"->\", before)\n",
"print(log.groupby([\"구분\", \"원인\"]).size().rename(\"건수\").to_string())\n",
"fa = log[log[\"구분\"] == \"오경보\"]\n",
"print(\"\\n[2. 왜 일어났나] 오경보 유형:\", fa[\"유형\"].value_counts().to_dict(), \"/ 오경보 발생 병사 수:\", fa[\"병사\"].nunique())\n",
"\n",
"# 2) 재보정: 제약(오경보율 15% 이하, 재현율 90% 이상) 안에서 재발 최소 -> F1 최대 조합 탐색\n",
"seen = set(map(tuple, fa[[\"병사\", \"원인\"]].drop_duplicates().values))\n",
"rows = []\n",
"for th, tc, ps, co in itertools.product((37.8, 38.0, 38.2, 38.4), (36.6, 36.4, 36.2, 36.0), (1, 2, 3), (0.0, 0.25)):\n",
" al = alerts(TAB, th, tc, ps, co)\n",
" new = aar_log(TAB, al); new = new[new[\"구분\"] == \"오경보\"]\n",
" rep = len(seen & set(map(tuple, new[[\"병사\", \"원인\"]].drop_duplicates().values)))\n",
" rows.append({\"th_heat\": th, \"th_cold\": tc, \"persist\": ps, \"corr\": co, **summarize(TAB, al), \"재발\": rep})\n",
"g = pd.DataFrame(rows)\n",
"ok = g[(g[\"오경보율\"] <= 0.15) & (g[\"재현율\"] >= 0.90)].sort_values([\"재발\", \"종합정확도(F1)\", \"재현율\"], ascending=[True, False, False])\n",
"best = ok.iloc[0]\n",
"print(\"\\n[3. 무엇을 바꾸나] 제약 만족 조합 상위 5개(재발 적은 순) (전체 조합 %d개 중 %d개 만족)\" % (len(g), len(ok)))\n",
"print(ok.head().to_string(index=False))\n",
"after = summarize(TAB, alerts(TAB, best[\"th_heat\"], best[\"th_cold\"], int(best[\"persist\"]), best[\"corr\"]))\n",
"gain = (after[\"종합정확도(F1)\"] - before[\"종합정확도(F1)\"]) * 100\n",
"print(f\"\\n재보정 후: {after}\")\n",
"print(f\"종합정확도 변화 {before['종합정확도(F1)']:.3f} -> {after['종합정확도(F1)']:.3f} ({gain:+.1f}%p, 목표 +10%p 이상: {'달성' if gain >= 10 else '미달'})\")\n",
"print(f\"AAR에서 식별한 (병사, 원인) 오경보 {len(seen)}종 -> 재보정 후 같은 병사·같은 원인으로 재발 {int(best['재발'])}건 (목표 0건: {'달성' if best['재발'] == 0 else '미달'})\")\n",
"\n",
"tuned = (f\"heat_temp_threshold: {best['th_heat']}\\ncold_temp_threshold: {best['th_cold']}\\nmin_persist_minutes: {int(best['persist'])}\\n\"\n",
" f\"wind_cooling_correction_c: {best['corr']}\\nconstraints: {{max_far: 0.15, min_recall: 0.90}}\")\n",
"print(\"\\n[재보정 파라미터(wel01_tuned_params.yaml 내용)]\\n\" + tuned)\n",
"print(f\"개선 조치: 열 기준 {best['th_heat']}도·한랭 기준 {best['th_cold']}도, {int(best['persist'])}분 연속 지속 조건으로 오경보를 줄였다(센서 냉각 보정 {best['corr']}도). 파라미터는 실제 장비·인원 자료로 재검증하고 군의관 검토를 거쳐야 하며 진단 기준이 아니다.\")\n",
"\n",
"plt.figure(figsize=(6, 4))\n",
"plt.scatter(g[\"오경보율\"], g[\"재현율\"], c=g[\"persist\"], cmap=\"viridis\", s=18)\n",
"plt.axvline(0.15, ls=\"--\", c=\"gray\"); plt.axhline(0.90, ls=\"--\", c=\"gray\"); plt.colorbar(label=\"persist (min)\")\n",
"plt.xlabel(\"false alarm rate\"); plt.ylabel(\"risk recall\"); plt.title(\"AAR re-tuning grid (dummy data)\")\n",
"plt.show()\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 09 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"[1. 무엇이 일어났나] 초기 설정 {'th_heat': 37.8, 'th_cold': 36.6, 'persist': 1, 'corr': 0.0} -> {'경보': 2570, '오경보율': 0.405, '재현율': 1.0, '종합정확도(F1)': 0.746}\n",
"구분 원인 \n",
"오경보 강풍 센서 냉각 오차 380\n",
" 기타 5\n",
" 센서 튐 영향 6\n",
" 임계값 근접(조기 경보) 650\n",
"\n",
"[2. 왜 일어났나] 오경보 유형: {'normal': 1041} / 오경보 발생 병사 수: 58\n",
"\n",
"[3. 무엇을 바꾸나] 제약 만족 조합 상위 5개(재발 적은 순) (전체 조합 96개 중 8개 만족)\n",
" th_heat th_cold persist corr 경보 오경보율 재현율 종합정확도(F1) 재발\n",
" 38.2 36.2 3 0.00 1487 0.073 0.902 0.914 20\n",
" 38.2 36.4 3 0.25 1545 0.103 0.906 0.902 21\n",
" 38.2 36.2 2 0.00 1543 0.086 0.923 0.919 23\n",
" 38.2 36.4 2 0.25 1603 0.116 0.927 0.905 27\n",
" 38.2 36.2 1 0.00 1601 0.102 0.940 0.918 29\n",
"\n",
"재보정 후: {'경보': 1487, '오경보율': 0.073, '재현율': 0.902, '종합정확도(F1)': 0.914}\n",
"종합정확도 변화 0.746 -> 0.914 (+16.8%p, 목표 +10%p 이상: 달성)\n",
"AAR에서 식별한 (병사, 원인) 오경보 62종 -> 재보정 후 같은 병사·같은 원인으로 재발 20건 (목표 0건: 미달)\n",
"\n",
"[재보정 파라미터(wel01_tuned_params.yaml 내용)]\n",
"heat_temp_threshold: 38.2\n",
"cold_temp_threshold: 36.2\n",
"min_persist_minutes: 3\n",
"wind_cooling_correction_c: 0.0\n",
"constraints: {max_far: 0.15, min_recall: 0.90}\n",
"개선 조치: 열 기준 38.2도·한랭 기준 36.2도, 3분 연속 지속 조건으로 오경보를 줄였다(센서 냉각 보정 0.0도). 파라미터는 실제 장비·인원 자료로 재검증하고 군의관 검토를 거쳐야 하며 진단 기준이 아니다.\n",
"```\n",
"</details>\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"\n",
"## 🎯 미션 10. 더미 생체신호 기반 온열·한랭 스트레스 경보 시스템 통합 패키지 완성\n",
"**소요시간**: 약 3.5시간\n",
"\n",
"### 📋 미션 배경 및 수행 지침\n",
"#### 미션 배경 및 당위성\n",
"\n",
"더미 생체신호 기반 열·한랭 스트레스 경보 시스템 통합 패키지를 완성한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **데이터 검증부터 전처리, 위험 점수, 지속 조건 경보, 군의관 검토 대기열, 상황판, 재보정값 적용까지 전체 파이프라인을 통합하고 브리핑 패키지를 완성한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
"\n",
"이 문제를 해결하기 위해 본 미션에서는 **재보정 파라미터로 전체 과정을 한 번에 실행하는 통합 코드를 완성하고 손상 데이터와 기기 두절 병사가 자동 처리되는지 확인하고 유효성을 점검하는** 실습을 체계적으로 수행합니다.\n",
"\n",
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
"\n",
"**이번 결과물:** `src/main_wel01.py, README.md, docs/wel01_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 dis\n",
"import hashlib\n",
"import json\n",
"import sys\n",
"import time\n",
"import numpy as np\n",
"import pandas as pd\n",
"import matplotlib.pyplot as plt\n",
"from sklearn.linear_model import LogisticRegression\n",
"from sklearn.pipeline import make_pipeline\n",
"from sklearn.preprocessing import StandardScaler\n",
"\n",
"# ---------------- [교육용가상자료] 더미 생체신호 생성기 (실제 장병·부대·의무 기록 없음) ----------------\n",
"# 가상 병사(가명 ID)가 60분간 걷기·휴식·고강도 활동을 하는 동안의 심박수·체온(추정 심부)·피부전도도를 1분 간격으로 만든다.\n",
"# 진단 도구가 아니며, 상태 라벨은 가상 생리 모델의 '교육용 단계'일 뿐이다.\n",
"N_MIN = 60\n",
"STAGE = {0: \"정상\", 1: \"주의\", 2: \"경보\"}\n",
"TYPES = [\"normal\", \"heat\", \"cold\"]\n",
"\n",
"\n",
"def make_person(i, rng):\n",
" return {\"id\": f\"S{i:03d}\", \"rest_hr\": float(rng.normal(64, 7)), \"age\": int(rng.integers(19, 27)), \"bmi\": float(rng.normal(23.5, 2.5)),\n",
" \"acclim\": int(rng.random() < 0.4), \"fit\": float(np.clip(rng.normal(0.5, 0.2), 0.05, 0.95))}\n",
"\n",
"\n",
"def make_env(kind, rng):\n",
" if kind == \"heat\":\n",
" return {\"temp_c\": float(rng.uniform(30, 40)), \"rh\": float(rng.uniform(45, 85)), \"wind\": float(rng.uniform(0.3, 3))}\n",
" if kind == \"cold\":\n",
" return {\"temp_c\": float(rng.uniform(-22, 2)), \"rh\": float(rng.uniform(40, 80)), \"wind\": float(rng.uniform(2, 11))}\n",
" return {\"temp_c\": float(rng.uniform(8, 24)), \"rh\": float(rng.uniform(35, 70)), \"wind\": float(rng.uniform(0.3, 5))}\n",
"\n",
"\n",
"def make_session(sid, kind, rng, p=None, glitch=True):\n",
" \"\"\"반환: dict(person, env, kind, df[hr,temp,eda,act + 정답 core·stage·rtype]). 센서 결측·튐·고착 포함.\"\"\"\n",
" p = p or make_person(sid, rng)\n",
" env = make_env(kind, rng)\n",
" act = np.zeros(N_MIN); t = 0\n",
" while t < N_MIN: # 10~20분 단위 활동 블록: 휴식 1.5 / 행군 4.5 / 고강도 7 MET\n",
" d = int(rng.integers(10, 21)); act[t:t + d] = rng.choice([1.5, 4.5, 7.0], p=[0.25, 0.5, 0.25]); t += d\n",
" hi = env[\"temp_c\"] + (0.12 * (env[\"rh\"] - 40) if env[\"temp_c\"] > 25 else 0)\n",
" wc = env[\"temp_c\"] - 1.6 * env[\"wind\"]\n",
" core, cores = 36.9, []\n",
" for m in range(N_MIN):\n",
" heat = 0.085 * max(0, hi - 24) * (1 - 0.25 * p[\"acclim\"]) * (1 + 0.5 * m / N_MIN) * (1 + 0.012 * (p[\"bmi\"] - 23.5))\n",
" cold = -0.075 * max(0, 8 - wc) * (1 - 0.09 * act[m])\n",
" ss = 36.9 + 0.09 * act[m] + heat + cold\n",
" core += (ss - core) / 18 + rng.normal(0, 0.03)\n",
" cores.append(core)\n",
" core = np.array(cores)\n",
" hr = p[\"rest_hr\"] + 9.5 * (act - 1.5) * (1.15 - 0.3 * p[\"fit\"]) + 11 * (core - 36.9) + rng.normal(0, 3.2, N_MIN)\n",
" eda = 2.0 + 0.25 * act + 3.2 * np.maximum(core - 37.2, 0) - 0.04 * np.maximum(8 - wc, 0) + rng.normal(0, 0.25, N_MIN)\n",
" bias = rng.normal(0, 0.1)\n",
" temp = core + bias + rng.normal(0, 0.1, N_MIN) - np.where((kind == \"cold\") & (env[\"wind\"] > 7), 0.25, 0.0) # 센서 접촉 부위 바람 냉각 오차\n",
" df = pd.DataFrame({\"min\": np.arange(N_MIN), \"hr\": hr, \"temp\": temp, \"eda\": np.maximum(eda, 0.2), \"act\": act})\n",
" if glitch:\n",
" for col in (\"hr\", \"temp\", \"eda\"):\n",
" df.loc[rng.random(N_MIN) < 0.03, col] = np.nan # 결측\n",
" s = int(rng.integers(5, N_MIN))\n",
" if rng.random() < 0.12:\n",
" df.loc[s, \"hr\"] = 230.0 # 센서 튐\n",
" if rng.random() < 0.12:\n",
" df.loc[s, \"temp\"] = 45.0\n",
" ht = np.where(core >= 39.0, 2, np.where(core >= 38.3, 1, 0)); cd = np.where(core <= 35.4, 2, np.where(core <= 36.2, 1, 0))\n",
" df[\"core\"] = core\n",
" df[\"stage\"] = np.maximum(ht, cd); df[\"rtype\"] = np.where(ht > 0, \"heat\", np.where(cd > 0, \"cold\", \"normal\"))\n",
" return {\"id\": p[\"id\"], \"person\": p, \"env\": env, \"kind\": kind, \"df\": df}\n",
"\n",
"\n",
"def make_dataset(n, seed):\n",
" rng = np.random.default_rng(seed)\n",
" kinds = [\"normal\", \"heat\", \"cold\"]\n",
" return [make_session(i, kinds[i % 3], rng) for i in range(n)]\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"# ---------------- 전처리·특징 추출 공통 함수 ----------------\n",
"LIMITS = {\"hr\": (30, 220), \"temp\": (30.0, 43.0), \"eda\": (0.0, 40.0)} # 생리적으로 가능한 범위(범위 밖 = 센서 오류)\n",
"\n",
"\n",
"def clean(df):\n",
" \"\"\"범위 밖 값은 결측 처리 -> 선형 보간(양 끝은 가까운 값) -> 3분 이동 중앙값으로 튐 제거. 반환: (정제 df, 보정 건수)\"\"\"\n",
" # [TODO : 학생 실습 10] 더미 생체신호 기반 온열·한랭 스트레스 경보 시스템 통합 패키지 완성 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def env_index(env):\n",
" # [TODO : 학생 실습 10] 더미 생체신호 기반 온열·한랭 스트레스 경보 시스템 통합 패키지 완성 관련 핵심 로직을 구현하세요.\n",
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
" # <<< 여기에 코드를 작성하세요 >>>\n",
" pass # 작성 후 pass를 제거하세요.\n",
"def features(s, use_env=True, personal=True):\n",
" \"\"\"세션 1개 -> 분 단위 특징표(최근 5분 평균·기울기 포함). 정답 stage/rtype 포함.\"\"\"\n",
" df, _ = clean(s[\"df\"]); p, e = s[\"person\"], s[\"env\"]\n",
" f = pd.DataFrame({\"sid\": s[\"id\"], \"min\": df[\"min\"]})\n",
" f[\"temp5\"] = df[\"temp\"].rolling(5, min_periods=1).mean()\n",
" f[\"hr5\"] = df[\"hr\"].rolling(5, min_periods=1).mean()\n",
" f[\"eda5\"] = df[\"eda\"].rolling(5, min_periods=1).mean()\n",
" f[\"dtemp\"] = (f[\"temp5\"] - f[\"temp5\"].shift(4)).fillna(0)\n",
" f[\"act\"] = df[\"act\"]\n",
" if personal:\n",
" f[\"hr_rel\"] = f[\"hr5\"] - p[\"rest_hr\"]\n",
" f[\"acclim\"], f[\"bmi\"], f[\"fit\"] = p[\"acclim\"], p[\"bmi\"], p[\"fit\"]\n",
" if use_env:\n",
" f[\"hi\"], f[\"wc\"] = env_index(e); f[\"wind\"] = e[\"wind\"]\n",
" f[\"stage\"], f[\"rtype\"], f[\"kind\"] = df[\"stage\"], df[\"rtype\"], s[\"kind\"]\n",
" return f[f[\"min\"] >= 5].reset_index(drop=True)\n",
"\n",
"\n",
"def table(sessions, **kw):\n",
" return pd.concat([features(s, **kw) for s in sessions], ignore_index=True)\n",
"# ---------------------------------------------------------------------------\n",
"\n",
"CONFIG = {\"alert_threshold\": 0.4, \"min_persist\": 3, \"heat_temp\": 38.2, \"cold_temp\": 36.2} # 미션 9 재보정값 + 모델 임계값(지속 조건이 오경보를 막으므로 미션 5의 0.5보다 낮춰 미탐을 줄임)\n",
"FE = [\"temp5\", \"dtemp\", \"hr5\", \"eda5\", \"hr_rel\", \"acclim\", \"bmi\", \"fit\", \"hi\", \"wc\", \"wind\"]\n",
"\n",
"\n",
"def validate(s):\n",
" errs = []\n",
" if s[\"kind\"] not in TYPES:\n",
" errs.append(\"kind 허용값 아님\")\n",
" if len(s[\"df\"]) != N_MIN or not {\"hr\", \"temp\", \"eda\"} <= set(s[\"df\"].columns):\n",
" errs.append(\"신호 길이·필드 오류\")\n",
" if not str(s[\"id\"]).startswith(\"S\"):\n",
" errs.append(\"soldier_id 형식 오류\")\n",
" return errs\n",
"\n",
"\n",
"def run_session(s, model):\n",
" \"\"\"세션 1개: 정제 -> 특징 -> 점수 -> 지속 조건 경보. 신호가 5분 이상 끊기면 '신호두절' 플래그.\"\"\"\n",
" f = features(s)\n",
" raw = s[\"df\"].set_index(\"min\")\n",
" f[\"lost\"] = [bool(raw.loc[max(m - 4, 0):m, [\"hr\", \"temp\"]].isna().all().all()) for m in f[\"min\"]]\n",
" f[\"score\"] = model.predict_proba(f[FE])[:, 1]\n",
" hot = (f[\"temp5\"] >= CONFIG[\"heat_temp\"]) | (f[\"temp5\"] <= CONFIG[\"cold_temp\"])\n",
" cond = (f[\"score\"] >= CONFIG[\"alert_threshold\"]) & (hot | (f[\"score\"] >= 0.8))\n",
" f[\"alert\"] = cond.rolling(CONFIG[\"min_persist\"], min_periods=CONFIG[\"min_persist\"]).min().fillna(0).astype(bool) & ~f[\"lost\"]\n",
" f[\"grade\"] = np.where(f[\"score\"] >= 0.8, \"High\", np.where(f[\"score\"] >= 0.5, \"Medium\", \"Low\"))\n",
" return f\n",
"\n",
"\n",
"class Queue:\n",
" def __init__(self):\n",
" self.items, self.log = [], []\n",
"\n",
" def submit(self, sid, minute, score, grade):\n",
" prev = self.log[-1][\"hash\"] if self.log else \"GENESIS\"\n",
" body = {\"event\": \"SUBMIT\", \"id\": f\"{sid}@{minute}\", \"score\": round(score, 3), \"by\": \"AI\"}\n",
" self.log.append({**body, \"prev\": prev, \"hash\": hashlib.sha256((prev + json.dumps(body, sort_keys=True)).encode()).hexdigest()})\n",
" self.items.append({\"id\": f\"{sid}@{minute}\", \"grade\": grade, \"status\": \"PENDING\"})\n",
"\n",
" def verify(self):\n",
" prev = \"GENESIS\"\n",
" for rec in self.log:\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, sort_keys=True)).encode()).hexdigest():\n",
" return False\n",
" prev = rec[\"hash\"]\n",
" return True\n",
"\n",
"\n",
"def episodes(f):\n",
" \"\"\"연속 경보 분을 하나의 사건으로 묶어 첫 분만 검토 대기열에 올린다(같은 사건 중복 통보 방지).\"\"\"\n",
" a = f[\"alert\"].values\n",
" return [i for i in range(len(a)) if a[i] and (i == 0 or not a[i - 1])]\n",
"\n",
"\n",
"t_start = time.perf_counter()\n",
"# ---------------- 1. 학습(점수 엔진 준비) ----------------\n",
"tr = table(make_dataset(150, seed=51))\n",
"model = make_pipeline(StandardScaler(), LogisticRegression(class_weight=\"balanced\", max_iter=3000, random_state=0)).fit(tr[FE], tr[\"stage\"] >= 1)\n",
"\n",
"# ---------------- 2. 운용 데이터 처리(손상 1건, 기기 두절 1건 주입) ----------------\n",
"rng = np.random.default_rng(61)\n",
"ops = [make_session(i, TYPES[i % 3], rng) for i in range(60)]\n",
"ops[4] = {**ops[4], \"kind\": \"tundra\"} # 허용되지 않은 유형\n",
"dead = ops[8][\"df\"].copy(); dead.loc[35:, [\"hr\", \"temp\", \"eda\"]] = np.nan; ops[8] = {**ops[8], \"df\": dead} # 기기 두절\n",
"queue, rejected, runs, results = Queue(), {}, [], []\n",
"codes = {f.__code__ for f in (validate, run_session, episodes, Queue.submit, Queue.verify, clean, env_index, features, table)}\n",
"lines_total = {c: {ln for _, ln in dis.findlinestarts(c) if ln != c.co_firstlineno} for c in codes}\n",
"trace_lines = set()\n",
"\n",
"\n",
"def tracer(frame, event, arg):\n",
" if frame.f_code in codes:\n",
" def local(fr, ev, ar):\n",
" if ev == \"line\":\n",
" trace_lines.add((fr.f_code, fr.f_lineno))\n",
" return local\n",
" return local\n",
"\n",
"\n",
"sys.settrace(tracer)\n",
"for s in ops:\n",
" errs = validate(s)\n",
" if errs:\n",
" rejected[s[\"id\"]] = errs; continue\n",
" t0 = time.perf_counter()\n",
" f = run_session(s, model)\n",
" for i in episodes(f):\n",
" queue.submit(s[\"id\"], int(f[\"min\"].iloc[i]), float(f[\"score\"].iloc[i]), f[\"grade\"].iloc[i])\n",
" runs.append({\"sid\": s[\"id\"], \"ms_per_min\": (time.perf_counter() - t0) * 1000 / len(f), \"events\": len(episodes(f)), \"lost\": int(f[\"lost\"].any())})\n",
" results.append(f)\n",
"queue.verify()\n",
"sys.settrace(None)\n",
"covered = sum(len({ln for c2, ln in trace_lines if c2 is c} & lines_total[c]) for c in codes)\n",
"cov = covered / sum(len(v) for v in lines_total.values())\n",
"\n",
"# ---------------- 3. 성능 평가 ----------------\n",
"allf = pd.concat(results, ignore_index=True)\n",
"truth = allf[\"stage\"] >= 1\n",
"far = (allf[\"alert\"] & ~truth).sum() / max(allf[\"alert\"].sum(), 1)\n",
"rec = (allf[\"alert\"] & truth).sum() / truth.sum()\n",
"risk_sids = set(allf.loc[truth, \"sid\"]); alerted = set(allf.loc[allf[\"alert\"] & truth, \"sid\"])\n",
"ms = np.mean([r[\"ms_per_min\"] for r in runs])\n",
"\n",
"# ---------------- 4. 자체 점검(통합 테스트) ----------------\n",
"chk = {\n",
" \"손상 데이터 차단\": \"S004\" in rejected,\n",
" \"처리 세션 59개 전량 성공\": len(runs) == 59,\n",
" \"신호두절 병사 식별\": any(r[\"lost\"] for r in runs) and bool(allf.loc[allf[\"sid\"] == \"S008\", \"lost\"].any()),\n",
" \"위험 구간 재현율 85% 이상\": rec >= 0.85,\n",
" \"오경보율 15% 이하\": far <= 0.15,\n",
" \"위험 병사 사건 식별 90% 이상\": len(alerted) / max(len(risk_sids), 1) >= 0.9,\n",
" \"자동 조치 0건\": all(i[\"status\"] == \"PENDING\" for i in queue.items),\n",
" \"감사 로그 건수 일치·무결\": len(queue.log) == len(queue.items) and queue.verify(),\n",
" \"분당 처리 50ms 이내\": ms < 50,\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)} 통과 ({passed / len(chk):.0%}), 파이프라인 실행 성공 {'예' if len(runs) + len(rejected) == len(ops) else '아니오'}, 핵심 모듈 실행 라인 커버리지 {cov:.1%}\")\n",
"\n",
"# ---------------- 5. 소대 상황판 ----------------\n",
"colr = {\"High\": \"tab:red\", \"Medium\": \"orange\", \"Low\": \"tab:gray\"}\n",
"fig, ax = plt.subplots(2, 2, figsize=(12, 8))\n",
"top = sorted(results, key=lambda f: -f[\"score\"].max())[:2]\n",
"for a, f in zip(ax[0], top):\n",
" a.plot(f[\"min\"], f[\"temp5\"], label=\"temp (5-min mean)\"); a.scatter(f[\"min\"][f[\"alert\"]], f[\"temp5\"][f[\"alert\"]], color=\"red\", s=12, label=\"alert\")\n",
" a.set_title(f\"{f['sid'].iloc[0]} temperature and alerts\"); a.set_xlabel(\"minute\"); a.legend()\n",
"g = pd.Series([i[\"grade\"] for i in queue.items]).value_counts().reindex([\"High\", \"Medium\", \"Low\"]).fillna(0)\n",
"ax[1, 0].bar(g.index, g.values, color=[colr[i] for i in g.index]); ax[1, 0].set_title(\"Review queue events by grade (pending human review)\")\n",
"ax[1, 1].axis(\"off\")\n",
"ax[1, 1].text(0, 0.9, f\"Risk recall {rec:.1%}\\nFalse alarm {far:.1%}\\nQueue {len(queue.items)} pending\\nLatency {ms:.2f} ms/min\\nSelf-check {passed}/{len(chk)}\", family=\"monospace\", fontsize=13, va=\"top\")\n",
"plt.suptitle(\"WEL01 integrated pipeline - dummy data, not diagnosis\"); plt.tight_layout(); plt.show()\n",
"\n",
"# ---------------- 6. 브리핑 요약 ----------------\n",
"print(f\"\"\"\n",
"[열·한랭 스트레스 경보 브리핑 요약 - 교육용가상자료, 진단 아님]\n",
"1. 처리: 가상 병사 {len(ops)}명 중 {len(runs)}명 처리, 손상 데이터 {len(rejected)}건 차단({', '.join(rejected)}), 신호두절 {sum(r['lost'] for r in runs)}명 확인 필요 표시\n",
"2. 성능: 위험 구간 재현율 {rec:.1%}, 오경보율 {far:.1%}, 분당 {ms:.2f} ms, 전체 소요 {time.perf_counter() - t_start:.1f} s\n",
"3. 경보: {len(queue.items)}건 군의관·소대장 검토 대기 (High {sum(i['grade'] == 'High' for i in queue.items)} / Medium {sum(i['grade'] == 'Medium' for i in queue.items)} / Low {sum(i['grade'] == 'Low' for i in queue.items)}), 자동 조치 0건\n",
"4. 운용 유의: 휴식 권고·임무 배제·의무대 인계는 군의관·소대장·지휘관의 승인 후에만 이루어지며, 의학적 판단은 의무대·군의관에게 연계한다.\n",
"5. 한계: 가상 생리 모델로 만든 더미 신호 기준 결과로, 진단 도구가 아니며 실제 장비·인원 자료로 재검증이 필요하다.\"\"\")\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<details>\n",
"<summary><b>💡 [참고] 미션 10 예상 실행 결과 (Output)</b></summary>\n",
"\n",
"```text\n",
"[PASS] 손상 데이터 차단\n",
"[PASS] 처리 세션 59개 전량 성공\n",
"[PASS] 신호두절 병사 식별\n",
"[PASS] 위험 구간 재현율 85% 이상\n",
"[PASS] 오경보율 15% 이하\n",
"[PASS] 위험 병사 사건 식별 90% 이상\n",
"[PASS] 자동 조치 0건\n",
"[PASS] 감사 로그 건수 일치·무결\n",
"[PASS] 분당 처리 50ms 이내\n",
"[PASS] 실행 라인 커버리지 85% 이상\n",
"자체 점검 10/10 통과 (100%), 파이프라인 실행 성공 예, 핵심 모듈 실행 라인 커버리지 92.7%\n",
"\n",
"[열·한랭 스트레스 경보 브리핑 요약 - 교육용가상자료, 진단 아님]\n",
"1. 처리: 가상 병사 60명 중 59명 처리, 손상 데이터 1건 차단(S004), 신호두절 1명 확인 필요 표시\n",
"2. 성능: 위험 구간 재현율 85.6%, 오경보율 8.3%, 분당 2.24 ms, 전체 소요 10.2 s\n",
"3. 경보: 40건 군의관·소대장 검토 대기 (High 24 / Medium 15 / Low 1), 자동 조치 0건\n",
"4. 운용 유의: 휴식 권고·임무 배제·의무대 인계는 군의관·소대장·지휘관의 승인 후에만 이루어지며, 의학적 판단은 의무대·군의관에게 연계한다.\n",
"5. 한계: 가상 생리 모델로 만든 더미 신호 기준 결과로, 진단 도구가 아니며 실제 장비·인원 자료로 재검증이 필요하다.\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