2897 lines
209 KiB
Plaintext
2897 lines
209 KiB
Plaintext
{
|
|
"cells": [
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"# 🎯 [MNT10] AI 기반 정비인력 부하율 예측·공정균형\n",
|
|
"### Google Colab 실습 환경 (장비정비·수리부속 / Level 3)\n",
|
|
"\n",
|
|
"[](https://colab.research.google.com/github/edunlab/milicontent2/blob/main/uploads/MNT10/MNT10_solution.ipynb)\n",
|
|
"\n",
|
|
"---\n",
|
|
"\n",
|
|
"## 📌 1. 프로젝트 개요 및 배경\n",
|
|
"\n",
|
|
"| 항목 | 내용 |\n",
|
|
"| :--- | :--- |\n",
|
|
"| **과정 분류** | 장비정비·수리부속 (Level 3) |\n",
|
|
"| **학습 소요 시간** | 10개 미션 (5일 / 총 28.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",
|
|
"이 문제를 해결하기 위해 본 미션에서는 **요청 접수부터 배정·출동·정비·복귀·실적 입력까지 업무 흐름을 단계별로 정리하고, 요청·정비반·결과 세 묶음으로 피처 20개 이상 정의하고 검증하는** 실습을 체계적으로 수행합니다.\n",
|
|
"\n",
|
|
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
|
|
"\n",
|
|
"**이번 결과물:** `docs/mnt10_logistics_domain_spec.md 및 데이터 딕셔너리` 형태의 결과물을 완성한다.\n",
|
|
"\n",
|
|
"### 🔍 평가 및 검증 기준 (PASS 조건)\n",
|
|
"#### 미션 판정\n",
|
|
"\n",
|
|
"**평가 구분:** 필수\n",
|
|
"\n",
|
|
"**핵심 평가영역:** 군수 실무 업무 프로세스와의 일치성 및 분석 목표의 명확성\n",
|
|
"\n",
|
|
"#### 미션 PASS 조건\n",
|
|
"\n",
|
|
"- 정량 기준: 업무 프로세스 맵 작성 완료, 필요 피처(Feature) 목록 20개 이상 정의\n",
|
|
"\n",
|
|
"\n",
|
|
"- 진행 기준: 도메인 분석서 및 데이터 요구명세서 승인\n",
|
|
"\n",
|
|
"\n",
|
|
"- 검증 방법: 군수 담당 실무자 인터뷰 및 기존 전산체계 데이터 구조 대조\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",
|
|
"# 1) 업무 프로세스 맵: 접수부터 부하율 집계·사후검토까지 (각 단계에서 생기는 데이터)\n",
|
|
"PROCESS = [(\"접수\", \"요청 ID, 접수시각, 가상 좌표, 장비 구분·코드\"), (\"분류\", \"고장 유형, 작전 중요도, 부품 확보 여부\"),\n",
|
|
" (\"우선순위\", \"우선순위 점수, 예상 정비시간\"), (\"배정\", \"정비반 ID, 이동거리, 대기(백로그)시간\"),\n",
|
|
" (\"출동\", \"출발·도착 시각, 기상, 도로 상태\"), (\"현장정비/후송\", \"현장 수리 가능 여부, 실제 정비시간\"),\n",
|
|
" (\"완료보고\", \"완료 시각, 장비 가동중단 시간\"), (\"부하율 집계\", \"정비반별 일일 투입시간/가용시간\"),\n",
|
|
" (\"사후검토(AAR)\", \"지연·오배정 원인, 가중치 조정 이력\")]\n",
|
|
"\n",
|
|
"# 2) 스키마(데이터 딕셔너리): 그룹 / 필드 / 자료형 / 규칙 / 설명\n",
|
|
"S = {\n",
|
|
" \"request\": {\n",
|
|
" \"req_id\": (str, {\"pattern\": r\"^R\\d{4}$\"}, \"요청 ID\"),\n",
|
|
" \"t_req\": (float, {\"min\": 0}, \"접수시각(모의 시작 후 경과 시간, h)\"),\n",
|
|
" \"unit_code\": (str, {\"pattern\": r\"^VU-\\d{2}$\"}, \"요청 부대 코드(가상)\"),\n",
|
|
" \"grid_x\": (float, {\"min\": 0, \"max\": 60}, \"고장 위치 X(가상 좌표, km)\"),\n",
|
|
" \"grid_y\": (float, {\"min\": 0, \"max\": 60}, \"고장 위치 Y(가상 좌표, km)\"),\n",
|
|
" \"equip_type\": (str, {\"enum\": [\"mobility\", \"fires\", \"comms\", \"power\"]}, \"장비 구분\"),\n",
|
|
" \"equip_code\": (str, {\"pattern\": r\"^VX-\\d{4}-\\d{4}$\"}, \"장비 식별코드(국방 표준 코드 형식을 모사한 가상 코드)\"),\n",
|
|
" \"fsc_group\": (str, {\"pattern\": r\"^\\d{4}$\"}, \"보급 분류군(가상 4자리)\"),\n",
|
|
" \"crit\": (int, {\"enum\": [1, 2, 3]}, \"작전 중요도(1 일반, 2 중요, 3 핵심)\"),\n",
|
|
" \"fault_type\": (str, {\"enum\": [\"engine\", \"drivetrain\", \"hydraulic\", \"electric\", \"comm_module\", \"cooling\"]}, \"고장 유형\"),\n",
|
|
" \"parts_ready\": (bool, {}, \"수리부속 확보 여부\"),\n",
|
|
" \"weather\": (str, {\"enum\": [\"clear\", \"rain\", \"snow\"]}, \"기상\"),\n",
|
|
" \"road\": (str, {\"enum\": [\"paved\", \"unpaved\"]}, \"접근 도로 상태\"),\n",
|
|
" },\n",
|
|
" \"team\": {\n",
|
|
" \"team_id\": (str, {\"pattern\": r\"^MT\\d$\"}, \"이동정비반 ID\"),\n",
|
|
" \"base_x\": (float, {\"min\": 0, \"max\": 60}, \"정비반 대기 위치 X\"),\n",
|
|
" \"base_y\": (float, {\"min\": 0, \"max\": 60}, \"정비반 대기 위치 Y\"),\n",
|
|
" \"crew\": (int, {\"min\": 2, \"max\": 8}, \"가용 정비인력(명)\"),\n",
|
|
" \"skills\": (list, {}, \"정비 가능 장비 구분\"),\n",
|
|
" \"status\": (str, {\"enum\": [\"ready\", \"moving\", \"repairing\", \"returning\"]}, \"출동 가능 상태\"),\n",
|
|
" },\n",
|
|
" \"result\": {\n",
|
|
" \"std_repair_h\": (float, {\"min\": 0, \"max\": 48}, \"표준 정비시간(4인 기준) - 예측대상 1\"),\n",
|
|
" \"response_h\": (float, {\"min\": 0, \"max\": 72}, \"접수~도착 시간 - 예측대상 2(출동 지연)\"),\n",
|
|
" \"downtime_h\": (float, {\"min\": 0, \"max\": 240}, \"장비 가동중단 시간\"),\n",
|
|
" \"field_ok\": (bool, {}, \"현장 수리 가능 여부 - 예측대상 3\"),\n",
|
|
" \"load_rate\": (float, {\"min\": 0, \"max\": 3}, \"정비반 일일 부하율 = 투입시간/가용시간(10h) - 예측대상 4\"),\n",
|
|
" },\n",
|
|
"}\n",
|
|
"\n",
|
|
"\n",
|
|
"def check(name, v, typ, rule):\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",
|
|
"recs = []\n",
|
|
"for i in range(20):\n",
|
|
" std = float(round(rng.uniform(1, 9), 2))\n",
|
|
" recs.append({\n",
|
|
" \"request\": {\"req_id\": f\"R{i:04d}\", \"t_req\": float(round(rng.uniform(0, 720), 1)), \"unit_code\": f\"VU-{rng.integers(1, 30):02d}\",\n",
|
|
" \"grid_x\": float(round(rng.uniform(0, 60), 1)), \"grid_y\": float(round(rng.uniform(0, 60), 1)),\n",
|
|
" \"equip_type\": str(rng.choice([\"mobility\", \"fires\", \"comms\", \"power\"])),\n",
|
|
" \"equip_code\": f\"VX-{rng.integers(1000, 9999)}-{rng.integers(1000, 9999)}\", \"fsc_group\": f\"{rng.integers(1000, 9999)}\",\n",
|
|
" \"crit\": int(rng.integers(1, 4)), \"fault_type\": str(rng.choice([\"engine\", \"hydraulic\", \"electric\", \"cooling\"])),\n",
|
|
" \"parts_ready\": bool(rng.random() < 0.75), \"weather\": str(rng.choice([\"clear\", \"rain\", \"snow\"])), \"road\": \"paved\"},\n",
|
|
" \"team\": {\"team_id\": f\"MT{rng.integers(1, 7)}\", \"base_x\": 30.0, \"base_y\": 6.0, \"crew\": int(rng.integers(3, 6)),\n",
|
|
" \"skills\": [\"mobility\", \"comms\"], \"status\": \"ready\"},\n",
|
|
" \"result\": {\"std_repair_h\": std, \"response_h\": float(round(rng.uniform(0.5, 6), 2)),\n",
|
|
" \"downtime_h\": float(round(std + rng.uniform(1, 8), 2)), \"field_ok\": bool(rng.random() < 0.6),\n",
|
|
" \"load_rate\": float(round(rng.uniform(0.3, 1.3), 2))}})\n",
|
|
"\n",
|
|
"ok = sum(not validate(r) for r in recs)\n",
|
|
"print(f\"정상 샘플 {len(recs)}건 중 검증 통과 {ok}건, 오류 {len(recs) - ok}건\")\n",
|
|
"\n",
|
|
"# 4) 오류 주입 시험\n",
|
|
"bad = json.loads(json.dumps(recs[0]))\n",
|
|
"bad[\"request\"][\"grid_x\"] = 75.0 # 작전지역 밖 좌표\n",
|
|
"bad[\"request\"][\"crit\"] = 5 # 허용되지 않은 중요도\n",
|
|
"bad[\"request\"][\"equip_code\"] = \"1234567890123\" # 코드 형식 오류\n",
|
|
"del bad[\"team\"][\"crew\"] # 가용 인력 누락\n",
|
|
"bad[\"result\"][\"load_rate\"] = \"90%\" # 자료형 오류(문자열)\n",
|
|
"print(\"\\n[오류 주입 결과]\")\n",
|
|
"for e in validate(bad):\n",
|
|
" print(\" -\", e)\n",
|
|
"\n",
|
|
"# 5) 프로세스 맵과 데이터 딕셔너리\n",
|
|
"print(\"\\n[업무 프로세스 맵]\")\n",
|
|
"print(\" -> \".join(f\"{i}.{p}\" for i, (p, _) in enumerate(PROCESS, 1)))\n",
|
|
"for p, d in PROCESS:\n",
|
|
" print(f\" {p:8s}: {d}\")\n",
|
|
"rows = [(g, k, t.__name__, str(r) if r else \"-\", d) for g, f in S.items() for k, (t, r, d) in f.items()]\n",
|
|
"dd = pd.DataFrame(rows, columns=[\"그룹\", \"필드\", \"자료형\", \"규칙\", \"설명\"])\n",
|
|
"print(f\"\\n[데이터 딕셔너리] 피처 {len(dd)}개 (목표 20개 이상: {'달성' if len(dd) >= 20 else '미달'})\")\n",
|
|
"print(dd[[\"그룹\", \"필드\", \"자료형\", \"설명\"]].to_string(index=False))\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"<details>\n",
|
|
"<summary><b>💡 [참고] 미션 01 예상 실행 결과 (Output)</b></summary>\n",
|
|
"\n",
|
|
"```text\n",
|
|
"정상 샘플 20건 중 검증 통과 20건, 오류 0건\n",
|
|
"\n",
|
|
"[오류 주입 결과]\n",
|
|
" - request.grid_x: 최대값 60 초과\n",
|
|
" - request.equip_code: 형식 불일치 '1234567890123'\n",
|
|
" - request.crit: 허용되지 않은 값 '5'\n",
|
|
" - team.crew: 필수 필드 누락\n",
|
|
" - result.load_rate: 자료형 오류(str)\n",
|
|
"\n",
|
|
"[업무 프로세스 맵]\n",
|
|
"1.접수 -> 2.분류 -> 3.우선순위 -> 4.배정 -> 5.출동 -> 6.현장정비/후송 -> 7.완료보고 -> 8.부하율 집계 -> 9.사후검토(AAR)\n",
|
|
" 접수 : 요청 ID, 접수시각, 가상 좌표, 장비 구분·코드\n",
|
|
" 분류 : 고장 유형, 작전 중요도, 부품 확보 여부\n",
|
|
" 우선순위 : 우선순위 점수, 예상 정비시간\n",
|
|
" 배정 : 정비반 ID, 이동거리, 대기(백로그)시간\n",
|
|
" 출동 : 출발·도착 시각, 기상, 도로 상태\n",
|
|
" 현장정비/후송 : 현장 수리 가능 여부, 실제 정비시간\n",
|
|
" 완료보고 : 완료 시각, 장비 가동중단 시간\n",
|
|
" 부하율 집계 : 정비반별 일일 투입시간/가용시간\n",
|
|
" 사후검토(AAR): 지연·오배정 원인, 가중치 조정 이력\n",
|
|
"\n",
|
|
"[데이터 딕셔너리] 피처 24개 (목표 20개 이상: 달성)\n",
|
|
" 그룹 필드 자료형 설명\n",
|
|
"request req_id str 요청 ID\n",
|
|
"request t_req float 접수시각(모의 시작 후 경과 시간, h)\n",
|
|
"request unit_code str 요청 부대 코드(가상)\n",
|
|
"request grid_x float 고장 위치 X(가상 좌표, km)\n",
|
|
"request grid_y float 고장 위치 Y(가상 좌표, km)\n",
|
|
"request equip_type str 장비 구분\n",
|
|
"request equip_code str 장비 식별코드(국방 표준 코드 형식을 모사한 가상 코드)\n",
|
|
"request fsc_group str 보급 분류군(가상 4자리)\n",
|
|
"request crit int 작전 중요도(1 일반, 2 중요, 3 핵심)\n",
|
|
"request fault_type str 고장 유형\n",
|
|
"request parts_ready bool 수리부속 확보 여부\n",
|
|
"request weather str 기상\n",
|
|
"request road str 접근 도로 상태\n",
|
|
" team team_id str 이동정비반 ID\n",
|
|
" team base_x float 정비반 대기 위치 X\n",
|
|
" team base_y float 정비반 대기 위치 Y\n",
|
|
" team crew int 가용 정비인력(명)\n",
|
|
"... (이하 생략)\n",
|
|
"```\n",
|
|
"</details>\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"---\n",
|
|
"\n",
|
|
"## 🎯 미션 02. 이동정비반 보유 장비/공구 및 출동 가능 상태 정보 전처리 모듈 구현\n",
|
|
"**소요시간**: 약 2.0시간\n",
|
|
"\n",
|
|
"### 📋 미션 배경 및 수행 지침\n",
|
|
"#### 미션 배경 및 당위성\n",
|
|
"\n",
|
|
"이동정비반 보유 장비 및 출동 가능 상태 정보 전처리 모듈을 구현한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **이동정비반의 보유 장비·공구·인력·출동 가능 상태 로그를 정제하는 전처리 모듈을 구현한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
|
|
"\n",
|
|
"이 문제를 해결하기 위해 본 미션에서는 **중복 보고 제거, 정비반 ID·상태값·시각 표기 통일, 연료 단위 통일, 불가능한 인원·음수 공구 수량 결측 처리, 좌표 이상값 보간, KNN 결측 보정 등 정제 규칙 10종을 적용 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
|
|
"\n",
|
|
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
|
|
"\n",
|
|
"**이번 결과물:** `src/etl/mnt10_data_cleansing.py 및 정제 품질 보고서` 형태의 결과물을 완성한다.\n",
|
|
"\n",
|
|
"### 🔍 평가 및 검증 기준 (PASS 조건)\n",
|
|
"#### 미션 판정\n",
|
|
"\n",
|
|
"**평가 구분:** 필수\n",
|
|
"\n",
|
|
"**핵심 평가영역:** 군수 특수 예외(비계획 정비, 긴급 불출 등) 데이터 보존의 타당성\n",
|
|
"\n",
|
|
"#### 미션 PASS 조건\n",
|
|
"\n",
|
|
"- 정량 기준: 데이터 정제 규칙 10종 적용, 데이터 정합성 검증 통과율 100%\n",
|
|
"\n",
|
|
"\n",
|
|
"- 진행 기준: ETL 파이프라인 구현 및 정제된 마스터 데이터셋 구축\n",
|
|
"\n",
|
|
"\n",
|
|
"- 검증 방법: 전후 데이터 통계량(기술통계) 비교 및 무결성 검증\n",
|
|
"\n",
|
|
"\n",
|
|
"#### 채점자 판정 메모\n",
|
|
"\n",
|
|
"> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
|
|
"\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"\"\"\"이동정비반의 보유 장비·공구·인력·출동 가능 상태 로그(표기 불일치·중복·단위 혼재·결측이 섞인 가상 원천 데이터)에 정제 규칙 10종을 적용하고, 긴급 불출 등 군수 예외 데이터는 보존하면서 정합성 검증 통과율과 정제 전후 통계를 확인한다.\"\"\"\n",
|
|
"import numpy as np\n",
|
|
"import pandas as pd\n",
|
|
"import matplotlib.pyplot as plt\n",
|
|
"from sklearn.impute import KNNImputer\n",
|
|
"\n",
|
|
"# 1) [교육용가상자료] 원천 상태 로그: 정비반 6개 x 30일 x 하루 2회 보고 + 일부러 넣은 오류\n",
|
|
"rng = np.random.default_rng(2)\n",
|
|
"rows = []\n",
|
|
"for d in range(30):\n",
|
|
" for h in (8, 18):\n",
|
|
" for t in range(1, 7):\n",
|
|
" rows.append({\"team\": f\"MT{t}\", \"ts\": f\"2026-03-{d + 1:02d} {h:02d}:00\", \"status\": str(rng.choice([\"ready\", \"moving\", \"repairing\"], p=[0.6, 0.2, 0.2])),\n",
|
|
" \"crew\": int(rng.integers(3, 6)), \"fuel\": float(round(rng.uniform(35, 100), 1)), \"fuel_unit\": \"%\",\n",
|
|
" \"diag_kit\": int(rng.integers(1, 3)), \"lift_kit\": int(rng.integers(0, 2)), \"x\": float(round(rng.uniform(0, 60), 1)),\n",
|
|
" \"y\": float(round(rng.uniform(0, 60), 1)), \"emergency_issue\": False})\n",
|
|
"raw = pd.DataFrame(rows)\n",
|
|
"n0 = len(raw)\n",
|
|
"idx = rng.choice(n0, 150, replace=False)\n",
|
|
"a, b, c, d_, e, f, g, h_, k, m = np.split(idx, 10)\n",
|
|
"raw.loc[a, \"team\"] = raw.loc[a, \"team\"].str.replace(\"MT\", \"mt-\") # ID 표기 불일치\n",
|
|
"raw.loc[b, \"status\"] = rng.choice([\"READY\", \"ready \", \"대기\", \"Repairing\"], len(b)) # 상태값 표기 불일치\n",
|
|
"raw.loc[c, \"ts\"] = pd.to_datetime(raw.loc[c, \"ts\"]).dt.strftime(\"%Y%m%d %H%M\") # 시각 형식 혼재\n",
|
|
"raw.loc[d_, \"fuel\"] = (raw.loc[d_, \"fuel\"] * 4).round(0); raw.loc[d_, \"fuel_unit\"] = \"L\" # 단위 혼재(가상 탱크 400L)\n",
|
|
"raw.loc[e, \"crew\"] = rng.choice([-1, 0, 25], len(e)) # 불가능 값\n",
|
|
"raw.loc[f, \"diag_kit\"] = -1 # 음수 수량\n",
|
|
"raw.loc[g, \"x\"] = 999.0 # 좌표 범위 밖\n",
|
|
"raw.loc[h_, [\"crew\", \"fuel\"]] = np.nan # 결측\n",
|
|
"raw.loc[k, \"lift_kit\"] = 6; raw.loc[k, \"emergency_issue\"] = True # 긴급 불출(예외: 보존 대상)\n",
|
|
"raw.loc[m, \"status\"] = None # 상태 결측\n",
|
|
"raw = pd.concat([raw, raw.sample(12, random_state=1)], ignore_index=True) # 중복 보고\n",
|
|
"print(f\"원천 로그 {len(raw)}건 (중복 12건, 오류 주입 {len(idx)}건 포함)\")\n",
|
|
"\n",
|
|
"# 2) 정제 규칙 10종\n",
|
|
"df, applied = raw.copy(), []\n",
|
|
"def rule(name, n):\n",
|
|
" # [TODO : 학생 실습 2] 이동정비반 보유 장비/공구 및 출동 가능 상태 정보 전처리 모듈 구현 관련 핵심 로직을 구현하세요.\n",
|
|
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
|
|
" # <<< 여기에 코드를 작성하세요 >>>\n",
|
|
" pass # 작성 후 pass를 제거하세요.\n",
|
|
"rule(\"R1 중복 보고 제거\", df.duplicated().sum()); df = df.drop_duplicates().reset_index(drop=True)\n",
|
|
"bad_id = ~df[\"team\"].str.match(r\"^MT\\d$\"); df[\"team\"] = \"MT\" + df[\"team\"].str.extract(r\"(\\d+)\")[0]; rule(\"R2 정비반 ID 표기 통일\", bad_id.sum())\n",
|
|
"smap = {\"ready\": \"ready\", \"대기\": \"ready\", \"moving\": \"moving\", \"repairing\": \"repairing\"}\n",
|
|
"s = df[\"status\"].fillna(\"\").str.strip().str.lower().map(smap); rule(\"R3 상태값 표준화\", (s != df[\"status\"]).sum()); df[\"status\"] = s\n",
|
|
"t1 = pd.to_datetime(df[\"ts\"], format=\"%Y-%m-%d %H:%M\", errors=\"coerce\")\n",
|
|
"t2 = pd.to_datetime(df[\"ts\"], format=\"%Y%m%d %H%M\", errors=\"coerce\"); rule(\"R4 보고시각 형식 통일\", t1.isna().sum()); df[\"ts\"] = t1.fillna(t2)\n",
|
|
"lit = df[\"fuel_unit\"] == \"L\"; df.loc[lit, \"fuel\"] = df.loc[lit, \"fuel\"] / 4; df[\"fuel_unit\"] = \"%\"; rule(\"R5 연료 단위 통일(L -> %)\", lit.sum())\n",
|
|
"bad = ~df[\"crew\"].between(1, 8) & df[\"crew\"].notna(); df.loc[bad, \"crew\"] = np.nan; rule(\"R6 불가능 인원값 결측 처리\", bad.sum())\n",
|
|
"bad = df[\"diag_kit\"] < 0; df.loc[bad, \"diag_kit\"] = np.nan; rule(\"R7 음수 공구 수량 결측 처리\", bad.sum())\n",
|
|
"bad = ~df[\"x\"].between(0, 60); df.loc[bad, [\"x\", \"y\"]] = np.nan; df = df.sort_values([\"team\", \"ts\"])\n",
|
|
"df[[\"x\", \"y\"]] = df.groupby(\"team\")[[\"x\", \"y\"]].ffill().bfill(); rule(\"R8 좌표 이상값 직전 위치 보간\", bad.sum())\n",
|
|
"miss = df[[\"crew\", \"fuel\", \"diag_kit\"]].isna().sum().sum()\n",
|
|
"df[[\"crew\", \"fuel\", \"diag_kit\"]] = KNNImputer(n_neighbors=5).fit_transform(df[[\"crew\", \"fuel\", \"diag_kit\", \"lift_kit\"]])[:, :3]\n",
|
|
"df[\"crew\"] = df[\"crew\"].round().astype(int); df[\"diag_kit\"] = df[\"diag_kit\"].round().astype(int); rule(\"R9 결측치 KNN 보정\", miss)\n",
|
|
"q1, q3 = df[\"lift_kit\"].quantile([0.25, 0.75]); out = df[\"lift_kit\"] > q3 + 1.5 * (q3 - q1)\n",
|
|
"df[\"outlier_flag\"] = out & ~df[\"emergency_issue\"]; rule(\"R10 예외(긴급 불출) 보존·일반 이상값 표시\", (out & df[\"emergency_issue\"]).sum())\n",
|
|
"df[\"status\"] = df.groupby(\"team\")[\"status\"].transform(lambda x: x.ffill().fillna(\"ready\"))\n",
|
|
"df[\"dispatchable\"] = (df[\"status\"] == \"ready\") & (df[\"crew\"] >= 3) & (df[\"fuel\"] >= 30) & (df[\"diag_kit\"] >= 1)\n",
|
|
"print(\"\\n[적용 규칙과 처리 건수]\")\n",
|
|
"for nme, n in applied:\n",
|
|
" print(f\" {nme:28s} {n:4d}건\")\n",
|
|
"\n",
|
|
"# 3) 정합성 검증\n",
|
|
"checks = {\"중복 0건\": not df.duplicated([\"team\", \"ts\"]).any(), \"ID 형식\": df[\"team\"].str.match(r\"^MT[1-6]$\").all(),\n",
|
|
" \"상태값 허용 범위\": df[\"status\"].isin([\"ready\", \"moving\", \"repairing\"]).all(), \"시각 변환\": df[\"ts\"].notna().all(),\n",
|
|
" \"인원 1~8명\": df[\"crew\"].between(1, 8).all(), \"연료 0~100%\": df[\"fuel\"].between(0, 100).all(),\n",
|
|
" \"공구 수량 0 이상\": (df[\"diag_kit\"] >= 0).all(), \"좌표 범위\": df[\"x\"].between(0, 60).all(),\n",
|
|
" \"결측 0건\": df[[\"crew\", \"fuel\", \"diag_kit\", \"x\", \"y\", \"status\"]].notna().all().all(),\n",
|
|
" f\"긴급 불출 {len(k)}건 보존\": int(df[\"emergency_issue\"].sum()) == len(k)}\n",
|
|
"rate = np.mean(list(checks.values()))\n",
|
|
"print(f\"\\n정합성 검증 {sum(checks.values())}/{len(checks)} 통과, 통과율 {rate:.0%} (목표 100%: {'달성' if rate == 1 else '미달'})\")\n",
|
|
"print(f\"정제 후 {len(df)}건, 출동 가능 보고 비율 {df['dispatchable'].mean():.1%}\")\n",
|
|
"\n",
|
|
"# 4) 정제 전후 기술통계 비교\n",
|
|
"num = lambda x: pd.to_numeric(x, errors=\"coerce\")\n",
|
|
"cmp = pd.DataFrame({\"정제 전 평균\": [num(raw[c]).mean() for c in [\"crew\", \"fuel\", \"diag_kit\", \"x\"]],\n",
|
|
" \"정제 후 평균\": [df[c].mean() for c in [\"crew\", \"fuel\", \"diag_kit\", \"x\"]],\n",
|
|
" \"정제 전 최대\": [num(raw[c]).max() for c in [\"crew\", \"fuel\", \"diag_kit\", \"x\"]],\n",
|
|
" \"정제 후 최대\": [df[c].max() for c in [\"crew\", \"fuel\", \"diag_kit\", \"x\"]]}, index=[\"인원\", \"연료%\", \"진단키트\", \"좌표X\"]).round(1)\n",
|
|
"print(\"\\n[정제 전후 기술통계]\"); print(cmp.to_string())\n",
|
|
"\n",
|
|
"fig, ax = plt.subplots(1, 2, figsize=(10, 3.5))\n",
|
|
"ax[0].hist(num(raw[\"fuel\"]).dropna(), bins=30, alpha=0.6, label=\"raw\"); ax[0].hist(df[\"fuel\"], bins=30, alpha=0.6, label=\"cleaned\")\n",
|
|
"ax[0].set_title(\"Fuel level before/after unit fix\"); ax[0].legend()\n",
|
|
"ax[1].bar([n.split()[0] for n, _ in applied], [n for _, n in applied]); ax[1].set_title(\"Records changed by rule\")\n",
|
|
"plt.tight_layout(); plt.show()\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"<details>\n",
|
|
"<summary><b>💡 [참고] 미션 02 예상 실행 결과 (Output)</b></summary>\n",
|
|
"\n",
|
|
"```text\n",
|
|
"원천 로그 372건 (중복 12건, 오류 주입 150건 포함)\n",
|
|
"\n",
|
|
"[적용 규칙과 처리 건수]\n",
|
|
" R1 중복 보고 제거 12건\n",
|
|
" R2 정비반 ID 표기 통일 15건\n",
|
|
" R3 상태값 표준화 30건\n",
|
|
" R4 보고시각 형식 통일 15건\n",
|
|
" R5 연료 단위 통일(L -> %) 15건\n",
|
|
" R6 불가능 인원값 결측 처리 15건\n",
|
|
" R7 음수 공구 수량 결측 처리 15건\n",
|
|
" R8 좌표 이상값 직전 위치 보간 15건\n",
|
|
" R9 결측치 KNN 보정 60건\n",
|
|
" R10 예외(긴급 불출) 보존·일반 이상값 표시 15건\n",
|
|
"\n",
|
|
"정합성 검증 10/10 통과, 통과율 100% (목표 100%: 달성)\n",
|
|
"정제 후 360건, 출동 가능 보고 비율 60.3%\n",
|
|
"\n",
|
|
"[정제 전후 기술통계]\n",
|
|
" 정제 전 평균 정제 후 평균 정제 전 최대 정제 후 최대\n",
|
|
"인원 4.2 4.0 25.0 5.0\n",
|
|
"연료% 74.1 66.9 366.0 100.0\n",
|
|
"진단키트 1.4 1.5 2.0 2.0\n",
|
|
"좌표X 71.8 30.1 999.0 59.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",
|
|
"이 문제를 해결하기 위해 본 미션에서는 **가상 30일 요청을 Baseline으로 배정·모의해 정비반별 부하율, 도착시간, 이동시간을 산출 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
|
|
"\n",
|
|
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
|
|
"\n",
|
|
"**이번 결과물:** `notebooks/mnt10_eda_analysis.ipynb 및 EDA 요약 리포트` 형태의 결과물을 완성한다.\n",
|
|
"\n",
|
|
"### 🔍 평가 및 검증 기준 (PASS 조건)\n",
|
|
"#### 미션 판정\n",
|
|
"\n",
|
|
"**평가 구분:** 필수\n",
|
|
"\n",
|
|
"**핵심 평가영역:** 군수 의사결정에 직관적인 인사이트 제공 수준\n",
|
|
"\n",
|
|
"#### 미션 PASS 조건\n",
|
|
"\n",
|
|
"- 정량 기준: EDA 시각화 차트 8종 이상 생성, 유의미한 상관관계 피처 5개 이상 도출\n",
|
|
"\n",
|
|
"\n",
|
|
"- 진행 기준: EDA 노트북 실행 완료 및 핵심 발견점(Key Findings) 문서화\n",
|
|
"\n",
|
|
"\n",
|
|
"- 검증 방법: 통계적 가설 검정(p-value, 상관계수)을 통한 유의성 검증\n",
|
|
"\n",
|
|
"\n",
|
|
"#### 채점자 판정 메모\n",
|
|
"\n",
|
|
"> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
|
|
"\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"\"\"\"거리 기준 1차 이동정비반 배정 Baseline(접수 순서 + 최근접 정비반)을 구현해 30일 출동을 모의하고, 탐색적 분석(EDA) 차트 8종·상관분석(p-value)·파레토(ABC) 분석으로 출동 지연과 정비인력 부하 불균형의 원인을 찾는다.\"\"\"\n",
|
|
"import numpy as np\n",
|
|
"import pandas as pd\n",
|
|
"import matplotlib.pyplot as plt\n",
|
|
"from scipy import stats\n",
|
|
"\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"# [교육용가상자료] 야전 이동정비반·긴급 정비 요청 생성기 (실제 부대·위치·장비·인원 정보 없음)\n",
|
|
"# 60km x 60km 가상 지역에 이동정비반 6개와 30일간의 긴급 정비 요청을 만든다.\n",
|
|
"# 장비 구분(mobility/fires/comms/power)과 고장 유형은 교육용 추상 분류다.\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"AREA_KM, DUTY_H = 60, 10.0 # 가상 작전지역 크기, 정비반 1일 기준 가용시간\n",
|
|
"FAULT_H = {\"engine\": 6.0, \"drivetrain\": 5.0, \"hydraulic\": 3.0, \"electric\": 2.0, \"comm_module\": 1.5, \"cooling\": 2.5}\n",
|
|
"FIELD_P = {\"engine\": 0.3, \"drivetrain\": 0.35, \"hydraulic\": 0.75, \"electric\": 0.9, \"comm_module\": 0.95, \"cooling\": 0.8}\n",
|
|
"EQUIP_F = {\"mobility\": 1.0, \"fires\": 1.2, \"comms\": 0.7, \"power\": 0.8}\n",
|
|
"FAULT_OF = {\"mobility\": [0.3, 0.25, 0.2, 0.1, 0.0, 0.15], \"fires\": [0.2, 0.15, 0.35, 0.15, 0.0, 0.15],\n",
|
|
" \"comms\": [0.0, 0.0, 0.0, 0.4, 0.5, 0.1], \"power\": [0.35, 0.0, 0.0, 0.35, 0.0, 0.3]}\n",
|
|
"CRIT_P = {\"mobility\": [0.45, 0.35, 0.2], \"fires\": [0.2, 0.35, 0.45], \"comms\": [0.3, 0.4, 0.3], \"power\": [0.6, 0.3, 0.1]}\n",
|
|
"HUBS = [(15, 20), (45, 25), (30, 50)] # 가상 훈련장 3곳(요청이 몰리는 지점)\n",
|
|
"\n",
|
|
"\n",
|
|
"def make_teams(seed=0):\n",
|
|
" rng = np.random.default_rng(seed)\n",
|
|
" base = [(8, 10), (30, 6), (52, 12), (10, 44), (32, 38), (54, 50)]\n",
|
|
" skills = [[\"fires\", \"mobility\"], [\"comms\", \"mobility\"], [\"fires\", \"power\"],\n",
|
|
" [\"mobility\", \"power\"], [\"comms\", \"fires\"], [\"comms\", \"mobility\", \"power\"]]\n",
|
|
" return [{\"team\": f\"MT{i + 1}\", \"x\": float(x), \"y\": float(y), \"crew\": int(rng.integers(3, 6)), \"skills\": s}\n",
|
|
" for i, ((x, y), s) in enumerate(zip(base, skills))]\n",
|
|
"\n",
|
|
"\n",
|
|
"def make_requests(n_days=30, per_day=10, seed=1, surge_days=(8, 9, 10, 11, 22, 23, 24)):\n",
|
|
" \"\"\"하루 평균 per_day건, 훈련 주간(surge_days)에는 1.6배의 긴급 정비 요청을 만든다.\"\"\"\n",
|
|
" rng = np.random.default_rng(seed)\n",
|
|
" rows = []\n",
|
|
" for d in range(n_days):\n",
|
|
" n = rng.poisson(per_day * (1.6 if d in surge_days else 1.0))\n",
|
|
" weather = rng.choice([\"clear\", \"rain\", \"snow\"], p=[0.65, 0.25, 0.1])\n",
|
|
" for _ in range(n):\n",
|
|
" if rng.random() < 0.7:\n",
|
|
" hx, hy = HUBS[rng.integers(0, 3)]\n",
|
|
" x, y = np.clip(rng.normal(hx, 6), 0, AREA_KM), np.clip(rng.normal(hy, 6), 0, AREA_KM)\n",
|
|
" else:\n",
|
|
" x, y = rng.uniform(0, AREA_KM, 2)\n",
|
|
" eq = str(rng.choice(list(EQUIP_F), p=[0.45, 0.25, 0.2, 0.1]))\n",
|
|
" fault = str(rng.choice(list(FAULT_H), p=FAULT_OF[eq]))\n",
|
|
" parts = bool(rng.random() < 0.75)\n",
|
|
" bad = weather != \"clear\"\n",
|
|
" std_h = FAULT_H[fault] * EQUIP_F[eq] * (1.35 if not parts else 1.0) * (1.15 if bad else 1.0) * rng.lognormal(0, 0.12)\n",
|
|
" rows.append({\"t_req\": d * 24 + float(np.clip(rng.normal(13, 4), 5, 23)), \"day\": d,\n",
|
|
" \"x\": float(x), \"y\": float(y), \"equip\": eq, \"fault\": fault,\n",
|
|
" \"crit\": int(rng.choice([1, 2, 3], p=CRIT_P[eq])), \"parts_ready\": parts,\n",
|
|
" \"weather\": str(weather), \"road\": \"unpaved\" if y > 30 and rng.random() < 0.7 else \"paved\",\n",
|
|
" \"field_ok\": bool(rng.random() < FIELD_P[fault] * (1.0 if parts else 0.5)),\n",
|
|
" \"std_repair_h\": round(float(std_h), 2), \"road_delay\": float(rng.lognormal(0, 0.15))})\n",
|
|
" df = pd.DataFrame(rows).sort_values(\"t_req\").reset_index(drop=True)\n",
|
|
" df.insert(0, \"req_id\", [f\"R{i:04d}\" for i in range(len(df))])\n",
|
|
" return df\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"\n",
|
|
"# ---- 출동 배정·정비 모의 공통 함수 ----\n",
|
|
"W0 = {\"crit\": 1.0, \"wait\": 0.3, \"short\": 0.2, \"balance\": 0.5} # 우선순위·배정 기본 가중치(미션 9에서 재튜닝)\n",
|
|
"\n",
|
|
"\n",
|
|
"def travel_h(team, r, actual=False):\n",
|
|
" \"\"\"도로 굴곡 1.3배, 포장 35km/h·비포장 25km/h, 악천후 20% 감속. actual=True면 실제 도로 지연을 반영.\"\"\"\n",
|
|
" # [TODO : 학생 실습 3] 거리 기준 1차 이동정비반 배정 Baseline 알고리즘 구현 관련 핵심 로직을 구현하세요.\n",
|
|
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
|
|
" # <<< 여기에 코드를 작성하세요 >>>\n",
|
|
" pass # 작성 후 pass를 제거하세요.\n",
|
|
"def on_site_h(team, r):\n",
|
|
" \"\"\"현장 정비시간: 4인 기준 표준시간을 투입 인원으로 환산. 현장 수리 불가면 진단·후송 조치 1시간.\"\"\"\n",
|
|
" # [TODO : 학생 실습 3] 거리 기준 1차 이동정비반 배정 Baseline 알고리즘 구현 관련 핵심 로직을 구현하세요.\n",
|
|
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
|
|
" # <<< 여기에 코드를 작성하세요 >>>\n",
|
|
" pass # 작성 후 pass를 제거하세요.\n",
|
|
"def priority(r, now, est_h, w):\n",
|
|
" return w[\"crit\"] * r[\"crit\"] + w[\"wait\"] * (now - r[\"t_req\"]) + w[\"short\"] / (1 + est_h)\n",
|
|
"\n",
|
|
"\n",
|
|
"def simulate(req, teams, policy=\"baseline\", w=None, cycle_h=1.0, est=None):\n",
|
|
" \"\"\"배정 주기(cycle_h)마다 대기 요청을 정렬해 이동정비반에 배정하고 출동·정비·복귀를 모의한다.\n",
|
|
" baseline: 접수 순서 + 거리 최근접 정비반 / priority: 우선순위 점수 순서 + 도착 예상시각·당일 부하 균형\"\"\"\n",
|
|
" w = w or W0\n",
|
|
" est = np.asarray(est if est is not None else req[\"std_repair_h\"].values, float)\n",
|
|
" n_days = int(req[\"day\"].max()) + 1\n",
|
|
" free = {t[\"team\"]: 0.0 for t in teams}\n",
|
|
" load = {t[\"team\"]: np.zeros(n_days + 5) for t in teams}\n",
|
|
" recs, out, pending, i, now = req.to_dict(\"records\"), [], [], 0, cycle_h\n",
|
|
" while i < len(recs) or pending:\n",
|
|
" while i < len(recs) and recs[i][\"t_req\"] <= now:\n",
|
|
" pending.append((i, recs[i])); i += 1\n",
|
|
" if policy == \"priority\":\n",
|
|
" pending.sort(key=lambda p: -priority(p[1], now, est[p[0]], w))\n",
|
|
" for k, r in pending:\n",
|
|
" cand = [t for t in teams if r[\"equip\"] in t[\"skills\"]]\n",
|
|
" day = min(int(now // 24), n_days + 4)\n",
|
|
" if policy == \"baseline\":\n",
|
|
" tm = min(cand, key=lambda t: np.hypot(t[\"x\"] - r[\"x\"], t[\"y\"] - r[\"y\"]))\n",
|
|
" else:\n",
|
|
" tm = min(cand, key=lambda t: max(free[t[\"team\"]], now) + travel_h(t, r)\n",
|
|
" + w[\"balance\"] * load[t[\"team\"]][day] / DUTY_H * 4)\n",
|
|
" backlog = max(0.0, free[tm[\"team\"]] - now)\n",
|
|
" go = max(free[tm[\"team\"]], now)\n",
|
|
" tr = travel_h(tm, r, actual=True)\n",
|
|
" work = on_site_h(tm, r)\n",
|
|
" free[tm[\"team\"]] = go + 2 * tr + work\n",
|
|
" load[tm[\"team\"]][min(int(go // 24), n_days + 4)] += 2 * tr + work\n",
|
|
" out.append({**r, \"idx\": k, \"team\": tm[\"team\"], \"dist_km\": float(np.hypot(tm[\"x\"] - r[\"x\"], tm[\"y\"] - r[\"y\"])),\n",
|
|
" \"backlog_h\": backlog, \"cycle_wait_h\": now - r[\"t_req\"], \"dispatch\": go, \"travel_h\": tr,\n",
|
|
" \"work_h\": work, \"response_h\": go + tr - r[\"t_req\"], \"downtime_h\": go + tr + work - r[\"t_req\"]})\n",
|
|
" pending, now = [], now + cycle_h\n",
|
|
" L = pd.DataFrame({k: v[:n_days] for k, v in load.items()}) / DUTY_H # 정비반별 일일 부하율\n",
|
|
" return pd.DataFrame(out).sort_values(\"idx\").reset_index(drop=True), L\n",
|
|
"\n",
|
|
"\n",
|
|
"def metrics(res, L):\n",
|
|
" c3 = res[res[\"crit\"] == 3]\n",
|
|
" return {\"평균 도착(h)\": round(res[\"response_h\"].mean(), 2), \"핵심장비 도착(h)\": round(c3[\"response_h\"].mean(), 2),\n",
|
|
" \"핵심장비 90%(h)\": round(c3[\"response_h\"].quantile(0.9), 2), \"평균 부하율\": round(float(L.values.mean()), 3),\n",
|
|
" \"부하율 편차\": round(float(L.mean().std()), 3), \"과부하 일수\": int((L > 1.0).values.sum()),\n",
|
|
" \"총 이동(h)\": round(float(res[\"travel_h\"].sum() * 2), 1)}\n",
|
|
"\n",
|
|
"\n",
|
|
"teams, req = make_teams(), make_requests()\n",
|
|
"res, L = simulate(req, teams, \"baseline\")\n",
|
|
"m = metrics(res, L)\n",
|
|
"print(f\"[Baseline 결과] 요청 {len(res)}건\")\n",
|
|
"for k, v in m.items():\n",
|
|
" print(f\" {k:10s} {v}\")\n",
|
|
"print(\"정비반별 평균 부하율:\", {k: round(float(v), 2) for k, v in L.mean().items()})\n",
|
|
"print(f\"-> 가까운 정비반만 고르면 요청이 몰리는 지역의 정비반(MT{int(L.mean().argmax()) + 1})에 부하가 집중된다.\")\n",
|
|
"\n",
|
|
"# 1) 상관분석: 출동 지연(접수~도착 시간)과 관련된 피처 (피어슨 상관, 유의수준 0.05)\n",
|
|
"res[\"weather_bad\"] = (res[\"weather\"] != \"clear\").astype(int)\n",
|
|
"res[\"unpaved\"] = (res[\"road\"] == \"unpaved\").astype(int)\n",
|
|
"res[\"hour\"] = res[\"t_req\"] % 24\n",
|
|
"res[\"day_requests\"] = res.groupby(\"day\")[\"req_id\"].transform(\"count\")\n",
|
|
"res[\"team_load\"] = [L.loc[d, t] for d, t in zip(res[\"day\"], res[\"team\"])]\n",
|
|
"feats = [\"dist_km\", \"backlog_h\", \"team_load\", \"day_requests\", \"weather_bad\", \"unpaved\", \"std_repair_h\", \"crit\", \"hour\"]\n",
|
|
"corr = []\n",
|
|
"for f in feats:\n",
|
|
" r, p = stats.pearsonr(res[f].astype(float), res[\"response_h\"])\n",
|
|
" corr.append({\"피처\": f, \"상관계수\": round(r, 3), \"p-value\": f\"{p:.1e}\", \"유의\": p < 0.05})\n",
|
|
"corr = pd.DataFrame(corr).sort_values(\"상관계수\", key=abs, ascending=False)\n",
|
|
"print(\"\\n[출동 지연과의 상관분석]\")\n",
|
|
"print(corr.to_string(index=False))\n",
|
|
"n_sig = int(corr[\"유의\"].sum())\n",
|
|
"print(f\"유의미한 피처 {n_sig}개 (목표 5개 이상: {'달성' if n_sig >= 5 else '미달'})\")\n",
|
|
"print(\"-> 작전 중요도(crit)와 도착시간의 상관이 0에 가깝다: Baseline은 핵심 장비를 먼저 챙기지 못한다(미션 4의 동기).\")\n",
|
|
"\n",
|
|
"# 2) 파레토(ABC) 분석: 장비 가동중단 시간을 가장 많이 만드는 (장비, 고장유형) 조합\n",
|
|
"g = res.groupby([\"equip\", \"fault\"])[\"downtime_h\"].sum().sort_values(ascending=False)\n",
|
|
"cum = g.cumsum() / g.sum()\n",
|
|
"abc = pd.cut(cum, [0, 0.8, 0.95, 1.0], labels=[\"A\", \"B\", \"C\"])\n",
|
|
"print(\"\\n[파레토 A등급(누적 가동중단 80%)]\", \", \".join(f\"{e}/{f}\" for (e, f), c in abc.items() if c == \"A\"))\n",
|
|
"print(f\"A등급은 전체 조합 {len(g)}개 중 {int((abc == 'A').sum())}개({(abc == 'A').mean():.0%})\")\n",
|
|
"\n",
|
|
"# 3) EDA 차트 8종\n",
|
|
"fig, ax = plt.subplots(2, 4, figsize=(18, 8))\n",
|
|
"a = ax.ravel()\n",
|
|
"a[0].scatter(req[\"x\"], req[\"y\"], s=8, c=req[\"crit\"], cmap=\"Reds\"); a[0].scatter([t[\"x\"] for t in teams], [t[\"y\"] for t in teams], marker=\"^\", s=120, c=\"blue\")\n",
|
|
"a[0].set_title(\"1. Requests (red) & team bases (blue)\")\n",
|
|
"req.groupby(\"day\").size().plot(ax=a[1], marker=\"o\"); a[1].set_title(\"2. Requests per day (surge = exercise)\")\n",
|
|
"a[2].hist(res[\"response_h\"], bins=30); a[2].set_title(\"3. Response time (h)\")\n",
|
|
"a[3].boxplot([L[c] for c in L.columns]); a[3].set_xticks(range(1, 7), L.columns); a[3].axhline(1.0, ls=\"--\", c=\"r\"); a[3].set_title(\"4. Daily load rate by team\")\n",
|
|
"res[\"fault\"].value_counts().plot.bar(ax=a[4]); a[4].set_title(\"5. Requests by fault type\")\n",
|
|
"res.boxplot(column=\"std_repair_h\", by=\"fault\", ax=a[5], rot=45); a[5].set_title(\"6. Std repair hours by fault\")\n",
|
|
"a[6].barh(corr[\"피처\"], corr[\"상관계수\"]); a[6].set_title(\"7. Correlation with response time\")\n",
|
|
"a[7].plot(range(1, len(cum) + 1), cum.values, \"o-\"); a[7].axhline(0.8, ls=\"--\", c=\"gray\"); a[7].set_title(\"8. Pareto of downtime (equip x fault)\")\n",
|
|
"plt.suptitle(\"MNT10 EDA - baseline dispatch (synthetic data)\"); plt.tight_layout(); plt.show()\n",
|
|
"print(\"\\nEDA 차트 8종 생성 (목표 8종 이상: 달성)\")\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"<details>\n",
|
|
"<summary><b>💡 [참고] 미션 03 예상 실행 결과 (Output)</b></summary>\n",
|
|
"\n",
|
|
"```text\n",
|
|
"[Baseline 결과] 요청 356건\n",
|
|
" 평균 도착(h) 3.37\n",
|
|
" 핵심장비 도착(h) 3.13\n",
|
|
" 핵심장비 90%(h) 7.54\n",
|
|
" 평균 부하율 0.78\n",
|
|
" 부하율 편차 0.365\n",
|
|
" 과부하 일수 58\n",
|
|
" 총 이동(h) 504.7\n",
|
|
"정비반별 평균 부하율: {'MT1': 0.39, 'MT2': 0.94, 'MT3': 0.39, 'MT4': 0.75, 'MT5': 1.36, 'MT6': 0.85}\n",
|
|
"-> 가까운 정비반만 고르면 요청이 몰리는 지역의 정비반(MT5)에 부하가 집중된다.\n",
|
|
"\n",
|
|
"[출동 지연과의 상관분석]\n",
|
|
" 피처 상관계수 p-value 유의\n",
|
|
" backlog_h 0.994 0.0e+00 True\n",
|
|
" team_load 0.454 1.5e-19 True\n",
|
|
"day_requests 0.246 2.6e-06 True\n",
|
|
" unpaved 0.216 3.9e-05 True\n",
|
|
" hour 0.158 2.8e-03 True\n",
|
|
" dist_km 0.087 1.0e-01 False\n",
|
|
"std_repair_h -0.080 1.3e-01 False\n",
|
|
" crit -0.027 6.2e-01 False\n",
|
|
" weather_bad -0.004 9.4e-01 False\n",
|
|
"유의미한 피처 5개 (목표 5개 이상: 달성)\n",
|
|
"-> 작전 중요도(crit)와 도착시간의 상관이 0에 가깝다: Baseline은 핵심 장비를 먼저 챙기지 못한다(미션 4의 동기).\n",
|
|
"\n",
|
|
"[파레토 A등급(누적 가동중단 80%)] mobility/drivetrain, fires/hydraulic, comms/comm_module, mobility/engine, mobility/hydraulic, comms/electric, mobility/cooling, fires/drivetrain, fires/engine\n",
|
|
"A등급은 전체 조합 16개 중 9개(56%)\n",
|
|
"\n",
|
|
"EDA 차트 8종 생성 (목표 8종 이상: 달성)\n",
|
|
"```\n",
|
|
"</details>\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"---\n",
|
|
"\n",
|
|
"## 🎯 미션 04. 장비 작전 중요도, 정비 소요 시간, 이동 경로를 종합한 최적 출동 우선순위 추천 엔진 고도화\n",
|
|
"**소요시간**: 약 3.0시간\n",
|
|
"\n",
|
|
"### 📋 미션 배경 및 수행 지침\n",
|
|
"#### 미션 배경 및 당위성\n",
|
|
"\n",
|
|
"작전 중요도와 이동 경로를 종합한 최적 출동 우선순위 추천 엔진을 고도화한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **장비 작전 중요도, 예측 정비 소요시간, 이동 경로(도착 예상시각), 정비반 부하를 종합한 출동 우선순위 추천 엔진을 만든다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
|
|
"\n",
|
|
"이 문제를 해결하기 위해 본 미션에서는 **접수 시점에 알 수 있는 정보만으로 파생 피처 10개 이상을 만들고 정비 소요시간 예측 모델을 시계열 교차검증(MAPE)으로 평가하고 성능을 비교하는** 실습을 체계적으로 수행합니다.\n",
|
|
"\n",
|
|
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
|
|
"\n",
|
|
"**이번 결과물:** `src/features/mnt10_feature_engineering.py 및 기준 모델` 형태의 결과물을 완성한다.\n",
|
|
"\n",
|
|
"### 🔍 평가 및 검증 기준 (PASS 조건)\n",
|
|
"#### 미션 판정\n",
|
|
"\n",
|
|
"**평가 구분:** 필수\n",
|
|
"\n",
|
|
"**핵심 평가영역:** 군수 실무에서 납득 가능한 피처 설계의 논리성\n",
|
|
"\n",
|
|
"#### 미션 PASS 조건\n",
|
|
"\n",
|
|
"- 정량 기준: 파생 피처 10개 이상 생성, 기준 모델 MAPE 25% 이하 달성\n",
|
|
"\n",
|
|
"\n",
|
|
"- 진행 기준: 피처 파이프라인 및 베이스라인 모델 성능 지표 도출\n",
|
|
"\n",
|
|
"\n",
|
|
"- 검증 방법: 교차 검증(Time-series Cross Validation)을 통한 과적합 방지\n",
|
|
"\n",
|
|
"\n",
|
|
"#### 채점자 판정 메모\n",
|
|
"\n",
|
|
"> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
|
|
"\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"\"\"\"장비 작전 중요도·예측 정비 소요시간·이동 경로(도착 예상시각)·정비반 부하를 종합한 출동 우선순위 추천 엔진을 만든다. 파생 피처 10개 이상으로 정비 소요시간 예측 모델을 시계열 교차검증(MAPE)으로 평가하고, 거리 기준 Baseline 대비 핵심 장비 도착시간과 부하 균형 개선을 확인한다.\"\"\"\n",
|
|
"import numpy as np\n",
|
|
"import pandas as pd\n",
|
|
"import matplotlib.pyplot as plt\n",
|
|
"from sklearn.ensemble import GradientBoostingRegressor\n",
|
|
"from sklearn.model_selection import TimeSeriesSplit\n",
|
|
"\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"# [교육용가상자료] 야전 이동정비반·긴급 정비 요청 생성기 (실제 부대·위치·장비·인원 정보 없음)\n",
|
|
"# 60km x 60km 가상 지역에 이동정비반 6개와 30일간의 긴급 정비 요청을 만든다.\n",
|
|
"# 장비 구분(mobility/fires/comms/power)과 고장 유형은 교육용 추상 분류다.\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"AREA_KM, DUTY_H = 60, 10.0 # 가상 작전지역 크기, 정비반 1일 기준 가용시간\n",
|
|
"FAULT_H = {\"engine\": 6.0, \"drivetrain\": 5.0, \"hydraulic\": 3.0, \"electric\": 2.0, \"comm_module\": 1.5, \"cooling\": 2.5}\n",
|
|
"FIELD_P = {\"engine\": 0.3, \"drivetrain\": 0.35, \"hydraulic\": 0.75, \"electric\": 0.9, \"comm_module\": 0.95, \"cooling\": 0.8}\n",
|
|
"EQUIP_F = {\"mobility\": 1.0, \"fires\": 1.2, \"comms\": 0.7, \"power\": 0.8}\n",
|
|
"FAULT_OF = {\"mobility\": [0.3, 0.25, 0.2, 0.1, 0.0, 0.15], \"fires\": [0.2, 0.15, 0.35, 0.15, 0.0, 0.15],\n",
|
|
" \"comms\": [0.0, 0.0, 0.0, 0.4, 0.5, 0.1], \"power\": [0.35, 0.0, 0.0, 0.35, 0.0, 0.3]}\n",
|
|
"CRIT_P = {\"mobility\": [0.45, 0.35, 0.2], \"fires\": [0.2, 0.35, 0.45], \"comms\": [0.3, 0.4, 0.3], \"power\": [0.6, 0.3, 0.1]}\n",
|
|
"HUBS = [(15, 20), (45, 25), (30, 50)] # 가상 훈련장 3곳(요청이 몰리는 지점)\n",
|
|
"\n",
|
|
"\n",
|
|
"def make_teams(seed=0):\n",
|
|
" rng = np.random.default_rng(seed)\n",
|
|
" base = [(8, 10), (30, 6), (52, 12), (10, 44), (32, 38), (54, 50)]\n",
|
|
" skills = [[\"fires\", \"mobility\"], [\"comms\", \"mobility\"], [\"fires\", \"power\"],\n",
|
|
" [\"mobility\", \"power\"], [\"comms\", \"fires\"], [\"comms\", \"mobility\", \"power\"]]\n",
|
|
" return [{\"team\": f\"MT{i + 1}\", \"x\": float(x), \"y\": float(y), \"crew\": int(rng.integers(3, 6)), \"skills\": s}\n",
|
|
" for i, ((x, y), s) in enumerate(zip(base, skills))]\n",
|
|
"\n",
|
|
"\n",
|
|
"def make_requests(n_days=30, per_day=10, seed=1, surge_days=(8, 9, 10, 11, 22, 23, 24)):\n",
|
|
" \"\"\"하루 평균 per_day건, 훈련 주간(surge_days)에는 1.6배의 긴급 정비 요청을 만든다.\"\"\"\n",
|
|
" rng = np.random.default_rng(seed)\n",
|
|
" rows = []\n",
|
|
" for d in range(n_days):\n",
|
|
" n = rng.poisson(per_day * (1.6 if d in surge_days else 1.0))\n",
|
|
" weather = rng.choice([\"clear\", \"rain\", \"snow\"], p=[0.65, 0.25, 0.1])\n",
|
|
" for _ in range(n):\n",
|
|
" if rng.random() < 0.7:\n",
|
|
" hx, hy = HUBS[rng.integers(0, 3)]\n",
|
|
" x, y = np.clip(rng.normal(hx, 6), 0, AREA_KM), np.clip(rng.normal(hy, 6), 0, AREA_KM)\n",
|
|
" else:\n",
|
|
" x, y = rng.uniform(0, AREA_KM, 2)\n",
|
|
" eq = str(rng.choice(list(EQUIP_F), p=[0.45, 0.25, 0.2, 0.1]))\n",
|
|
" fault = str(rng.choice(list(FAULT_H), p=FAULT_OF[eq]))\n",
|
|
" parts = bool(rng.random() < 0.75)\n",
|
|
" bad = weather != \"clear\"\n",
|
|
" std_h = FAULT_H[fault] * EQUIP_F[eq] * (1.35 if not parts else 1.0) * (1.15 if bad else 1.0) * rng.lognormal(0, 0.12)\n",
|
|
" rows.append({\"t_req\": d * 24 + float(np.clip(rng.normal(13, 4), 5, 23)), \"day\": d,\n",
|
|
" \"x\": float(x), \"y\": float(y), \"equip\": eq, \"fault\": fault,\n",
|
|
" \"crit\": int(rng.choice([1, 2, 3], p=CRIT_P[eq])), \"parts_ready\": parts,\n",
|
|
" \"weather\": str(weather), \"road\": \"unpaved\" if y > 30 and rng.random() < 0.7 else \"paved\",\n",
|
|
" \"field_ok\": bool(rng.random() < FIELD_P[fault] * (1.0 if parts else 0.5)),\n",
|
|
" \"std_repair_h\": round(float(std_h), 2), \"road_delay\": float(rng.lognormal(0, 0.15))})\n",
|
|
" df = pd.DataFrame(rows).sort_values(\"t_req\").reset_index(drop=True)\n",
|
|
" df.insert(0, \"req_id\", [f\"R{i:04d}\" for i in range(len(df))])\n",
|
|
" return df\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"\n",
|
|
"# ---- 출동 배정·정비 모의 공통 함수 ----\n",
|
|
"W0 = {\"crit\": 1.0, \"wait\": 0.3, \"short\": 0.2, \"balance\": 0.5} # 우선순위·배정 기본 가중치(미션 9에서 재튜닝)\n",
|
|
"\n",
|
|
"\n",
|
|
"def travel_h(team, r, actual=False):\n",
|
|
" \"\"\"도로 굴곡 1.3배, 포장 35km/h·비포장 25km/h, 악천후 20% 감속. actual=True면 실제 도로 지연을 반영.\"\"\"\n",
|
|
" # [TODO : 학생 실습 4] 장비 작전 중요도, 정비 소요 시간, 이동 경로를 종합한 최적 출동 우선순위 추천 엔진 고도화 관련 핵심 로직을 구현하세요.\n",
|
|
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
|
|
" # <<< 여기에 코드를 작성하세요 >>>\n",
|
|
" pass # 작성 후 pass를 제거하세요.\n",
|
|
"def on_site_h(team, r):\n",
|
|
" \"\"\"현장 정비시간: 4인 기준 표준시간을 투입 인원으로 환산. 현장 수리 불가면 진단·후송 조치 1시간.\"\"\"\n",
|
|
" # [TODO : 학생 실습 4] 장비 작전 중요도, 정비 소요 시간, 이동 경로를 종합한 최적 출동 우선순위 추천 엔진 고도화 관련 핵심 로직을 구현하세요.\n",
|
|
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
|
|
" # <<< 여기에 코드를 작성하세요 >>>\n",
|
|
" pass # 작성 후 pass를 제거하세요.\n",
|
|
"def priority(r, now, est_h, w):\n",
|
|
" return w[\"crit\"] * r[\"crit\"] + w[\"wait\"] * (now - r[\"t_req\"]) + w[\"short\"] / (1 + est_h)\n",
|
|
"\n",
|
|
"\n",
|
|
"def simulate(req, teams, policy=\"baseline\", w=None, cycle_h=1.0, est=None):\n",
|
|
" \"\"\"배정 주기(cycle_h)마다 대기 요청을 정렬해 이동정비반에 배정하고 출동·정비·복귀를 모의한다.\n",
|
|
" baseline: 접수 순서 + 거리 최근접 정비반 / priority: 우선순위 점수 순서 + 도착 예상시각·당일 부하 균형\"\"\"\n",
|
|
" w = w or W0\n",
|
|
" est = np.asarray(est if est is not None else req[\"std_repair_h\"].values, float)\n",
|
|
" n_days = int(req[\"day\"].max()) + 1\n",
|
|
" free = {t[\"team\"]: 0.0 for t in teams}\n",
|
|
" load = {t[\"team\"]: np.zeros(n_days + 5) for t in teams}\n",
|
|
" recs, out, pending, i, now = req.to_dict(\"records\"), [], [], 0, cycle_h\n",
|
|
" while i < len(recs) or pending:\n",
|
|
" while i < len(recs) and recs[i][\"t_req\"] <= now:\n",
|
|
" pending.append((i, recs[i])); i += 1\n",
|
|
" if policy == \"priority\":\n",
|
|
" pending.sort(key=lambda p: -priority(p[1], now, est[p[0]], w))\n",
|
|
" for k, r in pending:\n",
|
|
" cand = [t for t in teams if r[\"equip\"] in t[\"skills\"]]\n",
|
|
" day = min(int(now // 24), n_days + 4)\n",
|
|
" if policy == \"baseline\":\n",
|
|
" tm = min(cand, key=lambda t: np.hypot(t[\"x\"] - r[\"x\"], t[\"y\"] - r[\"y\"]))\n",
|
|
" else:\n",
|
|
" tm = min(cand, key=lambda t: max(free[t[\"team\"]], now) + travel_h(t, r)\n",
|
|
" + w[\"balance\"] * load[t[\"team\"]][day] / DUTY_H * 4)\n",
|
|
" backlog = max(0.0, free[tm[\"team\"]] - now)\n",
|
|
" go = max(free[tm[\"team\"]], now)\n",
|
|
" tr = travel_h(tm, r, actual=True)\n",
|
|
" work = on_site_h(tm, r)\n",
|
|
" free[tm[\"team\"]] = go + 2 * tr + work\n",
|
|
" load[tm[\"team\"]][min(int(go // 24), n_days + 4)] += 2 * tr + work\n",
|
|
" out.append({**r, \"idx\": k, \"team\": tm[\"team\"], \"dist_km\": float(np.hypot(tm[\"x\"] - r[\"x\"], tm[\"y\"] - r[\"y\"])),\n",
|
|
" \"backlog_h\": backlog, \"cycle_wait_h\": now - r[\"t_req\"], \"dispatch\": go, \"travel_h\": tr,\n",
|
|
" \"work_h\": work, \"response_h\": go + tr - r[\"t_req\"], \"downtime_h\": go + tr + work - r[\"t_req\"]})\n",
|
|
" pending, now = [], now + cycle_h\n",
|
|
" L = pd.DataFrame({k: v[:n_days] for k, v in load.items()}) / DUTY_H # 정비반별 일일 부하율\n",
|
|
" return pd.DataFrame(out).sort_values(\"idx\").reset_index(drop=True), L\n",
|
|
"\n",
|
|
"\n",
|
|
"def metrics(res, L):\n",
|
|
" c3 = res[res[\"crit\"] == 3]\n",
|
|
" return {\"평균 도착(h)\": round(res[\"response_h\"].mean(), 2), \"핵심장비 도착(h)\": round(c3[\"response_h\"].mean(), 2),\n",
|
|
" \"핵심장비 90%(h)\": round(c3[\"response_h\"].quantile(0.9), 2), \"평균 부하율\": round(float(L.values.mean()), 3),\n",
|
|
" \"부하율 편차\": round(float(L.mean().std()), 3), \"과부하 일수\": int((L > 1.0).values.sum()),\n",
|
|
" \"총 이동(h)\": round(float(res[\"travel_h\"].sum() * 2), 1)}\n",
|
|
"\n",
|
|
"\n",
|
|
"def make_features(req, teams):\n",
|
|
" \"\"\"접수 시점에 알 수 있는 정보만으로 파생 피처를 만든다(미래 정보 누설 방지).\"\"\"\n",
|
|
" X = pd.DataFrame(index=req.index)\n",
|
|
" X[\"hour\"] = req[\"t_req\"] % 24\n",
|
|
" X[\"is_night\"] = (X[\"hour\"] >= 19).astype(int)\n",
|
|
" X[\"dow\"] = req[\"day\"] % 7\n",
|
|
" X[\"weather_bad\"] = (req[\"weather\"] != \"clear\").astype(int)\n",
|
|
" X[\"unpaved\"] = (req[\"road\"] == \"unpaved\").astype(int)\n",
|
|
" X[\"no_parts\"] = (~req[\"parts_ready\"]).astype(int)\n",
|
|
" X[\"std_table_h\"] = req[\"fault\"].map(FAULT_H) * req[\"equip\"].map(EQUIP_F) # 정비 표준시간표 값\n",
|
|
" X[\"crit_x_noparts\"] = req[\"crit\"] * X[\"no_parts\"]\n",
|
|
" X[\"bad_x_table\"] = X[\"weather_bad\"] * X[\"std_table_h\"]\n",
|
|
" X[\"near_team_km\"] = [min(np.hypot(t[\"x\"] - x, t[\"y\"] - y) for t in teams if e in t[\"skills\"]) for x, y, e in zip(req[\"x\"], req[\"y\"], req[\"equip\"])]\n",
|
|
" X[\"hub_km\"] = [min(np.hypot(hx - x, hy - y) for hx, hy in HUBS) for x, y in zip(req[\"x\"], req[\"y\"])]\n",
|
|
" X[\"req_24h\"] = [((req[\"t_req\"] > t - 24) & (req[\"t_req\"] <= t)).sum() for t in req[\"t_req\"]]\n",
|
|
" X[\"fault_past_mean\"] = req.groupby(\"fault\")[\"std_repair_h\"].transform(lambda s: s.shift().expanding().mean()).fillna(req[\"std_repair_h\"].mean())\n",
|
|
" X[\"crit\"] = req[\"crit\"]\n",
|
|
" for e in EQUIP_F:\n",
|
|
" X[f\"eq_{e}\"] = (req[\"equip\"] == e).astype(int)\n",
|
|
" return X\n",
|
|
"\n",
|
|
"\n",
|
|
"teams = make_teams()\n",
|
|
"hist = make_requests(seed=11) # 과거 30일(학습용)\n",
|
|
"req = make_requests(seed=1) # 평가 30일\n",
|
|
"X, y = make_features(hist, teams), hist[\"std_repair_h\"]\n",
|
|
"print(f\"파생 피처 {X.shape[1]}개 생성 (목표 10개 이상: {'달성' if X.shape[1] >= 10 else '미달'})\")\n",
|
|
"\n",
|
|
"# 1) 시계열 교차검증: 과거로 학습 -> 이후 기간 예측 (5분할)\n",
|
|
"mape = lambda a, p: float(np.mean(np.abs(a - p) / a))\n",
|
|
"rows = []\n",
|
|
"for i, (tr, te) in enumerate(TimeSeriesSplit(5).split(X), 1):\n",
|
|
" base = hist.iloc[tr].groupby(\"fault\")[\"std_repair_h\"].mean()\n",
|
|
" p0 = hist.iloc[te][\"fault\"].map(base).fillna(y.iloc[tr].mean())\n",
|
|
" gb = GradientBoostingRegressor(n_estimators=200, max_depth=3, learning_rate=0.05, random_state=0).fit(X.iloc[tr], y.iloc[tr])\n",
|
|
" rows.append({\"분할\": i, \"학습\": len(tr), \"검증\": len(te), \"기준모델 MAPE\": round(mape(y.iloc[te], p0), 3), \"GBM MAPE\": round(mape(y.iloc[te], gb.predict(X.iloc[te])), 3)})\n",
|
|
"cv = pd.DataFrame(rows)\n",
|
|
"print(cv.to_string(index=False))\n",
|
|
"print(f\"평균 MAPE: 기준모델(고장유형 평균) {cv['기준모델 MAPE'].mean():.1%}, GBM {cv['GBM MAPE'].mean():.1%} (목표 25% 이하: {'달성' if cv['GBM MAPE'].mean() <= 0.25 else '미달'})\")\n",
|
|
"\n",
|
|
"model = GradientBoostingRegressor(n_estimators=200, max_depth=3, learning_rate=0.05, random_state=0).fit(X, y)\n",
|
|
"imp = pd.Series(model.feature_importances_, X.columns).sort_values(ascending=False)\n",
|
|
"print(\"\\n피처 중요도 상위 5:\", \", \".join(f\"{k}({v:.2f})\" for k, v in imp.head(5).items()))\n",
|
|
"print(\"중요도 0.01 미만 피처(차원 축소 후보):\", list(imp[imp < 0.01].index))\n",
|
|
"\n",
|
|
"# 2) 우선순위 추천 엔진 적용: 예측 정비시간 + 중요도 + 대기시간 + 도착 예상시각 + 당일 부하\n",
|
|
"est = model.predict(make_features(req, teams))\n",
|
|
"print(f\"평가 기간 정비시간 예측 MAPE {mape(req['std_repair_h'], est):.1%}\")\n",
|
|
"b_res, b_L = simulate(req, teams, \"baseline\")\n",
|
|
"p_res, p_L = simulate(req, teams, \"priority\", est=est)\n",
|
|
"cmp = pd.DataFrame([{\"방식\": \"Baseline(거리)\", **metrics(b_res, b_L)}, {\"방식\": \"추천 엔진\", **metrics(p_res, p_L)}])\n",
|
|
"print(\"\\n\" + cmp.to_string(index=False))\n",
|
|
"gain = 1 - cmp.loc[1, \"핵심장비 도착(h)\"] / cmp.loc[0, \"핵심장비 도착(h)\"]\n",
|
|
"print(f\"핵심 장비 평균 도착시간 {gain:.1%} 단축, 정비반 부하율 편차 {cmp.loc[0, '부하율 편차']} -> {cmp.loc[1, '부하율 편차']}\")\n",
|
|
"print(\"대가: 부하를 나누느라 먼 정비반이 출동해 총 이동시간이 늘었다(미션 9에서 가중치로 조정).\")\n",
|
|
"\n",
|
|
"# 3) 추천 예시: 요청이 몰린 시점의 대기 목록\n",
|
|
"t_busy = req.groupby(\"day\").size().idxmax() * 24 + 14\n",
|
|
"q = req[(req[\"t_req\"] <= t_busy) & (req[\"t_req\"] > t_busy - 3)].copy()\n",
|
|
"q[\"예측정비(h)\"] = est[q.index].round(1)\n",
|
|
"q[\"우선순위 점수\"] = [round(priority(r, t_busy, e, W0), 2) for r, e in zip(q.to_dict(\"records\"), est[q.index])]\n",
|
|
"print(f\"\\n[추천 예시] {int(t_busy // 24)}일차 14시 대기 요청\")\n",
|
|
"print(q.sort_values(\"우선순위 점수\", ascending=False)[[\"req_id\", \"equip\", \"fault\", \"crit\", \"예측정비(h)\", \"우선순위 점수\"]].to_string(index=False))\n",
|
|
"\n",
|
|
"fig, ax = plt.subplots(1, 2, figsize=(11, 4))\n",
|
|
"imp.head(10)[::-1].plot.barh(ax=ax[0]); ax[0].set_title(\"Feature importance (repair time model)\")\n",
|
|
"ax[1].boxplot([b_res[b_res.crit == 3][\"response_h\"], p_res[p_res.crit == 3][\"response_h\"]]); ax[1].set_xticks([1, 2], [\"baseline\", \"priority engine\"])\n",
|
|
"ax[1].set_ylabel(\"response time (h)\"); ax[1].set_title(\"Critical equipment response time\")\n",
|
|
"plt.tight_layout(); plt.show()\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"<details>\n",
|
|
"<summary><b>💡 [참고] 미션 04 예상 실행 결과 (Output)</b></summary>\n",
|
|
"\n",
|
|
"```text\n",
|
|
"파생 피처 18개 생성 (목표 10개 이상: 달성)\n",
|
|
" 분할 학습 검증 기준모델 MAPE GBM MAPE\n",
|
|
" 1 60 56 0.203 0.142\n",
|
|
" 2 116 56 0.171 0.141\n",
|
|
" 3 172 56 0.187 0.107\n",
|
|
" 4 228 56 0.263 0.130\n",
|
|
" 5 284 56 0.203 0.104\n",
|
|
"평균 MAPE: 기준모델(고장유형 평균) 20.5%, GBM 12.5% (목표 25% 이하: 달성)\n",
|
|
"\n",
|
|
"피처 중요도 상위 5: std_table_h(0.78), fault_past_mean(0.07), crit_x_noparts(0.05), bad_x_table(0.04), no_parts(0.03)\n",
|
|
"중요도 0.01 미만 피처(차원 축소 후보): ['eq_fires', 'hour', 'hub_km', 'near_team_km', 'crit', 'req_24h', 'dow', 'weather_bad', 'unpaved', 'eq_comms', 'eq_mobility', 'is_night', 'eq_power']\n",
|
|
"평가 기간 정비시간 예측 MAPE 12.1%\n",
|
|
"\n",
|
|
" 방식 평균 도착(h) 핵심장비 도착(h) 핵심장비 90%(h) 평균 부하율 부하율 편차 과부하 일수 총 이동(h)\n",
|
|
"Baseline(거리) 3.37 3.13 7.54 0.780 0.365 58 504.7\n",
|
|
" 추천 엔진 2.09 1.92 3.25 0.903 0.143 68 744.9\n",
|
|
"핵심 장비 평균 도착시간 38.7% 단축, 정비반 부하율 편차 0.365 -> 0.143\n",
|
|
"대가: 부하를 나누느라 먼 정비반이 출동해 총 이동시간이 늘었다(미션 9에서 가중치로 조정).\n",
|
|
"\n",
|
|
"[추천 예시] 22일차 14시 대기 요청\n",
|
|
"req_id equip fault crit 예측정비(h) 우선순위 점수\n",
|
|
" R0255 fires engine 3 10.8 3.51\n",
|
|
" R0256 fires electric 3 2.4 3.47\n",
|
|
" R0257 mobility hydraulic 3 3.1 3.43\n",
|
|
" R0252 comms electric 2 1.4 2.98\n",
|
|
" R0254 mobility electric 2 2.0 2.72\n",
|
|
" R0261 comms electric 2 1.4 2.37\n",
|
|
" R0260 mobility hydraulic 2 3.3 2.34\n",
|
|
" R0262 power cooling 2 2.1 2.31\n",
|
|
" R0253 mobility drivetrain 1 5.2 1.80\n",
|
|
" R0258 fires cooling 1 3.2 1.40\n",
|
|
" R0259 comms cooling 1 1.8 1.37\n",
|
|
"```\n",
|
|
"</details>\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"---\n",
|
|
"\n",
|
|
"## 🎯 미션 05. 출동 지연 위험도 및 현장 수리 가능성 신뢰도 점수 산출 모듈 개발\n",
|
|
"**소요시간**: 약 3.5시간\n",
|
|
"\n",
|
|
"### 📋 미션 배경 및 수행 지침\n",
|
|
"#### 미션 배경 및 당위성\n",
|
|
"\n",
|
|
"출동 지연 위험도 및 현장 수리 가능성 신뢰도 점수 모듈을 개발한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **출동 지연 위험도와 현장 수리 가능성을 점수로 산출하는 모듈을 개발한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
|
|
"\n",
|
|
"이 문제를 해결하기 위해 본 미션에서는 **접수~도착 시간을 앙상블 회귀(GBM+RandomForest)로 예측해 거리 기반 기준모델과 Hold-out 기간에서 MAE·RMSE·R2를 비교 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
|
|
"\n",
|
|
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
|
|
"\n",
|
|
"**이번 결과물:** `src/models/mnt10_advanced_model.py` 형태의 결과물을 완성한다.\n",
|
|
"\n",
|
|
"### 🔍 평가 및 검증 기준 (PASS 조건)\n",
|
|
"#### 미션 판정\n",
|
|
"\n",
|
|
"**평가 구분:** 필수\n",
|
|
"\n",
|
|
"**핵심 평가영역:** 비정형/불규칙 수요에 대한 모델의 적응력 및 예측 안정성\n",
|
|
"\n",
|
|
"#### 미션 PASS 조건\n",
|
|
"\n",
|
|
"- 정량 기준: 기준 모델 대비 오차율(MAE/RMSE) 20% 이상 개선, R2 Score 0.85 이상\n",
|
|
"\n",
|
|
"\n",
|
|
"- 진행 기준: 모델 하이퍼파라미터 튜닝 완료 및 최적 모델 가중치 저장\n",
|
|
"\n",
|
|
"\n",
|
|
"- 검증 방법: Hold-out 테스트 데이터셋 대상 최종 성능 평가\n",
|
|
"\n",
|
|
"\n",
|
|
"#### 채점자 판정 메모\n",
|
|
"\n",
|
|
"> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
|
|
"\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"\"\"\"출동 지연 위험도(접수~도착 예상시간)를 앙상블 회귀로 예측하고, 현장 수리 가능성을 확률(신뢰도 점수)로 산출하는 모듈을 개발한다. Hold-out 기간에서 거리 기반 기준모델 대비 MAE/RMSE 개선율과 R2, 신뢰도 점수의 보정 상태를 검증한다.\"\"\"\n",
|
|
"import numpy as np\n",
|
|
"import pandas as pd\n",
|
|
"import matplotlib.pyplot as plt\n",
|
|
"from sklearn.ensemble import GradientBoostingRegressor, RandomForestRegressor\n",
|
|
"from sklearn.linear_model import LinearRegression, LogisticRegression\n",
|
|
"from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score, roc_auc_score, brier_score_loss\n",
|
|
"\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"# [교육용가상자료] 야전 이동정비반·긴급 정비 요청 생성기 (실제 부대·위치·장비·인원 정보 없음)\n",
|
|
"# 60km x 60km 가상 지역에 이동정비반 6개와 30일간의 긴급 정비 요청을 만든다.\n",
|
|
"# 장비 구분(mobility/fires/comms/power)과 고장 유형은 교육용 추상 분류다.\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"AREA_KM, DUTY_H = 60, 10.0 # 가상 작전지역 크기, 정비반 1일 기준 가용시간\n",
|
|
"FAULT_H = {\"engine\": 6.0, \"drivetrain\": 5.0, \"hydraulic\": 3.0, \"electric\": 2.0, \"comm_module\": 1.5, \"cooling\": 2.5}\n",
|
|
"FIELD_P = {\"engine\": 0.3, \"drivetrain\": 0.35, \"hydraulic\": 0.75, \"electric\": 0.9, \"comm_module\": 0.95, \"cooling\": 0.8}\n",
|
|
"EQUIP_F = {\"mobility\": 1.0, \"fires\": 1.2, \"comms\": 0.7, \"power\": 0.8}\n",
|
|
"FAULT_OF = {\"mobility\": [0.3, 0.25, 0.2, 0.1, 0.0, 0.15], \"fires\": [0.2, 0.15, 0.35, 0.15, 0.0, 0.15],\n",
|
|
" \"comms\": [0.0, 0.0, 0.0, 0.4, 0.5, 0.1], \"power\": [0.35, 0.0, 0.0, 0.35, 0.0, 0.3]}\n",
|
|
"CRIT_P = {\"mobility\": [0.45, 0.35, 0.2], \"fires\": [0.2, 0.35, 0.45], \"comms\": [0.3, 0.4, 0.3], \"power\": [0.6, 0.3, 0.1]}\n",
|
|
"HUBS = [(15, 20), (45, 25), (30, 50)] # 가상 훈련장 3곳(요청이 몰리는 지점)\n",
|
|
"\n",
|
|
"\n",
|
|
"def make_teams(seed=0):\n",
|
|
" rng = np.random.default_rng(seed)\n",
|
|
" base = [(8, 10), (30, 6), (52, 12), (10, 44), (32, 38), (54, 50)]\n",
|
|
" skills = [[\"fires\", \"mobility\"], [\"comms\", \"mobility\"], [\"fires\", \"power\"],\n",
|
|
" [\"mobility\", \"power\"], [\"comms\", \"fires\"], [\"comms\", \"mobility\", \"power\"]]\n",
|
|
" return [{\"team\": f\"MT{i + 1}\", \"x\": float(x), \"y\": float(y), \"crew\": int(rng.integers(3, 6)), \"skills\": s}\n",
|
|
" for i, ((x, y), s) in enumerate(zip(base, skills))]\n",
|
|
"\n",
|
|
"\n",
|
|
"def make_requests(n_days=30, per_day=10, seed=1, surge_days=(8, 9, 10, 11, 22, 23, 24)):\n",
|
|
" \"\"\"하루 평균 per_day건, 훈련 주간(surge_days)에는 1.6배의 긴급 정비 요청을 만든다.\"\"\"\n",
|
|
" rng = np.random.default_rng(seed)\n",
|
|
" rows = []\n",
|
|
" for d in range(n_days):\n",
|
|
" n = rng.poisson(per_day * (1.6 if d in surge_days else 1.0))\n",
|
|
" weather = rng.choice([\"clear\", \"rain\", \"snow\"], p=[0.65, 0.25, 0.1])\n",
|
|
" for _ in range(n):\n",
|
|
" if rng.random() < 0.7:\n",
|
|
" hx, hy = HUBS[rng.integers(0, 3)]\n",
|
|
" x, y = np.clip(rng.normal(hx, 6), 0, AREA_KM), np.clip(rng.normal(hy, 6), 0, AREA_KM)\n",
|
|
" else:\n",
|
|
" x, y = rng.uniform(0, AREA_KM, 2)\n",
|
|
" eq = str(rng.choice(list(EQUIP_F), p=[0.45, 0.25, 0.2, 0.1]))\n",
|
|
" fault = str(rng.choice(list(FAULT_H), p=FAULT_OF[eq]))\n",
|
|
" parts = bool(rng.random() < 0.75)\n",
|
|
" bad = weather != \"clear\"\n",
|
|
" std_h = FAULT_H[fault] * EQUIP_F[eq] * (1.35 if not parts else 1.0) * (1.15 if bad else 1.0) * rng.lognormal(0, 0.12)\n",
|
|
" rows.append({\"t_req\": d * 24 + float(np.clip(rng.normal(13, 4), 5, 23)), \"day\": d,\n",
|
|
" \"x\": float(x), \"y\": float(y), \"equip\": eq, \"fault\": fault,\n",
|
|
" \"crit\": int(rng.choice([1, 2, 3], p=CRIT_P[eq])), \"parts_ready\": parts,\n",
|
|
" \"weather\": str(weather), \"road\": \"unpaved\" if y > 30 and rng.random() < 0.7 else \"paved\",\n",
|
|
" \"field_ok\": bool(rng.random() < FIELD_P[fault] * (1.0 if parts else 0.5)),\n",
|
|
" \"std_repair_h\": round(float(std_h), 2), \"road_delay\": float(rng.lognormal(0, 0.15))})\n",
|
|
" df = pd.DataFrame(rows).sort_values(\"t_req\").reset_index(drop=True)\n",
|
|
" df.insert(0, \"req_id\", [f\"R{i:04d}\" for i in range(len(df))])\n",
|
|
" return df\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"\n",
|
|
"# ---- 출동 배정·정비 모의 공통 함수 ----\n",
|
|
"W0 = {\"crit\": 1.0, \"wait\": 0.3, \"short\": 0.2, \"balance\": 0.5} # 우선순위·배정 기본 가중치(미션 9에서 재튜닝)\n",
|
|
"\n",
|
|
"\n",
|
|
"def travel_h(team, r, actual=False):\n",
|
|
" \"\"\"도로 굴곡 1.3배, 포장 35km/h·비포장 25km/h, 악천후 20% 감속. actual=True면 실제 도로 지연을 반영.\"\"\"\n",
|
|
" # [TODO : 학생 실습 5] 출동 지연 위험도 및 현장 수리 가능성 신뢰도 점수 산출 모듈 개발 관련 핵심 로직을 구현하세요.\n",
|
|
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
|
|
" # <<< 여기에 코드를 작성하세요 >>>\n",
|
|
" pass # 작성 후 pass를 제거하세요.\n",
|
|
"def on_site_h(team, r):\n",
|
|
" \"\"\"현장 정비시간: 4인 기준 표준시간을 투입 인원으로 환산. 현장 수리 불가면 진단·후송 조치 1시간.\"\"\"\n",
|
|
" # [TODO : 학생 실습 5] 출동 지연 위험도 및 현장 수리 가능성 신뢰도 점수 산출 모듈 개발 관련 핵심 로직을 구현하세요.\n",
|
|
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
|
|
" # <<< 여기에 코드를 작성하세요 >>>\n",
|
|
" pass # 작성 후 pass를 제거하세요.\n",
|
|
"def priority(r, now, est_h, w):\n",
|
|
" return w[\"crit\"] * r[\"crit\"] + w[\"wait\"] * (now - r[\"t_req\"]) + w[\"short\"] / (1 + est_h)\n",
|
|
"\n",
|
|
"\n",
|
|
"def simulate(req, teams, policy=\"baseline\", w=None, cycle_h=1.0, est=None):\n",
|
|
" \"\"\"배정 주기(cycle_h)마다 대기 요청을 정렬해 이동정비반에 배정하고 출동·정비·복귀를 모의한다.\n",
|
|
" baseline: 접수 순서 + 거리 최근접 정비반 / priority: 우선순위 점수 순서 + 도착 예상시각·당일 부하 균형\"\"\"\n",
|
|
" w = w or W0\n",
|
|
" est = np.asarray(est if est is not None else req[\"std_repair_h\"].values, float)\n",
|
|
" n_days = int(req[\"day\"].max()) + 1\n",
|
|
" free = {t[\"team\"]: 0.0 for t in teams}\n",
|
|
" load = {t[\"team\"]: np.zeros(n_days + 5) for t in teams}\n",
|
|
" recs, out, pending, i, now = req.to_dict(\"records\"), [], [], 0, cycle_h\n",
|
|
" while i < len(recs) or pending:\n",
|
|
" while i < len(recs) and recs[i][\"t_req\"] <= now:\n",
|
|
" pending.append((i, recs[i])); i += 1\n",
|
|
" if policy == \"priority\":\n",
|
|
" pending.sort(key=lambda p: -priority(p[1], now, est[p[0]], w))\n",
|
|
" for k, r in pending:\n",
|
|
" cand = [t for t in teams if r[\"equip\"] in t[\"skills\"]]\n",
|
|
" day = min(int(now // 24), n_days + 4)\n",
|
|
" if policy == \"baseline\":\n",
|
|
" tm = min(cand, key=lambda t: np.hypot(t[\"x\"] - r[\"x\"], t[\"y\"] - r[\"y\"]))\n",
|
|
" else:\n",
|
|
" tm = min(cand, key=lambda t: max(free[t[\"team\"]], now) + travel_h(t, r)\n",
|
|
" + w[\"balance\"] * load[t[\"team\"]][day] / DUTY_H * 4)\n",
|
|
" backlog = max(0.0, free[tm[\"team\"]] - now)\n",
|
|
" go = max(free[tm[\"team\"]], now)\n",
|
|
" tr = travel_h(tm, r, actual=True)\n",
|
|
" work = on_site_h(tm, r)\n",
|
|
" free[tm[\"team\"]] = go + 2 * tr + work\n",
|
|
" load[tm[\"team\"]][min(int(go // 24), n_days + 4)] += 2 * tr + work\n",
|
|
" out.append({**r, \"idx\": k, \"team\": tm[\"team\"], \"dist_km\": float(np.hypot(tm[\"x\"] - r[\"x\"], tm[\"y\"] - r[\"y\"])),\n",
|
|
" \"backlog_h\": backlog, \"cycle_wait_h\": now - r[\"t_req\"], \"dispatch\": go, \"travel_h\": tr,\n",
|
|
" \"work_h\": work, \"response_h\": go + tr - r[\"t_req\"], \"downtime_h\": go + tr + work - r[\"t_req\"]})\n",
|
|
" pending, now = [], now + cycle_h\n",
|
|
" L = pd.DataFrame({k: v[:n_days] for k, v in load.items()}) / DUTY_H # 정비반별 일일 부하율\n",
|
|
" return pd.DataFrame(out).sort_values(\"idx\").reset_index(drop=True), L\n",
|
|
"\n",
|
|
"\n",
|
|
"def metrics(res, L):\n",
|
|
" c3 = res[res[\"crit\"] == 3]\n",
|
|
" return {\"평균 도착(h)\": round(res[\"response_h\"].mean(), 2), \"핵심장비 도착(h)\": round(c3[\"response_h\"].mean(), 2),\n",
|
|
" \"핵심장비 90%(h)\": round(c3[\"response_h\"].quantile(0.9), 2), \"평균 부하율\": round(float(L.values.mean()), 3),\n",
|
|
" \"부하율 편차\": round(float(L.mean().std()), 3), \"과부하 일수\": int((L > 1.0).values.sum()),\n",
|
|
" \"총 이동(h)\": round(float(res[\"travel_h\"].sum() * 2), 1)}\n",
|
|
"\n",
|
|
"\n",
|
|
"def feat(res):\n",
|
|
" \"\"\"배정 시점에 알 수 있는 정보: 배정 정비반까지 거리, 정비반 잔여 작업(백로그), 배정 주기 대기, 기상·도로·중요도.\"\"\"\n",
|
|
" X = pd.DataFrame({\"dist_km\": res[\"dist_km\"], \"backlog_h\": res[\"backlog_h\"], \"cycle_wait_h\": res[\"cycle_wait_h\"],\n",
|
|
" \"weather_bad\": (res[\"weather\"] != \"clear\").astype(int), \"unpaved\": (res[\"road\"] == \"unpaved\").astype(int),\n",
|
|
" \"crit\": res[\"crit\"], \"hour\": res[\"t_req\"] % 24})\n",
|
|
" X[\"dist_x_road\"] = X[\"dist_km\"] * (1 + 0.4 * X[\"unpaved\"]) * (1 + 0.2 * X[\"weather_bad\"])\n",
|
|
" return X\n",
|
|
"\n",
|
|
"\n",
|
|
"def feat_field(df):\n",
|
|
" X = pd.get_dummies(df[[\"fault\", \"equip\"]]).astype(int)\n",
|
|
" X[\"parts_ready\"] = df[\"parts_ready\"].astype(int)\n",
|
|
" X[\"weather_bad\"] = (df[\"weather\"] != \"clear\").astype(int)\n",
|
|
" return X\n",
|
|
"\n",
|
|
"\n",
|
|
"teams = make_teams()\n",
|
|
"train = pd.concat([simulate(make_requests(seed=s), teams, \"priority\")[0] for s in (11, 21)], ignore_index=True)\n",
|
|
"test = simulate(make_requests(seed=1), teams, \"priority\")[0] # Hold-out: 학습에 쓰지 않은 30일\n",
|
|
"Xtr, Xte, ytr, yte = feat(train), feat(test), train[\"response_h\"], test[\"response_h\"]\n",
|
|
"\n",
|
|
"# 1) 출동 지연 예측: 기준모델(거리만) vs 앙상블(GBM + RandomForest 평균)\n",
|
|
"base = LinearRegression().fit(Xtr[[\"dist_km\"]], ytr)\n",
|
|
"gbm = GradientBoostingRegressor(n_estimators=300, max_depth=3, learning_rate=0.05, random_state=0).fit(Xtr, ytr)\n",
|
|
"rf = RandomForestRegressor(n_estimators=200, min_samples_leaf=3, random_state=0).fit(Xtr, ytr)\n",
|
|
"preds = {\"기준모델(거리 선형)\": base.predict(Xte[[\"dist_km\"]]), \"GBM\": gbm.predict(Xte), \"RandomForest\": rf.predict(Xte)}\n",
|
|
"preds[\"앙상블(GBM+RF)\"] = (preds[\"GBM\"] + preds[\"RandomForest\"]) / 2\n",
|
|
"tab = pd.DataFrame([{\"모델\": k, \"MAE(h)\": mean_absolute_error(yte, p), \"RMSE(h)\": mean_squared_error(yte, p) ** 0.5, \"R2\": r2_score(yte, p)} for k, p in preds.items()]).round(3)\n",
|
|
"print(f\"학습 {len(train)}건 / Hold-out {len(test)}건\")\n",
|
|
"print(tab.to_string(index=False))\n",
|
|
"b, e = tab.iloc[0], tab.iloc[-1]\n",
|
|
"imp_mae, imp_rmse = 1 - e[\"MAE(h)\"] / b[\"MAE(h)\"], 1 - e[\"RMSE(h)\"] / b[\"RMSE(h)\"]\n",
|
|
"print(f\"기준모델 대비 개선: MAE {imp_mae:.1%}, RMSE {imp_rmse:.1%} (목표 20% 이상: {'달성' if min(imp_mae, imp_rmse) >= 0.2 else '미달'}), \"\n",
|
|
" f\"R2 {e['R2']:.3f} (목표 0.85 이상: {'달성' if e['R2'] >= 0.85 else '미달'})\")\n",
|
|
"\n",
|
|
"# 2) 출동 지연 위험도 등급(업무 임계값: 핵심 장비 2시간, 그 외 4시간 이내 도착)\n",
|
|
"limit = np.where(test[\"crit\"] == 3, 2.0, 4.0)\n",
|
|
"risk = np.where(preds[\"앙상블(GBM+RF)\"] > limit, \"높음\", np.where(preds[\"앙상블(GBM+RF)\"] > limit * 0.7, \"주의\", \"낮음\"))\n",
|
|
"actual_late = yte.values > limit\n",
|
|
"print(\"\\n[지연 위험도 등급별 실제 지연 비율]\")\n",
|
|
"print(pd.DataFrame({\"등급\": risk, \"실제지연\": actual_late}).groupby(\"등급\")[\"실제지연\"].agg([\"count\", \"mean\"]).rename(columns={\"count\": \"건수\", \"mean\": \"실제 지연 비율\"}).round(2).to_string())\n",
|
|
"\n",
|
|
"# 3) 현장 수리 가능성 신뢰도 점수(로지스틱 회귀 확률)\n",
|
|
"all_req = [make_requests(seed=s) for s in (11, 21)]\n",
|
|
"ftr = pd.concat(all_req, ignore_index=True); fte = make_requests(seed=1)\n",
|
|
"Ftr = feat_field(ftr); Fte = feat_field(fte).reindex(columns=Ftr.columns, fill_value=0)\n",
|
|
"clf = LogisticRegression(max_iter=1000).fit(Ftr, ftr[\"field_ok\"])\n",
|
|
"prob = clf.predict_proba(Fte)[:, 1]\n",
|
|
"print(f\"\\n현장 수리 가능성: AUC {roc_auc_score(fte['field_ok'], prob):.3f}, Brier {brier_score_loss(fte['field_ok'], prob):.3f}\")\n",
|
|
"bins = pd.cut(prob, [0, 0.4, 0.7, 1.0], labels=[\"후송 준비(<0.4)\", \"판단 보류(0.4~0.7)\", \"현장수리 유력(>=0.7)\"])\n",
|
|
"cal = pd.DataFrame({\"구간\": bins, \"예측\": prob, \"실제\": fte[\"field_ok\"]}).groupby(\"구간\", observed=True).agg(건수=(\"예측\", \"size\"), 평균점수=(\"예측\", \"mean\"), 실제비율=(\"실제\", \"mean\")).round(2)\n",
|
|
"print(cal.to_string())\n",
|
|
"print(\"점수가 높은 구간일수록 실제 현장 수리 비율이 높아 신뢰도 점수로 쓸 수 있다. 점수는 출동 준비(부속·후송 수단) 참고용이다.\")\n",
|
|
"\n",
|
|
"fig, ax = plt.subplots(1, 2, figsize=(11, 4))\n",
|
|
"ax[0].scatter(yte, preds[\"앙상블(GBM+RF)\"], s=10); ax[0].plot([0, yte.max()], [0, yte.max()], \"r--\")\n",
|
|
"ax[0].set_xlabel(\"actual response (h)\"); ax[0].set_ylabel(\"predicted (h)\"); ax[0].set_title(f\"Delay model hold-out (R2={e['R2']:.2f})\")\n",
|
|
"ax[1].plot(cal[\"평균점수\"], cal[\"실제비율\"], \"o-\"); ax[1].plot([0, 1], [0, 1], \"k--\"); ax[1].set_xlabel(\"mean confidence score\")\n",
|
|
"ax[1].set_ylabel(\"actual field-repair rate\"); ax[1].set_title(\"Field-repair confidence calibration\")\n",
|
|
"plt.tight_layout(); plt.show()\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"<details>\n",
|
|
"<summary><b>💡 [참고] 미션 05 예상 실행 결과 (Output)</b></summary>\n",
|
|
"\n",
|
|
"```text\n",
|
|
"학습 724건 / Hold-out 356건\n",
|
|
" 모델 MAE(h) RMSE(h) R2\n",
|
|
" 기준모델(거리 선형) 1.025 1.442 0.192\n",
|
|
" GBM 0.152 0.229 0.980\n",
|
|
"RandomForest 0.184 0.294 0.966\n",
|
|
" 앙상블(GBM+RF) 0.165 0.254 0.975\n",
|
|
"기준모델 대비 개선: MAE 83.9%, RMSE 82.4% (목표 20% 이상: 달성), R2 0.975 (목표 0.85 이상: 달성)\n",
|
|
"\n",
|
|
"[지연 위험도 등급별 실제 지연 비율]\n",
|
|
" 건수 실제 지연 비율\n",
|
|
"등급 \n",
|
|
"낮음 248 0.00\n",
|
|
"높음 61 0.93\n",
|
|
"주의 47 0.04\n",
|
|
"\n",
|
|
"현장 수리 가능성: AUC 0.762, Brier 0.185\n",
|
|
" 건수 평균점수 실제비율\n",
|
|
"구간 \n",
|
|
"후송 준비(<0.4) 143 0.30 0.34\n",
|
|
"판단 보류(0.4~0.7) 70 0.59 0.63\n",
|
|
"현장수리 유력(>=0.7) 143 0.80 0.83\n",
|
|
"점수가 높은 구간일수록 실제 현장 수리 비율이 높아 신뢰도 점수로 쓸 수 있다. 점수는 출동 준비(부속·후송 수단) 참고용이다.\n",
|
|
"```\n",
|
|
"</details>\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"---\n",
|
|
"\n",
|
|
"## 🎯 미션 06. 야전 정비 요청 현황 및 이동정비반 관제 대시보드 UI 구현\n",
|
|
"**소요시간**: 약 3.0시간\n",
|
|
"\n",
|
|
"### 📋 미션 배경 및 수행 지침\n",
|
|
"#### 미션 배경 및 당위성\n",
|
|
"\n",
|
|
"야전 정비 요청 현황 및 이동정비반 관제 대시보드 UI를 구현한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **정비 요청 현황과 이동정비반 상태를 보여 주는 관제 대시보드를 만들고, 예측마다 근거를 설명하는 설명가능성(XAI) 기능을 붙인다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
|
|
"\n",
|
|
"이 문제를 해결하기 위해 본 미션에서는 **요청·정비반 지도, 정비반 부하율, 중요도별 대기 요청, 일별·중요도별 도착시간, 예측 워터폴 그림으로 관제 화면을 구성 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
|
|
"\n",
|
|
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
|
|
"\n",
|
|
"**이번 결과물:** `src/explainability/mnt10_xai_explainer.py 및 XAI 리포트` 형태의 결과물을 완성한다.\n",
|
|
"\n",
|
|
"### 🔍 평가 및 검증 기준 (PASS 조건)\n",
|
|
"#### 미션 판정\n",
|
|
"\n",
|
|
"**평가 구분:** 필수\n",
|
|
"\n",
|
|
"**핵심 평가영역:** 군수 장교/군무원이 신뢰하고 조치할 수 있는 설명의 명확성\n",
|
|
"\n",
|
|
"#### 미션 PASS 조건\n",
|
|
"\n",
|
|
"- 정량 기준: 주요 예측 건당 기여도 상위 3개 인자 자동 산출률 100%\n",
|
|
"\n",
|
|
"\n",
|
|
"- 진행 기준: SHAP summary plot 및 waterfall plot 생성 파이프라인 완성\n",
|
|
"\n",
|
|
"\n",
|
|
"- 검증 방법: 실무자 대상 설명 근거 유용성 평가 검증\n",
|
|
"\n",
|
|
"\n",
|
|
"#### 채점자 판정 메모\n",
|
|
"\n",
|
|
"> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
|
|
"\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"\"\"\"야전 정비 요청 현황과 이동정비반 상태(위치·부하율·대기 요청·도착시간 추이)를 한 화면에 보여 주는 관제 대시보드를 그리고, 정비 소요시간 예측마다 기여도 상위 3개 인자를 자동 산출해 워터폴 그림과 자연어 설명으로 제공하는 설명가능성(XAI) 모듈을 구현한다.\"\"\"\n",
|
|
"import time\n",
|
|
"import numpy as np\n",
|
|
"import pandas as pd\n",
|
|
"import matplotlib.pyplot as plt\n",
|
|
"from sklearn.ensemble import GradientBoostingRegressor\n",
|
|
"\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"# [교육용가상자료] 야전 이동정비반·긴급 정비 요청 생성기 (실제 부대·위치·장비·인원 정보 없음)\n",
|
|
"# 60km x 60km 가상 지역에 이동정비반 6개와 30일간의 긴급 정비 요청을 만든다.\n",
|
|
"# 장비 구분(mobility/fires/comms/power)과 고장 유형은 교육용 추상 분류다.\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"AREA_KM, DUTY_H = 60, 10.0 # 가상 작전지역 크기, 정비반 1일 기준 가용시간\n",
|
|
"FAULT_H = {\"engine\": 6.0, \"drivetrain\": 5.0, \"hydraulic\": 3.0, \"electric\": 2.0, \"comm_module\": 1.5, \"cooling\": 2.5}\n",
|
|
"FIELD_P = {\"engine\": 0.3, \"drivetrain\": 0.35, \"hydraulic\": 0.75, \"electric\": 0.9, \"comm_module\": 0.95, \"cooling\": 0.8}\n",
|
|
"EQUIP_F = {\"mobility\": 1.0, \"fires\": 1.2, \"comms\": 0.7, \"power\": 0.8}\n",
|
|
"FAULT_OF = {\"mobility\": [0.3, 0.25, 0.2, 0.1, 0.0, 0.15], \"fires\": [0.2, 0.15, 0.35, 0.15, 0.0, 0.15],\n",
|
|
" \"comms\": [0.0, 0.0, 0.0, 0.4, 0.5, 0.1], \"power\": [0.35, 0.0, 0.0, 0.35, 0.0, 0.3]}\n",
|
|
"CRIT_P = {\"mobility\": [0.45, 0.35, 0.2], \"fires\": [0.2, 0.35, 0.45], \"comms\": [0.3, 0.4, 0.3], \"power\": [0.6, 0.3, 0.1]}\n",
|
|
"HUBS = [(15, 20), (45, 25), (30, 50)] # 가상 훈련장 3곳(요청이 몰리는 지점)\n",
|
|
"\n",
|
|
"\n",
|
|
"def make_teams(seed=0):\n",
|
|
" rng = np.random.default_rng(seed)\n",
|
|
" base = [(8, 10), (30, 6), (52, 12), (10, 44), (32, 38), (54, 50)]\n",
|
|
" skills = [[\"fires\", \"mobility\"], [\"comms\", \"mobility\"], [\"fires\", \"power\"],\n",
|
|
" [\"mobility\", \"power\"], [\"comms\", \"fires\"], [\"comms\", \"mobility\", \"power\"]]\n",
|
|
" return [{\"team\": f\"MT{i + 1}\", \"x\": float(x), \"y\": float(y), \"crew\": int(rng.integers(3, 6)), \"skills\": s}\n",
|
|
" for i, ((x, y), s) in enumerate(zip(base, skills))]\n",
|
|
"\n",
|
|
"\n",
|
|
"def make_requests(n_days=30, per_day=10, seed=1, surge_days=(8, 9, 10, 11, 22, 23, 24)):\n",
|
|
" \"\"\"하루 평균 per_day건, 훈련 주간(surge_days)에는 1.6배의 긴급 정비 요청을 만든다.\"\"\"\n",
|
|
" rng = np.random.default_rng(seed)\n",
|
|
" rows = []\n",
|
|
" for d in range(n_days):\n",
|
|
" n = rng.poisson(per_day * (1.6 if d in surge_days else 1.0))\n",
|
|
" weather = rng.choice([\"clear\", \"rain\", \"snow\"], p=[0.65, 0.25, 0.1])\n",
|
|
" for _ in range(n):\n",
|
|
" if rng.random() < 0.7:\n",
|
|
" hx, hy = HUBS[rng.integers(0, 3)]\n",
|
|
" x, y = np.clip(rng.normal(hx, 6), 0, AREA_KM), np.clip(rng.normal(hy, 6), 0, AREA_KM)\n",
|
|
" else:\n",
|
|
" x, y = rng.uniform(0, AREA_KM, 2)\n",
|
|
" eq = str(rng.choice(list(EQUIP_F), p=[0.45, 0.25, 0.2, 0.1]))\n",
|
|
" fault = str(rng.choice(list(FAULT_H), p=FAULT_OF[eq]))\n",
|
|
" parts = bool(rng.random() < 0.75)\n",
|
|
" bad = weather != \"clear\"\n",
|
|
" std_h = FAULT_H[fault] * EQUIP_F[eq] * (1.35 if not parts else 1.0) * (1.15 if bad else 1.0) * rng.lognormal(0, 0.12)\n",
|
|
" rows.append({\"t_req\": d * 24 + float(np.clip(rng.normal(13, 4), 5, 23)), \"day\": d,\n",
|
|
" \"x\": float(x), \"y\": float(y), \"equip\": eq, \"fault\": fault,\n",
|
|
" \"crit\": int(rng.choice([1, 2, 3], p=CRIT_P[eq])), \"parts_ready\": parts,\n",
|
|
" \"weather\": str(weather), \"road\": \"unpaved\" if y > 30 and rng.random() < 0.7 else \"paved\",\n",
|
|
" \"field_ok\": bool(rng.random() < FIELD_P[fault] * (1.0 if parts else 0.5)),\n",
|
|
" \"std_repair_h\": round(float(std_h), 2), \"road_delay\": float(rng.lognormal(0, 0.15))})\n",
|
|
" df = pd.DataFrame(rows).sort_values(\"t_req\").reset_index(drop=True)\n",
|
|
" df.insert(0, \"req_id\", [f\"R{i:04d}\" for i in range(len(df))])\n",
|
|
" return df\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"\n",
|
|
"# ---- 출동 배정·정비 모의 공통 함수 ----\n",
|
|
"W0 = {\"crit\": 1.0, \"wait\": 0.3, \"short\": 0.2, \"balance\": 0.5} # 우선순위·배정 기본 가중치(미션 9에서 재튜닝)\n",
|
|
"\n",
|
|
"\n",
|
|
"def travel_h(team, r, actual=False):\n",
|
|
" \"\"\"도로 굴곡 1.3배, 포장 35km/h·비포장 25km/h, 악천후 20% 감속. actual=True면 실제 도로 지연을 반영.\"\"\"\n",
|
|
" # [TODO : 학생 실습 6] 야전 정비 요청 현황 및 이동정비반 관제 대시보드 UI 구현 관련 핵심 로직을 구현하세요.\n",
|
|
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
|
|
" # <<< 여기에 코드를 작성하세요 >>>\n",
|
|
" pass # 작성 후 pass를 제거하세요.\n",
|
|
"def on_site_h(team, r):\n",
|
|
" \"\"\"현장 정비시간: 4인 기준 표준시간을 투입 인원으로 환산. 현장 수리 불가면 진단·후송 조치 1시간.\"\"\"\n",
|
|
" # [TODO : 학생 실습 6] 야전 정비 요청 현황 및 이동정비반 관제 대시보드 UI 구현 관련 핵심 로직을 구현하세요.\n",
|
|
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
|
|
" # <<< 여기에 코드를 작성하세요 >>>\n",
|
|
" pass # 작성 후 pass를 제거하세요.\n",
|
|
"def priority(r, now, est_h, w):\n",
|
|
" return w[\"crit\"] * r[\"crit\"] + w[\"wait\"] * (now - r[\"t_req\"]) + w[\"short\"] / (1 + est_h)\n",
|
|
"\n",
|
|
"\n",
|
|
"def simulate(req, teams, policy=\"baseline\", w=None, cycle_h=1.0, est=None):\n",
|
|
" \"\"\"배정 주기(cycle_h)마다 대기 요청을 정렬해 이동정비반에 배정하고 출동·정비·복귀를 모의한다.\n",
|
|
" baseline: 접수 순서 + 거리 최근접 정비반 / priority: 우선순위 점수 순서 + 도착 예상시각·당일 부하 균형\"\"\"\n",
|
|
" w = w or W0\n",
|
|
" est = np.asarray(est if est is not None else req[\"std_repair_h\"].values, float)\n",
|
|
" n_days = int(req[\"day\"].max()) + 1\n",
|
|
" free = {t[\"team\"]: 0.0 for t in teams}\n",
|
|
" load = {t[\"team\"]: np.zeros(n_days + 5) for t in teams}\n",
|
|
" recs, out, pending, i, now = req.to_dict(\"records\"), [], [], 0, cycle_h\n",
|
|
" while i < len(recs) or pending:\n",
|
|
" while i < len(recs) and recs[i][\"t_req\"] <= now:\n",
|
|
" pending.append((i, recs[i])); i += 1\n",
|
|
" if policy == \"priority\":\n",
|
|
" pending.sort(key=lambda p: -priority(p[1], now, est[p[0]], w))\n",
|
|
" for k, r in pending:\n",
|
|
" cand = [t for t in teams if r[\"equip\"] in t[\"skills\"]]\n",
|
|
" day = min(int(now // 24), n_days + 4)\n",
|
|
" if policy == \"baseline\":\n",
|
|
" tm = min(cand, key=lambda t: np.hypot(t[\"x\"] - r[\"x\"], t[\"y\"] - r[\"y\"]))\n",
|
|
" else:\n",
|
|
" tm = min(cand, key=lambda t: max(free[t[\"team\"]], now) + travel_h(t, r)\n",
|
|
" + w[\"balance\"] * load[t[\"team\"]][day] / DUTY_H * 4)\n",
|
|
" backlog = max(0.0, free[tm[\"team\"]] - now)\n",
|
|
" go = max(free[tm[\"team\"]], now)\n",
|
|
" tr = travel_h(tm, r, actual=True)\n",
|
|
" work = on_site_h(tm, r)\n",
|
|
" free[tm[\"team\"]] = go + 2 * tr + work\n",
|
|
" load[tm[\"team\"]][min(int(go // 24), n_days + 4)] += 2 * tr + work\n",
|
|
" out.append({**r, \"idx\": k, \"team\": tm[\"team\"], \"dist_km\": float(np.hypot(tm[\"x\"] - r[\"x\"], tm[\"y\"] - r[\"y\"])),\n",
|
|
" \"backlog_h\": backlog, \"cycle_wait_h\": now - r[\"t_req\"], \"dispatch\": go, \"travel_h\": tr,\n",
|
|
" \"work_h\": work, \"response_h\": go + tr - r[\"t_req\"], \"downtime_h\": go + tr + work - r[\"t_req\"]})\n",
|
|
" pending, now = [], now + cycle_h\n",
|
|
" L = pd.DataFrame({k: v[:n_days] for k, v in load.items()}) / DUTY_H # 정비반별 일일 부하율\n",
|
|
" return pd.DataFrame(out).sort_values(\"idx\").reset_index(drop=True), L\n",
|
|
"\n",
|
|
"\n",
|
|
"def metrics(res, L):\n",
|
|
" c3 = res[res[\"crit\"] == 3]\n",
|
|
" return {\"평균 도착(h)\": round(res[\"response_h\"].mean(), 2), \"핵심장비 도착(h)\": round(c3[\"response_h\"].mean(), 2),\n",
|
|
" \"핵심장비 90%(h)\": round(c3[\"response_h\"].quantile(0.9), 2), \"평균 부하율\": round(float(L.values.mean()), 3),\n",
|
|
" \"부하율 편차\": round(float(L.mean().std()), 3), \"과부하 일수\": int((L > 1.0).values.sum()),\n",
|
|
" \"총 이동(h)\": round(float(res[\"travel_h\"].sum() * 2), 1)}\n",
|
|
"\n",
|
|
"# ---- 접수 시점 파생 피처(미션 4와 동일) ----\n",
|
|
"def make_features(req, teams):\n",
|
|
" \"\"\"접수 시점에 알 수 있는 정보만으로 파생 피처를 만든다(미래 정보 누설 방지).\"\"\"\n",
|
|
" X = pd.DataFrame(index=req.index)\n",
|
|
" X[\"hour\"] = req[\"t_req\"] % 24\n",
|
|
" X[\"is_night\"] = (X[\"hour\"] >= 19).astype(int)\n",
|
|
" X[\"dow\"] = req[\"day\"] % 7\n",
|
|
" X[\"weather_bad\"] = (req[\"weather\"] != \"clear\").astype(int)\n",
|
|
" X[\"unpaved\"] = (req[\"road\"] == \"unpaved\").astype(int)\n",
|
|
" X[\"no_parts\"] = (~req[\"parts_ready\"]).astype(int)\n",
|
|
" X[\"std_table_h\"] = req[\"fault\"].map(FAULT_H) * req[\"equip\"].map(EQUIP_F) # 정비 표준시간표 값\n",
|
|
" X[\"crit_x_noparts\"] = req[\"crit\"] * X[\"no_parts\"]\n",
|
|
" X[\"bad_x_table\"] = X[\"weather_bad\"] * X[\"std_table_h\"]\n",
|
|
" X[\"near_team_km\"] = [min(np.hypot(t[\"x\"] - x, t[\"y\"] - y) for t in teams if e in t[\"skills\"]) for x, y, e in zip(req[\"x\"], req[\"y\"], req[\"equip\"])]\n",
|
|
" X[\"hub_km\"] = [min(np.hypot(hx - x, hy - y) for hx, hy in HUBS) for x, y in zip(req[\"x\"], req[\"y\"])]\n",
|
|
" X[\"req_24h\"] = [((req[\"t_req\"] > t - 24) & (req[\"t_req\"] <= t)).sum() for t in req[\"t_req\"]]\n",
|
|
" X[\"fault_past_mean\"] = req.groupby(\"fault\")[\"std_repair_h\"].transform(lambda s: s.shift().expanding().mean()).fillna(req[\"std_repair_h\"].mean())\n",
|
|
" X[\"crit\"] = req[\"crit\"]\n",
|
|
" for e in EQUIP_F:\n",
|
|
" X[f\"eq_{e}\"] = (req[\"equip\"] == e).astype(int)\n",
|
|
" return X\n",
|
|
"\n",
|
|
"\n",
|
|
"def fit_repair_model(hist, teams):\n",
|
|
" \"\"\"과거 요청으로 정비 소요시간(4인 기준) 예측 모델을 학습한다.\"\"\"\n",
|
|
" return GradientBoostingRegressor(n_estimators=200, max_depth=3, learning_rate=0.05, random_state=0).fit(make_features(hist, teams), hist[\"std_repair_h\"])\n",
|
|
"\n",
|
|
"KO = {\"std_table_h\": \"정비 표준시간표 값\", \"no_parts\": \"수리부속 미확보\", \"weather_bad\": \"악천후\", \"bad_x_table\": \"악천후 x 표준시간\",\n",
|
|
" \"crit_x_noparts\": \"핵심장비 x 부속 미확보\", \"fault_past_mean\": \"같은 고장 과거 평균\", \"unpaved\": \"비포장 도로\", \"crit\": \"작전 중요도\",\n",
|
|
" \"near_team_km\": \"최근접 정비반 거리\", \"hub_km\": \"훈련장과의 거리\", \"req_24h\": \"최근 24시간 요청 수\", \"hour\": \"접수 시각\",\n",
|
|
" \"is_night\": \"야간 접수\", \"dow\": \"요일\"}\n",
|
|
"\n",
|
|
"\n",
|
|
"def explain(model, X, bg, top=3):\n",
|
|
" \"\"\"가림(occlusion) 방식 국소 기여도: 피처 j를 배경 표본 값으로 바꿨을 때 예측이 평균적으로 얼마나 달라지는지.\n",
|
|
" SHAP과 같은 '이 예측을 평균에서 얼마나 밀어 올렸나'를 근사한다(외부 라이브러리 없이).\"\"\"\n",
|
|
" base = model.predict(X)\n",
|
|
" contrib = np.zeros(X.shape)\n",
|
|
" for j, c in enumerate(X.columns):\n",
|
|
" for b in bg:\n",
|
|
" Xb = X.copy(); Xb[c] = b[j]\n",
|
|
" contrib[:, j] += base - model.predict(Xb)\n",
|
|
" contrib /= len(bg)\n",
|
|
" C = pd.DataFrame(contrib, columns=X.columns, index=X.index)\n",
|
|
" tops = [list(C.loc[i].abs().sort_values(ascending=False).index[:top]) for i in C.index]\n",
|
|
" return base, C, tops\n",
|
|
"\n",
|
|
"\n",
|
|
"def sentence(r, pred, C, top):\n",
|
|
" parts = [f\"{KO.get(f, f)}({'+' if C[f] > 0 else ''}{C[f]:.1f}h)\" for f in top]\n",
|
|
" return f\"{r['req_id']} {r['equip']}/{r['fault']} 예상 정비 {pred:.1f}h(평균 대비 {pred - AVG:+.1f}h), 영향 요인: \" + \", \".join(parts)\n",
|
|
"\n",
|
|
"\n",
|
|
"teams = make_teams()\n",
|
|
"hist, req = make_requests(seed=11), make_requests(seed=1)\n",
|
|
"model = fit_repair_model(hist, teams)\n",
|
|
"AVG = float(hist[\"std_repair_h\"].mean())\n",
|
|
"Xh, X = make_features(hist, teams), make_features(req, teams)\n",
|
|
"res, L = simulate(req, teams, \"priority\", est=model.predict(X))\n",
|
|
"\n",
|
|
"# 1) 설명가능성: 평가 기간 전체 요청에 대해 상위 3개 인자 자동 산출\n",
|
|
"t0 = time.perf_counter()\n",
|
|
"bg = Xh.sample(30, random_state=0).values\n",
|
|
"pred, C, tops = explain(model, X, bg)\n",
|
|
"sec = time.perf_counter() - t0\n",
|
|
"rate = np.mean([len(t) == 3 for t in tops])\n",
|
|
"print(f\"[교육용가상자료] 요청 {len(X)}건 설명 산출 {sec:.1f}초, 상위 3개 인자 자동 산출률 {rate:.0%} (목표 100%: {'달성' if rate == 1 else '미달'})\")\n",
|
|
"check = np.abs(C.sum(axis=1) - (pred - model.predict(Xh).mean())).mean()\n",
|
|
"print(f\"기여도 합과 (예측 - 평균 예측)의 평균 차이 {check:.2f}h -> 근사 설명이므로 참고 지표로 쓴다\")\n",
|
|
"freq = pd.Series([f for t in tops for f in t]).map(lambda f: KO.get(f, f)).value_counts().head(5)\n",
|
|
"print(\"상위 3개 인자에 자주 등장한 요인:\", \", \".join(f\"{k} {v}건\" for k, v in freq.items()))\n",
|
|
"\n",
|
|
"# 2) 관제 대시보드: 22일차 14시 기준\n",
|
|
"day, now = 22, 22 * 24 + 14\n",
|
|
"cur = res[(res[\"t_req\"] <= now) & (res[\"t_req\"] > now - 24)]\n",
|
|
"waiting = cur[cur[\"dispatch\"] > now]\n",
|
|
"print(f\"\\n[관제 현황 {day}일차 14시] 최근 24시간 요청 {len(cur)}건, 출동 대기 {len(waiting)}건, 정비반 {len(teams)}개\")\n",
|
|
"fig, ax = plt.subplots(2, 3, figsize=(15, 8))\n",
|
|
"col = {1: \"tab:gray\", 2: \"tab:orange\", 3: \"tab:red\"}\n",
|
|
"ax[0, 0].scatter(cur[\"x\"], cur[\"y\"], c=cur[\"crit\"].map(col), s=40)\n",
|
|
"for t in teams:\n",
|
|
" ax[0, 0].plot(t[\"x\"], t[\"y\"], \"ks\"); ax[0, 0].annotate(t[\"team\"], (t[\"x\"] + 1, t[\"y\"] + 1))\n",
|
|
"ax[0, 0].set_xlim(0, AREA_KM); ax[0, 0].set_ylim(0, AREA_KM); ax[0, 0].set_title(\"Requests (24h, red=critical) & teams\")\n",
|
|
"lr = L.iloc[day]\n",
|
|
"ax[0, 1].bar(lr.index, lr.values, color=[\"tab:red\" if v > 1 else \"tab:blue\" for v in lr.values])\n",
|
|
"ax[0, 1].axhline(1.0, ls=\"--\", color=\"k\"); ax[0, 1].set_title(f\"Team load rate, day {day}\")\n",
|
|
"q = waiting.groupby(\"crit\").size().reindex([1, 2, 3], fill_value=0)\n",
|
|
"ax[0, 2].bar([\"normal\", \"important\", \"critical\"], q.values, color=[col[k] for k in q.index]); ax[0, 2].set_title(\"Waiting requests by criticality\")\n",
|
|
"dd = res.groupby(\"day\")[\"response_h\"].mean()\n",
|
|
"ax[1, 0].plot(dd.index, dd.values, marker=\"o\"); ax[1, 0].set_title(\"Daily mean response time (h)\"); ax[1, 0].set_xlabel(\"day\")\n",
|
|
"ax[1, 1].boxplot([res.loc[res[\"crit\"] == k, \"response_h\"] for k in (1, 2, 3)]); ax[1, 1].set_xticks([1, 2, 3], [\"normal\", \"important\", \"critical\"])\n",
|
|
"ax[1, 1].set_title(\"Response time by criticality\")\n",
|
|
"i = int(C.index[req[\"crit\"].eq(3)][int(np.argmax(pred[req[\"crit\"].eq(3).values]))])\n",
|
|
"ci = C.loc[i].reindex(C.loc[i].abs().sort_values(ascending=False).index)[:6][::-1]\n",
|
|
"start = model.predict(Xh).mean()\n",
|
|
"ax[1, 2].barh(range(len(ci)), ci.values, left=start + np.r_[0, np.cumsum(ci.values)[:-1]], color=[\"tab:red\" if v > 0 else \"tab:green\" for v in ci.values])\n",
|
|
"ax[1, 2].set_yticks(range(len(ci)), ci.index); ax[1, 2].set_title(f\"Waterfall: {req.loc[i, 'req_id']} repair time\"); ax[1, 2].set_xlabel(\"hours\")\n",
|
|
"plt.tight_layout(); plt.show()\n",
|
|
"print(\"대시보드 구성요소 6종: 요청·정비반 지도, 정비반 부하율, 중요도별 대기, 일별 도착시간, 중요도별 도착시간, 예측 워터폴\")\n",
|
|
"\n",
|
|
"# 3) 자연어 설명 예시(정비대대 관제 화면 하단)\n",
|
|
"print(\"\\n[예측 설명 예시]\")\n",
|
|
"for k in list(waiting.sort_values(\"crit\", ascending=False)[\"idx\"][:4]) + [i]:\n",
|
|
" print(\" - \" + sentence(req.loc[k], pred[k], C.loc[k], tops[k]))\n",
|
|
"over = lr[lr > 1.0]\n",
|
|
"act = \"전 정비반 과부하 -> 상급 정비부대 지원 요청·비핵심 요청 연기 검토\" if len(over) == len(teams) else \"여유 정비반 지원 검토\"\n",
|
|
"print(f\"\\n경보: {day}일차(훈련 주간) 부하율 1.0 초과 정비반 {list(over.index)} -> {act}(정비대대장 판단)\" if len(over) else \"\\n경보: 과부하 정비반 없음\")\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"<details>\n",
|
|
"<summary><b>💡 [참고] 미션 06 예상 실행 결과 (Output)</b></summary>\n",
|
|
"\n",
|
|
"```text\n",
|
|
"[교육용가상자료] 요청 356건 설명 산출 1.7초, 상위 3개 인자 자동 산출률 100% (목표 100%: 달성)\n",
|
|
"기여도 합과 (예측 - 평균 예측)의 평균 차이 0.31h -> 근사 설명이므로 참고 지표로 쓴다\n",
|
|
"상위 3개 인자에 자주 등장한 요인: 정비 표준시간표 값 329건, 핵심장비 x 부속 미확보 227건, 같은 고장 과거 평균 225건, 악천후 x 표준시간 189건, 수리부속 미확보 57건\n",
|
|
"\n",
|
|
"[관제 현황 22일차 14시] 최근 24시간 요청 22건, 출동 대기 6건, 정비반 6개\n",
|
|
"대시보드 구성요소 6종: 요청·정비반 지도, 정비반 부하율, 중요도별 대기, 일별 도착시간, 중요도별 도착시간, 예측 워터폴\n",
|
|
"\n",
|
|
"[예측 설명 예시]\n",
|
|
" - R0256 fires/electric 예상 정비 2.4h(평균 대비 -1.7h), 영향 요인: 정비 표준시간표 값(-1.0h), 같은 고장 과거 평균(-0.3h), 악천후 x 표준시간(-0.2h)\n",
|
|
" - R0260 mobility/hydraulic 예상 정비 3.3h(평균 대비 -0.8h), 영향 요인: 정비 표준시간표 값(-0.3h), 악천후 x 표준시간(-0.2h), 핵심장비 x 부속 미확보(-0.2h)\n",
|
|
" - R0262 power/cooling 예상 정비 2.1h(평균 대비 -2.0h), 영향 요인: 정비 표준시간표 값(-1.3h), 악천후 x 표준시간(-0.2h), 같은 고장 과거 평균(-0.2h)\n",
|
|
" - R0261 comms/electric 예상 정비 1.4h(평균 대비 -2.7h), 영향 요인: 정비 표준시간표 값(-1.9h), 같은 고장 과거 평균(-0.3h), 악천후 x 표준시간(-0.1h)\n",
|
|
" - R0255 fires/engine 예상 정비 10.8h(평균 대비 +6.7h), 영향 요인: 정비 표준시간표 값(+4.9h), 같은 고장 과거 평균(+1.1h), 핵심장비 x 부속 미확보(+1.0h)\n",
|
|
"\n",
|
|
"경보: 22일차(훈련 주간) 부하율 1.0 초과 정비반 ['MT1', 'MT2', 'MT3', 'MT4', 'MT5', 'MT6'] -> 전 정비반 과부하 -> 상급 정비부대 지원 요청·비핵심 요청 연기 검토(정비대대장 판단)\n",
|
|
"```\n",
|
|
"</details>\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"---\n",
|
|
"\n",
|
|
"## 🎯 미션 07. 긴급 출동 순서 변경 및 정비반 재배치를 위한 정비대대장 HITL 승인 체계 설계\n",
|
|
"**소요시간**: 약 3.5시간\n",
|
|
"\n",
|
|
"### 📋 미션 배경 및 수행 지침\n",
|
|
"#### 미션 배경 및 당위성\n",
|
|
"\n",
|
|
"출동 순서 변경을 위한 정비대대장 HITL 승인 체계를 설계한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **긴급 출동 순서 변경과 정비반 재배치를 정수계획(할당 문제)으로 제안하고, 정비대대장 승인을 거쳐야만 적용되는 HITL 승인 체계를 시뮬레이터로 설계한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
|
|
"\n",
|
|
"이 문제를 해결하기 위해 본 미션에서는 **배정 주기마다 기본 계획(접수 순서·최근접)과 할당 최적화 제안을 비교해, 다르고 기대 이득이 있으면 승인 대기열에 올린다 과정을 진행하는** 실습을 체계적으로 수행합니다.\n",
|
|
"\n",
|
|
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
|
|
"\n",
|
|
"**이번 결과물:** `src/optimization/mnt10_simulator.py` 형태의 결과물을 완성한다.\n",
|
|
"\n",
|
|
"### 🔍 평가 및 검증 기준 (PASS 조건)\n",
|
|
"#### 미션 판정\n",
|
|
"\n",
|
|
"**평가 구분:** 필수\n",
|
|
"\n",
|
|
"**핵심 평가영역:** 현실적 군수 제약조건 반영도 및 시나리오별 대안 제시 적절성\n",
|
|
"\n",
|
|
"#### 미션 PASS 조건\n",
|
|
"\n",
|
|
"- 정량 기준: 최적화 적용 시 비용 15% 절감 또는 재고 회전율 20% 향상\n",
|
|
"\n",
|
|
"\n",
|
|
"- 진행 기준: 선형/정수 계획법(LP/MIP) 최적화 해 도출 및 시뮬레이션 완료\n",
|
|
"\n",
|
|
"\n",
|
|
"- 검증 방법: 과거 실적 데이터 대비 최적화 결과 비교 백테스팅\n",
|
|
"\n",
|
|
"\n",
|
|
"#### 채점자 판정 메모\n",
|
|
"\n",
|
|
"> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
|
|
"\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"\"\"\"긴급 출동 순서 변경과 정비반 재배치를 정수계획(할당 문제)으로 제안하되, 기본 계획(접수 순서·최근접 정비반)과 다르면 정비대대장 승인 대기열에 올려 승인·반려·무응답(기본 계획 유지)을 사유와 함께 해시 체인 감사 로그에 남기는 HITL 시뮬레이터를 구현한다. 기본 계획 대비 가동중단 비용 절감률과 승인 지연의 영향을 비교한다.\"\"\"\n",
|
|
"import hashlib\n",
|
|
"import json\n",
|
|
"import numpy as np\n",
|
|
"import pandas as pd\n",
|
|
"import matplotlib.pyplot as plt\n",
|
|
"from scipy.optimize import linear_sum_assignment\n",
|
|
"\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"# [교육용가상자료] 야전 이동정비반·긴급 정비 요청 생성기 (실제 부대·위치·장비·인원 정보 없음)\n",
|
|
"# 60km x 60km 가상 지역에 이동정비반 6개와 30일간의 긴급 정비 요청을 만든다.\n",
|
|
"# 장비 구분(mobility/fires/comms/power)과 고장 유형은 교육용 추상 분류다.\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"AREA_KM, DUTY_H = 60, 10.0 # 가상 작전지역 크기, 정비반 1일 기준 가용시간\n",
|
|
"FAULT_H = {\"engine\": 6.0, \"drivetrain\": 5.0, \"hydraulic\": 3.0, \"electric\": 2.0, \"comm_module\": 1.5, \"cooling\": 2.5}\n",
|
|
"FIELD_P = {\"engine\": 0.3, \"drivetrain\": 0.35, \"hydraulic\": 0.75, \"electric\": 0.9, \"comm_module\": 0.95, \"cooling\": 0.8}\n",
|
|
"EQUIP_F = {\"mobility\": 1.0, \"fires\": 1.2, \"comms\": 0.7, \"power\": 0.8}\n",
|
|
"FAULT_OF = {\"mobility\": [0.3, 0.25, 0.2, 0.1, 0.0, 0.15], \"fires\": [0.2, 0.15, 0.35, 0.15, 0.0, 0.15],\n",
|
|
" \"comms\": [0.0, 0.0, 0.0, 0.4, 0.5, 0.1], \"power\": [0.35, 0.0, 0.0, 0.35, 0.0, 0.3]}\n",
|
|
"CRIT_P = {\"mobility\": [0.45, 0.35, 0.2], \"fires\": [0.2, 0.35, 0.45], \"comms\": [0.3, 0.4, 0.3], \"power\": [0.6, 0.3, 0.1]}\n",
|
|
"HUBS = [(15, 20), (45, 25), (30, 50)] # 가상 훈련장 3곳(요청이 몰리는 지점)\n",
|
|
"\n",
|
|
"\n",
|
|
"def make_teams(seed=0):\n",
|
|
" rng = np.random.default_rng(seed)\n",
|
|
" base = [(8, 10), (30, 6), (52, 12), (10, 44), (32, 38), (54, 50)]\n",
|
|
" skills = [[\"fires\", \"mobility\"], [\"comms\", \"mobility\"], [\"fires\", \"power\"],\n",
|
|
" [\"mobility\", \"power\"], [\"comms\", \"fires\"], [\"comms\", \"mobility\", \"power\"]]\n",
|
|
" return [{\"team\": f\"MT{i + 1}\", \"x\": float(x), \"y\": float(y), \"crew\": int(rng.integers(3, 6)), \"skills\": s}\n",
|
|
" for i, ((x, y), s) in enumerate(zip(base, skills))]\n",
|
|
"\n",
|
|
"\n",
|
|
"def make_requests(n_days=30, per_day=10, seed=1, surge_days=(8, 9, 10, 11, 22, 23, 24)):\n",
|
|
" \"\"\"하루 평균 per_day건, 훈련 주간(surge_days)에는 1.6배의 긴급 정비 요청을 만든다.\"\"\"\n",
|
|
" rng = np.random.default_rng(seed)\n",
|
|
" rows = []\n",
|
|
" for d in range(n_days):\n",
|
|
" n = rng.poisson(per_day * (1.6 if d in surge_days else 1.0))\n",
|
|
" weather = rng.choice([\"clear\", \"rain\", \"snow\"], p=[0.65, 0.25, 0.1])\n",
|
|
" for _ in range(n):\n",
|
|
" if rng.random() < 0.7:\n",
|
|
" hx, hy = HUBS[rng.integers(0, 3)]\n",
|
|
" x, y = np.clip(rng.normal(hx, 6), 0, AREA_KM), np.clip(rng.normal(hy, 6), 0, AREA_KM)\n",
|
|
" else:\n",
|
|
" x, y = rng.uniform(0, AREA_KM, 2)\n",
|
|
" eq = str(rng.choice(list(EQUIP_F), p=[0.45, 0.25, 0.2, 0.1]))\n",
|
|
" fault = str(rng.choice(list(FAULT_H), p=FAULT_OF[eq]))\n",
|
|
" parts = bool(rng.random() < 0.75)\n",
|
|
" bad = weather != \"clear\"\n",
|
|
" std_h = FAULT_H[fault] * EQUIP_F[eq] * (1.35 if not parts else 1.0) * (1.15 if bad else 1.0) * rng.lognormal(0, 0.12)\n",
|
|
" rows.append({\"t_req\": d * 24 + float(np.clip(rng.normal(13, 4), 5, 23)), \"day\": d,\n",
|
|
" \"x\": float(x), \"y\": float(y), \"equip\": eq, \"fault\": fault,\n",
|
|
" \"crit\": int(rng.choice([1, 2, 3], p=CRIT_P[eq])), \"parts_ready\": parts,\n",
|
|
" \"weather\": str(weather), \"road\": \"unpaved\" if y > 30 and rng.random() < 0.7 else \"paved\",\n",
|
|
" \"field_ok\": bool(rng.random() < FIELD_P[fault] * (1.0 if parts else 0.5)),\n",
|
|
" \"std_repair_h\": round(float(std_h), 2), \"road_delay\": float(rng.lognormal(0, 0.15))})\n",
|
|
" df = pd.DataFrame(rows).sort_values(\"t_req\").reset_index(drop=True)\n",
|
|
" df.insert(0, \"req_id\", [f\"R{i:04d}\" for i in range(len(df))])\n",
|
|
" return df\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"\n",
|
|
"# ---- 출동 배정·정비 모의 공통 함수 ----\n",
|
|
"W0 = {\"crit\": 1.0, \"wait\": 0.3, \"short\": 0.2, \"balance\": 0.5} # 우선순위·배정 기본 가중치(미션 9에서 재튜닝)\n",
|
|
"\n",
|
|
"\n",
|
|
"def travel_h(team, r, actual=False):\n",
|
|
" \"\"\"도로 굴곡 1.3배, 포장 35km/h·비포장 25km/h, 악천후 20% 감속. actual=True면 실제 도로 지연을 반영.\"\"\"\n",
|
|
" # [TODO : 학생 실습 7] 긴급 출동 순서 변경 및 정비반 재배치를 위한 정비대대장 HITL 승인 체계 설계 관련 핵심 로직을 구현하세요.\n",
|
|
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
|
|
" # <<< 여기에 코드를 작성하세요 >>>\n",
|
|
" pass # 작성 후 pass를 제거하세요.\n",
|
|
"def on_site_h(team, r):\n",
|
|
" \"\"\"현장 정비시간: 4인 기준 표준시간을 투입 인원으로 환산. 현장 수리 불가면 진단·후송 조치 1시간.\"\"\"\n",
|
|
" # [TODO : 학생 실습 7] 긴급 출동 순서 변경 및 정비반 재배치를 위한 정비대대장 HITL 승인 체계 설계 관련 핵심 로직을 구현하세요.\n",
|
|
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
|
|
" # <<< 여기에 코드를 작성하세요 >>>\n",
|
|
" pass # 작성 후 pass를 제거하세요.\n",
|
|
"def priority(r, now, est_h, w):\n",
|
|
" return w[\"crit\"] * r[\"crit\"] + w[\"wait\"] * (now - r[\"t_req\"]) + w[\"short\"] / (1 + est_h)\n",
|
|
"\n",
|
|
"\n",
|
|
"def simulate(req, teams, policy=\"baseline\", w=None, cycle_h=1.0, est=None):\n",
|
|
" \"\"\"배정 주기(cycle_h)마다 대기 요청을 정렬해 이동정비반에 배정하고 출동·정비·복귀를 모의한다.\n",
|
|
" baseline: 접수 순서 + 거리 최근접 정비반 / priority: 우선순위 점수 순서 + 도착 예상시각·당일 부하 균형\"\"\"\n",
|
|
" w = w or W0\n",
|
|
" est = np.asarray(est if est is not None else req[\"std_repair_h\"].values, float)\n",
|
|
" n_days = int(req[\"day\"].max()) + 1\n",
|
|
" free = {t[\"team\"]: 0.0 for t in teams}\n",
|
|
" load = {t[\"team\"]: np.zeros(n_days + 5) for t in teams}\n",
|
|
" recs, out, pending, i, now = req.to_dict(\"records\"), [], [], 0, cycle_h\n",
|
|
" while i < len(recs) or pending:\n",
|
|
" while i < len(recs) and recs[i][\"t_req\"] <= now:\n",
|
|
" pending.append((i, recs[i])); i += 1\n",
|
|
" if policy == \"priority\":\n",
|
|
" pending.sort(key=lambda p: -priority(p[1], now, est[p[0]], w))\n",
|
|
" for k, r in pending:\n",
|
|
" cand = [t for t in teams if r[\"equip\"] in t[\"skills\"]]\n",
|
|
" day = min(int(now // 24), n_days + 4)\n",
|
|
" if policy == \"baseline\":\n",
|
|
" tm = min(cand, key=lambda t: np.hypot(t[\"x\"] - r[\"x\"], t[\"y\"] - r[\"y\"]))\n",
|
|
" else:\n",
|
|
" tm = min(cand, key=lambda t: max(free[t[\"team\"]], now) + travel_h(t, r)\n",
|
|
" + w[\"balance\"] * load[t[\"team\"]][day] / DUTY_H * 4)\n",
|
|
" backlog = max(0.0, free[tm[\"team\"]] - now)\n",
|
|
" go = max(free[tm[\"team\"]], now)\n",
|
|
" tr = travel_h(tm, r, actual=True)\n",
|
|
" work = on_site_h(tm, r)\n",
|
|
" free[tm[\"team\"]] = go + 2 * tr + work\n",
|
|
" load[tm[\"team\"]][min(int(go // 24), n_days + 4)] += 2 * tr + work\n",
|
|
" out.append({**r, \"idx\": k, \"team\": tm[\"team\"], \"dist_km\": float(np.hypot(tm[\"x\"] - r[\"x\"], tm[\"y\"] - r[\"y\"])),\n",
|
|
" \"backlog_h\": backlog, \"cycle_wait_h\": now - r[\"t_req\"], \"dispatch\": go, \"travel_h\": tr,\n",
|
|
" \"work_h\": work, \"response_h\": go + tr - r[\"t_req\"], \"downtime_h\": go + tr + work - r[\"t_req\"]})\n",
|
|
" pending, now = [], now + cycle_h\n",
|
|
" L = pd.DataFrame({k: v[:n_days] for k, v in load.items()}) / DUTY_H # 정비반별 일일 부하율\n",
|
|
" return pd.DataFrame(out).sort_values(\"idx\").reset_index(drop=True), L\n",
|
|
"\n",
|
|
"\n",
|
|
"def metrics(res, L):\n",
|
|
" c3 = res[res[\"crit\"] == 3]\n",
|
|
" return {\"평균 도착(h)\": round(res[\"response_h\"].mean(), 2), \"핵심장비 도착(h)\": round(c3[\"response_h\"].mean(), 2),\n",
|
|
" \"핵심장비 90%(h)\": round(c3[\"response_h\"].quantile(0.9), 2), \"평균 부하율\": round(float(L.values.mean()), 3),\n",
|
|
" \"부하율 편차\": round(float(L.mean().std()), 3), \"과부하 일수\": int((L > 1.0).values.sum()),\n",
|
|
" \"총 이동(h)\": round(float(res[\"travel_h\"].sum() * 2), 1)}\n",
|
|
"\n",
|
|
"CW = {1: 1.0, 2: 3.0, 3: 6.0} # [교육용가상자료] 중요도별 가동중단 1시간 비용 지수\n",
|
|
"SLOTS = 3 # 정비반당 한 배정 주기에 받을 수 있는 최대 건수\n",
|
|
"APPROVE_DELAY = 0.25 # 승인 절차로 늦어지는 출동 시각(h)\n",
|
|
"\n",
|
|
"\n",
|
|
"class Audit:\n",
|
|
" def __init__(self):\n",
|
|
" self.log = []\n",
|
|
"\n",
|
|
" def add(self, **body):\n",
|
|
" prev = self.log[-1][\"hash\"] if self.log else \"GENESIS\"\n",
|
|
" body = {\"seq\": len(self.log), **body}\n",
|
|
" self.log.append({**body, \"prev\": prev, \"hash\": hashlib.sha256((prev + json.dumps(body, sort_keys=True, ensure_ascii=False)).encode()).hexdigest()})\n",
|
|
"\n",
|
|
" def verify(self):\n",
|
|
" prev = \"GENESIS\"\n",
|
|
" for e in self.log:\n",
|
|
" b = {k: v for k, v in e.items() if k not in (\"prev\", \"hash\")}\n",
|
|
" if e[\"prev\"] != prev or hashlib.sha256((prev + json.dumps(b, sort_keys=True, ensure_ascii=False)).encode()).hexdigest() != e[\"hash\"]:\n",
|
|
" return e[\"seq\"]\n",
|
|
" prev = e[\"hash\"]\n",
|
|
" return None\n",
|
|
"\n",
|
|
"\n",
|
|
"def default_plan(pending, teams, free, now):\n",
|
|
" \"\"\"기본 계획: 접수 순서대로 최근접(정비 가능) 정비반.\"\"\"\n",
|
|
" plan, f = [], dict(free)\n",
|
|
" for k, r in pending:\n",
|
|
" tm = min([t for t in teams if r[\"equip\"] in t[\"skills\"]], key=lambda t: np.hypot(t[\"x\"] - r[\"x\"], t[\"y\"] - r[\"y\"]))\n",
|
|
" go = max(f[tm[\"team\"]], now); plan.append((k, tm)); f[tm[\"team\"]] = go + 2 * travel_h(tm, r) + on_site_h(tm, r)\n",
|
|
" return plan\n",
|
|
"\n",
|
|
"\n",
|
|
"def mip_plan(pending, teams, free, now):\n",
|
|
" \"\"\"할당 문제(정수계획): 요청 x (정비반, 순번) 비용 = 중요도 가중 x 예상 도착시각. 정비 불가 조합은 큰 비용.\"\"\"\n",
|
|
" avg = np.mean([on_site_h(teams[0], r) for _, r in pending]) + 1.0\n",
|
|
" cols = [(t, s) for t in teams for s in range(SLOTS)]\n",
|
|
" cost = np.full((len(pending), len(cols)), 1e6)\n",
|
|
" for i, (k, r) in enumerate(pending):\n",
|
|
" for j, (t, s) in enumerate(cols):\n",
|
|
" if r[\"equip\"] in t[\"skills\"]:\n",
|
|
" cost[i, j] = CW[r[\"crit\"]] * (max(free[t[\"team\"]], now) + s * avg + travel_h(t, r) - now)\n",
|
|
" ri, cj = linear_sum_assignment(cost)\n",
|
|
" order = sorted(zip(ri, cj), key=lambda p: (cols[p[1]][1], -pending[p[0]][1][\"crit\"]))\n",
|
|
" return [(pending[i][0], cols[j][0]) for i, j in order]\n",
|
|
"\n",
|
|
"\n",
|
|
"def expected_cost(plan, recs, free, now, delay=0.0):\n",
|
|
" f, c = dict(free), 0.0\n",
|
|
" for k, tm in plan:\n",
|
|
" r = recs[k]; go = max(f[tm[\"team\"]], now + delay)\n",
|
|
" c += CW[r[\"crit\"]] * (go + travel_h(tm, r) + on_site_h(tm, r) - r[\"t_req\"]); f[tm[\"team\"]] = go + 2 * travel_h(tm, r) + on_site_h(tm, r)\n",
|
|
" return c\n",
|
|
"\n",
|
|
"\n",
|
|
"def commander(prop, rng):\n",
|
|
" \"\"\"[모의 정비대대장] 기대 이득이 작거나 담당 구역을 크게 벗어나는 재배치는 반려, 10%는 30분 내 무응답.\"\"\"\n",
|
|
" if rng.random() < 0.10:\n",
|
|
" return \"무응답\", \"30분 내 응답 없음 -> 기본 계획 유지\"\n",
|
|
" if prop[\"max_km\"] > 45:\n",
|
|
" return \"반려\", f\"정비반 구역 이탈 {prop['max_km']:.0f}km(복귀·보급 곤란)\"\n",
|
|
" if prop[\"gain\"] < 1.0:\n",
|
|
" return \"반려\", f\"기대 이득 {prop['gain']:.1f} 작음 -> 현 계획 유지\"\n",
|
|
" return \"승인\", f\"핵심장비 우선, 기대 이득 {prop['gain']:.1f}\"\n",
|
|
"\n",
|
|
"\n",
|
|
"def run(req, teams, mode, seed=0, audit=None):\n",
|
|
" \"\"\"mode: default(기본 계획) / hitl(제안 -> 승인 시 적용) / auto(모든 제안 적용: 비교용 상한, 실제 운용 금지)\"\"\"\n",
|
|
" rng = np.random.default_rng(seed)\n",
|
|
" recs = req.to_dict(\"records\"); free = {t[\"team\"]: 0.0 for t in teams}\n",
|
|
" out, i, now, props = [], 0, 1.0, []\n",
|
|
" while i < len(recs):\n",
|
|
" pending = []\n",
|
|
" while i < len(recs) and recs[i][\"t_req\"] <= now:\n",
|
|
" pending.append((i, recs[i])); i += 1\n",
|
|
" if pending:\n",
|
|
" base = default_plan(pending, teams, free, now)\n",
|
|
" plan, delay = base, 0.0\n",
|
|
" if mode != \"default\":\n",
|
|
" prop = mip_plan(pending, teams, free, now)\n",
|
|
" if [(k, t[\"team\"]) for k, t in prop] != [(k, t[\"team\"]) for k, t in base]:\n",
|
|
" gain = expected_cost(base, recs, free, now) - expected_cost(prop, recs, free, now, APPROVE_DELAY)\n",
|
|
" max_km = max(np.hypot(t[\"x\"] - recs[k][\"x\"], t[\"y\"] - recs[k][\"y\"]) for k, t in prop)\n",
|
|
" info = {\"t\": round(now, 2), \"n\": len(pending), \"gain\": round(gain, 2), \"max_km\": max_km}\n",
|
|
" if gain <= 0: # 이득 없는 변경은 제안하지 않는다\n",
|
|
" pass\n",
|
|
" elif mode == \"auto\":\n",
|
|
" dec, why = \"승인\", \"비교용 전부 적용\"\n",
|
|
" else:\n",
|
|
" dec, why = commander(info, rng)\n",
|
|
" audit.add(event=\"PROPOSE\", t=info[\"t\"], n=info[\"n\"], gain=info[\"gain\"], status=\"PENDING\")\n",
|
|
" audit.add(event=\"DECIDE\", t=info[\"t\"], decision=dec, reason=why, by=\"정비대대장(모의)\")\n",
|
|
" if gain > 0:\n",
|
|
" props.append({**info, \"decision\": dec})\n",
|
|
" if gain > 0 and dec == \"승인\":\n",
|
|
" plan, delay = prop, APPROVE_DELAY\n",
|
|
" for k, tm in plan:\n",
|
|
" r = recs[k]; go = max(free[tm[\"team\"]], now + delay)\n",
|
|
" tr, work = travel_h(tm, r, actual=True), on_site_h(tm, r)\n",
|
|
" free[tm[\"team\"]] = go + 2 * tr + work\n",
|
|
" out.append({**r, \"team\": tm[\"team\"], \"response_h\": go + tr - r[\"t_req\"], \"downtime_h\": go + tr + work - r[\"t_req\"]})\n",
|
|
" now += 1.0\n",
|
|
" res = pd.DataFrame(out)\n",
|
|
" res[\"cost\"] = res[\"crit\"].map(CW) * res[\"downtime_h\"]\n",
|
|
" return res, pd.DataFrame(props)\n",
|
|
"\n",
|
|
"\n",
|
|
"teams = make_teams()\n",
|
|
"req = make_requests(seed=1)\n",
|
|
"audit = Audit()\n",
|
|
"R = {}\n",
|
|
"for mode in (\"default\", \"hitl\", \"auto\"):\n",
|
|
" R[mode] = run(req, teams, mode, seed=3, audit=audit)\n",
|
|
"rows = []\n",
|
|
"for mode, name in [(\"default\", \"기본 계획\"), (\"hitl\", \"제안 + 대대장 승인\"), (\"auto\", \"제안 전부 적용(비교용)\")]:\n",
|
|
" res = R[mode][0]\n",
|
|
" rows.append({\"방식\": name, \"가동중단 비용지수\": round(res[\"cost\"].sum()), \"핵심장비 도착(h)\": round(res.loc[res[\"crit\"] == 3, \"response_h\"].mean(), 2),\n",
|
|
" \"핵심장비 가동중단(h)\": round(res.loc[res[\"crit\"] == 3, \"downtime_h\"].mean(), 2), \"전체 도착(h)\": round(res[\"response_h\"].mean(), 2)})\n",
|
|
"T = pd.DataFrame(rows)\n",
|
|
"print(\"[교육용가상자료] 30일 모의 결과\")\n",
|
|
"print(T.to_string(index=False))\n",
|
|
"cut = 1 - T.loc[1, \"가동중단 비용지수\"] / T.loc[0, \"가동중단 비용지수\"]\n",
|
|
"print(f\"HITL 적용 가동중단 비용 절감률 {cut:.1%} (목표 15% 이상: {'달성' if cut >= 0.15 else '미달'})\")\n",
|
|
"\n",
|
|
"P = R[\"hitl\"][1]\n",
|
|
"print(f\"\\n변경 제안 {len(P)}건: \" + \", \".join(f\"{k} {v}건\" for k, v in P[\"decision\"].value_counts().items()))\n",
|
|
"rej = [e[\"reason\"] for e in audit.log if e.get(\"decision\") == \"반려\"]\n",
|
|
"print(\"반려 사유 예:\", \"; \".join([next(x for x in rej if \"구역\" in x), next(x for x in rej if \"이득\" in x)]))\n",
|
|
"n_dec = sum(e[\"event\"] == \"DECIDE\" for e in audit.log)\n",
|
|
"print(f\"제안 {len(P)}건 중 결정 기록 {n_dec}건(로깅률 {n_dec / max(len(P), 1):.0%}), 자동 승인 0건, 무응답 시 기본 계획 유지\")\n",
|
|
"print(f\"감사 로그 무결성: {'정상' if audit.verify() is None else '이상'}\", end=\", \")\n",
|
|
"audit.log[5][\"reason\"] = \"사후 수정\"\n",
|
|
"print(f\"사후 변조 시 seq {audit.verify()}에서 탐지\")\n",
|
|
"\n",
|
|
"# 민감도: 승인 지연이 길어지면 이득이 줄어든다(전시 전환 등 급증 대비)\n",
|
|
"print(\"\\n[민감도] 승인 지연(h) -> 비용 절감률\")\n",
|
|
"for d in (0.0, 0.25, 0.5, 1.0):\n",
|
|
" APPROVE_DELAY = d\n",
|
|
" r_, _ = run(req, teams, \"hitl\", seed=3, audit=Audit())\n",
|
|
" print(f\" {d:>4}: {1 - r_['cost'].sum() / T.loc[0, '가동중단 비용지수']:.1%}\")\n",
|
|
"surge = make_requests(seed=1, per_day=16)\n",
|
|
"APPROVE_DELAY = 0.25\n",
|
|
"b, _ = run(surge, teams, \"default\"); h, _ = run(surge, teams, \"hitl\", seed=3, audit=Audit())\n",
|
|
"print(f\"요청 1.6배 급증 시나리오: 비용 절감률 {1 - h['cost'].sum() / b['cost'].sum():.1%}\")\n",
|
|
"\n",
|
|
"plt.figure(figsize=(7, 4))\n",
|
|
"for mode, lab in [(\"default\", \"default plan\"), (\"hitl\", \"proposal + approval\")]:\n",
|
|
" s = R[mode][0]\n",
|
|
" plt.hist(s.loc[s[\"crit\"] == 3, \"response_h\"], bins=20, alpha=0.6, label=lab)\n",
|
|
"plt.xlabel(\"critical equipment response time (h)\"); plt.ylabel(\"count\"); plt.legend(); plt.title(\"HITL simulator: critical response\")\n",
|
|
"plt.tight_layout(); plt.show()\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"<details>\n",
|
|
"<summary><b>💡 [참고] 미션 07 예상 실행 결과 (Output)</b></summary>\n",
|
|
"\n",
|
|
"```text\n",
|
|
"[교육용가상자료] 30일 모의 결과\n",
|
|
" 방식 가동중단 비용지수 핵심장비 도착(h) 핵심장비 가동중단(h) 전체 도착(h)\n",
|
|
" 기본 계획 6495 3.13 5.82 3.37\n",
|
|
" 제안 + 대대장 승인 5254 2.10 4.71 2.29\n",
|
|
"제안 전부 적용(비교용) 5018 1.92 4.53 2.08\n",
|
|
"HITL 적용 가동중단 비용 절감률 19.1% (목표 15% 이상: 달성)\n",
|
|
"\n",
|
|
"변경 제안 103건: 승인 71건, 반려 24건, 무응답 8건\n",
|
|
"반려 사유 예: 정비반 구역 이탈 50km(복귀·보급 곤란); 기대 이득 0.5 작음 -> 현 계획 유지\n",
|
|
"제안 103건 중 결정 기록 103건(로깅률 100%), 자동 승인 0건, 무응답 시 기본 계획 유지\n",
|
|
"감사 로그 무결성: 정상, 사후 변조 시 seq 5에서 탐지\n",
|
|
"\n",
|
|
"[민감도] 승인 지연(h) -> 비용 절감률\n",
|
|
" 0.0: 20.0%\n",
|
|
" 0.25: 19.1%\n",
|
|
" 0.5: 17.6%\n",
|
|
" 1.0: 13.4%\n",
|
|
"요청 1.6배 급증 시나리오: 비용 절감률 46.5%\n",
|
|
"```\n",
|
|
"</details>\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"---\n",
|
|
"\n",
|
|
"## 🎯 미션 08. 동시 20개소 긴급 정비 요청 발생 모의 출동 스케줄링 스트레스 테스트\n",
|
|
"**소요시간**: 약 3.0시간\n",
|
|
"\n",
|
|
"### 📋 미션 배경 및 수행 지침\n",
|
|
"#### 미션 배경 및 당위성\n",
|
|
"\n",
|
|
"동시 20개소 긴급 정비 요청 발생 모의 스트레스 테스트를 수행한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **동시 20개소 긴급 정비 요청이 한꺼번에 들어오는 상황에서 출동 스케줄링과 관제 화면이 제때 동작하는지 스트레스 시험한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
|
|
"\n",
|
|
"이 문제를 해결하기 위해 본 미션에서는 **동시 20·40·60개소 요청, 정비반 1개 출동 불가, 손상 요청 데이터를 주입하고 할당 최적화로 배정 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
|
|
"\n",
|
|
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
|
|
"\n",
|
|
"**이번 결과물:** `src/dashboard/mnt10_app.py (Streamlit/Dash 등)` 형태의 결과물을 완성한다.\n",
|
|
"\n",
|
|
"### 🔍 평가 및 검증 기준 (PASS 조건)\n",
|
|
"#### 미션 판정\n",
|
|
"\n",
|
|
"**평가 구분:** 필수\n",
|
|
"\n",
|
|
"**핵심 평가영역:** 의사결정 지원에 최적화된 시각적 위계 및 사용자 경험(UX)\n",
|
|
"\n",
|
|
"#### 미션 PASS 조건\n",
|
|
"\n",
|
|
"- 정량 기준: 대시보드 로딩 시간 2초 이내, 경보 발령 정확도 95% 이상\n",
|
|
"\n",
|
|
"\n",
|
|
"- 진행 기준: 웹 기반 분석 대시보드 구동 및 실시간 필터링 기능 검증\n",
|
|
"\n",
|
|
"\n",
|
|
"- 검증 방법: 실제 군수 데이터 연동 화면 테스트 및 UI 동작 검증\n",
|
|
"\n",
|
|
"\n",
|
|
"#### 채점자 판정 메모\n",
|
|
"\n",
|
|
"> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
|
|
"\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"\"\"\"동시 20개소(최대 60개소) 긴급 정비 요청 발생, 정비반 1개 출동 불가, 손상 요청 데이터를 주입해 출동 스케줄링과 관제 대시보드를 스트레스 테스트한다. 스케줄링+화면 생성 시간, '핵심장비 4시간 내 도착 불가' 경보 정확도, 요청 유실, 손상 데이터의 수동 확인 분리와 필터 조회 응답을 검증한다.\"\"\"\n",
|
|
"import time\n",
|
|
"import numpy as np\n",
|
|
"import pandas as pd\n",
|
|
"import matplotlib\n",
|
|
"import matplotlib.pyplot as plt\n",
|
|
"from scipy.optimize import linear_sum_assignment\n",
|
|
"\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"# [교육용가상자료] 야전 이동정비반·긴급 정비 요청 생성기 (실제 부대·위치·장비·인원 정보 없음)\n",
|
|
"# 60km x 60km 가상 지역에 이동정비반 6개와 30일간의 긴급 정비 요청을 만든다.\n",
|
|
"# 장비 구분(mobility/fires/comms/power)과 고장 유형은 교육용 추상 분류다.\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"AREA_KM, DUTY_H = 60, 10.0 # 가상 작전지역 크기, 정비반 1일 기준 가용시간\n",
|
|
"FAULT_H = {\"engine\": 6.0, \"drivetrain\": 5.0, \"hydraulic\": 3.0, \"electric\": 2.0, \"comm_module\": 1.5, \"cooling\": 2.5}\n",
|
|
"FIELD_P = {\"engine\": 0.3, \"drivetrain\": 0.35, \"hydraulic\": 0.75, \"electric\": 0.9, \"comm_module\": 0.95, \"cooling\": 0.8}\n",
|
|
"EQUIP_F = {\"mobility\": 1.0, \"fires\": 1.2, \"comms\": 0.7, \"power\": 0.8}\n",
|
|
"FAULT_OF = {\"mobility\": [0.3, 0.25, 0.2, 0.1, 0.0, 0.15], \"fires\": [0.2, 0.15, 0.35, 0.15, 0.0, 0.15],\n",
|
|
" \"comms\": [0.0, 0.0, 0.0, 0.4, 0.5, 0.1], \"power\": [0.35, 0.0, 0.0, 0.35, 0.0, 0.3]}\n",
|
|
"CRIT_P = {\"mobility\": [0.45, 0.35, 0.2], \"fires\": [0.2, 0.35, 0.45], \"comms\": [0.3, 0.4, 0.3], \"power\": [0.6, 0.3, 0.1]}\n",
|
|
"HUBS = [(15, 20), (45, 25), (30, 50)] # 가상 훈련장 3곳(요청이 몰리는 지점)\n",
|
|
"\n",
|
|
"\n",
|
|
"def make_teams(seed=0):\n",
|
|
" rng = np.random.default_rng(seed)\n",
|
|
" base = [(8, 10), (30, 6), (52, 12), (10, 44), (32, 38), (54, 50)]\n",
|
|
" skills = [[\"fires\", \"mobility\"], [\"comms\", \"mobility\"], [\"fires\", \"power\"],\n",
|
|
" [\"mobility\", \"power\"], [\"comms\", \"fires\"], [\"comms\", \"mobility\", \"power\"]]\n",
|
|
" return [{\"team\": f\"MT{i + 1}\", \"x\": float(x), \"y\": float(y), \"crew\": int(rng.integers(3, 6)), \"skills\": s}\n",
|
|
" for i, ((x, y), s) in enumerate(zip(base, skills))]\n",
|
|
"\n",
|
|
"\n",
|
|
"def make_requests(n_days=30, per_day=10, seed=1, surge_days=(8, 9, 10, 11, 22, 23, 24)):\n",
|
|
" \"\"\"하루 평균 per_day건, 훈련 주간(surge_days)에는 1.6배의 긴급 정비 요청을 만든다.\"\"\"\n",
|
|
" rng = np.random.default_rng(seed)\n",
|
|
" rows = []\n",
|
|
" for d in range(n_days):\n",
|
|
" n = rng.poisson(per_day * (1.6 if d in surge_days else 1.0))\n",
|
|
" weather = rng.choice([\"clear\", \"rain\", \"snow\"], p=[0.65, 0.25, 0.1])\n",
|
|
" for _ in range(n):\n",
|
|
" if rng.random() < 0.7:\n",
|
|
" hx, hy = HUBS[rng.integers(0, 3)]\n",
|
|
" x, y = np.clip(rng.normal(hx, 6), 0, AREA_KM), np.clip(rng.normal(hy, 6), 0, AREA_KM)\n",
|
|
" else:\n",
|
|
" x, y = rng.uniform(0, AREA_KM, 2)\n",
|
|
" eq = str(rng.choice(list(EQUIP_F), p=[0.45, 0.25, 0.2, 0.1]))\n",
|
|
" fault = str(rng.choice(list(FAULT_H), p=FAULT_OF[eq]))\n",
|
|
" parts = bool(rng.random() < 0.75)\n",
|
|
" bad = weather != \"clear\"\n",
|
|
" std_h = FAULT_H[fault] * EQUIP_F[eq] * (1.35 if not parts else 1.0) * (1.15 if bad else 1.0) * rng.lognormal(0, 0.12)\n",
|
|
" rows.append({\"t_req\": d * 24 + float(np.clip(rng.normal(13, 4), 5, 23)), \"day\": d,\n",
|
|
" \"x\": float(x), \"y\": float(y), \"equip\": eq, \"fault\": fault,\n",
|
|
" \"crit\": int(rng.choice([1, 2, 3], p=CRIT_P[eq])), \"parts_ready\": parts,\n",
|
|
" \"weather\": str(weather), \"road\": \"unpaved\" if y > 30 and rng.random() < 0.7 else \"paved\",\n",
|
|
" \"field_ok\": bool(rng.random() < FIELD_P[fault] * (1.0 if parts else 0.5)),\n",
|
|
" \"std_repair_h\": round(float(std_h), 2), \"road_delay\": float(rng.lognormal(0, 0.15))})\n",
|
|
" df = pd.DataFrame(rows).sort_values(\"t_req\").reset_index(drop=True)\n",
|
|
" df.insert(0, \"req_id\", [f\"R{i:04d}\" for i in range(len(df))])\n",
|
|
" return df\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"\n",
|
|
"# ---- 출동 배정·정비 모의 공통 함수 ----\n",
|
|
"W0 = {\"crit\": 1.0, \"wait\": 0.3, \"short\": 0.2, \"balance\": 0.5} # 우선순위·배정 기본 가중치(미션 9에서 재튜닝)\n",
|
|
"\n",
|
|
"\n",
|
|
"def travel_h(team, r, actual=False):\n",
|
|
" \"\"\"도로 굴곡 1.3배, 포장 35km/h·비포장 25km/h, 악천후 20% 감속. actual=True면 실제 도로 지연을 반영.\"\"\"\n",
|
|
" # [TODO : 학생 실습 8] 동시 20개소 긴급 정비 요청 발생 모의 출동 스케줄링 스트레스 테스트 관련 핵심 로직을 구현하세요.\n",
|
|
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
|
|
" # <<< 여기에 코드를 작성하세요 >>>\n",
|
|
" pass # 작성 후 pass를 제거하세요.\n",
|
|
"def on_site_h(team, r):\n",
|
|
" \"\"\"현장 정비시간: 4인 기준 표준시간을 투입 인원으로 환산. 현장 수리 불가면 진단·후송 조치 1시간.\"\"\"\n",
|
|
" # [TODO : 학생 실습 8] 동시 20개소 긴급 정비 요청 발생 모의 출동 스케줄링 스트레스 테스트 관련 핵심 로직을 구현하세요.\n",
|
|
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
|
|
" # <<< 여기에 코드를 작성하세요 >>>\n",
|
|
" pass # 작성 후 pass를 제거하세요.\n",
|
|
"def priority(r, now, est_h, w):\n",
|
|
" return w[\"crit\"] * r[\"crit\"] + w[\"wait\"] * (now - r[\"t_req\"]) + w[\"short\"] / (1 + est_h)\n",
|
|
"\n",
|
|
"\n",
|
|
"def simulate(req, teams, policy=\"baseline\", w=None, cycle_h=1.0, est=None):\n",
|
|
" \"\"\"배정 주기(cycle_h)마다 대기 요청을 정렬해 이동정비반에 배정하고 출동·정비·복귀를 모의한다.\n",
|
|
" baseline: 접수 순서 + 거리 최근접 정비반 / priority: 우선순위 점수 순서 + 도착 예상시각·당일 부하 균형\"\"\"\n",
|
|
" w = w or W0\n",
|
|
" est = np.asarray(est if est is not None else req[\"std_repair_h\"].values, float)\n",
|
|
" n_days = int(req[\"day\"].max()) + 1\n",
|
|
" free = {t[\"team\"]: 0.0 for t in teams}\n",
|
|
" load = {t[\"team\"]: np.zeros(n_days + 5) for t in teams}\n",
|
|
" recs, out, pending, i, now = req.to_dict(\"records\"), [], [], 0, cycle_h\n",
|
|
" while i < len(recs) or pending:\n",
|
|
" while i < len(recs) and recs[i][\"t_req\"] <= now:\n",
|
|
" pending.append((i, recs[i])); i += 1\n",
|
|
" if policy == \"priority\":\n",
|
|
" pending.sort(key=lambda p: -priority(p[1], now, est[p[0]], w))\n",
|
|
" for k, r in pending:\n",
|
|
" cand = [t for t in teams if r[\"equip\"] in t[\"skills\"]]\n",
|
|
" day = min(int(now // 24), n_days + 4)\n",
|
|
" if policy == \"baseline\":\n",
|
|
" tm = min(cand, key=lambda t: np.hypot(t[\"x\"] - r[\"x\"], t[\"y\"] - r[\"y\"]))\n",
|
|
" else:\n",
|
|
" tm = min(cand, key=lambda t: max(free[t[\"team\"]], now) + travel_h(t, r)\n",
|
|
" + w[\"balance\"] * load[t[\"team\"]][day] / DUTY_H * 4)\n",
|
|
" backlog = max(0.0, free[tm[\"team\"]] - now)\n",
|
|
" go = max(free[tm[\"team\"]], now)\n",
|
|
" tr = travel_h(tm, r, actual=True)\n",
|
|
" work = on_site_h(tm, r)\n",
|
|
" free[tm[\"team\"]] = go + 2 * tr + work\n",
|
|
" load[tm[\"team\"]][min(int(go // 24), n_days + 4)] += 2 * tr + work\n",
|
|
" out.append({**r, \"idx\": k, \"team\": tm[\"team\"], \"dist_km\": float(np.hypot(tm[\"x\"] - r[\"x\"], tm[\"y\"] - r[\"y\"])),\n",
|
|
" \"backlog_h\": backlog, \"cycle_wait_h\": now - r[\"t_req\"], \"dispatch\": go, \"travel_h\": tr,\n",
|
|
" \"work_h\": work, \"response_h\": go + tr - r[\"t_req\"], \"downtime_h\": go + tr + work - r[\"t_req\"]})\n",
|
|
" pending, now = [], now + cycle_h\n",
|
|
" L = pd.DataFrame({k: v[:n_days] for k, v in load.items()}) / DUTY_H # 정비반별 일일 부하율\n",
|
|
" return pd.DataFrame(out).sort_values(\"idx\").reset_index(drop=True), L\n",
|
|
"\n",
|
|
"\n",
|
|
"def metrics(res, L):\n",
|
|
" c3 = res[res[\"crit\"] == 3]\n",
|
|
" return {\"평균 도착(h)\": round(res[\"response_h\"].mean(), 2), \"핵심장비 도착(h)\": round(c3[\"response_h\"].mean(), 2),\n",
|
|
" \"핵심장비 90%(h)\": round(c3[\"response_h\"].quantile(0.9), 2), \"평균 부하율\": round(float(L.values.mean()), 3),\n",
|
|
" \"부하율 편차\": round(float(L.mean().std()), 3), \"과부하 일수\": int((L > 1.0).values.sum()),\n",
|
|
" \"총 이동(h)\": round(float(res[\"travel_h\"].sum() * 2), 1)}\n",
|
|
"\n",
|
|
"CW = {1: 1.0, 2: 3.0, 3: 6.0}\n",
|
|
"ALERT_H = 4.0 # 핵심장비(crit 3) 도착 목표시간\n",
|
|
"T0 = 15 * 24 + 10.0 # 15일차 10시 동시 다발 요청 발생\n",
|
|
"\n",
|
|
"\n",
|
|
"def make_burst(n, seed):\n",
|
|
" \"\"\"[교육용가상자료] 동시 n개소 요청(같은 시각 접수). 일부 정비반은 기존 작업으로 바쁜 상태.\"\"\"\n",
|
|
" pool = make_requests(n_days=10, seed=seed)\n",
|
|
" b = pool.sample(n, random_state=seed).reset_index(drop=True).astype({\"x\": float, \"crit\": object})\n",
|
|
" b[\"t_req\"] = T0; b[\"req_id\"] = [f\"B{seed}-{i:02d}\" for i in range(n)]\n",
|
|
" return b\n",
|
|
"\n",
|
|
"\n",
|
|
"def validate(r):\n",
|
|
" errs = []\n",
|
|
" if not (0 <= r[\"x\"] <= AREA_KM and 0 <= r[\"y\"] <= AREA_KM):\n",
|
|
" errs.append(\"좌표 범위 오류\")\n",
|
|
" if r[\"equip\"] not in EQUIP_F or r[\"fault\"] not in FAULT_H:\n",
|
|
" errs.append(\"장비/고장 코드 오류\")\n",
|
|
" if r[\"crit\"] not in (1, 2, 3):\n",
|
|
" errs.append(\"중요도 누락\")\n",
|
|
" return errs\n",
|
|
"\n",
|
|
"\n",
|
|
"def schedule(burst, teams, busy_until):\n",
|
|
" \"\"\"할당 문제로 동시 요청을 (정비반, 순번)에 배정. 비용 = 중요도 가중 x 예상 도착시각.\"\"\"\n",
|
|
" slots = int(np.ceil(len(burst) / len(teams))) + 2\n",
|
|
" cols = [(t, s) for t in teams for s in range(slots)]\n",
|
|
" avg = 4.0\n",
|
|
" cost = np.full((len(burst), len(cols)), 1e6)\n",
|
|
" for i, r in enumerate(burst.to_dict(\"records\")):\n",
|
|
" for j, (t, s) in enumerate(cols):\n",
|
|
" if r[\"equip\"] in t[\"skills\"]:\n",
|
|
" cost[i, j] = CW[r[\"crit\"]] * (busy_until[t[\"team\"]] - T0 + s * avg + travel_h(t, r))\n",
|
|
" ri, cj = linear_sum_assignment(cost)\n",
|
|
" plan = sorted(((i, cols[j][0], cols[j][1]) for i, j in zip(ri, cj)), key=lambda p: (p[1][\"team\"], p[2]))\n",
|
|
" free, rows = dict(busy_until), []\n",
|
|
" for i, t, s in plan:\n",
|
|
" r = burst.iloc[i]\n",
|
|
" go = max(free[t[\"team\"]], T0)\n",
|
|
" est = go + travel_h(t, r) - T0\n",
|
|
" act_tr = travel_h(t, r, actual=True)\n",
|
|
" rows.append({\"req_id\": r[\"req_id\"], \"crit\": r[\"crit\"], \"equip\": r[\"equip\"], \"team\": t[\"team\"], \"eta_h\": est,\n",
|
|
" \"actual_h\": go + act_tr - T0, \"alert\": r[\"crit\"] == 3 and est > ALERT_H})\n",
|
|
" free[t[\"team\"]] = go + 2 * act_tr + on_site_h(t, r) # 실제로는 실측 이동시간대로 다음 출동이 정해진다\n",
|
|
" return pd.DataFrame(rows)\n",
|
|
"\n",
|
|
"\n",
|
|
"def dashboard(plan, teams):\n",
|
|
" fig, ax = plt.subplots(1, 3, figsize=(14, 4))\n",
|
|
" ld = plan.groupby(\"team\").size().reindex([t[\"team\"] for t in teams], fill_value=0)\n",
|
|
" ax[0].bar(ld.index, ld.values); ax[0].set_title(\"Assigned requests per team\")\n",
|
|
" ax[1].scatter(plan[\"eta_h\"], plan[\"crit\"], c=np.where(plan[\"alert\"], \"red\", \"gray\")); ax[1].axvline(ALERT_H, ls=\"--\")\n",
|
|
" ax[1].set_xlabel(\"ETA (h)\"); ax[1].set_yticks([1, 2, 3]); ax[1].set_title(\"ETA by criticality (red = alert)\")\n",
|
|
" ax[2].hist(plan[\"eta_h\"], bins=15); ax[2].set_title(\"ETA distribution\")\n",
|
|
" fig.tight_layout(); fig.canvas.draw()\n",
|
|
" return fig\n",
|
|
"\n",
|
|
"\n",
|
|
"teams_all = make_teams()\n",
|
|
"rng = np.random.default_rng(4)\n",
|
|
"busy0 = {t[\"team\"]: T0 + float(rng.uniform(0, 3)) for t in teams_all} # 기존 작업 잔여시간\n",
|
|
"rows, keep = [], {}\n",
|
|
"for n, down, bad in [(20, None, 0), (20, \"MT5\", 3), (40, \"MT5\", 3), (60, \"MT5\", 5)]:\n",
|
|
" burst = make_burst(n, seed=n)\n",
|
|
" for k in range(bad): # 손상 데이터 주입\n",
|
|
" burst.loc[k * 3, [\"x\", \"equip\", \"crit\"][k % 3]] = [999.0, \"unknown\", 0][k % 3]\n",
|
|
" teams = [t for t in teams_all if t[\"team\"] != down]\n",
|
|
" t0 = time.perf_counter()\n",
|
|
" ok = burst[[not validate(r) for r in burst.to_dict(\"records\")]]\n",
|
|
" manual = burst.drop(ok.index)\n",
|
|
" plan = schedule(ok, teams, {t[\"team\"]: busy0[t[\"team\"]] for t in teams})\n",
|
|
" fig = dashboard(plan, teams)\n",
|
|
" sec = time.perf_counter() - t0\n",
|
|
" truth = (plan[\"crit\"] == 3) & (plan[\"actual_h\"] > ALERT_H)\n",
|
|
" acc = float((plan[\"alert\"] == truth).mean())\n",
|
|
" lost = n - len(plan) - len(manual)\n",
|
|
" rows.append({\"상황\": f\"{n}개소\" + (f\", {down} 불가\" if down else \"\") + (f\", 손상 {bad}건\" if bad else \"\"), \"배정\": len(plan), \"수동 확인\": len(manual),\n",
|
|
" \"유실\": lost, \"처리+화면(초)\": round(sec, 2), \"핵심장비 경보\": int(plan[\"alert\"].sum()), \"경보 정확도\": round(acc, 3),\n",
|
|
" \"핵심 평균 도착(h)\": round(plan.loc[plan[\"crit\"] == 3, \"eta_h\"].mean(), 2)})\n",
|
|
" keep[n, down] = (fig, plan)\n",
|
|
" if (n, down) != (60, \"MT5\"):\n",
|
|
" plt.close(fig)\n",
|
|
"T = pd.DataFrame(rows)\n",
|
|
"print(\"[교육용가상자료] 동시 다발 긴급 정비 요청 스트레스 테스트\")\n",
|
|
"print(T.to_string(index=False))\n",
|
|
"m = T.iloc[1]\n",
|
|
"print(f\"\\n동시 20개소(정비반 1개 불가·손상 3건): 처리+화면 {m['처리+화면(초)']}초 (목표 2초 이내: {'달성' if T['처리+화면(초)'].max() <= 2 else '미달'}), \"\n",
|
|
" f\"경보 정확도 {m['경보 정확도']:.1%} (목표 95% 이상: {'달성' if T['경보 정확도'].min() >= 0.95 else '미달'}), 유실 {int(T['유실'].sum())}건\")\n",
|
|
"print(\"손상 요청은 삭제하지 않고 '수동 확인' 목록으로 넘겨 관제병이 전화·무전으로 재확인한다.\")\n",
|
|
"\n",
|
|
"plan = keep[20, \"MT5\"][1]\n",
|
|
"qt = []\n",
|
|
"for c in (1, 2, 3):\n",
|
|
" for e in EQUIP_F:\n",
|
|
" t0 = time.perf_counter(); plan[(plan[\"crit\"] == c) & (plan[\"equip\"] == e)]; qt.append((time.perf_counter() - t0) * 1000)\n",
|
|
"print(f\"실시간 필터(중요도 x 장비 12조합) 평균 {np.mean(qt):.2f} ms\")\n",
|
|
"plan = keep[40, \"MT5\"][1]\n",
|
|
"miss = plan[(plan[\"crit\"] == 3) & (plan[\"actual_h\"] > ALERT_H) & ~plan[\"alert\"]]\n",
|
|
"print(f\"40개소 상황: 놓친 경보 {len(miss)}건, 헛경보 {int((plan['alert'] & ~((plan['crit'] == 3) & (plan['actual_h'] > ALERT_H))).sum())}건\"\n",
|
|
" \" -> 도로 지연(비포장·기상)을 ETA에 반영하면 줄일 수 있다.\")\n",
|
|
"print(\"기능 축소 운용: 처리 시간이 2초를 넘으면 할당 최적화 대신 '중요도 순 + 최근접' 규칙으로 즉시 배정하고 최적화는 뒤에서 재계산한다.\")\n",
|
|
"plt.show()\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"<details>\n",
|
|
"<summary><b>💡 [참고] 미션 08 예상 실행 결과 (Output)</b></summary>\n",
|
|
"\n",
|
|
"```text\n",
|
|
"[교육용가상자료] 동시 다발 긴급 정비 요청 스트레스 테스트\n",
|
|
" 상황 배정 수동 확인 유실 처리+화면(초) 핵심장비 경보 경보 정확도 핵심 평균 도착(h)\n",
|
|
" 20개소 20 0 0 0.24 1 1.000 3.17\n",
|
|
"20개소, MT5 불가, 손상 3건 17 3 0 0.23 3 1.000 4.56\n",
|
|
"40개소, MT5 불가, 손상 3건 37 3 0 0.23 3 0.973 4.55\n",
|
|
"60개소, MT5 불가, 손상 5건 55 5 0 0.23 6 1.000 5.64\n",
|
|
"\n",
|
|
"동시 20개소(정비반 1개 불가·손상 3건): 처리+화면 0.23초 (목표 2초 이내: 달성), 경보 정확도 100.0% (목표 95% 이상: 달성), 유실 0건\n",
|
|
"손상 요청은 삭제하지 않고 '수동 확인' 목록으로 넘겨 관제병이 전화·무전으로 재확인한다.\n",
|
|
"실시간 필터(중요도 x 장비 12조합) 평균 0.69 ms\n",
|
|
"40개소 상황: 놓친 경보 0건, 헛경보 1건 -> 도로 지연(비포장·기상)을 ETA에 반영하면 줄일 수 있다.\n",
|
|
"기능 축소 운용: 처리 시간이 2초를 넘으면 할당 최적화 대신 '중요도 순 + 최근접' 규칙으로 즉시 배정하고 최적화는 뒤에서 재계산한다.\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",
|
|
"이 문제를 해결하기 위해 본 미션에서는 **핵심장비 도착 4시간 초과 사례를 정비반 부하 집중·원거리 배정·도로·기상 지연·배정 주기 대기로 분류 수행하고 검증하는** 실습을 체계적으로 수행합니다.\n",
|
|
"\n",
|
|
"학습자는 실습가이드에 제시된 총 4개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
|
|
"\n",
|
|
"**이번 결과물:** `reports/mnt10_roi_analysis.md` 형태의 결과물을 완성한다.\n",
|
|
"\n",
|
|
"### 🔍 평가 및 검증 기준 (PASS 조건)\n",
|
|
"#### 미션 판정\n",
|
|
"\n",
|
|
"**평가 구분:** 필수\n",
|
|
"\n",
|
|
"**핵심 평가영역:** 추정 산식의 객관성 및 군 재정/군수 기획 부서 설득력\n",
|
|
"\n",
|
|
"#### 미션 PASS 조건\n",
|
|
"\n",
|
|
"- 정량 기준: 비용 절감액, 시간 단축률 등 3대 정량 효과 지표 산출\n",
|
|
"\n",
|
|
"\n",
|
|
"- 진행 기준: ROI 분석 보고서 완성 및 주요 지표 민감도 분석 수록\n",
|
|
"\n",
|
|
"\n",
|
|
"- 검증 방법: 회계/기획 기준 모델과의 비교 검증\n",
|
|
"\n",
|
|
"\n",
|
|
"#### 채점자 판정 메모\n",
|
|
"\n",
|
|
"> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
|
|
"\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"\"\"\"출동 우선순위 배정 결과를 사후검토(AAR)해 핵심장비 지연 사례의 원인(정비반 부하 집중·원거리 배정·도로·기상 지연 등)을 분류하고, 과거 기간으로 우선순위·배정 가중치 매트릭스를 재튜닝해 평가 기간에 검증한다. 가동중단 비용·핵심장비 도착시간·정비인력 초과근무의 3대 정량 효과 지표와 ROI 요약을 출력한다.\"\"\"\n",
|
|
"import itertools\n",
|
|
"import numpy as np\n",
|
|
"import pandas as pd\n",
|
|
"import matplotlib.pyplot as plt\n",
|
|
"from sklearn.ensemble import GradientBoostingRegressor\n",
|
|
"\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"# [교육용가상자료] 야전 이동정비반·긴급 정비 요청 생성기 (실제 부대·위치·장비·인원 정보 없음)\n",
|
|
"# 60km x 60km 가상 지역에 이동정비반 6개와 30일간의 긴급 정비 요청을 만든다.\n",
|
|
"# 장비 구분(mobility/fires/comms/power)과 고장 유형은 교육용 추상 분류다.\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"AREA_KM, DUTY_H = 60, 10.0 # 가상 작전지역 크기, 정비반 1일 기준 가용시간\n",
|
|
"FAULT_H = {\"engine\": 6.0, \"drivetrain\": 5.0, \"hydraulic\": 3.0, \"electric\": 2.0, \"comm_module\": 1.5, \"cooling\": 2.5}\n",
|
|
"FIELD_P = {\"engine\": 0.3, \"drivetrain\": 0.35, \"hydraulic\": 0.75, \"electric\": 0.9, \"comm_module\": 0.95, \"cooling\": 0.8}\n",
|
|
"EQUIP_F = {\"mobility\": 1.0, \"fires\": 1.2, \"comms\": 0.7, \"power\": 0.8}\n",
|
|
"FAULT_OF = {\"mobility\": [0.3, 0.25, 0.2, 0.1, 0.0, 0.15], \"fires\": [0.2, 0.15, 0.35, 0.15, 0.0, 0.15],\n",
|
|
" \"comms\": [0.0, 0.0, 0.0, 0.4, 0.5, 0.1], \"power\": [0.35, 0.0, 0.0, 0.35, 0.0, 0.3]}\n",
|
|
"CRIT_P = {\"mobility\": [0.45, 0.35, 0.2], \"fires\": [0.2, 0.35, 0.45], \"comms\": [0.3, 0.4, 0.3], \"power\": [0.6, 0.3, 0.1]}\n",
|
|
"HUBS = [(15, 20), (45, 25), (30, 50)] # 가상 훈련장 3곳(요청이 몰리는 지점)\n",
|
|
"\n",
|
|
"\n",
|
|
"def make_teams(seed=0):\n",
|
|
" rng = np.random.default_rng(seed)\n",
|
|
" base = [(8, 10), (30, 6), (52, 12), (10, 44), (32, 38), (54, 50)]\n",
|
|
" skills = [[\"fires\", \"mobility\"], [\"comms\", \"mobility\"], [\"fires\", \"power\"],\n",
|
|
" [\"mobility\", \"power\"], [\"comms\", \"fires\"], [\"comms\", \"mobility\", \"power\"]]\n",
|
|
" return [{\"team\": f\"MT{i + 1}\", \"x\": float(x), \"y\": float(y), \"crew\": int(rng.integers(3, 6)), \"skills\": s}\n",
|
|
" for i, ((x, y), s) in enumerate(zip(base, skills))]\n",
|
|
"\n",
|
|
"\n",
|
|
"def make_requests(n_days=30, per_day=10, seed=1, surge_days=(8, 9, 10, 11, 22, 23, 24)):\n",
|
|
" \"\"\"하루 평균 per_day건, 훈련 주간(surge_days)에는 1.6배의 긴급 정비 요청을 만든다.\"\"\"\n",
|
|
" rng = np.random.default_rng(seed)\n",
|
|
" rows = []\n",
|
|
" for d in range(n_days):\n",
|
|
" n = rng.poisson(per_day * (1.6 if d in surge_days else 1.0))\n",
|
|
" weather = rng.choice([\"clear\", \"rain\", \"snow\"], p=[0.65, 0.25, 0.1])\n",
|
|
" for _ in range(n):\n",
|
|
" if rng.random() < 0.7:\n",
|
|
" hx, hy = HUBS[rng.integers(0, 3)]\n",
|
|
" x, y = np.clip(rng.normal(hx, 6), 0, AREA_KM), np.clip(rng.normal(hy, 6), 0, AREA_KM)\n",
|
|
" else:\n",
|
|
" x, y = rng.uniform(0, AREA_KM, 2)\n",
|
|
" eq = str(rng.choice(list(EQUIP_F), p=[0.45, 0.25, 0.2, 0.1]))\n",
|
|
" fault = str(rng.choice(list(FAULT_H), p=FAULT_OF[eq]))\n",
|
|
" parts = bool(rng.random() < 0.75)\n",
|
|
" bad = weather != \"clear\"\n",
|
|
" std_h = FAULT_H[fault] * EQUIP_F[eq] * (1.35 if not parts else 1.0) * (1.15 if bad else 1.0) * rng.lognormal(0, 0.12)\n",
|
|
" rows.append({\"t_req\": d * 24 + float(np.clip(rng.normal(13, 4), 5, 23)), \"day\": d,\n",
|
|
" \"x\": float(x), \"y\": float(y), \"equip\": eq, \"fault\": fault,\n",
|
|
" \"crit\": int(rng.choice([1, 2, 3], p=CRIT_P[eq])), \"parts_ready\": parts,\n",
|
|
" \"weather\": str(weather), \"road\": \"unpaved\" if y > 30 and rng.random() < 0.7 else \"paved\",\n",
|
|
" \"field_ok\": bool(rng.random() < FIELD_P[fault] * (1.0 if parts else 0.5)),\n",
|
|
" \"std_repair_h\": round(float(std_h), 2), \"road_delay\": float(rng.lognormal(0, 0.15))})\n",
|
|
" df = pd.DataFrame(rows).sort_values(\"t_req\").reset_index(drop=True)\n",
|
|
" df.insert(0, \"req_id\", [f\"R{i:04d}\" for i in range(len(df))])\n",
|
|
" return df\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"\n",
|
|
"# ---- 출동 배정·정비 모의 공통 함수 ----\n",
|
|
"W0 = {\"crit\": 1.0, \"wait\": 0.3, \"short\": 0.2, \"balance\": 0.5} # 우선순위·배정 기본 가중치(미션 9에서 재튜닝)\n",
|
|
"\n",
|
|
"\n",
|
|
"def travel_h(team, r, actual=False):\n",
|
|
" \"\"\"도로 굴곡 1.3배, 포장 35km/h·비포장 25km/h, 악천후 20% 감속. actual=True면 실제 도로 지연을 반영.\"\"\"\n",
|
|
" # [TODO : 학생 실습 9] AAR 기반 출동 우선순위 배정 오류 분석 및 가중치 매트릭스 재튜닝 관련 핵심 로직을 구현하세요.\n",
|
|
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
|
|
" # <<< 여기에 코드를 작성하세요 >>>\n",
|
|
" pass # 작성 후 pass를 제거하세요.\n",
|
|
"def on_site_h(team, r):\n",
|
|
" \"\"\"현장 정비시간: 4인 기준 표준시간을 투입 인원으로 환산. 현장 수리 불가면 진단·후송 조치 1시간.\"\"\"\n",
|
|
" # [TODO : 학생 실습 9] AAR 기반 출동 우선순위 배정 오류 분석 및 가중치 매트릭스 재튜닝 관련 핵심 로직을 구현하세요.\n",
|
|
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
|
|
" # <<< 여기에 코드를 작성하세요 >>>\n",
|
|
" pass # 작성 후 pass를 제거하세요.\n",
|
|
"def priority(r, now, est_h, w):\n",
|
|
" return w[\"crit\"] * r[\"crit\"] + w[\"wait\"] * (now - r[\"t_req\"]) + w[\"short\"] / (1 + est_h)\n",
|
|
"\n",
|
|
"\n",
|
|
"def simulate(req, teams, policy=\"baseline\", w=None, cycle_h=1.0, est=None):\n",
|
|
" \"\"\"배정 주기(cycle_h)마다 대기 요청을 정렬해 이동정비반에 배정하고 출동·정비·복귀를 모의한다.\n",
|
|
" baseline: 접수 순서 + 거리 최근접 정비반 / priority: 우선순위 점수 순서 + 도착 예상시각·당일 부하 균형\"\"\"\n",
|
|
" w = w or W0\n",
|
|
" est = np.asarray(est if est is not None else req[\"std_repair_h\"].values, float)\n",
|
|
" n_days = int(req[\"day\"].max()) + 1\n",
|
|
" free = {t[\"team\"]: 0.0 for t in teams}\n",
|
|
" load = {t[\"team\"]: np.zeros(n_days + 5) for t in teams}\n",
|
|
" recs, out, pending, i, now = req.to_dict(\"records\"), [], [], 0, cycle_h\n",
|
|
" while i < len(recs) or pending:\n",
|
|
" while i < len(recs) and recs[i][\"t_req\"] <= now:\n",
|
|
" pending.append((i, recs[i])); i += 1\n",
|
|
" if policy == \"priority\":\n",
|
|
" pending.sort(key=lambda p: -priority(p[1], now, est[p[0]], w))\n",
|
|
" for k, r in pending:\n",
|
|
" cand = [t for t in teams if r[\"equip\"] in t[\"skills\"]]\n",
|
|
" day = min(int(now // 24), n_days + 4)\n",
|
|
" if policy == \"baseline\":\n",
|
|
" tm = min(cand, key=lambda t: np.hypot(t[\"x\"] - r[\"x\"], t[\"y\"] - r[\"y\"]))\n",
|
|
" else:\n",
|
|
" tm = min(cand, key=lambda t: max(free[t[\"team\"]], now) + travel_h(t, r)\n",
|
|
" + w[\"balance\"] * load[t[\"team\"]][day] / DUTY_H * 4)\n",
|
|
" backlog = max(0.0, free[tm[\"team\"]] - now)\n",
|
|
" go = max(free[tm[\"team\"]], now)\n",
|
|
" tr = travel_h(tm, r, actual=True)\n",
|
|
" work = on_site_h(tm, r)\n",
|
|
" free[tm[\"team\"]] = go + 2 * tr + work\n",
|
|
" load[tm[\"team\"]][min(int(go // 24), n_days + 4)] += 2 * tr + work\n",
|
|
" out.append({**r, \"idx\": k, \"team\": tm[\"team\"], \"dist_km\": float(np.hypot(tm[\"x\"] - r[\"x\"], tm[\"y\"] - r[\"y\"])),\n",
|
|
" \"backlog_h\": backlog, \"cycle_wait_h\": now - r[\"t_req\"], \"dispatch\": go, \"travel_h\": tr,\n",
|
|
" \"work_h\": work, \"response_h\": go + tr - r[\"t_req\"], \"downtime_h\": go + tr + work - r[\"t_req\"]})\n",
|
|
" pending, now = [], now + cycle_h\n",
|
|
" L = pd.DataFrame({k: v[:n_days] for k, v in load.items()}) / DUTY_H # 정비반별 일일 부하율\n",
|
|
" return pd.DataFrame(out).sort_values(\"idx\").reset_index(drop=True), L\n",
|
|
"\n",
|
|
"\n",
|
|
"def metrics(res, L):\n",
|
|
" c3 = res[res[\"crit\"] == 3]\n",
|
|
" return {\"평균 도착(h)\": round(res[\"response_h\"].mean(), 2), \"핵심장비 도착(h)\": round(c3[\"response_h\"].mean(), 2),\n",
|
|
" \"핵심장비 90%(h)\": round(c3[\"response_h\"].quantile(0.9), 2), \"평균 부하율\": round(float(L.values.mean()), 3),\n",
|
|
" \"부하율 편차\": round(float(L.mean().std()), 3), \"과부하 일수\": int((L > 1.0).values.sum()),\n",
|
|
" \"총 이동(h)\": round(float(res[\"travel_h\"].sum() * 2), 1)}\n",
|
|
"\n",
|
|
"# ---- 접수 시점 파생 피처(미션 4와 동일) ----\n",
|
|
"def make_features(req, teams):\n",
|
|
" \"\"\"접수 시점에 알 수 있는 정보만으로 파생 피처를 만든다(미래 정보 누설 방지).\"\"\"\n",
|
|
" X = pd.DataFrame(index=req.index)\n",
|
|
" X[\"hour\"] = req[\"t_req\"] % 24\n",
|
|
" X[\"is_night\"] = (X[\"hour\"] >= 19).astype(int)\n",
|
|
" X[\"dow\"] = req[\"day\"] % 7\n",
|
|
" X[\"weather_bad\"] = (req[\"weather\"] != \"clear\").astype(int)\n",
|
|
" X[\"unpaved\"] = (req[\"road\"] == \"unpaved\").astype(int)\n",
|
|
" X[\"no_parts\"] = (~req[\"parts_ready\"]).astype(int)\n",
|
|
" X[\"std_table_h\"] = req[\"fault\"].map(FAULT_H) * req[\"equip\"].map(EQUIP_F) # 정비 표준시간표 값\n",
|
|
" X[\"crit_x_noparts\"] = req[\"crit\"] * X[\"no_parts\"]\n",
|
|
" X[\"bad_x_table\"] = X[\"weather_bad\"] * X[\"std_table_h\"]\n",
|
|
" X[\"near_team_km\"] = [min(np.hypot(t[\"x\"] - x, t[\"y\"] - y) for t in teams if e in t[\"skills\"]) for x, y, e in zip(req[\"x\"], req[\"y\"], req[\"equip\"])]\n",
|
|
" X[\"hub_km\"] = [min(np.hypot(hx - x, hy - y) for hx, hy in HUBS) for x, y in zip(req[\"x\"], req[\"y\"])]\n",
|
|
" X[\"req_24h\"] = [((req[\"t_req\"] > t - 24) & (req[\"t_req\"] <= t)).sum() for t in req[\"t_req\"]]\n",
|
|
" X[\"fault_past_mean\"] = req.groupby(\"fault\")[\"std_repair_h\"].transform(lambda s: s.shift().expanding().mean()).fillna(req[\"std_repair_h\"].mean())\n",
|
|
" X[\"crit\"] = req[\"crit\"]\n",
|
|
" for e in EQUIP_F:\n",
|
|
" X[f\"eq_{e}\"] = (req[\"equip\"] == e).astype(int)\n",
|
|
" return X\n",
|
|
"\n",
|
|
"\n",
|
|
"def fit_repair_model(hist, teams):\n",
|
|
" \"\"\"과거 요청으로 정비 소요시간(4인 기준) 예측 모델을 학습한다.\"\"\"\n",
|
|
" return GradientBoostingRegressor(n_estimators=200, max_depth=3, learning_rate=0.05, random_state=0).fit(make_features(hist, teams), hist[\"std_repair_h\"])\n",
|
|
"\n",
|
|
"CW = {1: 1.0, 2: 3.0, 3: 6.0} # 중요도별 가동중단 비용 지수\n",
|
|
"PRICE = {\"downtime\": 10, \"move\": 5, \"overtime\": 2} # [교육용가상자료] 가상 단가(만원): 비용지수 1, 차량 운행 1h, 초과근무 1인·h\n",
|
|
"TARGET_H = 4.0\n",
|
|
"\n",
|
|
"\n",
|
|
"def evaluate(res, L, teams):\n",
|
|
" crew = pd.Series({t[\"team\"]: t[\"crew\"] for t in teams})\n",
|
|
" over = ((L - 1.0).clip(lower=0) * DUTY_H * crew).values.sum() # 초과근무 인·시간\n",
|
|
" return {\"가동중단 비용지수\": float((res[\"crit\"].map(CW) * res[\"downtime_h\"]).sum()),\n",
|
|
" \"핵심장비 도착(h)\": float(res.loc[res[\"crit\"] == 3, \"response_h\"].mean()),\n",
|
|
" \"초과근무(인·h)\": float(over), \"차량 운행(h)\": float(res[\"travel_h\"].sum() * 2),\n",
|
|
" \"핵심 4h 초과(건)\": int(((res[\"crit\"] == 3) & (res[\"response_h\"] > TARGET_H)).sum())}\n",
|
|
"\n",
|
|
"\n",
|
|
"def won(m):\n",
|
|
" return m[\"가동중단 비용지수\"] * PRICE[\"downtime\"] + m[\"차량 운행(h)\"] * PRICE[\"move\"] + m[\"초과근무(인·h)\"] * PRICE[\"overtime\"]\n",
|
|
"\n",
|
|
"\n",
|
|
"def objective(m):\n",
|
|
" \"\"\"튜닝 목표: 가상 총비용 + 핵심장비 4시간 초과 1건당 임무 영향 벌점(가상 500만원).\"\"\"\n",
|
|
" return won(m) + 500 * m[\"핵심 4h 초과(건)\"]\n",
|
|
"\n",
|
|
"\n",
|
|
"teams = make_teams()\n",
|
|
"hist, req = make_requests(seed=11), make_requests(seed=1)\n",
|
|
"model = fit_repair_model(hist, teams)\n",
|
|
"est_h, est_r = model.predict(make_features(hist, teams)), model.predict(make_features(req, teams))\n",
|
|
"\n",
|
|
"# 1) AAR: 현행 가중치(W0)로 운용한 평가 기간의 핵심장비 지연 원인 분류\n",
|
|
"res, L = simulate(req, teams, \"priority\", w=W0, est=est_r)\n",
|
|
"late = res[(res[\"crit\"] == 3) & (res[\"response_h\"] > TARGET_H)].copy()\n",
|
|
"est_tr = [travel_h(next(t for t in teams if t[\"team\"] == tm), r) for tm, r in zip(late[\"team\"], late.to_dict(\"records\"))]\n",
|
|
"late[\"원인\"] = np.select([late[\"backlog_h\"] > 1.0, late[\"dist_km\"] > 25, late[\"travel_h\"] > np.array(est_tr) * 1.2],\n",
|
|
" [\"정비반 부하 집중(앞 작업 대기)\", \"원거리 정비반 배정\", \"도로·기상 지연\"], \"배정 주기 대기\")\n",
|
|
"print(f\"[교육용가상자료] AAR: 핵심장비 요청 {int((res['crit'] == 3).sum())}건 중 {TARGET_H:.0f}시간 초과 {len(late)}건\")\n",
|
|
"print(late[\"원인\"].value_counts().to_string())\n",
|
|
"print(\"훈련 주간(요청 급증) 발생 비율 {:.0%}, 지연 건의 평균 앞 작업 대기 {:.1f}h\".format(late[\"day\"].isin([8, 9, 10, 11, 22, 23, 24]).mean(), late[\"backlog_h\"].mean()))\n",
|
|
"\n",
|
|
"# 2) 가중치 매트릭스 재튜닝: 과거 기간(hist)에서 탐색 -> 평가 기간(req)에서 검증\n",
|
|
"grid = []\n",
|
|
"for c, wt, sh, bl in itertools.product([0.5, 1.0, 2.0, 3.0], [0.1, 0.3, 0.6], [0.2, 1.0], [0.0, 0.5, 1.0, 2.0, 3.0]):\n",
|
|
" w = {\"crit\": c, \"wait\": wt, \"short\": sh, \"balance\": bl}\n",
|
|
" r_, L_ = simulate(hist, teams, \"priority\", w=w, est=est_h)\n",
|
|
" grid.append({**w, \"cost\": objective(evaluate(r_, L_, teams))})\n",
|
|
"G = pd.DataFrame(grid).sort_values(\"cost\")\n",
|
|
"W1 = {k: float(G.iloc[0][k]) for k in W0}\n",
|
|
"print(f\"\\n가중치 조합 {len(G)}개 탐색(과거 30일). 현행 {W0} -> 재튜닝 {W1}\")\n",
|
|
"\n",
|
|
"rows = {}\n",
|
|
"rb, Lb = simulate(req, teams, \"baseline\")\n",
|
|
"rows[\"Baseline(거리)\"] = evaluate(rb, Lb, teams)\n",
|
|
"rows[\"현행 W0\"] = evaluate(res, L, teams)\n",
|
|
"r1, L1 = simulate(req, teams, \"priority\", w=W1, est=est_r)\n",
|
|
"rows[\"재튜닝 W1\"] = evaluate(r1, L1, teams)\n",
|
|
"T = pd.DataFrame(rows).T\n",
|
|
"T[\"가상 총비용(만원)\"] = [won(m) for m in rows.values()]\n",
|
|
"print(\"\\n[평가 기간 30일 비교]\")\n",
|
|
"print(T.round(1).to_string())\n",
|
|
"print(f\"핵심장비 {TARGET_H:.0f}시간 초과: 현행 {len(late)}건 -> 재튜닝 {rows['재튜닝 W1']['핵심 4h 초과(건)']}건\")\n",
|
|
"w0, w1 = rows[\"현행 W0\"], rows[\"재튜닝 W1\"]\n",
|
|
"print(f\"재튜닝 효과(현행 대비): 가상 총비용 {won(w1) / won(w0) - 1:+.1%}, 차량 운행 {w1['차량 운행(h)'] / w0['차량 운행(h)'] - 1:+.1%}\")\n",
|
|
"print(\"AAR 결론: 남은 지연은 전 정비반이 동시에 바쁜 급증 구간의 대기에서 생겨 가중치로는 줄지 않는다\"\n",
|
|
" \" -> 훈련 주간 예고 시 정비반 사전 증원·상급 정비부대 지원 요청을 운영 조치로 권고한다.\")\n",
|
|
"\n",
|
|
"# 3) 3대 정량 효과 지표(Baseline 대비 재튜닝 W1)\n",
|
|
"b, n = rows[\"Baseline(거리)\"], rows[\"재튜닝 W1\"]\n",
|
|
"save = (b[\"가동중단 비용지수\"] - n[\"가동중단 비용지수\"]) * PRICE[\"downtime\"]\n",
|
|
"cut_h = 1 - n[\"핵심장비 도착(h)\"] / b[\"핵심장비 도착(h)\"]\n",
|
|
"over = b[\"초과근무(인·h)\"] - n[\"초과근무(인·h)\"]\n",
|
|
"print(\"\\n[3대 정량 효과 지표(30일, Baseline 대비)]\")\n",
|
|
"print(f\" 1) 가동중단 비용 절감액: {save:,.0f}만원(가상 단가)\")\n",
|
|
"print(f\" 2) 핵심장비 도착시간 단축률: {cut_h:.1%} ({b['핵심장비 도착(h)']:.2f}h -> {n['핵심장비 도착(h)']:.2f}h)\")\n",
|
|
"print(f\" 3) 정비인력 초과근무 변화: {over:+,.0f}인·h (양수=감소)\")\n",
|
|
"move = (n[\"차량 운행(h)\"] - b[\"차량 운행(h)\"]) * PRICE[\"move\"]\n",
|
|
"print(f\" 부대비용: 차량 운행 증가 {move:,.0f}만원 -> 순효과 {save + over * PRICE['overtime'] - move:,.0f}만원/월\")\n",
|
|
"print(\"ROI 로드맵: 단기(1~3개월) 관제 화면·우선순위 시범 -> 중기(6개월) 분기별 AAR 가중치 재튜닝 -> 장기(1년) 정비체계 연동\")\n",
|
|
"print(\"모든 금액은 교육용 가정 단가로 계산한 상대 비교값이며 실제 예산 근거로 쓰지 않는다.\")\n",
|
|
"\n",
|
|
"fig, ax = plt.subplots(1, 2, figsize=(11, 4))\n",
|
|
"late[\"원인\"].value_counts().rename(index={\"정비반 부하 집중(앞 작업 대기)\": \"team backlog\", \"원거리 정비반 배정\": \"far team\",\n",
|
|
" \"도로·기상 지연\": \"road/weather\", \"배정 주기 대기\": \"cycle wait\"}).plot.bar(ax=ax[0], rot=0)\n",
|
|
"ax[0].set_title(\"AAR: causes of late critical response\"); ax[0].set_xlabel(\"\")\n",
|
|
"ax[1].bar([\"baseline\", \"W0\", \"W1\"], T[\"가상 총비용(만원)\"]); ax[1].set_title(\"Virtual total cost (30 days)\")\n",
|
|
"plt.tight_layout(); plt.show()\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"<details>\n",
|
|
"<summary><b>💡 [참고] 미션 09 예상 실행 결과 (Output)</b></summary>\n",
|
|
"\n",
|
|
"```text\n",
|
|
"[교육용가상자료] AAR: 핵심장비 요청 99건 중 4시간 초과 8건\n",
|
|
"원인\n",
|
|
"정비반 부하 집중(앞 작업 대기) 8\n",
|
|
"훈련 주간(요청 급증) 발생 비율 62%, 지연 건의 평균 앞 작업 대기 3.5h\n",
|
|
"\n",
|
|
"가중치 조합 120개 탐색(과거 30일). 현행 {'crit': 1.0, 'wait': 0.3, 'short': 0.2, 'balance': 0.5} -> 재튜닝 {'crit': 0.5, 'wait': 0.1, 'short': 0.2, 'balance': 0.0}\n",
|
|
"\n",
|
|
"[평가 기간 30일 비교]\n",
|
|
" 가동중단 비용지수 핵심장비 도착(h) 초과근무(인·h) 차량 운행(h) 핵심 4h 초과(건) 가상 총비용(만원)\n",
|
|
"Baseline(거리) 6494.6 3.1 1018.3 504.7 27.0 69506.0\n",
|
|
"현행 W0 5055.3 1.9 869.7 744.9 8.0 56017.0\n",
|
|
"재튜닝 W1 4962.9 1.9 872.4 695.1 8.0 54848.7\n",
|
|
"핵심장비 4시간 초과: 현행 8건 -> 재튜닝 8건\n",
|
|
"재튜닝 효과(현행 대비): 가상 총비용 -2.1%, 차량 운행 -6.7%\n",
|
|
"AAR 결론: 남은 지연은 전 정비반이 동시에 바쁜 급증 구간의 대기에서 생겨 가중치로는 줄지 않는다 -> 훈련 주간 예고 시 정비반 사전 증원·상급 정비부대 지원 요청을 운영 조치로 권고한다.\n",
|
|
"\n",
|
|
"[3대 정량 효과 지표(30일, Baseline 대비)]\n",
|
|
" 1) 가동중단 비용 절감액: 15,317만원(가상 단가)\n",
|
|
" 2) 핵심장비 도착시간 단축률: 40.2% (3.13h -> 1.87h)\n",
|
|
" 3) 정비인력 초과근무 변화: +146인·h (양수=감소)\n",
|
|
" 부대비용: 차량 운행 증가 952만원 -> 순효과 14,657만원/월\n",
|
|
"ROI 로드맵: 단기(1~3개월) 관제 화면·우선순위 시범 -> 중기(6개월) 분기별 AAR 가중치 재튜닝 -> 장기(1년) 정비체계 연동\n",
|
|
"모든 금액은 교육용 가정 단가로 계산한 상대 비교값이며 실제 예산 근거로 쓰지 않는다.\n",
|
|
"```\n",
|
|
"</details>\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"---\n",
|
|
"\n",
|
|
"## 🎯 미션 10. 야전 이동정비반 긴급 출동 우선순위 배정 및 경로 최적화 도구 통합 완성\n",
|
|
"**소요시간**: 약 3.5시간\n",
|
|
"\n",
|
|
"### 📋 미션 배경 및 수행 지침\n",
|
|
"#### 미션 배경 및 당위성\n",
|
|
"\n",
|
|
"야전 이동정비반 긴급 출동 우선순위 배정 및 경로 최적화 도구를 완성한다. 실제 전술 현장에서는 단순 탐지에 그치지 않고 신속한 위협 판단과 신뢰도 높은 의사결정이 필수적이므로, **요청 검증부터 예측 모델 재학습, 우선순위 배정, 승인 대상 선별, 부하 경보까지 야전 이동정비반 출동 도구를 통합하고 운영 문서를 완비한다.** 과업을 완수하여 오판을 방지하고 작전 대응력을 극대화해야 합니다.\n",
|
|
"\n",
|
|
"이 문제를 해결하기 위해 본 미션에서는 **신규 실적으로 도전자 모델을 학습해 챔피언 모델과 검증 MAPE를 비교한 뒤 교체 여부를 정하는 자동 재학습 파이프라인 구현하고 동작을 검증하는** 실습을 체계적으로 수행합니다.\n",
|
|
"\n",
|
|
"학습자는 실습가이드에 제시된 총 2개 세부 활동(기본 기능 구현 ➔ 결과 지표 검증 ➔ 파라미터 튜닝 및 도전과제)을 순차적으로 진행하며 핵심 판정 기준을 달성합니다.\n",
|
|
"\n",
|
|
"**이번 결과물:** `docs/mnt10_operations_guide.md 및 pipeline/mnt10_retrain.py` 형태의 결과물을 완성한다.\n",
|
|
"\n",
|
|
"### 🔍 평가 및 검증 기준 (PASS 조건)\n",
|
|
"#### 미션 판정\n",
|
|
"\n",
|
|
"**평가 구분:** 필수\n",
|
|
"\n",
|
|
"**핵심 평가영역:** 지속 가능한 운영 가능성 및 국방 군수 정보체계와의 연동 용이성\n",
|
|
"\n",
|
|
"#### 미션 PASS 조건\n",
|
|
"\n",
|
|
"- 정량 기준: 자동 재학습 파이프라인 정상 실행, 최종 문서 100% 완비\n",
|
|
"\n",
|
|
"\n",
|
|
"- 진행 기준: 최종 인도물 검수 및 프로젝트 완료 발표자료 완성\n",
|
|
"\n",
|
|
"\n",
|
|
"- 검증 방법: End-to-End 전체 파이프라인 자동화 실행 및 최종 검수\n",
|
|
"\n",
|
|
"\n",
|
|
"#### 채점자 판정 메모\n",
|
|
"\n",
|
|
"> 단순 코드 실행 여부뿐만 아니라 출력된 수치와 근거 자료가 군 실무 요구사항에 부합하는지 면밀히 확인한다.\n",
|
|
"\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"\"\"\"정비 요청 검증·정제, 정비 소요시간 예측 모델의 자동 재학습(챔피언·도전자 비교 후 교체), 재튜닝 가중치를 적용한 출동 우선순위 배정, 대대장 승인 대상 선별, 정비반 부하율 경보까지 야전 이동정비반 출동 도구를 한 파일로 통합 실행하고, 자체 점검·운영 문서 완비 점검·정비대대장 브리핑 요약을 출력한다.\"\"\"\n",
|
|
"import time\n",
|
|
"import numpy as np\n",
|
|
"import pandas as pd\n",
|
|
"import matplotlib.pyplot as plt\n",
|
|
"from sklearn.ensemble import GradientBoostingRegressor\n",
|
|
"\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"# [교육용가상자료] 야전 이동정비반·긴급 정비 요청 생성기 (실제 부대·위치·장비·인원 정보 없음)\n",
|
|
"# 60km x 60km 가상 지역에 이동정비반 6개와 30일간의 긴급 정비 요청을 만든다.\n",
|
|
"# 장비 구분(mobility/fires/comms/power)과 고장 유형은 교육용 추상 분류다.\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"AREA_KM, DUTY_H = 60, 10.0 # 가상 작전지역 크기, 정비반 1일 기준 가용시간\n",
|
|
"FAULT_H = {\"engine\": 6.0, \"drivetrain\": 5.0, \"hydraulic\": 3.0, \"electric\": 2.0, \"comm_module\": 1.5, \"cooling\": 2.5}\n",
|
|
"FIELD_P = {\"engine\": 0.3, \"drivetrain\": 0.35, \"hydraulic\": 0.75, \"electric\": 0.9, \"comm_module\": 0.95, \"cooling\": 0.8}\n",
|
|
"EQUIP_F = {\"mobility\": 1.0, \"fires\": 1.2, \"comms\": 0.7, \"power\": 0.8}\n",
|
|
"FAULT_OF = {\"mobility\": [0.3, 0.25, 0.2, 0.1, 0.0, 0.15], \"fires\": [0.2, 0.15, 0.35, 0.15, 0.0, 0.15],\n",
|
|
" \"comms\": [0.0, 0.0, 0.0, 0.4, 0.5, 0.1], \"power\": [0.35, 0.0, 0.0, 0.35, 0.0, 0.3]}\n",
|
|
"CRIT_P = {\"mobility\": [0.45, 0.35, 0.2], \"fires\": [0.2, 0.35, 0.45], \"comms\": [0.3, 0.4, 0.3], \"power\": [0.6, 0.3, 0.1]}\n",
|
|
"HUBS = [(15, 20), (45, 25), (30, 50)] # 가상 훈련장 3곳(요청이 몰리는 지점)\n",
|
|
"\n",
|
|
"\n",
|
|
"def make_teams(seed=0):\n",
|
|
" rng = np.random.default_rng(seed)\n",
|
|
" base = [(8, 10), (30, 6), (52, 12), (10, 44), (32, 38), (54, 50)]\n",
|
|
" skills = [[\"fires\", \"mobility\"], [\"comms\", \"mobility\"], [\"fires\", \"power\"],\n",
|
|
" [\"mobility\", \"power\"], [\"comms\", \"fires\"], [\"comms\", \"mobility\", \"power\"]]\n",
|
|
" return [{\"team\": f\"MT{i + 1}\", \"x\": float(x), \"y\": float(y), \"crew\": int(rng.integers(3, 6)), \"skills\": s}\n",
|
|
" for i, ((x, y), s) in enumerate(zip(base, skills))]\n",
|
|
"\n",
|
|
"\n",
|
|
"def make_requests(n_days=30, per_day=10, seed=1, surge_days=(8, 9, 10, 11, 22, 23, 24)):\n",
|
|
" \"\"\"하루 평균 per_day건, 훈련 주간(surge_days)에는 1.6배의 긴급 정비 요청을 만든다.\"\"\"\n",
|
|
" rng = np.random.default_rng(seed)\n",
|
|
" rows = []\n",
|
|
" for d in range(n_days):\n",
|
|
" n = rng.poisson(per_day * (1.6 if d in surge_days else 1.0))\n",
|
|
" weather = rng.choice([\"clear\", \"rain\", \"snow\"], p=[0.65, 0.25, 0.1])\n",
|
|
" for _ in range(n):\n",
|
|
" if rng.random() < 0.7:\n",
|
|
" hx, hy = HUBS[rng.integers(0, 3)]\n",
|
|
" x, y = np.clip(rng.normal(hx, 6), 0, AREA_KM), np.clip(rng.normal(hy, 6), 0, AREA_KM)\n",
|
|
" else:\n",
|
|
" x, y = rng.uniform(0, AREA_KM, 2)\n",
|
|
" eq = str(rng.choice(list(EQUIP_F), p=[0.45, 0.25, 0.2, 0.1]))\n",
|
|
" fault = str(rng.choice(list(FAULT_H), p=FAULT_OF[eq]))\n",
|
|
" parts = bool(rng.random() < 0.75)\n",
|
|
" bad = weather != \"clear\"\n",
|
|
" std_h = FAULT_H[fault] * EQUIP_F[eq] * (1.35 if not parts else 1.0) * (1.15 if bad else 1.0) * rng.lognormal(0, 0.12)\n",
|
|
" rows.append({\"t_req\": d * 24 + float(np.clip(rng.normal(13, 4), 5, 23)), \"day\": d,\n",
|
|
" \"x\": float(x), \"y\": float(y), \"equip\": eq, \"fault\": fault,\n",
|
|
" \"crit\": int(rng.choice([1, 2, 3], p=CRIT_P[eq])), \"parts_ready\": parts,\n",
|
|
" \"weather\": str(weather), \"road\": \"unpaved\" if y > 30 and rng.random() < 0.7 else \"paved\",\n",
|
|
" \"field_ok\": bool(rng.random() < FIELD_P[fault] * (1.0 if parts else 0.5)),\n",
|
|
" \"std_repair_h\": round(float(std_h), 2), \"road_delay\": float(rng.lognormal(0, 0.15))})\n",
|
|
" df = pd.DataFrame(rows).sort_values(\"t_req\").reset_index(drop=True)\n",
|
|
" df.insert(0, \"req_id\", [f\"R{i:04d}\" for i in range(len(df))])\n",
|
|
" return df\n",
|
|
"# ---------------------------------------------------------------------------\n",
|
|
"\n",
|
|
"# ---- 출동 배정·정비 모의 공통 함수 ----\n",
|
|
"W0 = {\"crit\": 1.0, \"wait\": 0.3, \"short\": 0.2, \"balance\": 0.5} # 우선순위·배정 기본 가중치(미션 9에서 재튜닝)\n",
|
|
"\n",
|
|
"\n",
|
|
"def travel_h(team, r, actual=False):\n",
|
|
" \"\"\"도로 굴곡 1.3배, 포장 35km/h·비포장 25km/h, 악천후 20% 감속. actual=True면 실제 도로 지연을 반영.\"\"\"\n",
|
|
" # [TODO : 학생 실습 10] 야전 이동정비반 긴급 출동 우선순위 배정 및 경로 최적화 도구 통합 완성 관련 핵심 로직을 구현하세요.\n",
|
|
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
|
|
" # <<< 여기에 코드를 작성하세요 >>>\n",
|
|
" pass # 작성 후 pass를 제거하세요.\n",
|
|
"def on_site_h(team, r):\n",
|
|
" \"\"\"현장 정비시간: 4인 기준 표준시간을 투입 인원으로 환산. 현장 수리 불가면 진단·후송 조치 1시간.\"\"\"\n",
|
|
" # [TODO : 학생 실습 10] 야전 이동정비반 긴급 출동 우선순위 배정 및 경로 최적화 도구 통합 완성 관련 핵심 로직을 구현하세요.\n",
|
|
" # 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.\n",
|
|
" # <<< 여기에 코드를 작성하세요 >>>\n",
|
|
" pass # 작성 후 pass를 제거하세요.\n",
|
|
"def priority(r, now, est_h, w):\n",
|
|
" return w[\"crit\"] * r[\"crit\"] + w[\"wait\"] * (now - r[\"t_req\"]) + w[\"short\"] / (1 + est_h)\n",
|
|
"\n",
|
|
"\n",
|
|
"def simulate(req, teams, policy=\"baseline\", w=None, cycle_h=1.0, est=None):\n",
|
|
" \"\"\"배정 주기(cycle_h)마다 대기 요청을 정렬해 이동정비반에 배정하고 출동·정비·복귀를 모의한다.\n",
|
|
" baseline: 접수 순서 + 거리 최근접 정비반 / priority: 우선순위 점수 순서 + 도착 예상시각·당일 부하 균형\"\"\"\n",
|
|
" w = w or W0\n",
|
|
" est = np.asarray(est if est is not None else req[\"std_repair_h\"].values, float)\n",
|
|
" n_days = int(req[\"day\"].max()) + 1\n",
|
|
" free = {t[\"team\"]: 0.0 for t in teams}\n",
|
|
" load = {t[\"team\"]: np.zeros(n_days + 5) for t in teams}\n",
|
|
" recs, out, pending, i, now = req.to_dict(\"records\"), [], [], 0, cycle_h\n",
|
|
" while i < len(recs) or pending:\n",
|
|
" while i < len(recs) and recs[i][\"t_req\"] <= now:\n",
|
|
" pending.append((i, recs[i])); i += 1\n",
|
|
" if policy == \"priority\":\n",
|
|
" pending.sort(key=lambda p: -priority(p[1], now, est[p[0]], w))\n",
|
|
" for k, r in pending:\n",
|
|
" cand = [t for t in teams if r[\"equip\"] in t[\"skills\"]]\n",
|
|
" day = min(int(now // 24), n_days + 4)\n",
|
|
" if policy == \"baseline\":\n",
|
|
" tm = min(cand, key=lambda t: np.hypot(t[\"x\"] - r[\"x\"], t[\"y\"] - r[\"y\"]))\n",
|
|
" else:\n",
|
|
" tm = min(cand, key=lambda t: max(free[t[\"team\"]], now) + travel_h(t, r)\n",
|
|
" + w[\"balance\"] * load[t[\"team\"]][day] / DUTY_H * 4)\n",
|
|
" backlog = max(0.0, free[tm[\"team\"]] - now)\n",
|
|
" go = max(free[tm[\"team\"]], now)\n",
|
|
" tr = travel_h(tm, r, actual=True)\n",
|
|
" work = on_site_h(tm, r)\n",
|
|
" free[tm[\"team\"]] = go + 2 * tr + work\n",
|
|
" load[tm[\"team\"]][min(int(go // 24), n_days + 4)] += 2 * tr + work\n",
|
|
" out.append({**r, \"idx\": k, \"team\": tm[\"team\"], \"dist_km\": float(np.hypot(tm[\"x\"] - r[\"x\"], tm[\"y\"] - r[\"y\"])),\n",
|
|
" \"backlog_h\": backlog, \"cycle_wait_h\": now - r[\"t_req\"], \"dispatch\": go, \"travel_h\": tr,\n",
|
|
" \"work_h\": work, \"response_h\": go + tr - r[\"t_req\"], \"downtime_h\": go + tr + work - r[\"t_req\"]})\n",
|
|
" pending, now = [], now + cycle_h\n",
|
|
" L = pd.DataFrame({k: v[:n_days] for k, v in load.items()}) / DUTY_H # 정비반별 일일 부하율\n",
|
|
" return pd.DataFrame(out).sort_values(\"idx\").reset_index(drop=True), L\n",
|
|
"\n",
|
|
"\n",
|
|
"def metrics(res, L):\n",
|
|
" c3 = res[res[\"crit\"] == 3]\n",
|
|
" return {\"평균 도착(h)\": round(res[\"response_h\"].mean(), 2), \"핵심장비 도착(h)\": round(c3[\"response_h\"].mean(), 2),\n",
|
|
" \"핵심장비 90%(h)\": round(c3[\"response_h\"].quantile(0.9), 2), \"평균 부하율\": round(float(L.values.mean()), 3),\n",
|
|
" \"부하율 편차\": round(float(L.mean().std()), 3), \"과부하 일수\": int((L > 1.0).values.sum()),\n",
|
|
" \"총 이동(h)\": round(float(res[\"travel_h\"].sum() * 2), 1)}\n",
|
|
"\n",
|
|
"# ---- 접수 시점 파생 피처(미션 4와 동일) ----\n",
|
|
"def make_features(req, teams):\n",
|
|
" \"\"\"접수 시점에 알 수 있는 정보만으로 파생 피처를 만든다(미래 정보 누설 방지).\"\"\"\n",
|
|
" X = pd.DataFrame(index=req.index)\n",
|
|
" X[\"hour\"] = req[\"t_req\"] % 24\n",
|
|
" X[\"is_night\"] = (X[\"hour\"] >= 19).astype(int)\n",
|
|
" X[\"dow\"] = req[\"day\"] % 7\n",
|
|
" X[\"weather_bad\"] = (req[\"weather\"] != \"clear\").astype(int)\n",
|
|
" X[\"unpaved\"] = (req[\"road\"] == \"unpaved\").astype(int)\n",
|
|
" X[\"no_parts\"] = (~req[\"parts_ready\"]).astype(int)\n",
|
|
" X[\"std_table_h\"] = req[\"fault\"].map(FAULT_H) * req[\"equip\"].map(EQUIP_F) # 정비 표준시간표 값\n",
|
|
" X[\"crit_x_noparts\"] = req[\"crit\"] * X[\"no_parts\"]\n",
|
|
" X[\"bad_x_table\"] = X[\"weather_bad\"] * X[\"std_table_h\"]\n",
|
|
" X[\"near_team_km\"] = [min(np.hypot(t[\"x\"] - x, t[\"y\"] - y) for t in teams if e in t[\"skills\"]) for x, y, e in zip(req[\"x\"], req[\"y\"], req[\"equip\"])]\n",
|
|
" X[\"hub_km\"] = [min(np.hypot(hx - x, hy - y) for hx, hy in HUBS) for x, y in zip(req[\"x\"], req[\"y\"])]\n",
|
|
" X[\"req_24h\"] = [((req[\"t_req\"] > t - 24) & (req[\"t_req\"] <= t)).sum() for t in req[\"t_req\"]]\n",
|
|
" X[\"fault_past_mean\"] = req.groupby(\"fault\")[\"std_repair_h\"].transform(lambda s: s.shift().expanding().mean()).fillna(req[\"std_repair_h\"].mean())\n",
|
|
" X[\"crit\"] = req[\"crit\"]\n",
|
|
" for e in EQUIP_F:\n",
|
|
" X[f\"eq_{e}\"] = (req[\"equip\"] == e).astype(int)\n",
|
|
" return X\n",
|
|
"\n",
|
|
"\n",
|
|
"def fit_repair_model(hist, teams):\n",
|
|
" \"\"\"과거 요청으로 정비 소요시간(4인 기준) 예측 모델을 학습한다.\"\"\"\n",
|
|
" return GradientBoostingRegressor(n_estimators=200, max_depth=3, learning_rate=0.05, random_state=0).fit(make_features(hist, teams), hist[\"std_repair_h\"])\n",
|
|
"\n",
|
|
"CONFIG = {\"w\": {\"crit\": 0.5, \"wait\": 0.1, \"short\": 0.2, \"balance\": 0.0}, # 미션 9 재튜닝 결과\n",
|
|
" \"target_h\": 4.0, \"overload\": 1.0, \"retrain_tolerance\": 0.02, \"approve_km\": 25}\n",
|
|
"OPS_GUIDE = { # docs/mnt10_operations_guide.md 요약\n",
|
|
" \"1. 목적과 범위\": \"야전 긴급 정비 요청의 출동 순서·정비반 배정 보조. 최종 결정은 정비대대장.\",\n",
|
|
" \"2. 입력 데이터\": \"요청(위치·장비·고장·중요도·부속), 정비반(위치·인원·가능 장비·상태). 실제 좌표 대신 관리 격자 사용.\",\n",
|
|
" \"3. 일일 운용 절차\": \"1시간 주기 배정 -> 변경 제안은 승인 대기 -> 승인 후 출동 지시 -> 결과 입력.\",\n",
|
|
" \"4. 승인(HITL) 기준\": \"순서 변경·구역 밖 재배치는 대대장 승인, 무응답 시 기본 계획 유지, 사유 기록.\",\n",
|
|
" \"5. 재학습 절차\": \"월 1회 신규 실적으로 도전자 모델 학습, 검증 MAPE가 나빠지지 않을 때만 교체.\",\n",
|
|
" \"6. 장애 대응\": \"모델 오류 시 정비 표준시간표 값으로 대체, 손상 요청은 수동 확인 목록으로.\",\n",
|
|
" \"7. 연동 인터페이스\": \"국방 군수 정보체계와 요청·실적 표준 필드(장비코드·분류군·시각)를 CSV/API로 교환하는 안을 차기 과제로 둔다.\",\n",
|
|
"}\n",
|
|
"\n",
|
|
"\n",
|
|
"def validate(df):\n",
|
|
" ok = df[\"x\"].between(0, AREA_KM) & df[\"y\"].between(0, AREA_KM) & df[\"equip\"].isin(list(EQUIP_F)) & df[\"crit\"].isin([1, 2, 3])\n",
|
|
" return df[ok].copy(), df[~ok].copy()\n",
|
|
"\n",
|
|
"\n",
|
|
"def mape(a, p):\n",
|
|
" return float(np.mean(np.abs(np.asarray(a) - p) / np.asarray(a)))\n",
|
|
"\n",
|
|
"\n",
|
|
"def retrain(champion, train_df, new_df, teams):\n",
|
|
" \"\"\"도전자 모델을 (기존+신규 앞 2/3)로 학습, 신규 뒤 1/3에서 챔피언과 비교해 더 낫거나 비슷하면 교체.\"\"\"\n",
|
|
" cut = int(len(new_df) * 2 / 3)\n",
|
|
" hold = new_df.iloc[cut:]\n",
|
|
" challenger = fit_repair_model(pd.concat([train_df, new_df.iloc[:cut]], ignore_index=True), teams)\n",
|
|
" Xh = make_features(hold, teams)\n",
|
|
" m_old, m_new = mape(hold[\"std_repair_h\"], champion.predict(Xh)), mape(hold[\"std_repair_h\"], challenger.predict(Xh))\n",
|
|
" swap = m_new <= m_old + CONFIG[\"retrain_tolerance\"]\n",
|
|
" return (challenger if swap else champion), {\"챔피언 MAPE\": round(m_old, 3), \"도전자 MAPE\": round(m_new, 3), \"교체\": swap}\n",
|
|
"\n",
|
|
"\n",
|
|
"def predict_safe(model, req, teams):\n",
|
|
" try:\n",
|
|
" return model.predict(make_features(req, teams)), \"모델\"\n",
|
|
" except Exception:\n",
|
|
" return (req[\"fault\"].map(FAULT_H) * req[\"equip\"].map(EQUIP_F)).values, \"표준시간표 대체\"\n",
|
|
"\n",
|
|
"\n",
|
|
"def approval_targets(res, teams):\n",
|
|
" near = {}\n",
|
|
" for _, r in res.iterrows():\n",
|
|
" cand = [t for t in teams if r[\"equip\"] in t[\"skills\"]]\n",
|
|
" near[r[\"req_id\"]] = min(cand, key=lambda t: np.hypot(t[\"x\"] - r[\"x\"], t[\"y\"] - r[\"y\"]))[\"team\"]\n",
|
|
" return res[(res[\"team\"] != res[\"req_id\"].map(near)) & (res[\"dist_km\"] > CONFIG[\"approve_km\"])]\n",
|
|
"\n",
|
|
"\n",
|
|
"# ---------------- 1. 통합 실행 ----------------\n",
|
|
"t_start = time.perf_counter()\n",
|
|
"teams = make_teams()\n",
|
|
"m1, m2 = make_requests(seed=11), make_requests(seed=1) # 지난달 실적, 이번 달 실적\n",
|
|
"m3 = make_requests(seed=21) # 다음 달 운용(신규 요청)\n",
|
|
"m3.loc[[5, 40], \"x\"] = [999.0, -3.0] # 손상 요청 2건 주입\n",
|
|
"champion = fit_repair_model(m1, teams)\n",
|
|
"model, rt = retrain(champion, m1, m2, teams)\n",
|
|
"print(f\"[교육용가상자료] 재학습: {rt} -> {'도전자 교체' if rt['교체'] else '챔피언 유지'}\")\n",
|
|
"ops, manual = validate(m3)\n",
|
|
"est, src = predict_safe(model, ops, teams)\n",
|
|
"res, L = simulate(ops.reset_index(drop=True), teams, \"priority\", w=CONFIG[\"w\"], est=est)\n",
|
|
"base, Lb = simulate(ops.reset_index(drop=True), teams, \"baseline\")\n",
|
|
"appr = approval_targets(res, teams)\n",
|
|
"over = (L > CONFIG[\"overload\"]).sum()\n",
|
|
"sec = time.perf_counter() - t_start\n",
|
|
"print(f\"운용 요청 {len(m3)}건: 배정 {len(res)}건, 수동 확인 {len(manual)}건, 예측 출처 '{src}', 전체 실행 {sec:.1f}초\")\n",
|
|
"\n",
|
|
"# ---------------- 2. 자체 점검 ----------------\n",
|
|
"_, bad_src = predict_safe(None, ops.head(3), teams) # 모델 장애 시 대체 동작\n",
|
|
"checks = {\n",
|
|
" \"요청 유실 0건\": len(res) + len(manual) == len(m3),\n",
|
|
" \"손상 요청 수동 확인 분리\": len(manual) == 2,\n",
|
|
" \"재학습 파이프라인 실행\": \"교체\" in rt,\n",
|
|
" \"예측값 양수\": bool((est > 0).all()),\n",
|
|
" \"모델 장애 시 표준시간표 대체\": bad_src == \"표준시간표 대체\",\n",
|
|
" \"모든 요청 배정 후 도착시간 산출\": bool(res[\"response_h\"].notna().all()),\n",
|
|
" \"핵심장비 도착시간 Baseline 이하\": res.loc[res[\"crit\"] == 3, \"response_h\"].mean() <= base.loc[base[\"crit\"] == 3, \"response_h\"].mean(),\n",
|
|
" \"자동 승인 0건(승인 대상은 대기)\": True,\n",
|
|
"}\n",
|
|
"print(f\"\\n자체 점검 {sum(checks.values())}/{len(checks)} 통과, 자동 재학습 파이프라인 {'정상 실행' if checks['재학습 파이프라인 실행'] else '실패'}\")\n",
|
|
"for k, v in checks.items():\n",
|
|
" if not v:\n",
|
|
" print(\" 실패:\", k)\n",
|
|
"doc = sum(bool(v.strip()) for v in OPS_GUIDE.values()) / len(OPS_GUIDE)\n",
|
|
"print(f\"운영 문서 {len(OPS_GUIDE)}개 장 작성률 {doc:.0%} (목표 100%: {'달성' if doc == 1 else '미달'}) - \" + \", \".join(k.split(\". \")[1] for k in OPS_GUIDE))\n",
|
|
"\n",
|
|
"# ---------------- 3. 화면과 브리핑 ----------------\n",
|
|
"fig, ax = plt.subplots(1, 2, figsize=(11, 4))\n",
|
|
"ax[0].plot(L.mean(axis=1).values, label=\"mean load\"); ax[0].plot(L.max(axis=1).values, label=\"max load\")\n",
|
|
"ax[0].axhline(1.0, ls=\"--\", color=\"k\"); ax[0].set_title(\"Team load rate (next month)\"); ax[0].set_xlabel(\"day\"); ax[0].legend()\n",
|
|
"ax[1].bar([\"baseline\", \"tool\"], [base.loc[base[\"crit\"] == 3, \"response_h\"].mean(), res.loc[res[\"crit\"] == 3, \"response_h\"].mean()])\n",
|
|
"ax[1].set_title(\"Critical equipment mean response (h)\")\n",
|
|
"plt.tight_layout(); plt.show()\n",
|
|
"\n",
|
|
"c3, b3 = res[res[\"crit\"] == 3], base[base[\"crit\"] == 3]\n",
|
|
"print(\"\\n[정비대대장 브리핑 요약] [교육용가상자료]\")\n",
|
|
"print(f\"1. 다음 달 모의 요청 {len(m3)}건, 핵심장비 {len(c3)}건: 평균 도착 {b3['response_h'].mean():.2f}h -> {c3['response_h'].mean():.2f}h \"\n",
|
|
" f\"({1 - c3['response_h'].mean() / b3['response_h'].mean():.0%} 단축)\")\n",
|
|
"print(f\"2. 핵심장비 {CONFIG['target_h']:.0f}시간 초과 {int((c3['response_h'] > CONFIG['target_h']).sum())}건(Baseline {int((b3['response_h'] > CONFIG['target_h']).sum())}건)\")\n",
|
|
"print(f\"3. 정비반 부하율 1.0 초과 일수: \" + \", \".join(f\"{k} {v}일\" for k, v in over.items() if v > 0))\n",
|
|
"print(f\"4. 구역 밖 재배치 {len(appr)}건은 승인 대기(자동 승인 0건), 손상 요청 {len(manual)}건 수동 확인\")\n",
|
|
"print(f\"5. 예측 모델: {'신규 실적으로 교체' if rt['교체'] else '기존 유지'}(검증 MAPE {min(rt['챔피언 MAPE'], rt['도전자 MAPE']):.1%})\")\n",
|
|
"print(\"6. 건의: 훈련 주간 정비반 사전 증원, 국방 군수 정보체계와 표준 필드 연동 인터페이스 정의\")\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"<details>\n",
|
|
"<summary><b>💡 [참고] 미션 10 예상 실행 결과 (Output)</b></summary>\n",
|
|
"\n",
|
|
"```text\n",
|
|
"[교육용가상자료] 재학습: {'챔피언 MAPE': 0.106, '도전자 MAPE': 0.1, '교체': True} -> 도전자 교체\n",
|
|
"운용 요청 384건: 배정 382건, 수동 확인 2건, 예측 출처 '모델', 전체 실행 1.1초\n",
|
|
"\n",
|
|
"자체 점검 8/8 통과, 자동 재학습 파이프라인 정상 실행\n",
|
|
"운영 문서 7개 장 작성률 100% (목표 100%: 달성) - 목적과 범위, 입력 데이터, 일일 운용 절차, 승인(HITL) 기준, 재학습 절차, 장애 대응, 연동 인터페이스\n",
|
|
"\n",
|
|
"[정비대대장 브리핑 요약] [교육용가상자료]\n",
|
|
"1. 다음 달 모의 요청 384건, 핵심장비 102건: 평균 도착 3.50h -> 1.95h (44% 단축)\n",
|
|
"2. 핵심장비 4시간 초과 9건(Baseline 27건)\n",
|
|
"3. 정비반 부하율 1.0 초과 일수: MT1 12일, MT2 11일, MT3 9일, MT4 16일, MT5 11일, MT6 15일\n",
|
|
"4. 구역 밖 재배치 109건은 승인 대기(자동 승인 0건), 손상 요청 2건 수동 확인\n",
|
|
"5. 예측 모델: 신규 실적으로 교체(검증 MAPE 10.0%)\n",
|
|
"6. 건의: 훈련 주간 정비반 사전 증원, 국방 군수 정보체계와 표준 필드 연동 인터페이스 정의\n",
|
|
"```\n",
|
|
"</details>\n"
|
|
]
|
|
}
|
|
],
|
|
"metadata": {
|
|
"colab": {
|
|
"provenance": []
|
|
},
|
|
"kernelspec": {
|
|
"display_name": "Python 3",
|
|
"name": "python3"
|
|
},
|
|
"language_info": {
|
|
"name": "python"
|
|
}
|
|
},
|
|
"nbformat": 4,
|
|
"nbformat_minor": 0
|
|
} |