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