237 KiB
237 KiB
🎯 [WEL03] 가상 수질검사 데이터 위험 신호 대시보드
Google Colab 실습 환경 (장병 복지·지원 / Level 2)
📌 1. 프로젝트 개요 및 배경
| 항목 | 내용 |
|---|---|
| 과정 분류 | 장병 복지·지원 (Level 2) |
| 학습 소요 시간 | 10개 미션 (5일 / 총 27.5시간) |
| 환경 사양 | Python 3.10+, numpy, pandas, matplotlib, scipy, scikit-learn |
📖 프로젝트 목표
전장에서는 급수차, 정수 장비, 우물, 하천 취수원 등 급수원의 수질 이상을 빨리 알아채지 못하면 장병의 건강과 부대 활동에 바로 영향을 주지만, 센서 값은 잡음과 고장이 많아 오경보로 불필요하게 급수를 끊는 일도 생긴다. 러시아-우크라이나 전쟁 사례에서 학습 초점으로 삼은 전장 위생·건강 위험 관리를 반영해, 본 프로젝트는 가상 수질검사 데이터를 분석해 위험 신호를 데이터로 판단하는 과제를 다룬다. 교육용 가상 급수원 8곳의 pH, 탁도, 잔류염소, 중금속, 대장균 측정 자료를 코드로 만들어 스키마 정의, 센서 결측·이상치 정제, 기준치 Baseline 판정, 복합 오염 징후 분류 모델, 신뢰도 지수와 오경보 감축, 상황판 대시보드, 급수 중단 HITL 결재선, 대량 오염 부하 시험, AAR 임계치 재조정, 통합 패키지까지 단계별로 구현한다. 최종 목표는 수질 위험 신호를 근거와 함께 빠르게 제시하되, 실제 음용 판정과 급수 중단은 공식 기준과 의무(예방의학) 담당 확인, 지휘관 결심에 따르도록 지원하는 도구를 완성하는 것이다.
⚠️ [교육용 가상 데이터셋 원칙]
- 본 노트북의 모든 데이터(제원, 로그, 시나리오 등)는 보안 규정을 준수하여 코드 내 난수 기반 가상 데이터로 생성됩니다.
- 실제 부대 위치, 작전 인가, 실제 전장 정보, 군사 기밀 데이터는 일체 포함되지 않습니다.
In [ ]:
# ============================================================================
# [Google Colab 환경 설정] 필수 라이브러리 설치 및 임포트
# ============================================================================
!pip install -q numpy pandas matplotlib scipy scikit-learn
import os
import sys
import time
import math
import json
import re
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import warnings
warnings.filterwarnings('ignore')
# 그래프 영문 폰트 기본 설정 (환경 독립성 보장)
plt.rcParams['font.sans-serif'] = 'DejaVu Sans'
plt.rcParams['axes.unicode_minus'] = False
print(f"실행 환경 준비 완료 (Python {sys.version.split()[0]})")
In [ ]:
"""야전 급수원 가상 수질검사 항목(pH, 탁도, 잔류염소, 중금속, 대장균)과 급수원·센서·전술 파라미터를 담은 입력 데이터 스키마를 정의하고, 가상 샘플 20건과 오류 주입 데이터로 검증기를 시험한다."""
import json
import re
import numpy as np
import pandas as pd
# ---------------------------------------------------------------------------
# [교육용가상자료] 야전 급수원 가상 수질 센서 데이터 생성기 (실제 부대·위치·검사 결과 없음)
# 가상 급수원 8곳 x 14일 x 1시간 간격. 실제 수질(true_*)·센서 측정값·오염 사건·센서 고장 정답을 함께 만든다.
# 기준치는 교육용 가정값이며, 실제 음용 판정은 공인 검사 결과와 의무(예방의학) 담당 기준에 따른다.
# ---------------------------------------------------------------------------
PARAMS = ["ph", "turbidity", "chlorine", "metal", "ecoli"]
SOURCES = [("W01", "water_truck", 1.0, 2.0), ("W02", "water_truck", 3.5, 1.2), ("W03", "purification_unit", 2.2, 4.1),
("W04", "purification_unit", 5.0, 3.3), ("W05", "well", 0.8, 5.6), ("W06", "well", 4.4, 6.0),
("W07", "river_intake", 6.3, 1.8), ("W08", "river_intake", 6.9, 4.9)]
BASE = {"water_truck": (7.2, 0.25, 0.75, 1.5), "purification_unit": (7.0, 0.20, 0.85, 1.2),
"well": (6.8, 0.45, 0.50, 3.0), "river_intake": (7.3, 0.55, 0.60, 2.0)} # pH, 탁도(NTU), 잔류염소(mg/L), 중금속(ug/L)
LIMITS = {"ph": (5.8, 8.5), "turbidity": (1.0, 2.0), "chlorine": (0.2, 0.1), "metal": (5.0, 10.0), "ecoli": (1, 1)} # (주의, 위험) 가정 기준
EVENT_TYPES = { # 오염 사건: 변수별 최대 변화량, 최대치 도달 시간(h)
"sewage": ({"turbidity": 2.8, "chlorine": -0.5, "ecoli": 30, "ph": -0.3}, 4),
"metal": ({"metal": 22.0, "ph": -0.5}, 4),
"disinfect_fail": ({"chlorine": -0.8, "ecoli": 6}, 6),
"chemical": ({"ph": 1.7, "turbidity": 1.0}, 3),
"creeping_metal": ({"metal": 13.0}, 18),
}
FAULT_TYPES = ["stuck", "spike", "drift", "dropout"]
NOISE = {"ph": 0.04, "turbidity": 0.06, "chlorine": 0.03, "metal": 0.35}
DIGITS = {"ph": 2, "turbidity": 2, "chlorine": 2, "metal": 1}
def make_dataset(days=14, seed=7, n_events=6, n_faults=10, events=None, faults=None, sources=None):
"""events: [(급수원 번호, 사건 유형, 시작 시각 h, 지속 h, 강도)], faults: [(급수원 번호, 고장 유형, 항목, 시작 h, 지속 h)]"""
rng = np.random.default_rng(seed)
sources = sources or SOURCES
T = days * 24
tt = np.arange(T)
hour = tt % 24
night = (hour < 6) | (hour >= 20)
rain = np.zeros(T)
for _ in range(max(1, days // 4)):
s0, d = int(rng.integers(0, T - 14)), int(rng.integers(6, 14))
rain[s0:s0 + d] = np.maximum(rain[s0:s0 + d], rng.uniform(0.6, 1.0))
if events is None:
events = [(int(rng.integers(len(sources))), str(rng.choice(list(EVENT_TYPES))), int(rng.integers(min(24, T // 4), max(T // 4 + 1, T - 48))),
int(rng.integers(10, 30)), 1.0) for _ in range(n_events)]
if faults is None:
faults = [(int(rng.integers(len(sources))), str(rng.choice(FAULT_TYPES)), str(rng.choice(["ph", "turbidity", "chlorine", "metal"])),
int(rng.integers(12, max(13, T - 30))), int(rng.integers(6, 24))) for _ in range(n_faults)]
out = []
for si, (sid, stype, x, y) in enumerate(sources):
ph0, tb0, cl0, mt0 = BASE[stype]
tr = {"ph": ph0 + 0.08 * np.sin(2 * np.pi * hour / 24) + rng.normal(0, 0.03, T),
"turbidity": tb0 * (1 + 0.1 * rng.normal(0, 1, T)),
"chlorine": (cl0 * np.exp(-0.022 * hour) if stype == "water_truck" else cl0 + rng.normal(0, 0.03, T)),
"metal": mt0 + rng.normal(0, 0.2, T), "ecoli": np.zeros(T)}
if stype in ("river_intake", "well"):
tr["turbidity"] = tr["turbidity"] + rain * (0.9 if stype == "river_intake" else 0.45)
f_all, etype = np.zeros(T), np.array(["none"] * T, dtype=object)
for (s_i, et, onset, dur, scale) in events:
if s_i != si:
continue
delta, ramp = EVENT_TYPES[et]
f = np.clip((tt - onset + 1) / ramp, 0, 1) * ((tt >= onset) & (tt < onset + dur + ramp))
for k, v in delta.items():
tr[k] = tr[k] + v * f * scale
etype[f > 0] = et
f_all = np.maximum(f_all, f)
tr["chlorine"] = np.clip(tr["chlorine"], 0.02, None)
tr["turbidity"] = np.clip(tr["turbidity"], 0.05, None)
tr["metal"] = np.clip(tr["metal"], 0.1, None)
grade = np.where(f_all >= 0.2, 2, 0)
mild = (tr["turbidity"] > LIMITS["turbidity"][0]) | (tr["chlorine"] < LIMITS["chlorine"][0]) | (tr["metal"] > LIMITS["metal"][0])
grade = np.where((grade == 0) & mild, 1, grade)
ms = {}
for p in ["ph", "turbidity", "chlorine", "metal"]:
sd = NOISE[p] * np.where(night, 1.5, 1.0) * (tr[p] if p == "turbidity" else 1)
ms[p] = np.round(np.clip(tr[p] + rng.normal(0, 1, T) * sd, 0, None), DIGITS[p])
ms["ecoli"] = (rng.poisson(tr["ecoli"]) + (rng.random(T) < 0.004)).astype(float)
fault, fparam = np.array(["none"] * T, dtype=object), np.array(["-"] * T, dtype=object)
for (s_i, ft, p, s0, d) in faults:
if s_i != si:
continue
seg = slice(s0, min(T, s0 + d))
if ft == "stuck":
ms[p][seg] = ms[p][s0]
elif ft == "spike":
seg = slice(s0, s0 + 1)
ms[p][s0] = {"ph": 11.8, "turbidity": 38.0, "chlorine": 0.0, "metal": 75.0}[p]
elif ft == "drift":
ramp = np.linspace(0, 1, seg.stop - s0)
ms[p][seg] = np.clip(ms[p][seg] + ramp * {"ph": -1.6, "turbidity": 2.6, "chlorine": -0.75, "metal": 12.0}[p], 0, None).round(DIGITS[p])
elif ft == "dropout":
for q in PARAMS:
ms[q][seg] = np.nan
p = "all"
fault[seg], fparam[seg] = ft, p
df = pd.DataFrame({"source_id": sid, "source_type": stype, "x_km": x, "y_km": y, "t": tt,
"cond": np.where(rain > 0, "storm", np.where(night, "night", "day")),
**{p: ms[p] for p in PARAMS}, **{f"true_{p}": tr[p] for p in PARAMS},
"event": etype, "fault": fault, "fault_param": fparam, "grade": grade})
out.append(df)
return pd.concat(out, ignore_index=True)
# ---------------------------------------------------------------------------
# 1) 스키마 정의: 급수원 정보, 수질 측정값(5항목), 센서 상태, 전술 파라미터(기상·위치오차·통신지연)
SCHEMA = {
"required": ["record_id", "source", "measured_at", "quality", "sensor", "tactical"],
"fields": {
"record_id": {"type": str, "pattern": r"^WQ-W\d{2}-\d{4}$"},
"measured_at": {"type": str, "pattern": r"^D\+\d{2} \d{2}:00$"},
},
"source": {
"source_id": {"type": str, "pattern": r"^W\d{2}$"},
"source_type": {"type": str, "enum": ["water_truck", "purification_unit", "well", "river_intake"]},
"grid_km": {"type": list, "length": 2},
},
"quality": { # 단위와 물리적으로 가능한 범위(기준치가 아니라 측정 가능 범위)
"ph": {"type": (int, float), "min": 0, "max": 14, "unit": "-"},
"turbidity_ntu": {"type": (int, float), "min": 0, "max": 1000, "unit": "NTU"},
"chlorine_mgL": {"type": (int, float), "min": 0, "max": 10, "unit": "mg/L"},
"metal_ugL": {"type": (int, float), "min": 0, "max": 1000, "unit": "ug/L"},
"ecoli_cfu": {"type": int, "min": 0, "max": 100000, "unit": "CFU/100mL"},
},
"sensor": {
"status": {"type": str, "enum": ["ok", "calibration_due", "fault", "offline"]},
"last_calibrated_h": {"type": int, "min": 0, "max": 720},
},
"tactical": {
"weather": {"type": str, "enum": ["clear", "rain", "night", "snow", "dust"]},
"position_error_m": {"type": (int, float), "min": 0, "max": 500},
"comm_delay_ms": {"type": (int, float), "min": 0, "max": 10000},
},
}
def check(name, value, rule):
"""값 하나를 규칙(type, enum, min, max, length, pattern)으로 검사해 오류 목록을 돌려준다."""
# [TODO : 학생 실습 1] 야전 급수원 가상 수질검사 항목(pH, 탁도, 잔류염소, 중금속, 대장균) 데이터셋 정의 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def validate(rec):
# [TODO : 학생 실습 1] 야전 급수원 가상 수질검사 항목(pH, 탁도, 잔류염소, 중금속, 대장균) 데이터셋 정의 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
data = make_dataset(days=3, seed=1, n_events=1, n_faults=0)
rng = np.random.default_rng(1)
rows = data.sample(20, random_state=1).sort_values(["source_id", "t"])
records = []
for _, r in rows.iterrows():
records.append({
"record_id": f"WQ-{r['source_id']}-{int(r['t']):04d}", "measured_at": f"D+{int(r['t']) // 24:02d} {int(r['t']) % 24:02d}:00",
"source": {"source_id": r["source_id"], "source_type": r["source_type"], "grid_km": [r["x_km"], r["y_km"]]},
"quality": {"ph": float(r["ph"]), "turbidity_ntu": float(r["turbidity"]), "chlorine_mgL": float(r["chlorine"]),
"metal_ugL": float(r["metal"]), "ecoli_cfu": int(r["ecoli"])},
"sensor": {"status": "ok", "last_calibrated_h": int(rng.integers(1, 200))},
"tactical": {"weather": {"storm": "rain", "night": "night", "day": "clear"}[r["cond"]],
"position_error_m": float(round(rng.uniform(3, 40), 1)), "comm_delay_ms": int(rng.integers(30, 900))}})
ok = sum(1 for r in records if not validate(r))
print(f"정상 샘플 {len(records)}건 중 검증 통과 {ok}건, 오류 {len(records) - ok}건")
parsed = json.loads(json.dumps(records, ensure_ascii=False)) # 직렬화 -> 파싱 왕복 시험
print("JSON 직렬화·파싱 왕복 후 값 일치:", parsed == records)
# 3) 오류 주입 시험: 검증기가 잘못된 데이터를 잡아내는지 확인
bad = json.loads(json.dumps(records[0]))
bad["record_id"] = "WQ-1" # 형식 오류
bad["quality"]["ph"] = 15.2 # 물리 범위 초과
bad["quality"]["ecoli_cfu"] = "검출" # 자료형 오류(숫자여야 함)
bad["source"]["source_type"] = "bottle" # 허용되지 않은 급수원 유형
del bad["tactical"]["comm_delay_ms"] # 전술 파라미터 누락
print("\n[오류 주입 결과]")
errs = validate(bad)
for e in errs:
print(" -", e)
print(f"주입 오류 5종 중 검출 {len(errs)}종 -> {'모두 검출' if len(errs) >= 5 else '누락 있음'}")
# 4) 데이터 명세 요약(명세서 초안)
spec = []
for g in ("fields", "source", "quality", "sensor", "tactical"):
for k, r in SCHEMA[g].items():
t = r["type"].__name__ if isinstance(r["type"], type) else "number"
spec.append((g, k, t, r.get("unit", ""), str({x: y for x, y in r.items() if x not in ("type", "unit")})))
print("\n[데이터 명세 요약]")
print(pd.DataFrame(spec, columns=["그룹", "필드", "자료형", "단위", "규칙"]).to_string(index=False))
done = sum(1 for s in spec if s[2] and s[4] != "{}")
print(f"스키마 필드 정의 완성도 {done}/{len(spec)} ({done / len(spec):.0%}): 모든 필드에 자료형·검증 규칙 정의")
print("\n[가정 판정 기준(교육용)] (주의, 위험):", {k: v for k, v in LIMITS.items()})
print("급수원 유형별 샘플 수:", rows["source_type"].value_counts().to_dict())
In [ ]:
"""수질 센서 시계열의 물리 범위 밖 값·고정값(stuck)·단발 스파이크를 정제하고 결측을 보정하는 전처리 모듈을 구현한 뒤, 처리 속도·오류율·보정 정확도와 단위 테스트 5종을 확인한다."""
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
# ---------------------------------------------------------------------------
# [교육용가상자료] 야전 급수원 가상 수질 센서 데이터 생성기 (실제 부대·위치·검사 결과 없음)
# 가상 급수원 8곳 x 14일 x 1시간 간격. 실제 수질(true_*)·센서 측정값·오염 사건·센서 고장 정답을 함께 만든다.
# 기준치는 교육용 가정값이며, 실제 음용 판정은 공인 검사 결과와 의무(예방의학) 담당 기준에 따른다.
# ---------------------------------------------------------------------------
PARAMS = ["ph", "turbidity", "chlorine", "metal", "ecoli"]
SOURCES = [("W01", "water_truck", 1.0, 2.0), ("W02", "water_truck", 3.5, 1.2), ("W03", "purification_unit", 2.2, 4.1),
("W04", "purification_unit", 5.0, 3.3), ("W05", "well", 0.8, 5.6), ("W06", "well", 4.4, 6.0),
("W07", "river_intake", 6.3, 1.8), ("W08", "river_intake", 6.9, 4.9)]
BASE = {"water_truck": (7.2, 0.25, 0.75, 1.5), "purification_unit": (7.0, 0.20, 0.85, 1.2),
"well": (6.8, 0.45, 0.50, 3.0), "river_intake": (7.3, 0.55, 0.60, 2.0)} # pH, 탁도(NTU), 잔류염소(mg/L), 중금속(ug/L)
LIMITS = {"ph": (5.8, 8.5), "turbidity": (1.0, 2.0), "chlorine": (0.2, 0.1), "metal": (5.0, 10.0), "ecoli": (1, 1)} # (주의, 위험) 가정 기준
EVENT_TYPES = { # 오염 사건: 변수별 최대 변화량, 최대치 도달 시간(h)
"sewage": ({"turbidity": 2.8, "chlorine": -0.5, "ecoli": 30, "ph": -0.3}, 4),
"metal": ({"metal": 22.0, "ph": -0.5}, 4),
"disinfect_fail": ({"chlorine": -0.8, "ecoli": 6}, 6),
"chemical": ({"ph": 1.7, "turbidity": 1.0}, 3),
"creeping_metal": ({"metal": 13.0}, 18),
}
FAULT_TYPES = ["stuck", "spike", "drift", "dropout"]
NOISE = {"ph": 0.04, "turbidity": 0.06, "chlorine": 0.03, "metal": 0.35}
DIGITS = {"ph": 2, "turbidity": 2, "chlorine": 2, "metal": 1}
def make_dataset(days=14, seed=7, n_events=6, n_faults=10, events=None, faults=None, sources=None):
"""events: [(급수원 번호, 사건 유형, 시작 시각 h, 지속 h, 강도)], faults: [(급수원 번호, 고장 유형, 항목, 시작 h, 지속 h)]"""
rng = np.random.default_rng(seed)
sources = sources or SOURCES
T = days * 24
tt = np.arange(T)
hour = tt % 24
night = (hour < 6) | (hour >= 20)
rain = np.zeros(T)
for _ in range(max(1, days // 4)):
s0, d = int(rng.integers(0, T - 14)), int(rng.integers(6, 14))
rain[s0:s0 + d] = np.maximum(rain[s0:s0 + d], rng.uniform(0.6, 1.0))
if events is None:
events = [(int(rng.integers(len(sources))), str(rng.choice(list(EVENT_TYPES))), int(rng.integers(min(24, T // 4), max(T // 4 + 1, T - 48))),
int(rng.integers(10, 30)), 1.0) for _ in range(n_events)]
if faults is None:
faults = [(int(rng.integers(len(sources))), str(rng.choice(FAULT_TYPES)), str(rng.choice(["ph", "turbidity", "chlorine", "metal"])),
int(rng.integers(12, max(13, T - 30))), int(rng.integers(6, 24))) for _ in range(n_faults)]
out = []
for si, (sid, stype, x, y) in enumerate(sources):
ph0, tb0, cl0, mt0 = BASE[stype]
tr = {"ph": ph0 + 0.08 * np.sin(2 * np.pi * hour / 24) + rng.normal(0, 0.03, T),
"turbidity": tb0 * (1 + 0.1 * rng.normal(0, 1, T)),
"chlorine": (cl0 * np.exp(-0.022 * hour) if stype == "water_truck" else cl0 + rng.normal(0, 0.03, T)),
"metal": mt0 + rng.normal(0, 0.2, T), "ecoli": np.zeros(T)}
if stype in ("river_intake", "well"):
tr["turbidity"] = tr["turbidity"] + rain * (0.9 if stype == "river_intake" else 0.45)
f_all, etype = np.zeros(T), np.array(["none"] * T, dtype=object)
for (s_i, et, onset, dur, scale) in events:
if s_i != si:
continue
delta, ramp = EVENT_TYPES[et]
f = np.clip((tt - onset + 1) / ramp, 0, 1) * ((tt >= onset) & (tt < onset + dur + ramp))
for k, v in delta.items():
tr[k] = tr[k] + v * f * scale
etype[f > 0] = et
f_all = np.maximum(f_all, f)
tr["chlorine"] = np.clip(tr["chlorine"], 0.02, None)
tr["turbidity"] = np.clip(tr["turbidity"], 0.05, None)
tr["metal"] = np.clip(tr["metal"], 0.1, None)
grade = np.where(f_all >= 0.2, 2, 0)
mild = (tr["turbidity"] > LIMITS["turbidity"][0]) | (tr["chlorine"] < LIMITS["chlorine"][0]) | (tr["metal"] > LIMITS["metal"][0])
grade = np.where((grade == 0) & mild, 1, grade)
ms = {}
for p in ["ph", "turbidity", "chlorine", "metal"]:
sd = NOISE[p] * np.where(night, 1.5, 1.0) * (tr[p] if p == "turbidity" else 1)
ms[p] = np.round(np.clip(tr[p] + rng.normal(0, 1, T) * sd, 0, None), DIGITS[p])
ms["ecoli"] = (rng.poisson(tr["ecoli"]) + (rng.random(T) < 0.004)).astype(float)
fault, fparam = np.array(["none"] * T, dtype=object), np.array(["-"] * T, dtype=object)
for (s_i, ft, p, s0, d) in faults:
if s_i != si:
continue
seg = slice(s0, min(T, s0 + d))
if ft == "stuck":
ms[p][seg] = ms[p][s0]
elif ft == "spike":
seg = slice(s0, s0 + 1)
ms[p][s0] = {"ph": 11.8, "turbidity": 38.0, "chlorine": 0.0, "metal": 75.0}[p]
elif ft == "drift":
ramp = np.linspace(0, 1, seg.stop - s0)
ms[p][seg] = np.clip(ms[p][seg] + ramp * {"ph": -1.6, "turbidity": 2.6, "chlorine": -0.75, "metal": 12.0}[p], 0, None).round(DIGITS[p])
elif ft == "dropout":
for q in PARAMS:
ms[q][seg] = np.nan
p = "all"
fault[seg], fparam[seg] = ft, p
df = pd.DataFrame({"source_id": sid, "source_type": stype, "x_km": x, "y_km": y, "t": tt,
"cond": np.where(rain > 0, "storm", np.where(night, "night", "day")),
**{p: ms[p] for p in PARAMS}, **{f"true_{p}": tr[p] for p in PARAMS},
"event": etype, "fault": fault, "fault_param": fparam, "grade": grade})
out.append(df)
return pd.concat(out, ignore_index=True)
# ---------------------------------------------------------------------------
# ---- 전처리 공통 함수: 물리 범위 점검 -> 고정값(stuck) 표시 -> 단발 스파이크 제거 -> 결측 보정 ----
PHYS = {"ph": (0, 14), "turbidity": (0, 1000), "chlorine": (0, 10), "metal": (0, 1000), "ecoli": (0, 100000)}
SPIKE_K = {"day": 6.0, "night": 7.0, "storm": 9.0} # 상황별(주간/야간/악천후) 스파이크 판정 배수
MIN_JUMP = {"ph": 0.8, "turbidity": 1.5, "chlorine": 0.3, "metal": 6.0}
STUCK_RUN, SHORT_GAP = 6, 3
def clean(df, spike_k=None, min_jump=None):
"""급수원별 시계열을 정제하고 항목별 처리 내역(<항목>_flag: range/stuck/spike/interp/long_gap)을 남긴다."""
# [TODO : 학생 실습 2] 수질 센서 결측치 대체 및 시계열 센서 이상치 정제 전처리 모듈 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
data = make_dataset(days=14, seed=7, n_events=8, n_faults=16)
# 1) 처리 속도와 오류율
t0 = time.perf_counter()
c = clean(data)
dt = time.perf_counter() - t0
remain = int(c[PARAMS].isna().sum().sum())
print(f"전처리 {len(data):,}건, {len(data) / dt:,.0f}건/초, 처리 후 남은 결측 {remain}건, 오류율 {remain / len(data):.1%}")
# 2) 전처리 전후 비교: 센서 고장 유형별로 무엇이 처리되었는가
flag_cols = [f"{p}_flag" for p in PARAMS]
c["처리내역"] = c[flag_cols].apply(lambda r: ",".join(sorted({x for x in r if x})) or "-", axis=1)
print("\n[실제 고장 유형 x 전처리 처리 내역] (행 수)")
print(pd.crosstab(c["fault"], c["처리내역"]).to_string())
print(f"결측 셀 수: 전처리 전 {int(data[PARAMS].isna().sum().sum())} -> 후 {remain}")
# 3) 결측 보정 정확도: 정상 구간에서 값을 일부 지워 보고, 실제 값과 비교
rng = np.random.default_rng(3)
test = data.copy()
normal_idx = test.index[(test["fault"] == "none") & (test["event"] == "none")]
holes = rng.choice(normal_idx, 120, replace=False)
test.loc[holes, "chlorine"] = np.nan
filled = clean(test).set_index(["source_id", "t"])
truth = data.loc[holes].set_index(["source_id", "t"])["true_chlorine"]
mae = (filled.loc[truth.index, "chlorine"] - truth).abs().mean()
mean_fill = (data["chlorine"].mean() - truth).abs().mean()
print(f"\n잔류염소 결측 120건 보정 오차(MAE): 선형 보간 {mae:.3f} mg/L vs 전체 평균 대체 {mean_fill:.3f} mg/L")
# 4) 단위 테스트 5종
def unit_tests():
# [TODO : 학생 실습 2] 수질 센서 결측치 대체 및 시계열 센서 이상치 정제 전처리 모듈 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
res = unit_tests()
print("\n[단위 테스트]")
for k, v in res.items():
print(f" [{'PASS' if v else 'FAIL'}] {k}")
print(f"단위 테스트 {sum(res.values())}/5 통과")
# 5) 상황별(주간/야간/악천후) 임계값 동적 조정 효과: 최소 변화폭 하한을 끄고 배수만 비교
# (고장이 없는 행이 스파이크로 잘못 제거된 건수 / 실제 스파이크 고장 검출 건수)
no_floor = {p: 0.0 for p in MIN_JUMP}
for name, k in [("고정 임계(모든 상황 3배)", {"day": 3.0, "night": 3.0, "storm": 3.0}), ("상황별 동적 임계", SPIKE_K)]:
cc = clean(data, spike_k=k, min_jump=no_floor)
sp = (cc[flag_cols] == "spike").any(axis=1)
wrong, caught = int((sp & (cc["fault"] != "spike")).sum()), int((sp & (cc["fault"] == "spike")).sum())
print(f"{name}: 정상 변동 오제거 {wrong}건(야간·악천후 {int((sp & (cc['fault'] != 'spike') & (cc['cond'] != 'day')).sum())}건), 스파이크 검출 {caught}건")
print("운용 설정은 동적 임계 + 항목별 최소 변화폭 하한(MIN_JUMP)을 함께 쓴다.")
# 6) 전후 비교 그림
sid = data.loc[data["fault"].isin(["spike", "dropout"]), "source_id"].iloc[0]
raw, cl = data[data.source_id == sid], c[c.source_id == sid]
fig, ax = plt.subplots(2, 1, figsize=(10, 6), sharex=True)
for a, p in zip(ax, ["turbidity", "chlorine"]):
a.plot(raw["t"], raw[p], color="lightgray", lw=3, label="raw sensor")
a.plot(cl["t"], cl[p], color="tab:blue", lw=1, label="cleaned")
a.set_ylabel(p)
a.legend(loc="upper right")
ax[1].set_xlabel("hour")
plt.suptitle(f"{sid} before / after preprocessing (synthetic)")
plt.tight_layout()
plt.show()
In [ ]:
"""가정 기준치(주의·위험)를 넘는 항목이 있으면 음용 부적합 의심으로 판정하는 1차 필터링 Baseline 알고리즘을 구현하고, 기준 벤치마크 데이터에서 정확도·오차 유형·판정 근거·처리 지연을 확인한다."""
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
# ---------------------------------------------------------------------------
# [교육용가상자료] 야전 급수원 가상 수질 센서 데이터 생성기 (실제 부대·위치·검사 결과 없음)
# 가상 급수원 8곳 x 14일 x 1시간 간격. 실제 수질(true_*)·센서 측정값·오염 사건·센서 고장 정답을 함께 만든다.
# 기준치는 교육용 가정값이며, 실제 음용 판정은 공인 검사 결과와 의무(예방의학) 담당 기준에 따른다.
# ---------------------------------------------------------------------------
PARAMS = ["ph", "turbidity", "chlorine", "metal", "ecoli"]
SOURCES = [("W01", "water_truck", 1.0, 2.0), ("W02", "water_truck", 3.5, 1.2), ("W03", "purification_unit", 2.2, 4.1),
("W04", "purification_unit", 5.0, 3.3), ("W05", "well", 0.8, 5.6), ("W06", "well", 4.4, 6.0),
("W07", "river_intake", 6.3, 1.8), ("W08", "river_intake", 6.9, 4.9)]
BASE = {"water_truck": (7.2, 0.25, 0.75, 1.5), "purification_unit": (7.0, 0.20, 0.85, 1.2),
"well": (6.8, 0.45, 0.50, 3.0), "river_intake": (7.3, 0.55, 0.60, 2.0)} # pH, 탁도(NTU), 잔류염소(mg/L), 중금속(ug/L)
LIMITS = {"ph": (5.8, 8.5), "turbidity": (1.0, 2.0), "chlorine": (0.2, 0.1), "metal": (5.0, 10.0), "ecoli": (1, 1)} # (주의, 위험) 가정 기준
EVENT_TYPES = { # 오염 사건: 변수별 최대 변화량, 최대치 도달 시간(h)
"sewage": ({"turbidity": 2.8, "chlorine": -0.5, "ecoli": 30, "ph": -0.3}, 4),
"metal": ({"metal": 22.0, "ph": -0.5}, 4),
"disinfect_fail": ({"chlorine": -0.8, "ecoli": 6}, 6),
"chemical": ({"ph": 1.7, "turbidity": 1.0}, 3),
"creeping_metal": ({"metal": 13.0}, 18),
}
FAULT_TYPES = ["stuck", "spike", "drift", "dropout"]
NOISE = {"ph": 0.04, "turbidity": 0.06, "chlorine": 0.03, "metal": 0.35}
DIGITS = {"ph": 2, "turbidity": 2, "chlorine": 2, "metal": 1}
def make_dataset(days=14, seed=7, n_events=6, n_faults=10, events=None, faults=None, sources=None):
"""events: [(급수원 번호, 사건 유형, 시작 시각 h, 지속 h, 강도)], faults: [(급수원 번호, 고장 유형, 항목, 시작 h, 지속 h)]"""
rng = np.random.default_rng(seed)
sources = sources or SOURCES
T = days * 24
tt = np.arange(T)
hour = tt % 24
night = (hour < 6) | (hour >= 20)
rain = np.zeros(T)
for _ in range(max(1, days // 4)):
s0, d = int(rng.integers(0, T - 14)), int(rng.integers(6, 14))
rain[s0:s0 + d] = np.maximum(rain[s0:s0 + d], rng.uniform(0.6, 1.0))
if events is None:
events = [(int(rng.integers(len(sources))), str(rng.choice(list(EVENT_TYPES))), int(rng.integers(min(24, T // 4), max(T // 4 + 1, T - 48))),
int(rng.integers(10, 30)), 1.0) for _ in range(n_events)]
if faults is None:
faults = [(int(rng.integers(len(sources))), str(rng.choice(FAULT_TYPES)), str(rng.choice(["ph", "turbidity", "chlorine", "metal"])),
int(rng.integers(12, max(13, T - 30))), int(rng.integers(6, 24))) for _ in range(n_faults)]
out = []
for si, (sid, stype, x, y) in enumerate(sources):
ph0, tb0, cl0, mt0 = BASE[stype]
tr = {"ph": ph0 + 0.08 * np.sin(2 * np.pi * hour / 24) + rng.normal(0, 0.03, T),
"turbidity": tb0 * (1 + 0.1 * rng.normal(0, 1, T)),
"chlorine": (cl0 * np.exp(-0.022 * hour) if stype == "water_truck" else cl0 + rng.normal(0, 0.03, T)),
"metal": mt0 + rng.normal(0, 0.2, T), "ecoli": np.zeros(T)}
if stype in ("river_intake", "well"):
tr["turbidity"] = tr["turbidity"] + rain * (0.9 if stype == "river_intake" else 0.45)
f_all, etype = np.zeros(T), np.array(["none"] * T, dtype=object)
for (s_i, et, onset, dur, scale) in events:
if s_i != si:
continue
delta, ramp = EVENT_TYPES[et]
f = np.clip((tt - onset + 1) / ramp, 0, 1) * ((tt >= onset) & (tt < onset + dur + ramp))
for k, v in delta.items():
tr[k] = tr[k] + v * f * scale
etype[f > 0] = et
f_all = np.maximum(f_all, f)
tr["chlorine"] = np.clip(tr["chlorine"], 0.02, None)
tr["turbidity"] = np.clip(tr["turbidity"], 0.05, None)
tr["metal"] = np.clip(tr["metal"], 0.1, None)
grade = np.where(f_all >= 0.2, 2, 0)
mild = (tr["turbidity"] > LIMITS["turbidity"][0]) | (tr["chlorine"] < LIMITS["chlorine"][0]) | (tr["metal"] > LIMITS["metal"][0])
grade = np.where((grade == 0) & mild, 1, grade)
ms = {}
for p in ["ph", "turbidity", "chlorine", "metal"]:
sd = NOISE[p] * np.where(night, 1.5, 1.0) * (tr[p] if p == "turbidity" else 1)
ms[p] = np.round(np.clip(tr[p] + rng.normal(0, 1, T) * sd, 0, None), DIGITS[p])
ms["ecoli"] = (rng.poisson(tr["ecoli"]) + (rng.random(T) < 0.004)).astype(float)
fault, fparam = np.array(["none"] * T, dtype=object), np.array(["-"] * T, dtype=object)
for (s_i, ft, p, s0, d) in faults:
if s_i != si:
continue
seg = slice(s0, min(T, s0 + d))
if ft == "stuck":
ms[p][seg] = ms[p][s0]
elif ft == "spike":
seg = slice(s0, s0 + 1)
ms[p][s0] = {"ph": 11.8, "turbidity": 38.0, "chlorine": 0.0, "metal": 75.0}[p]
elif ft == "drift":
ramp = np.linspace(0, 1, seg.stop - s0)
ms[p][seg] = np.clip(ms[p][seg] + ramp * {"ph": -1.6, "turbidity": 2.6, "chlorine": -0.75, "metal": 12.0}[p], 0, None).round(DIGITS[p])
elif ft == "dropout":
for q in PARAMS:
ms[q][seg] = np.nan
p = "all"
fault[seg], fparam[seg] = ft, p
df = pd.DataFrame({"source_id": sid, "source_type": stype, "x_km": x, "y_km": y, "t": tt,
"cond": np.where(rain > 0, "storm", np.where(night, "night", "day")),
**{p: ms[p] for p in PARAMS}, **{f"true_{p}": tr[p] for p in PARAMS},
"event": etype, "fault": fault, "fault_param": fparam, "grade": grade})
out.append(df)
return pd.concat(out, ignore_index=True)
# ---------------------------------------------------------------------------
# ---- 전처리 공통 함수: 물리 범위 점검 -> 고정값(stuck) 표시 -> 단발 스파이크 제거 -> 결측 보정 ----
PHYS = {"ph": (0, 14), "turbidity": (0, 1000), "chlorine": (0, 10), "metal": (0, 1000), "ecoli": (0, 100000)}
SPIKE_K = {"day": 6.0, "night": 7.0, "storm": 9.0} # 상황별(주간/야간/악천후) 스파이크 판정 배수
MIN_JUMP = {"ph": 0.8, "turbidity": 1.5, "chlorine": 0.3, "metal": 6.0}
STUCK_RUN, SHORT_GAP = 6, 3
def clean(df, spike_k=None, min_jump=None):
"""급수원별 시계열을 정제하고 항목별 처리 내역(<항목>_flag: range/stuck/spike/interp/long_gap)을 남긴다."""
# [TODO : 학생 실습 3] 음용 적합/부적합 판정 1차 기준치 필터링 Baseline 알고리즘 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
GRADE_NAME = {0: "안전", 1: "주의", 2: "위험"}
def rule_grade(df):
"""기준치 1차 필터: 위험(음용 부적합 의심) = 한 항목이라도 위험 기준 초과, 주의 = 주의 기준 초과."""
# [TODO : 학생 실습 3] 음용 적합/부적합 판정 1차 기준치 필터링 Baseline 알고리즘 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def reasons(row):
r = []
if not LIMITS["ph"][0] <= row["ph"] <= LIMITS["ph"][1]:
r.append(f"pH {row['ph']:.2f}")
if row["turbidity"] > LIMITS["turbidity"][0]:
r.append(f"탁도 {row['turbidity']:.2f}NTU")
if row["chlorine"] < LIMITS["chlorine"][0]:
r.append(f"잔류염소 {row['chlorine']:.2f}mg/L")
if row["metal"] > LIMITS["metal"][0]:
r.append(f"중금속 {row['metal']:.1f}ug/L")
if row["ecoli"] >= 1:
r.append(f"대장균 {row['ecoli']:.0f}")
return ", ".join(r) or "-"
FEATS = [f"{p}{s}" for p in ["ph", "turbidity", "chlorine", "metal"] for s in ["", "_d1", "_d3", "_dev", "_sd6"]] + \
["ecoli", "ecoli_3h", "storm", "night", "n_flag", "truck", "purif", "well", "river"]
def add_features(c):
"""시간 순서를 지키는(과거 값만 쓰는) 급수원별 특징: 변화량, 24시간 기준선 대비 편차, 6시간 변동성, 대장균 누적 등."""
c = c.sort_values(["source_id", "t"]).reset_index(drop=True)
g = c.groupby("source_id")
for p in ["ph", "turbidity", "chlorine", "metal"]:
c[f"{p}_d1"] = g[p].diff().fillna(0)
c[f"{p}_d3"] = g[p].diff(3).fillna(0)
base = g[p].transform(lambda s: s.shift(1).rolling(24, min_periods=4).median())
c[f"{p}_dev"] = (c[p] - base).fillna(0)
c[f"{p}_sd6"] = g[p].transform(lambda s: s.rolling(6, min_periods=2).std()).fillna(0)
c["ecoli_3h"] = g["ecoli"].transform(lambda s: s.rolling(3, min_periods=1).sum())
c["storm"], c["night"] = (c["cond"] == "storm").astype(int), (c["cond"] == "night").astype(int)
flags = [f"{p}_flag" for p in PARAMS if f"{p}_flag" in c]
c["n_flag"] = sum(c[f].isin(["stuck", "spike", "long_gap", "range"]).astype(int) for f in flags) if flags else 0
for k, name in [("truck", "water_truck"), ("purif", "purification_unit"), ("well", "well"), ("river", "river_intake")]:
c[k] = (c["source_type"] == name).astype(int)
return c
# 기준 벤치마크 데이터셋: 가상 급수원 8곳 x 14일, 오염 사건 10건, 센서 고장 14건 (정답 등급 grade 포함)
bench = make_dataset(days=14, seed=11, n_events=10, n_faults=14)
truth_raw = bench.sort_values(["source_id", "t"])["grade"].to_numpy()
# 1) 전처리 없이(결측은 0으로) vs 전처리 후 Baseline 판정
t0 = time.perf_counter()
c = clean(bench)
c["pred"] = rule_grade(c)
dt = time.perf_counter() - t0
raw = bench.sort_values(["source_id", "t"]).fillna({p: 0 for p in PARAMS})
acc_raw = (rule_grade(raw) == truth_raw).mean()
acc = (c["pred"] == c["grade"]).mean()
print(f"[교육용가상자료] 벤치마크 {len(c):,}건(급수원 8곳 x 336시간)")
print(f"정확도: 전처리 없음 {acc_raw:.1%} -> 전처리 + 기준치 필터 {acc:.1%} (목표 75% 이상: {'달성' if acc >= 0.75 else '미달'})")
print(f"처리 지연: 전체 {dt * 1000:.0f}ms, 1건당 {dt / len(c) * 1000:.3f}ms")
# 2) 혼동행렬과 오차 유형
cm = pd.crosstab(c["grade"].map(GRADE_NAME), c["pred"].map(GRADE_NAME), rownames=["실제"], colnames=["판정"])
cm = cm.reindex(index=["안전", "주의", "위험"], columns=["안전", "주의", "위험"], fill_value=0)
print("\n[혼동행렬]")
print(cm.to_string())
danger = c["grade"] == 2
rec = ((c["pred"] == 2) & danger).sum() / danger.sum()
far = ((c["pred"] == 2) & ~danger).sum() / max((c["pred"] == 2).sum(), 1)
print(f"위험 재현율 {rec:.1%}, 위험 경보 중 오경보 비율 {far:.1%}")
miss = c[danger & (c["pred"] != 2)]
fa = c[~danger & (c["pred"] == 2)]
print("놓친 위험(사건 유형별):", miss["event"].value_counts().to_dict())
print("오경보(센서 고장 유형별):", fa["fault"].value_counts().to_dict())
# 3) 판정 근거(투명성): 위험 판정 사례 일부의 초과 항목
print("\n[위험 판정 근거 예시]")
show = c[c["pred"] == 2].groupby("source_id").head(1).head(5)
for _, r in show.iterrows():
print(f" {r['source_id']} D+{int(r['t']) // 24:02d} {int(r['t']) % 24:02d}시: {reasons(r)} -> 실제 {GRADE_NAME[r['grade']]}({r['event']})")
# 4) 휴리스틱 1차 튜닝: 위험 판정을 '2시간 연속 초과'로 바꾸면 오경보는 줄지만 놓침이 늘어난다
c["pred2"] = c["pred"]
prev = c.groupby("source_id")["pred"].shift(1).fillna(0)
c.loc[(c["pred"] == 2) & (prev < 2), "pred2"] = 1
for name, col in [("1시간 즉시 판정", "pred"), ("2시간 연속 확인", "pred2")]:
p = c[col]
print(f"{name}: 정확도 {(p == c['grade']).mean():.1%}, 위험 재현율 {((p == 2) & danger).sum() / danger.sum():.1%}, "
f"오경보 {int(((p == 2) & ~danger).sum())}건")
print("-> 음용수는 놓침이 더 위험하므로 Baseline은 즉시 판정을 유지하고, 오경보는 미션 5 신뢰도 지수로 다룬다.")
print("※ 기준치는 교육용 가정값이다. 실제 음용 적합 판정은 공식 수질 기준과 의무(예방의학)·환경 담당 확인에 따른다.")
sid = miss["source_id"].mode().iloc[0] if len(miss) else "W01"
s = c[c["source_id"] == sid]
fig, ax = plt.subplots(figsize=(10, 3.5))
ax.plot(s["t"], s["grade"], color="black", lw=2, label="true grade")
ax.plot(s["t"], s["pred"] + 0.05, color="tab:red", lw=1, label="baseline grade")
ax.set_yticks([0, 1, 2], ["safe", "caution", "danger"]); ax.set_xlabel("hour"); ax.legend()
ax.set_title(f"{sid} baseline vs true grade (synthetic)")
plt.tight_layout(); plt.show()
In [ ]:
"""시계열 특징(변화량·기준선 편차·변동성·대장균 누적·센서 처리 표시)으로 급수 안전도 위험 등급(안전·주의·위험)을 분류하는 랜덤 포레스트 모델을 학습하고, 복합 위협 시나리오 10종(오염 사건 7종, 센서 교란 3종)으로 재현율·오경보·처리 지연을 검증한다."""
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.ensemble import RandomForestClassifier
# ---------------------------------------------------------------------------
# [교육용가상자료] 야전 급수원 가상 수질 센서 데이터 생성기 (실제 부대·위치·검사 결과 없음)
# 가상 급수원 8곳 x 14일 x 1시간 간격. 실제 수질(true_*)·센서 측정값·오염 사건·센서 고장 정답을 함께 만든다.
# 기준치는 교육용 가정값이며, 실제 음용 판정은 공인 검사 결과와 의무(예방의학) 담당 기준에 따른다.
# ---------------------------------------------------------------------------
PARAMS = ["ph", "turbidity", "chlorine", "metal", "ecoli"]
SOURCES = [("W01", "water_truck", 1.0, 2.0), ("W02", "water_truck", 3.5, 1.2), ("W03", "purification_unit", 2.2, 4.1),
("W04", "purification_unit", 5.0, 3.3), ("W05", "well", 0.8, 5.6), ("W06", "well", 4.4, 6.0),
("W07", "river_intake", 6.3, 1.8), ("W08", "river_intake", 6.9, 4.9)]
BASE = {"water_truck": (7.2, 0.25, 0.75, 1.5), "purification_unit": (7.0, 0.20, 0.85, 1.2),
"well": (6.8, 0.45, 0.50, 3.0), "river_intake": (7.3, 0.55, 0.60, 2.0)} # pH, 탁도(NTU), 잔류염소(mg/L), 중금속(ug/L)
LIMITS = {"ph": (5.8, 8.5), "turbidity": (1.0, 2.0), "chlorine": (0.2, 0.1), "metal": (5.0, 10.0), "ecoli": (1, 1)} # (주의, 위험) 가정 기준
EVENT_TYPES = { # 오염 사건: 변수별 최대 변화량, 최대치 도달 시간(h)
"sewage": ({"turbidity": 2.8, "chlorine": -0.5, "ecoli": 30, "ph": -0.3}, 4),
"metal": ({"metal": 22.0, "ph": -0.5}, 4),
"disinfect_fail": ({"chlorine": -0.8, "ecoli": 6}, 6),
"chemical": ({"ph": 1.7, "turbidity": 1.0}, 3),
"creeping_metal": ({"metal": 13.0}, 18),
}
FAULT_TYPES = ["stuck", "spike", "drift", "dropout"]
NOISE = {"ph": 0.04, "turbidity": 0.06, "chlorine": 0.03, "metal": 0.35}
DIGITS = {"ph": 2, "turbidity": 2, "chlorine": 2, "metal": 1}
def make_dataset(days=14, seed=7, n_events=6, n_faults=10, events=None, faults=None, sources=None):
"""events: [(급수원 번호, 사건 유형, 시작 시각 h, 지속 h, 강도)], faults: [(급수원 번호, 고장 유형, 항목, 시작 h, 지속 h)]"""
rng = np.random.default_rng(seed)
sources = sources or SOURCES
T = days * 24
tt = np.arange(T)
hour = tt % 24
night = (hour < 6) | (hour >= 20)
rain = np.zeros(T)
for _ in range(max(1, days // 4)):
s0, d = int(rng.integers(0, T - 14)), int(rng.integers(6, 14))
rain[s0:s0 + d] = np.maximum(rain[s0:s0 + d], rng.uniform(0.6, 1.0))
if events is None:
events = [(int(rng.integers(len(sources))), str(rng.choice(list(EVENT_TYPES))), int(rng.integers(min(24, T // 4), max(T // 4 + 1, T - 48))),
int(rng.integers(10, 30)), 1.0) for _ in range(n_events)]
if faults is None:
faults = [(int(rng.integers(len(sources))), str(rng.choice(FAULT_TYPES)), str(rng.choice(["ph", "turbidity", "chlorine", "metal"])),
int(rng.integers(12, max(13, T - 30))), int(rng.integers(6, 24))) for _ in range(n_faults)]
out = []
for si, (sid, stype, x, y) in enumerate(sources):
ph0, tb0, cl0, mt0 = BASE[stype]
tr = {"ph": ph0 + 0.08 * np.sin(2 * np.pi * hour / 24) + rng.normal(0, 0.03, T),
"turbidity": tb0 * (1 + 0.1 * rng.normal(0, 1, T)),
"chlorine": (cl0 * np.exp(-0.022 * hour) if stype == "water_truck" else cl0 + rng.normal(0, 0.03, T)),
"metal": mt0 + rng.normal(0, 0.2, T), "ecoli": np.zeros(T)}
if stype in ("river_intake", "well"):
tr["turbidity"] = tr["turbidity"] + rain * (0.9 if stype == "river_intake" else 0.45)
f_all, etype = np.zeros(T), np.array(["none"] * T, dtype=object)
for (s_i, et, onset, dur, scale) in events:
if s_i != si:
continue
delta, ramp = EVENT_TYPES[et]
f = np.clip((tt - onset + 1) / ramp, 0, 1) * ((tt >= onset) & (tt < onset + dur + ramp))
for k, v in delta.items():
tr[k] = tr[k] + v * f * scale
etype[f > 0] = et
f_all = np.maximum(f_all, f)
tr["chlorine"] = np.clip(tr["chlorine"], 0.02, None)
tr["turbidity"] = np.clip(tr["turbidity"], 0.05, None)
tr["metal"] = np.clip(tr["metal"], 0.1, None)
grade = np.where(f_all >= 0.2, 2, 0)
mild = (tr["turbidity"] > LIMITS["turbidity"][0]) | (tr["chlorine"] < LIMITS["chlorine"][0]) | (tr["metal"] > LIMITS["metal"][0])
grade = np.where((grade == 0) & mild, 1, grade)
ms = {}
for p in ["ph", "turbidity", "chlorine", "metal"]:
sd = NOISE[p] * np.where(night, 1.5, 1.0) * (tr[p] if p == "turbidity" else 1)
ms[p] = np.round(np.clip(tr[p] + rng.normal(0, 1, T) * sd, 0, None), DIGITS[p])
ms["ecoli"] = (rng.poisson(tr["ecoli"]) + (rng.random(T) < 0.004)).astype(float)
fault, fparam = np.array(["none"] * T, dtype=object), np.array(["-"] * T, dtype=object)
for (s_i, ft, p, s0, d) in faults:
if s_i != si:
continue
seg = slice(s0, min(T, s0 + d))
if ft == "stuck":
ms[p][seg] = ms[p][s0]
elif ft == "spike":
seg = slice(s0, s0 + 1)
ms[p][s0] = {"ph": 11.8, "turbidity": 38.0, "chlorine": 0.0, "metal": 75.0}[p]
elif ft == "drift":
ramp = np.linspace(0, 1, seg.stop - s0)
ms[p][seg] = np.clip(ms[p][seg] + ramp * {"ph": -1.6, "turbidity": 2.6, "chlorine": -0.75, "metal": 12.0}[p], 0, None).round(DIGITS[p])
elif ft == "dropout":
for q in PARAMS:
ms[q][seg] = np.nan
p = "all"
fault[seg], fparam[seg] = ft, p
df = pd.DataFrame({"source_id": sid, "source_type": stype, "x_km": x, "y_km": y, "t": tt,
"cond": np.where(rain > 0, "storm", np.where(night, "night", "day")),
**{p: ms[p] for p in PARAMS}, **{f"true_{p}": tr[p] for p in PARAMS},
"event": etype, "fault": fault, "fault_param": fparam, "grade": grade})
out.append(df)
return pd.concat(out, ignore_index=True)
# ---------------------------------------------------------------------------
# ---- 전처리 공통 함수: 물리 범위 점검 -> 고정값(stuck) 표시 -> 단발 스파이크 제거 -> 결측 보정 ----
PHYS = {"ph": (0, 14), "turbidity": (0, 1000), "chlorine": (0, 10), "metal": (0, 1000), "ecoli": (0, 100000)}
SPIKE_K = {"day": 6.0, "night": 7.0, "storm": 9.0} # 상황별(주간/야간/악천후) 스파이크 판정 배수
MIN_JUMP = {"ph": 0.8, "turbidity": 1.5, "chlorine": 0.3, "metal": 6.0}
STUCK_RUN, SHORT_GAP = 6, 3
def clean(df, spike_k=None, min_jump=None):
"""급수원별 시계열을 정제하고 항목별 처리 내역(<항목>_flag: range/stuck/spike/interp/long_gap)을 남긴다."""
# [TODO : 학생 실습 4] 복합 오염원 유입 징후 및 급수 안전도 위험 등급 분류 머신러닝 모델 고도화 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
GRADE_NAME = {0: "안전", 1: "주의", 2: "위험"}
def rule_grade(df):
"""기준치 1차 필터: 위험(음용 부적합 의심) = 한 항목이라도 위험 기준 초과, 주의 = 주의 기준 초과."""
# [TODO : 학생 실습 4] 복합 오염원 유입 징후 및 급수 안전도 위험 등급 분류 머신러닝 모델 고도화 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def reasons(row):
r = []
if not LIMITS["ph"][0] <= row["ph"] <= LIMITS["ph"][1]:
r.append(f"pH {row['ph']:.2f}")
if row["turbidity"] > LIMITS["turbidity"][0]:
r.append(f"탁도 {row['turbidity']:.2f}NTU")
if row["chlorine"] < LIMITS["chlorine"][0]:
r.append(f"잔류염소 {row['chlorine']:.2f}mg/L")
if row["metal"] > LIMITS["metal"][0]:
r.append(f"중금속 {row['metal']:.1f}ug/L")
if row["ecoli"] >= 1:
r.append(f"대장균 {row['ecoli']:.0f}")
return ", ".join(r) or "-"
FEATS = [f"{p}{s}" for p in ["ph", "turbidity", "chlorine", "metal"] for s in ["", "_d1", "_d3", "_dev", "_sd6"]] + \
["ecoli", "ecoli_3h", "storm", "night", "n_flag", "truck", "purif", "well", "river"]
def add_features(c):
"""시간 순서를 지키는(과거 값만 쓰는) 급수원별 특징: 변화량, 24시간 기준선 대비 편차, 6시간 변동성, 대장균 누적 등."""
c = c.sort_values(["source_id", "t"]).reset_index(drop=True)
g = c.groupby("source_id")
for p in ["ph", "turbidity", "chlorine", "metal"]:
c[f"{p}_d1"] = g[p].diff().fillna(0)
c[f"{p}_d3"] = g[p].diff(3).fillna(0)
base = g[p].transform(lambda s: s.shift(1).rolling(24, min_periods=4).median())
c[f"{p}_dev"] = (c[p] - base).fillna(0)
c[f"{p}_sd6"] = g[p].transform(lambda s: s.rolling(6, min_periods=2).std()).fillna(0)
c["ecoli_3h"] = g["ecoli"].transform(lambda s: s.rolling(3, min_periods=1).sum())
c["storm"], c["night"] = (c["cond"] == "storm").astype(int), (c["cond"] == "night").astype(int)
flags = [f"{p}_flag" for p in PARAMS if f"{p}_flag" in c]
c["n_flag"] = sum(c[f].isin(["stuck", "spike", "long_gap", "range"]).astype(int) for f in flags) if flags else 0
for k, name in [("truck", "water_truck"), ("purif", "purification_unit"), ("well", "well"), ("river", "river_intake")]:
c[k] = (c["source_type"] == name).astype(int)
return c
def prepare(df):
c = add_features(clean(df))
c["rule"] = rule_grade(c)
return c
# 1) 학습(가상 2주 x 2세트)과 평가(벤치마크 seed 11)
train = pd.concat([prepare(make_dataset(seed=s, n_events=14, n_faults=14)) for s in (7, 8)], ignore_index=True)
test = prepare(make_dataset(seed=11, n_events=10, n_faults=14))
model = RandomForestClassifier(n_estimators=200, max_depth=10, class_weight="balanced_subsample", random_state=0, n_jobs=1)
model.fit(train[FEATS], train["grade"])
t0 = time.perf_counter()
test["ml"] = model.predict(test[FEATS])
batch_ms = (time.perf_counter() - t0) * 1000
t0 = time.perf_counter()
for i in range(20):
model.predict(test[FEATS].iloc[[i]])
single_ms = (time.perf_counter() - t0) / 20 * 1000
d = test["grade"] == 2
print(f"[교육용가상자료] 학습 {len(train):,}건 / 평가 {len(test):,}건, 특징 {len(FEATS)}개")
for name, col in [("Baseline 기준치", "rule"), ("랜덤 포레스트", "ml")]:
p = test[col]
print(f"{name}: 정확도 {(p == test['grade']).mean():.1%}, 위험 재현율 {((p == 2) & d).sum() / d.sum():.1%}, "
f"위험 오경보 {int(((p == 2) & ~d).sum())}건")
rec = ((test["ml"] == 2) & d).sum() / d.sum()
print(f"복합 위협 재현율 {rec:.1%} (목표 85% 이상: {'달성' if rec >= 0.85 else '미달'}), "
f"처리 지연 1건 {single_ms:.1f}ms / 전체 {batch_ms:.0f}ms (목표 200ms 이하: {'달성' if single_ms <= 200 else '미달'})")
print("Baseline이 놓친 서서히 오르는 중금속(creeping_metal) 중 모델 검출:",
f"{int(((test['ml'] == 2) & d & (test['rule'] != 2) & (test['event'] == 'creeping_metal')).sum())}"
f"/{int((d & (test['rule'] != 2) & (test['event'] == 'creeping_metal')).sum())}시간")
# 2) 오경보 유발 패턴: 위험 오경보가 어떤 센서 고장에서 나왔는가 -> 특징 공간 재설계 단서
fa = test[(test["ml"] == 2) & ~d]
print("모델 위험 오경보의 센서 고장 유형:", fa["fault"].value_counts().to_dict(), "-> drift(서서히 틀어짐)는 미션 5 교차 검증으로 보완")
imp = pd.Series(model.feature_importances_, FEATS).sort_values(ascending=False).head(6)
print("중요 특징 Top6:", ", ".join(f"{k} {v:.2f}" for k, v in imp.items()))
# 3) 복합 위협 시나리오 10종: (급수원, 사건) / 센서 교란은 오염이 없으므로 위험 경보가 나오면 안 된다
SCEN = [("하수 유입", [(6, "sewage", 30, 14, 1.0)], []), ("중금속 급증", [(4, "metal", 30, 14, 1.0)], []),
("소독 실패", [(0, "disinfect_fail", 30, 14, 1.0)], []), ("화학 오염(pH 급변)", [(2, "chemical", 30, 14, 1.0)], []),
("중금속 서서히 증가", [(5, "creeping_metal", 20, 24, 1.0)], []),
("하수+중금속 동시", [(7, "sewage", 30, 14, 1.0), (7, "metal", 34, 10, 1.0)], []),
("약한 하수 유입(강도 60%)", [(3, "sewage", 30, 14, 0.6)], []),
("교란: 탁도 센서 드리프트", [], [(1, "drift", "turbidity", 24, 18)]),
("교란: 잔류염소 단발 스파이크", [], [(0, "spike", "chlorine", 40, 1)]),
("교란: 통신 단절(전 항목 결측)", [], [(6, "dropout", "ph", 30, 10)])]
rows = []
for name, ev, ft in SCEN:
c = prepare(make_dataset(days=3, seed=21, events=ev, faults=ft))
c["ml"] = model.predict(c[FEATS])
if ev:
sid = SOURCES[ev[0][0]][0]; on = ev[0][2]
hit = c[(c["source_id"] == sid) & (c["ml"] == 2) & (c["t"] >= on)]
lag = int(hit["t"].min() - on) if len(hit) else None
ok = lag is not None and lag <= 8
rows.append((name, "오염", f"{lag}h" if lag is not None else "미탐", "PASS" if ok else "FAIL"))
else:
sid = SOURCES[ft[0][0]][0]
n = int(((c["source_id"] == sid) & (c["ml"] == 2)).sum())
rows.append((name, "교란", f"위험 오경보 {n}h", "PASS" if n <= 2 else "FAIL"))
res = pd.DataFrame(rows, columns=["시나리오", "구분", "탐지 지연/오경보", "판정"])
print("\n[복합 위협 시나리오 10종]")
print(res.to_string(index=False))
n_pass = int((res["판정"] == "PASS").sum())
print(f"통과 {n_pass}/10 (오염: 시작 후 8시간 내 위험 판정, 교란: 위험 오경보 2시간 이하)")
plt.figure(figsize=(8, 4))
imp.iloc[::-1].plot.barh(color="tab:blue")
plt.title("Random forest feature importance (top 6, synthetic)"); plt.xlabel("importance")
plt.tight_layout(); plt.show()
In [ ]:
"""위험 경보마다 다항목 동시 변화·연속 지속·센서 상태·단일 항목 드리프트 여부로 수질 안전 신뢰도 지수(0~100, High/Med/Low)를 산출해, 센서 오류에 의한 급수 중단(단수) 오경보를 줄이고 ROC-AUC·임계값 민감도·산출 일관성을 확인한다."""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import roc_auc_score, roc_curve
# ---------------------------------------------------------------------------
# [교육용가상자료] 야전 급수원 가상 수질 센서 데이터 생성기 (실제 부대·위치·검사 결과 없음)
# 가상 급수원 8곳 x 14일 x 1시간 간격. 실제 수질(true_*)·센서 측정값·오염 사건·센서 고장 정답을 함께 만든다.
# 기준치는 교육용 가정값이며, 실제 음용 판정은 공인 검사 결과와 의무(예방의학) 담당 기준에 따른다.
# ---------------------------------------------------------------------------
PARAMS = ["ph", "turbidity", "chlorine", "metal", "ecoli"]
SOURCES = [("W01", "water_truck", 1.0, 2.0), ("W02", "water_truck", 3.5, 1.2), ("W03", "purification_unit", 2.2, 4.1),
("W04", "purification_unit", 5.0, 3.3), ("W05", "well", 0.8, 5.6), ("W06", "well", 4.4, 6.0),
("W07", "river_intake", 6.3, 1.8), ("W08", "river_intake", 6.9, 4.9)]
BASE = {"water_truck": (7.2, 0.25, 0.75, 1.5), "purification_unit": (7.0, 0.20, 0.85, 1.2),
"well": (6.8, 0.45, 0.50, 3.0), "river_intake": (7.3, 0.55, 0.60, 2.0)} # pH, 탁도(NTU), 잔류염소(mg/L), 중금속(ug/L)
LIMITS = {"ph": (5.8, 8.5), "turbidity": (1.0, 2.0), "chlorine": (0.2, 0.1), "metal": (5.0, 10.0), "ecoli": (1, 1)} # (주의, 위험) 가정 기준
EVENT_TYPES = { # 오염 사건: 변수별 최대 변화량, 최대치 도달 시간(h)
"sewage": ({"turbidity": 2.8, "chlorine": -0.5, "ecoli": 30, "ph": -0.3}, 4),
"metal": ({"metal": 22.0, "ph": -0.5}, 4),
"disinfect_fail": ({"chlorine": -0.8, "ecoli": 6}, 6),
"chemical": ({"ph": 1.7, "turbidity": 1.0}, 3),
"creeping_metal": ({"metal": 13.0}, 18),
}
FAULT_TYPES = ["stuck", "spike", "drift", "dropout"]
NOISE = {"ph": 0.04, "turbidity": 0.06, "chlorine": 0.03, "metal": 0.35}
DIGITS = {"ph": 2, "turbidity": 2, "chlorine": 2, "metal": 1}
def make_dataset(days=14, seed=7, n_events=6, n_faults=10, events=None, faults=None, sources=None):
"""events: [(급수원 번호, 사건 유형, 시작 시각 h, 지속 h, 강도)], faults: [(급수원 번호, 고장 유형, 항목, 시작 h, 지속 h)]"""
rng = np.random.default_rng(seed)
sources = sources or SOURCES
T = days * 24
tt = np.arange(T)
hour = tt % 24
night = (hour < 6) | (hour >= 20)
rain = np.zeros(T)
for _ in range(max(1, days // 4)):
s0, d = int(rng.integers(0, T - 14)), int(rng.integers(6, 14))
rain[s0:s0 + d] = np.maximum(rain[s0:s0 + d], rng.uniform(0.6, 1.0))
if events is None:
events = [(int(rng.integers(len(sources))), str(rng.choice(list(EVENT_TYPES))), int(rng.integers(min(24, T // 4), max(T // 4 + 1, T - 48))),
int(rng.integers(10, 30)), 1.0) for _ in range(n_events)]
if faults is None:
faults = [(int(rng.integers(len(sources))), str(rng.choice(FAULT_TYPES)), str(rng.choice(["ph", "turbidity", "chlorine", "metal"])),
int(rng.integers(12, max(13, T - 30))), int(rng.integers(6, 24))) for _ in range(n_faults)]
out = []
for si, (sid, stype, x, y) in enumerate(sources):
ph0, tb0, cl0, mt0 = BASE[stype]
tr = {"ph": ph0 + 0.08 * np.sin(2 * np.pi * hour / 24) + rng.normal(0, 0.03, T),
"turbidity": tb0 * (1 + 0.1 * rng.normal(0, 1, T)),
"chlorine": (cl0 * np.exp(-0.022 * hour) if stype == "water_truck" else cl0 + rng.normal(0, 0.03, T)),
"metal": mt0 + rng.normal(0, 0.2, T), "ecoli": np.zeros(T)}
if stype in ("river_intake", "well"):
tr["turbidity"] = tr["turbidity"] + rain * (0.9 if stype == "river_intake" else 0.45)
f_all, etype = np.zeros(T), np.array(["none"] * T, dtype=object)
for (s_i, et, onset, dur, scale) in events:
if s_i != si:
continue
delta, ramp = EVENT_TYPES[et]
f = np.clip((tt - onset + 1) / ramp, 0, 1) * ((tt >= onset) & (tt < onset + dur + ramp))
for k, v in delta.items():
tr[k] = tr[k] + v * f * scale
etype[f > 0] = et
f_all = np.maximum(f_all, f)
tr["chlorine"] = np.clip(tr["chlorine"], 0.02, None)
tr["turbidity"] = np.clip(tr["turbidity"], 0.05, None)
tr["metal"] = np.clip(tr["metal"], 0.1, None)
grade = np.where(f_all >= 0.2, 2, 0)
mild = (tr["turbidity"] > LIMITS["turbidity"][0]) | (tr["chlorine"] < LIMITS["chlorine"][0]) | (tr["metal"] > LIMITS["metal"][0])
grade = np.where((grade == 0) & mild, 1, grade)
ms = {}
for p in ["ph", "turbidity", "chlorine", "metal"]:
sd = NOISE[p] * np.where(night, 1.5, 1.0) * (tr[p] if p == "turbidity" else 1)
ms[p] = np.round(np.clip(tr[p] + rng.normal(0, 1, T) * sd, 0, None), DIGITS[p])
ms["ecoli"] = (rng.poisson(tr["ecoli"]) + (rng.random(T) < 0.004)).astype(float)
fault, fparam = np.array(["none"] * T, dtype=object), np.array(["-"] * T, dtype=object)
for (s_i, ft, p, s0, d) in faults:
if s_i != si:
continue
seg = slice(s0, min(T, s0 + d))
if ft == "stuck":
ms[p][seg] = ms[p][s0]
elif ft == "spike":
seg = slice(s0, s0 + 1)
ms[p][s0] = {"ph": 11.8, "turbidity": 38.0, "chlorine": 0.0, "metal": 75.0}[p]
elif ft == "drift":
ramp = np.linspace(0, 1, seg.stop - s0)
ms[p][seg] = np.clip(ms[p][seg] + ramp * {"ph": -1.6, "turbidity": 2.6, "chlorine": -0.75, "metal": 12.0}[p], 0, None).round(DIGITS[p])
elif ft == "dropout":
for q in PARAMS:
ms[q][seg] = np.nan
p = "all"
fault[seg], fparam[seg] = ft, p
df = pd.DataFrame({"source_id": sid, "source_type": stype, "x_km": x, "y_km": y, "t": tt,
"cond": np.where(rain > 0, "storm", np.where(night, "night", "day")),
**{p: ms[p] for p in PARAMS}, **{f"true_{p}": tr[p] for p in PARAMS},
"event": etype, "fault": fault, "fault_param": fparam, "grade": grade})
out.append(df)
return pd.concat(out, ignore_index=True)
# ---------------------------------------------------------------------------
# ---- 전처리 공통 함수: 물리 범위 점검 -> 고정값(stuck) 표시 -> 단발 스파이크 제거 -> 결측 보정 ----
PHYS = {"ph": (0, 14), "turbidity": (0, 1000), "chlorine": (0, 10), "metal": (0, 1000), "ecoli": (0, 100000)}
SPIKE_K = {"day": 6.0, "night": 7.0, "storm": 9.0} # 상황별(주간/야간/악천후) 스파이크 판정 배수
MIN_JUMP = {"ph": 0.8, "turbidity": 1.5, "chlorine": 0.3, "metal": 6.0}
STUCK_RUN, SHORT_GAP = 6, 3
def clean(df, spike_k=None, min_jump=None):
"""급수원별 시계열을 정제하고 항목별 처리 내역(<항목>_flag: range/stuck/spike/interp/long_gap)을 남긴다."""
# [TODO : 학생 실습 5] 센서 오류에 의한 단수 오경보 방지 및 수질 안전 신뢰도 지수 산출 모듈 개발 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
GRADE_NAME = {0: "안전", 1: "주의", 2: "위험"}
def rule_grade(df):
"""기준치 1차 필터: 위험(음용 부적합 의심) = 한 항목이라도 위험 기준 초과, 주의 = 주의 기준 초과."""
# [TODO : 학생 실습 5] 센서 오류에 의한 단수 오경보 방지 및 수질 안전 신뢰도 지수 산출 모듈 개발 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def reasons(row):
r = []
if not LIMITS["ph"][0] <= row["ph"] <= LIMITS["ph"][1]:
r.append(f"pH {row['ph']:.2f}")
if row["turbidity"] > LIMITS["turbidity"][0]:
r.append(f"탁도 {row['turbidity']:.2f}NTU")
if row["chlorine"] < LIMITS["chlorine"][0]:
r.append(f"잔류염소 {row['chlorine']:.2f}mg/L")
if row["metal"] > LIMITS["metal"][0]:
r.append(f"중금속 {row['metal']:.1f}ug/L")
if row["ecoli"] >= 1:
r.append(f"대장균 {row['ecoli']:.0f}")
return ", ".join(r) or "-"
FEATS = [f"{p}{s}" for p in ["ph", "turbidity", "chlorine", "metal"] for s in ["", "_d1", "_d3", "_dev", "_sd6"]] + \
["ecoli", "ecoli_3h", "storm", "night", "n_flag", "truck", "purif", "well", "river"]
def add_features(c):
"""시간 순서를 지키는(과거 값만 쓰는) 급수원별 특징: 변화량, 24시간 기준선 대비 편차, 6시간 변동성, 대장균 누적 등."""
c = c.sort_values(["source_id", "t"]).reset_index(drop=True)
g = c.groupby("source_id")
for p in ["ph", "turbidity", "chlorine", "metal"]:
c[f"{p}_d1"] = g[p].diff().fillna(0)
c[f"{p}_d3"] = g[p].diff(3).fillna(0)
base = g[p].transform(lambda s: s.shift(1).rolling(24, min_periods=4).median())
c[f"{p}_dev"] = (c[p] - base).fillna(0)
c[f"{p}_sd6"] = g[p].transform(lambda s: s.rolling(6, min_periods=2).std()).fillna(0)
c["ecoli_3h"] = g["ecoli"].transform(lambda s: s.rolling(3, min_periods=1).sum())
c["storm"], c["night"] = (c["cond"] == "storm").astype(int), (c["cond"] == "night").astype(int)
flags = [f"{p}_flag" for p in PARAMS if f"{p}_flag" in c]
c["n_flag"] = sum(c[f].isin(["stuck", "spike", "long_gap", "range"]).astype(int) for f in flags) if flags else 0
for k, name in [("truck", "water_truck"), ("purif", "purification_unit"), ("well", "well"), ("river", "river_intake")]:
c[k] = (c["source_type"] == name).astype(int)
return c
SCALE = {"ph": 0.35, "turbidity": 0.6, "chlorine": 0.2, "metal": 3.0} # 항목별 '의미 있는 편차' 크기(가정)
W = {"corrob": 40, "persist": 25, "health": 20, "nondrift": 15} # 신뢰도 가중치 매트릭스(합 100)
CUT = {"High": 60, "Med": 35} # High만 급수 중단 검토 요청으로 올린다
def prepare(df):
c = add_features(clean(df))
c["rule"] = rule_grade(c)
return c
def confidence(c, alert):
"""경보(alert=True) 행마다 신뢰도 점수와 구성 요소를 계산한다."""
dev = pd.DataFrame({p: (c[f"{p}_dev"].abs() > SCALE[p]) | (c[f"{p}_d3"].abs() > SCALE[p]) for p in SCALE})
n_dev = dev.sum(axis=1) + (c["ecoli_3h"] >= 2)
corrob = np.minimum(n_dev, 3) / 3
a = pd.Series(alert.astype(int), index=c.index)
run = a.groupby([c["source_id"], (a != a.groupby(c["source_id"]).shift()).cumsum()]).cumsum()
persist = np.minimum(run, 3) / 3
flagged = c["n_flag"].groupby(c["source_id"]).transform(lambda s: s.rolling(6, min_periods=1).max()) > 0
health = 1 - flagged.astype(float)
# 드리프트 의심: 한 항목만 24시간 기준선에서 벗어나고, 6시간 변동성이 작으며(매끄러운 변화) 대장균이 없다
g = c.groupby("source_id")
jitter = {p: g[f"{p}_d1"].transform(lambda s: s.rolling(6, min_periods=3).std()).fillna(0) for p in SCALE}
smooth = pd.concat([jitter[p] < SCALE[p] * 0.5 for p in SCALE], axis=1).all(axis=1) # 변화량이 일정 = 기울기만 있는 변화
nondrift = 1 - ((n_dev <= 1) & smooth & (c["ecoli_3h"] < 2)).astype(float)
score = W["corrob"] * corrob + W["persist"] * persist + W["health"] * health + W["nondrift"] * nondrift
level = np.where(score >= CUT["High"], "High", np.where(score >= CUT["Med"], "Med", "Low"))
return pd.DataFrame({"score": score.round(1), "level": level, "n_dev": n_dev}, index=c.index)
train = pd.concat([prepare(make_dataset(seed=s, n_events=14, n_faults=14)) for s in (7, 8)], ignore_index=True)
model = RandomForestClassifier(n_estimators=200, max_depth=10, class_weight="balanced_subsample", random_state=0, n_jobs=1)
model.fit(train[FEATS], train["grade"])
test = prepare(make_dataset(seed=11, n_events=10, n_faults=14))
alert = ((test["rule"] == 2) | (model.predict(test[FEATS]) == 2)).to_numpy() # Baseline 또는 모델이 위험 판정
conf = confidence(test, alert)
A = test[alert].join(conf)
A["real"] = A["grade"] == 2
far_before = (~A["real"]).mean()
stop = A[A["level"] == "High"]
far_after = (~stop["real"]).mean()
cut = 1 - far_after / far_before
print(f"[교육용가상자료] 위험 경보 {len(A)}건(실제 위험 {int(A['real'].sum())}건, 오경보 {int((~A['real']).sum())}건)")
print(f"급수 중단 검토 경보 오경보율(FAR): 신뢰도 적용 전 {far_before:.1%} -> High만 상신 {far_after:.1%} "
f"({cut:.0%} 감축, 목표 40% 이상: {'달성' if cut >= 0.4 else '미달'})")
print("\n[신뢰도 등급별 경보 수와 조치]")
tab = A.groupby("level").agg(경보=("real", "size"), 실제위험=("real", "sum"), 평균점수=("score", "mean"))
tab["조치"] = tab.index.map({"High": "급수 중단 검토 요청(HITL)", "Med": "즉시 재채수·현장 키트 검사", "Low": "센서 점검·재측정"})
print(tab.reindex(["High", "Med", "Low"]).round(1).to_string())
print("오경보의 원인(센서 고장 유형) x 등급:")
print(pd.crosstab(A.loc[~A["real"], "fault"], A.loc[~A["real"], "level"]).to_string())
miss_real = int((A["real"] & (A["level"] == "Low")).sum())
print(f"실제 위험인데 Low로 낮춰진 경보 {miss_real}건 -> Low도 재측정 대상이므로 무시되지 않는다")
auc = roc_auc_score(A["real"], A["score"])
print(f"\nROC-AUC(신뢰도 점수로 실제 위험 구분) {auc:.3f}")
print("[임계값 민감도] High 기준 / 상신 경보 / 오경보율 / 실제 위험 상신 비율")
for th in [45, 55, 60, 70, 80]:
s = A[A["score"] >= th]
print(f" {th:>3} / {len(s):>4} / {(~s['real']).mean():6.1%} / {s['real'].sum() / A['real'].sum():6.1%}")
# 산출 일관성: 같은 사건·고장 조건에서 측정 잡음만 바꿔 5회 재산출했을 때 실제 위험 경보의 등급 일치율
EV = [(6, "sewage", 30, 14, 1.0), (4, "metal", 60, 12, 1.0), (2, "chemical", 40, 10, 1.0), (5, "creeping_metal", 20, 24, 1.0)]
FT = [(1, "drift", "turbidity", 24, 18), (0, "stuck", "chlorine", 30, 10)]
levels = []
for s in range(5):
c = prepare(make_dataset(days=4, seed=100 + s, events=EV, faults=FT))
al = ((c["rule"] == 2) | (model.predict(c[FEATS]) == 2)).to_numpy()
levels.append(confidence(c, al)["level"].where(al, "-").to_numpy())
L = np.array(levels)
ref = pd.Series(L[0]).mode().iloc[0]
mask = (L != "-").sum(axis=0) >= 3
agree = np.mean([pd.Series(col[col != "-"]).value_counts(normalize=True).iloc[0] for col in L[:, mask].T])
print(f"\n잡음 재현 5회 등급 일치율 {agree:.1%} (경보 3회 이상 시점 {int(mask.sum())}개, 목표 90% 이상: {'달성' if agree >= 0.9 else '미달'})")
print("※ 신뢰도 등급은 판단 보조다. 급수 중단·재개는 의무(예방의학)·환경 담당 확인과 지휘관 승인으로 결정한다.")
fpr, tpr, _ = roc_curve(A["real"], A["score"])
fig, ax = plt.subplots(1, 2, figsize=(10, 4))
ax[0].plot(fpr, tpr, color="tab:blue"); ax[0].plot([0, 1], [0, 1], "k--", lw=0.8)
ax[0].set_title(f"ROC of confidence score (AUC {auc:.2f})"); ax[0].set_xlabel("false positive rate"); ax[0].set_ylabel("true positive rate")
ax[1].hist([A.loc[A["real"], "score"], A.loc[~A["real"], "score"]], bins=12, label=["real danger", "false alarm"], color=["tab:red", "tab:gray"])
ax[1].axvline(CUT["High"], color="k", ls="--"); ax[1].set_xlabel("confidence score"); ax[1].legend(); ax[1].set_title("Score distribution (synthetic)")
plt.tight_layout(); plt.show()
In [ ]:
"""가상 급수원 8곳의 수질 측정 스트림을 1시간 단위로 갱신하며 급수원 상태 지도, 항목별 기준 대비 비율 히트맵, 경보 급수원 추이, 상태 현황판, 긴급 경보 목록 5종을 그리는 상황판을 구현하고, 갱신 주기·응답 시간을 측정한다(야간 모드 지원)."""
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
# ---------------------------------------------------------------------------
# [교육용가상자료] 야전 급수원 가상 수질 센서 데이터 생성기 (실제 부대·위치·검사 결과 없음)
# 가상 급수원 8곳 x 14일 x 1시간 간격. 실제 수질(true_*)·센서 측정값·오염 사건·센서 고장 정답을 함께 만든다.
# 기준치는 교육용 가정값이며, 실제 음용 판정은 공인 검사 결과와 의무(예방의학) 담당 기준에 따른다.
# ---------------------------------------------------------------------------
PARAMS = ["ph", "turbidity", "chlorine", "metal", "ecoli"]
SOURCES = [("W01", "water_truck", 1.0, 2.0), ("W02", "water_truck", 3.5, 1.2), ("W03", "purification_unit", 2.2, 4.1),
("W04", "purification_unit", 5.0, 3.3), ("W05", "well", 0.8, 5.6), ("W06", "well", 4.4, 6.0),
("W07", "river_intake", 6.3, 1.8), ("W08", "river_intake", 6.9, 4.9)]
BASE = {"water_truck": (7.2, 0.25, 0.75, 1.5), "purification_unit": (7.0, 0.20, 0.85, 1.2),
"well": (6.8, 0.45, 0.50, 3.0), "river_intake": (7.3, 0.55, 0.60, 2.0)} # pH, 탁도(NTU), 잔류염소(mg/L), 중금속(ug/L)
LIMITS = {"ph": (5.8, 8.5), "turbidity": (1.0, 2.0), "chlorine": (0.2, 0.1), "metal": (5.0, 10.0), "ecoli": (1, 1)} # (주의, 위험) 가정 기준
EVENT_TYPES = { # 오염 사건: 변수별 최대 변화량, 최대치 도달 시간(h)
"sewage": ({"turbidity": 2.8, "chlorine": -0.5, "ecoli": 30, "ph": -0.3}, 4),
"metal": ({"metal": 22.0, "ph": -0.5}, 4),
"disinfect_fail": ({"chlorine": -0.8, "ecoli": 6}, 6),
"chemical": ({"ph": 1.7, "turbidity": 1.0}, 3),
"creeping_metal": ({"metal": 13.0}, 18),
}
FAULT_TYPES = ["stuck", "spike", "drift", "dropout"]
NOISE = {"ph": 0.04, "turbidity": 0.06, "chlorine": 0.03, "metal": 0.35}
DIGITS = {"ph": 2, "turbidity": 2, "chlorine": 2, "metal": 1}
def make_dataset(days=14, seed=7, n_events=6, n_faults=10, events=None, faults=None, sources=None):
"""events: [(급수원 번호, 사건 유형, 시작 시각 h, 지속 h, 강도)], faults: [(급수원 번호, 고장 유형, 항목, 시작 h, 지속 h)]"""
rng = np.random.default_rng(seed)
sources = sources or SOURCES
T = days * 24
tt = np.arange(T)
hour = tt % 24
night = (hour < 6) | (hour >= 20)
rain = np.zeros(T)
for _ in range(max(1, days // 4)):
s0, d = int(rng.integers(0, T - 14)), int(rng.integers(6, 14))
rain[s0:s0 + d] = np.maximum(rain[s0:s0 + d], rng.uniform(0.6, 1.0))
if events is None:
events = [(int(rng.integers(len(sources))), str(rng.choice(list(EVENT_TYPES))), int(rng.integers(min(24, T // 4), max(T // 4 + 1, T - 48))),
int(rng.integers(10, 30)), 1.0) for _ in range(n_events)]
if faults is None:
faults = [(int(rng.integers(len(sources))), str(rng.choice(FAULT_TYPES)), str(rng.choice(["ph", "turbidity", "chlorine", "metal"])),
int(rng.integers(12, max(13, T - 30))), int(rng.integers(6, 24))) for _ in range(n_faults)]
out = []
for si, (sid, stype, x, y) in enumerate(sources):
ph0, tb0, cl0, mt0 = BASE[stype]
tr = {"ph": ph0 + 0.08 * np.sin(2 * np.pi * hour / 24) + rng.normal(0, 0.03, T),
"turbidity": tb0 * (1 + 0.1 * rng.normal(0, 1, T)),
"chlorine": (cl0 * np.exp(-0.022 * hour) if stype == "water_truck" else cl0 + rng.normal(0, 0.03, T)),
"metal": mt0 + rng.normal(0, 0.2, T), "ecoli": np.zeros(T)}
if stype in ("river_intake", "well"):
tr["turbidity"] = tr["turbidity"] + rain * (0.9 if stype == "river_intake" else 0.45)
f_all, etype = np.zeros(T), np.array(["none"] * T, dtype=object)
for (s_i, et, onset, dur, scale) in events:
if s_i != si:
continue
delta, ramp = EVENT_TYPES[et]
f = np.clip((tt - onset + 1) / ramp, 0, 1) * ((tt >= onset) & (tt < onset + dur + ramp))
for k, v in delta.items():
tr[k] = tr[k] + v * f * scale
etype[f > 0] = et
f_all = np.maximum(f_all, f)
tr["chlorine"] = np.clip(tr["chlorine"], 0.02, None)
tr["turbidity"] = np.clip(tr["turbidity"], 0.05, None)
tr["metal"] = np.clip(tr["metal"], 0.1, None)
grade = np.where(f_all >= 0.2, 2, 0)
mild = (tr["turbidity"] > LIMITS["turbidity"][0]) | (tr["chlorine"] < LIMITS["chlorine"][0]) | (tr["metal"] > LIMITS["metal"][0])
grade = np.where((grade == 0) & mild, 1, grade)
ms = {}
for p in ["ph", "turbidity", "chlorine", "metal"]:
sd = NOISE[p] * np.where(night, 1.5, 1.0) * (tr[p] if p == "turbidity" else 1)
ms[p] = np.round(np.clip(tr[p] + rng.normal(0, 1, T) * sd, 0, None), DIGITS[p])
ms["ecoli"] = (rng.poisson(tr["ecoli"]) + (rng.random(T) < 0.004)).astype(float)
fault, fparam = np.array(["none"] * T, dtype=object), np.array(["-"] * T, dtype=object)
for (s_i, ft, p, s0, d) in faults:
if s_i != si:
continue
seg = slice(s0, min(T, s0 + d))
if ft == "stuck":
ms[p][seg] = ms[p][s0]
elif ft == "spike":
seg = slice(s0, s0 + 1)
ms[p][s0] = {"ph": 11.8, "turbidity": 38.0, "chlorine": 0.0, "metal": 75.0}[p]
elif ft == "drift":
ramp = np.linspace(0, 1, seg.stop - s0)
ms[p][seg] = np.clip(ms[p][seg] + ramp * {"ph": -1.6, "turbidity": 2.6, "chlorine": -0.75, "metal": 12.0}[p], 0, None).round(DIGITS[p])
elif ft == "dropout":
for q in PARAMS:
ms[q][seg] = np.nan
p = "all"
fault[seg], fparam[seg] = ft, p
df = pd.DataFrame({"source_id": sid, "source_type": stype, "x_km": x, "y_km": y, "t": tt,
"cond": np.where(rain > 0, "storm", np.where(night, "night", "day")),
**{p: ms[p] for p in PARAMS}, **{f"true_{p}": tr[p] for p in PARAMS},
"event": etype, "fault": fault, "fault_param": fparam, "grade": grade})
out.append(df)
return pd.concat(out, ignore_index=True)
# ---------------------------------------------------------------------------
# ---- 전처리 공통 함수: 물리 범위 점검 -> 고정값(stuck) 표시 -> 단발 스파이크 제거 -> 결측 보정 ----
PHYS = {"ph": (0, 14), "turbidity": (0, 1000), "chlorine": (0, 10), "metal": (0, 1000), "ecoli": (0, 100000)}
SPIKE_K = {"day": 6.0, "night": 7.0, "storm": 9.0} # 상황별(주간/야간/악천후) 스파이크 판정 배수
MIN_JUMP = {"ph": 0.8, "turbidity": 1.5, "chlorine": 0.3, "metal": 6.0}
STUCK_RUN, SHORT_GAP = 6, 3
def clean(df, spike_k=None, min_jump=None):
"""급수원별 시계열을 정제하고 항목별 처리 내역(<항목>_flag: range/stuck/spike/interp/long_gap)을 남긴다."""
# [TODO : 학생 실습 6] 야전 급수차 및 정수 시설별 수질 상태 실시간 시각화 대시보드 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
GRADE_NAME = {0: "안전", 1: "주의", 2: "위험"}
def rule_grade(df):
"""기준치 1차 필터: 위험(음용 부적합 의심) = 한 항목이라도 위험 기준 초과, 주의 = 주의 기준 초과."""
# [TODO : 학생 실습 6] 야전 급수차 및 정수 시설별 수질 상태 실시간 시각화 대시보드 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def reasons(row):
r = []
if not LIMITS["ph"][0] <= row["ph"] <= LIMITS["ph"][1]:
r.append(f"pH {row['ph']:.2f}")
if row["turbidity"] > LIMITS["turbidity"][0]:
r.append(f"탁도 {row['turbidity']:.2f}NTU")
if row["chlorine"] < LIMITS["chlorine"][0]:
r.append(f"잔류염소 {row['chlorine']:.2f}mg/L")
if row["metal"] > LIMITS["metal"][0]:
r.append(f"중금속 {row['metal']:.1f}ug/L")
if row["ecoli"] >= 1:
r.append(f"대장균 {row['ecoli']:.0f}")
return ", ".join(r) or "-"
FEATS = [f"{p}{s}" for p in ["ph", "turbidity", "chlorine", "metal"] for s in ["", "_d1", "_d3", "_dev", "_sd6"]] + \
["ecoli", "ecoli_3h", "storm", "night", "n_flag", "truck", "purif", "well", "river"]
def add_features(c):
"""시간 순서를 지키는(과거 값만 쓰는) 급수원별 특징: 변화량, 24시간 기준선 대비 편차, 6시간 변동성, 대장균 누적 등."""
c = c.sort_values(["source_id", "t"]).reset_index(drop=True)
g = c.groupby("source_id")
for p in ["ph", "turbidity", "chlorine", "metal"]:
c[f"{p}_d1"] = g[p].diff().fillna(0)
c[f"{p}_d3"] = g[p].diff(3).fillna(0)
base = g[p].transform(lambda s: s.shift(1).rolling(24, min_periods=4).median())
c[f"{p}_dev"] = (c[p] - base).fillna(0)
c[f"{p}_sd6"] = g[p].transform(lambda s: s.rolling(6, min_periods=2).std()).fillna(0)
c["ecoli_3h"] = g["ecoli"].transform(lambda s: s.rolling(3, min_periods=1).sum())
c["storm"], c["night"] = (c["cond"] == "storm").astype(int), (c["cond"] == "night").astype(int)
flags = [f"{p}_flag" for p in PARAMS if f"{p}_flag" in c]
c["n_flag"] = sum(c[f].isin(["stuck", "spike", "long_gap", "range"]).astype(int) for f in flags) if flags else 0
for k, name in [("truck", "water_truck"), ("purif", "purification_unit"), ("well", "well"), ("river", "river_intake")]:
c[k] = (c["source_type"] == name).astype(int)
return c
DARK = True # 야간 작전 모드(어두운 배경)
NOW = 290 # 상황판 기준 시각(모의 시작 후 경과 시간 h)
COLOR = {0: "tab:green", 1: "gold", 2: "red"}
data = make_dataset(days=14, seed=11, n_events=10, n_faults=14)
data = data[data["t"] <= NOW]
def ratio(row):
"""항목별 '위험 기준 대비 비율'(1 이상이면 위험). 잔류염소·pH는 방향을 맞춰 환산한다."""
lo, hi = LIMITS["ph"]
return {"pH": max((row["ph"] - 7.0) / (hi - 7.0), (7.0 - row["ph"]) / (7.0 - lo)),
"turb": row["turbidity"] / LIMITS["turbidity"][1], "Cl": LIMITS["chlorine"][1] / max(row["chlorine"], 0.01),
"metal": row["metal"] / LIMITS["metal"][1], "E.coli": row["ecoli"] / LIMITS["ecoli"][1]}
# 1) 실시간 갱신 모의: 최근 24시간 창을 유지하며 1시간씩 새 측정값이 들어올 때 상태 계산 시간
window = data[data["t"] <= NOW - 24].copy()
upd = []
for h in range(NOW - 23, NOW + 1):
t0 = time.perf_counter()
window = pd.concat([window[window["t"] > h - 48], data[data["t"] == h]])
c = clean(window)
c["g"] = rule_grade(c)
latest = c[c["t"] == h]
upd.append((time.perf_counter() - t0) * 1000)
print(f"[교육용가상자료] 실시간 갱신 모의 24회: 1회 평균 {np.mean(upd):.1f}ms, 최대 {np.max(upd):.1f}ms "
f"(갱신 주기 1초 이내: {'달성' if max(upd) < 1000 else '미달'})")
# 2) 현재 상태표
state = latest.set_index("source_id")
hist = c.groupby("source_id")["g"].apply(lambda s: int((s.tail(6) == 2).sum()))
board = pd.DataFrame({"유형": state["source_type"], "상태": state["g"].map(GRADE_NAME), "최근6h 위험": hist,
"근거": [reasons(r) for _, r in state.iterrows()]})
print(f"\n[D+{NOW // 24:02d} {NOW % 24:02d}시 급수원 상태판]")
print(board.to_string())
alerts = board[(board["상태"] != "안전") | (board["최근6h 위험"] > 0)]
print(f"긴급 경보 강조 대상 {len(alerts)}곳: {', '.join(alerts.index)}")
# 3) 상황판 그리기(컴포넌트 5종) + 렌더링 시간
plt.style.use("dark_background" if DARK else "default")
t0 = time.perf_counter()
fig = plt.figure(figsize=(13, 7.5))
gs = fig.add_gridspec(2, 3)
ax1 = fig.add_subplot(gs[0, 0])
for sid, r in state.iterrows():
ax1.scatter(r["x_km"], r["y_km"], s=260, color=COLOR[r["g"]], edgecolor="white" if r["g"] == 2 else "none", lw=2)
ax1.annotate(sid, (r["x_km"], r["y_km"]), xytext=(6, 6), textcoords="offset points", fontsize=8)
ax1.set_title("Water source status map (virtual grid km)"); ax1.set_xlim(0, 7.5); ax1.set_ylim(0, 7)
ax2 = fig.add_subplot(gs[0, 1])
R = pd.DataFrame([ratio(r) for _, r in state.iterrows()], index=state.index)
im = ax2.imshow(R.clip(0, 2).to_numpy(), cmap="RdYlGn_r", vmin=0, vmax=2, aspect="auto")
ax2.set_xticks(range(R.shape[1]), R.columns); ax2.set_yticks(range(len(R)), R.index)
ax2.set_title("Ratio to danger limit (>=1 danger)"); fig.colorbar(im, ax=ax2, fraction=0.046)
ax3 = fig.add_subplot(gs[0, 2])
cnt = state["g"].value_counts().reindex([0, 1, 2], fill_value=0)
ax3.bar(["safe", "caution", "danger"], cnt.values, color=[COLOR[i] for i in range(3)])
ax3.set_title(f"Status count at hour {NOW}")
ax4 = fig.add_subplot(gs[1, :2])
focus = alerts.index[0] if len(alerts) else state.index[0]
s = c[c["source_id"] == focus]
for p, col in [("turbidity", "tab:cyan"), ("chlorine", "tab:orange"), ("metal", "violet")]:
ax4.plot(s["t"], s[p] / LIMITS[p][1] if p != "chlorine" else LIMITS[p][1] / s[p].clip(lower=0.01), label=p, color=col)
ax4.axhline(1, color="red", ls="--", lw=1); ax4.set_ylim(0, 3); ax4.legend(loc="upper left")
ax4.set_title(f"{focus} last 48h ratio to danger limit"); ax4.set_xlabel("hour")
ax5 = fig.add_subplot(gs[1, 2]); ax5.axis("off")
lines = [f"{sid}: {('DANGER' if r['상태'] == '위험' else 'CHECK')} 6h danger {r['최근6h 위험']}" for sid, r in alerts.iterrows()]
ax5.text(0, 1, "ALERT LIST\n" + "\n".join(lines or ["no alert"]), va="top", family="monospace", fontsize=10,
color="red" if len(alerts) else "white" if DARK else "black")
fig.suptitle("Field water quality risk board (synthetic, decision support only)")
fig.tight_layout()
fig.canvas.draw()
draw_ms = (time.perf_counter() - t0) * 1000
# 4) UI 응답성: 급수원 클릭(선택) 시 상세 조회 응답 시간
resp = []
for sid in state.index:
t0 = time.perf_counter()
d = c[c["source_id"] == sid].tail(6)
_ = (d[["ph", "turbidity", "chlorine", "metal", "ecoli"]].agg(["min", "max"]), reasons(d.iloc[-1]))
resp.append((time.perf_counter() - t0) * 1000)
print(f"\n시각화 컴포넌트 5종 렌더링 {draw_ms:.0f}ms, 급수원 상세 조회 응답 평균 {np.mean(resp):.1f}ms "
f"(UI 응답성 100ms 이내: {'달성' if max(resp) < 100 else '미달'})")
print("※ 상황판의 위험 표시는 가정 기준치 기반 판단 보조이며, 급수 중단 여부는 의무(예방의학)·환경 담당 확인 후 지휘관이 결정한다.")
plt.show()
In [ ]:
"""수질 위험 경보가 나면 시스템은 급수 중단을 자동 확정하지 않고 우선순위 검토 큐에 올리며, 측정 담당 긴급 채수 → 예방의학 담당 검토 → 지휘관 급수 중단 승인·반려(반려 사유 필수) 결재선과 권한 통제, 해시 체인 감사 로그, 큐 대기시간을 시뮬레이션한다."""
import hashlib
import json
import numpy as np
import pandas as pd
# ---------------------------------------------------------------------------
# [교육용가상자료] 야전 급수원 가상 수질 센서 데이터 생성기 (실제 부대·위치·검사 결과 없음)
# 가상 급수원 8곳 x 14일 x 1시간 간격. 실제 수질(true_*)·센서 측정값·오염 사건·센서 고장 정답을 함께 만든다.
# 기준치는 교육용 가정값이며, 실제 음용 판정은 공인 검사 결과와 의무(예방의학) 담당 기준에 따른다.
# ---------------------------------------------------------------------------
PARAMS = ["ph", "turbidity", "chlorine", "metal", "ecoli"]
SOURCES = [("W01", "water_truck", 1.0, 2.0), ("W02", "water_truck", 3.5, 1.2), ("W03", "purification_unit", 2.2, 4.1),
("W04", "purification_unit", 5.0, 3.3), ("W05", "well", 0.8, 5.6), ("W06", "well", 4.4, 6.0),
("W07", "river_intake", 6.3, 1.8), ("W08", "river_intake", 6.9, 4.9)]
BASE = {"water_truck": (7.2, 0.25, 0.75, 1.5), "purification_unit": (7.0, 0.20, 0.85, 1.2),
"well": (6.8, 0.45, 0.50, 3.0), "river_intake": (7.3, 0.55, 0.60, 2.0)} # pH, 탁도(NTU), 잔류염소(mg/L), 중금속(ug/L)
LIMITS = {"ph": (5.8, 8.5), "turbidity": (1.0, 2.0), "chlorine": (0.2, 0.1), "metal": (5.0, 10.0), "ecoli": (1, 1)} # (주의, 위험) 가정 기준
EVENT_TYPES = { # 오염 사건: 변수별 최대 변화량, 최대치 도달 시간(h)
"sewage": ({"turbidity": 2.8, "chlorine": -0.5, "ecoli": 30, "ph": -0.3}, 4),
"metal": ({"metal": 22.0, "ph": -0.5}, 4),
"disinfect_fail": ({"chlorine": -0.8, "ecoli": 6}, 6),
"chemical": ({"ph": 1.7, "turbidity": 1.0}, 3),
"creeping_metal": ({"metal": 13.0}, 18),
}
FAULT_TYPES = ["stuck", "spike", "drift", "dropout"]
NOISE = {"ph": 0.04, "turbidity": 0.06, "chlorine": 0.03, "metal": 0.35}
DIGITS = {"ph": 2, "turbidity": 2, "chlorine": 2, "metal": 1}
def make_dataset(days=14, seed=7, n_events=6, n_faults=10, events=None, faults=None, sources=None):
"""events: [(급수원 번호, 사건 유형, 시작 시각 h, 지속 h, 강도)], faults: [(급수원 번호, 고장 유형, 항목, 시작 h, 지속 h)]"""
rng = np.random.default_rng(seed)
sources = sources or SOURCES
T = days * 24
tt = np.arange(T)
hour = tt % 24
night = (hour < 6) | (hour >= 20)
rain = np.zeros(T)
for _ in range(max(1, days // 4)):
s0, d = int(rng.integers(0, T - 14)), int(rng.integers(6, 14))
rain[s0:s0 + d] = np.maximum(rain[s0:s0 + d], rng.uniform(0.6, 1.0))
if events is None:
events = [(int(rng.integers(len(sources))), str(rng.choice(list(EVENT_TYPES))), int(rng.integers(min(24, T // 4), max(T // 4 + 1, T - 48))),
int(rng.integers(10, 30)), 1.0) for _ in range(n_events)]
if faults is None:
faults = [(int(rng.integers(len(sources))), str(rng.choice(FAULT_TYPES)), str(rng.choice(["ph", "turbidity", "chlorine", "metal"])),
int(rng.integers(12, max(13, T - 30))), int(rng.integers(6, 24))) for _ in range(n_faults)]
out = []
for si, (sid, stype, x, y) in enumerate(sources):
ph0, tb0, cl0, mt0 = BASE[stype]
tr = {"ph": ph0 + 0.08 * np.sin(2 * np.pi * hour / 24) + rng.normal(0, 0.03, T),
"turbidity": tb0 * (1 + 0.1 * rng.normal(0, 1, T)),
"chlorine": (cl0 * np.exp(-0.022 * hour) if stype == "water_truck" else cl0 + rng.normal(0, 0.03, T)),
"metal": mt0 + rng.normal(0, 0.2, T), "ecoli": np.zeros(T)}
if stype in ("river_intake", "well"):
tr["turbidity"] = tr["turbidity"] + rain * (0.9 if stype == "river_intake" else 0.45)
f_all, etype = np.zeros(T), np.array(["none"] * T, dtype=object)
for (s_i, et, onset, dur, scale) in events:
if s_i != si:
continue
delta, ramp = EVENT_TYPES[et]
f = np.clip((tt - onset + 1) / ramp, 0, 1) * ((tt >= onset) & (tt < onset + dur + ramp))
for k, v in delta.items():
tr[k] = tr[k] + v * f * scale
etype[f > 0] = et
f_all = np.maximum(f_all, f)
tr["chlorine"] = np.clip(tr["chlorine"], 0.02, None)
tr["turbidity"] = np.clip(tr["turbidity"], 0.05, None)
tr["metal"] = np.clip(tr["metal"], 0.1, None)
grade = np.where(f_all >= 0.2, 2, 0)
mild = (tr["turbidity"] > LIMITS["turbidity"][0]) | (tr["chlorine"] < LIMITS["chlorine"][0]) | (tr["metal"] > LIMITS["metal"][0])
grade = np.where((grade == 0) & mild, 1, grade)
ms = {}
for p in ["ph", "turbidity", "chlorine", "metal"]:
sd = NOISE[p] * np.where(night, 1.5, 1.0) * (tr[p] if p == "turbidity" else 1)
ms[p] = np.round(np.clip(tr[p] + rng.normal(0, 1, T) * sd, 0, None), DIGITS[p])
ms["ecoli"] = (rng.poisson(tr["ecoli"]) + (rng.random(T) < 0.004)).astype(float)
fault, fparam = np.array(["none"] * T, dtype=object), np.array(["-"] * T, dtype=object)
for (s_i, ft, p, s0, d) in faults:
if s_i != si:
continue
seg = slice(s0, min(T, s0 + d))
if ft == "stuck":
ms[p][seg] = ms[p][s0]
elif ft == "spike":
seg = slice(s0, s0 + 1)
ms[p][s0] = {"ph": 11.8, "turbidity": 38.0, "chlorine": 0.0, "metal": 75.0}[p]
elif ft == "drift":
ramp = np.linspace(0, 1, seg.stop - s0)
ms[p][seg] = np.clip(ms[p][seg] + ramp * {"ph": -1.6, "turbidity": 2.6, "chlorine": -0.75, "metal": 12.0}[p], 0, None).round(DIGITS[p])
elif ft == "dropout":
for q in PARAMS:
ms[q][seg] = np.nan
p = "all"
fault[seg], fparam[seg] = ft, p
df = pd.DataFrame({"source_id": sid, "source_type": stype, "x_km": x, "y_km": y, "t": tt,
"cond": np.where(rain > 0, "storm", np.where(night, "night", "day")),
**{p: ms[p] for p in PARAMS}, **{f"true_{p}": tr[p] for p in PARAMS},
"event": etype, "fault": fault, "fault_param": fparam, "grade": grade})
out.append(df)
return pd.concat(out, ignore_index=True)
# ---------------------------------------------------------------------------
# ---- 전처리 공통 함수: 물리 범위 점검 -> 고정값(stuck) 표시 -> 단발 스파이크 제거 -> 결측 보정 ----
PHYS = {"ph": (0, 14), "turbidity": (0, 1000), "chlorine": (0, 10), "metal": (0, 1000), "ecoli": (0, 100000)}
SPIKE_K = {"day": 6.0, "night": 7.0, "storm": 9.0} # 상황별(주간/야간/악천후) 스파이크 판정 배수
MIN_JUMP = {"ph": 0.8, "turbidity": 1.5, "chlorine": 0.3, "metal": 6.0}
STUCK_RUN, SHORT_GAP = 6, 3
def clean(df, spike_k=None, min_jump=None):
"""급수원별 시계열을 정제하고 항목별 처리 내역(<항목>_flag: range/stuck/spike/interp/long_gap)을 남긴다."""
# [TODO : 학생 실습 7] 수질 위험 경보 발생 시 급수관 긴급 채수 검사 및 급수 중단 HITL 결재선 설계 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
GRADE_NAME = {0: "안전", 1: "주의", 2: "위험"}
def rule_grade(df):
"""기준치 1차 필터: 위험(음용 부적합 의심) = 한 항목이라도 위험 기준 초과, 주의 = 주의 기준 초과."""
# [TODO : 학생 실습 7] 수질 위험 경보 발생 시 급수관 긴급 채수 검사 및 급수 중단 HITL 결재선 설계 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def reasons(row):
r = []
if not LIMITS["ph"][0] <= row["ph"] <= LIMITS["ph"][1]:
r.append(f"pH {row['ph']:.2f}")
if row["turbidity"] > LIMITS["turbidity"][0]:
r.append(f"탁도 {row['turbidity']:.2f}NTU")
if row["chlorine"] < LIMITS["chlorine"][0]:
r.append(f"잔류염소 {row['chlorine']:.2f}mg/L")
if row["metal"] > LIMITS["metal"][0]:
r.append(f"중금속 {row['metal']:.1f}ug/L")
if row["ecoli"] >= 1:
r.append(f"대장균 {row['ecoli']:.0f}")
return ", ".join(r) or "-"
FEATS = [f"{p}{s}" for p in ["ph", "turbidity", "chlorine", "metal"] for s in ["", "_d1", "_d3", "_dev", "_sd6"]] + \
["ecoli", "ecoli_3h", "storm", "night", "n_flag", "truck", "purif", "well", "river"]
def add_features(c):
"""시간 순서를 지키는(과거 값만 쓰는) 급수원별 특징: 변화량, 24시간 기준선 대비 편차, 6시간 변동성, 대장균 누적 등."""
c = c.sort_values(["source_id", "t"]).reset_index(drop=True)
g = c.groupby("source_id")
for p in ["ph", "turbidity", "chlorine", "metal"]:
c[f"{p}_d1"] = g[p].diff().fillna(0)
c[f"{p}_d3"] = g[p].diff(3).fillna(0)
base = g[p].transform(lambda s: s.shift(1).rolling(24, min_periods=4).median())
c[f"{p}_dev"] = (c[p] - base).fillna(0)
c[f"{p}_sd6"] = g[p].transform(lambda s: s.rolling(6, min_periods=2).std()).fillna(0)
c["ecoli_3h"] = g["ecoli"].transform(lambda s: s.rolling(3, min_periods=1).sum())
c["storm"], c["night"] = (c["cond"] == "storm").astype(int), (c["cond"] == "night").astype(int)
flags = [f"{p}_flag" for p in PARAMS if f"{p}_flag" in c]
c["n_flag"] = sum(c[f].isin(["stuck", "spike", "long_gap", "range"]).astype(int) for f in flags) if flags else 0
for k, name in [("truck", "water_truck"), ("purif", "purification_unit"), ("well", "well"), ("river", "river_intake")]:
c[k] = (c["source_type"] == name).astype(int)
return c
ROLE_ACTIONS = {"측정병": {"sample"}, "예방의학담당": {"review"}, "지휘관": {"approve_stop", "reject"}, "시스템": {"enqueue"}}
USERS = {"op01": "측정병", "med01": "예방의학담당", "cdr01": "지휘관", "sys": "시스템"}
FLOW = {"enqueue": ("-", "채수대기"), "sample": ("채수대기", "검토대기"), "review": ("검토대기", "승인대기"),
"approve_stop": ("승인대기", "급수중단"), "reject": ("승인대기", "급수유지")}
class AuditLog:
def __init__(self):
self.rows = []
def add(self, **kw):
prev = self.rows[-1]["hash"] if self.rows else "0" * 16
body = json.dumps(kw, ensure_ascii=False, sort_keys=True)
self.rows.append({**kw, "prev": prev, "hash": hashlib.sha256((prev + body).encode()).hexdigest()[:16]})
def verify(self):
prev = "0" * 16
for r in self.rows:
body = json.dumps({k: v for k, v in r.items() if k not in ("prev", "hash")}, ensure_ascii=False, sort_keys=True)
if r["prev"] != prev or hashlib.sha256((prev + body).encode()).hexdigest()[:16] != r["hash"]:
return False
prev = r["hash"]
return True
class Workflow:
def __init__(self):
self.log, self.state, self.n_try = AuditLog(), {}, 0
def act(self, user, action, case, t, note=""):
"""권한·순서·사유를 검사한 뒤 상태를 바꾼다. 거부도 감사 로그에 남긴다."""
self.n_try += 1
role = USERS.get(user, "미등록")
cur = self.state.get(case, "-")
err = None
if action not in ROLE_ACTIONS.get(role, set()):
err = f"권한 없음({role} 역할에 {action} 권한 없음)"
elif FLOW[action][0] != cur:
err = f"순서 위반(현재 {cur})"
elif action in ("reject", "approve_stop", "review") and len(note.strip()) < 5:
err = "사유·근거 입력 필수"
if err is None:
self.state[case] = FLOW[action][1]
self.log.add(t=t, user=user, role=role, case=case, action=action, result="거부" if err else "처리",
detail=err or note, state=self.state.get(case, "-"))
return err is None
# 1) 경보 생성: 가상 벤치마크에서 위험 판정이 2시간 이상 이어진 구간을 '경보 사건'으로 묶는다
c = clean(make_dataset(days=14, seed=11, n_events=10, n_faults=14))
c["g"] = rule_grade(c)
cases = []
for sid, s in c.groupby("source_id"):
d = (s["g"] == 2).astype(int).to_numpy()
seg = np.split(np.arange(len(d)), np.where(np.diff(d) != 0)[0] + 1)
for idx in seg:
if d[idx[0]] == 1 and len(idx) >= 2:
r = s.iloc[idx[0]]
cases.append({"case": f"AL-{len(cases) + 1:02d}", "source": sid, "type": r["source_type"], "t": int(r["t"]),
"hours": len(idx), "real": bool((s.iloc[idx]["grade"] == 2).mean() >= 0.5), "basis": reasons(s.iloc[idx[-1]])})
Q = pd.DataFrame(cases)
# 우선순위: 지속시간 + 급수차·정수장(공급 인원 많음) 가중 + 대장균·중금속 포함 가중
Q["priority"] = Q["hours"].clip(upper=12) + Q["type"].isin(["water_truck", "purification_unit"]) * 4 \
+ Q["basis"].str.contains("대장균|중금속") * 3
Q = Q.sort_values(["priority", "t"], ascending=[False, True]).reset_index(drop=True)
print(f"[교육용가상자료] 경보 사건 {len(Q)}건이 검토 큐에 등록됨(자동 급수 중단 0건)")
# 2) 결재 시뮬레이션: 채수 담당 30분 내 채수, 예방의학 담당 재검사 결과로 의견, 지휘관이 승인·반려
wf = Workflow()
rng = np.random.default_rng(5)
wait = []
for _, q in Q.iterrows():
t = q["t"] * 60
wf.act("sys", "enqueue", q["case"], t, "기준치 초과 경보 자동 등록")
t += int(rng.integers(10, 30)); wf.act("op01", "sample", q["case"], t, "현장 키트 재채수")
t += int(rng.integers(20, 60))
opinion = "재검사 위험 확인, 급수 중단 권고" if q["real"] else "재검사 정상, 센서 점검 권고"
wf.act("med01", "review", q["case"], t, opinion)
t += int(rng.integers(5, 25))
if q["real"]:
wf.act("cdr01", "approve_stop", q["case"], t, f"{q['basis']} 확인, 대체 급수원 전환")
else:
wf.act("cdr01", "reject", q["case"], t, "재검사 정상, 급수 유지하고 센서 교정")
wait.append(t - q["t"] * 60)
st = pd.Series(wf.state).value_counts()
print("결재 결과:", st.to_dict(), f"/ 경보부터 결정까지 평균 {np.mean(wait):.0f}분, 최대 {max(wait)}분")
print("\n[검토 큐 상위 5건]")
print(Q.head(5)[["case", "source", "type", "t", "hours", "priority", "basis"]].to_string(index=False))
# 3) 권한 우회·절차 위반 시도 6종 (사이사이 정상 절차를 진행)
wf.act("sys", "enqueue", "AL-X", 99999, "시험 경보")
STEPS = [("위반", "op01", "approve_stop", "측정병이 직접 급수 중단", "중단"), ("위반", "sys", "approve_stop", "시스템 자동 확정", "자동"),
("위반", "cdr01", "approve_stop", "채수·검토 없이 지휘관 승인", "즉시 중단"), ("위반", "hacker", "review", "미등록 사용자 검토", "위험 확인됨"),
("정상", "op01", "sample", "정상 채수", "재채수"), ("위반", "med01", "review", "근거 없는 검토 의견", ""),
("정상", "med01", "review", "정상 검토", "재검사 정상 확인"), ("위반", "cdr01", "reject", "사유 없는 반려", "")]
blocked = n_v = 0
print("\n[권한 우회·절차 위반 시도]")
for kind, u, a, desc, note in STEPS:
ok = wf.act(u, a, "AL-X", 99999, note)
if kind == "위반":
n_v += 1; blocked += not ok
print(f" {desc:16s} -> {'차단' if not ok else '통과(문제)'}: {wf.log.rows[-1]['detail']}")
print(f"우회·위반 시도 차단 {blocked}/{n_v} ({blocked / n_v:.0%})")
# 4) 감사 로그 무결성과 로깅 성공률
n_log = len(wf.log.rows)
print(f"\n감사 로그 {n_log}건 / 요청 {wf.n_try}건 -> 로깅 성공률 {n_log / wf.n_try:.0%}, 해시 체인 검증: {'정상' if wf.log.verify() else '위변조'}")
wf.log.rows[5]["detail"] = "기록 임의 수정"
print(f"5번째 기록 임의 수정 후 검증: {'정상' if wf.log.verify() else '위변조 탐지'}")
print("※ 급수 중단·재개는 지휘관 결정 사항이며, 시스템은 경보와 근거만 제공한다. 실제 판정은 공식 기준과 의무·환경 담당 확인에 따른다.")
In [ ]:
"""가상 급수원을 8곳에서 48곳(트래픽 500% 증가)으로 늘리고 30곳에 오염물질이 동시에 유입되는 시나리오로 수집·판정 파이프라인 부하 시험을 수행해, 시스템 중단·데이터 유실·메모리 증가·경보 반응 시간과 과부하 시 기능 축소(Fail-safe) 동작을 검증한다."""
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
# ---------------------------------------------------------------------------
# [교육용가상자료] 야전 급수원 가상 수질 센서 데이터 생성기 (실제 부대·위치·검사 결과 없음)
# 가상 급수원 8곳 x 14일 x 1시간 간격. 실제 수질(true_*)·센서 측정값·오염 사건·센서 고장 정답을 함께 만든다.
# 기준치는 교육용 가정값이며, 실제 음용 판정은 공인 검사 결과와 의무(예방의학) 담당 기준에 따른다.
# ---------------------------------------------------------------------------
PARAMS = ["ph", "turbidity", "chlorine", "metal", "ecoli"]
SOURCES = [("W01", "water_truck", 1.0, 2.0), ("W02", "water_truck", 3.5, 1.2), ("W03", "purification_unit", 2.2, 4.1),
("W04", "purification_unit", 5.0, 3.3), ("W05", "well", 0.8, 5.6), ("W06", "well", 4.4, 6.0),
("W07", "river_intake", 6.3, 1.8), ("W08", "river_intake", 6.9, 4.9)]
BASE = {"water_truck": (7.2, 0.25, 0.75, 1.5), "purification_unit": (7.0, 0.20, 0.85, 1.2),
"well": (6.8, 0.45, 0.50, 3.0), "river_intake": (7.3, 0.55, 0.60, 2.0)} # pH, 탁도(NTU), 잔류염소(mg/L), 중금속(ug/L)
LIMITS = {"ph": (5.8, 8.5), "turbidity": (1.0, 2.0), "chlorine": (0.2, 0.1), "metal": (5.0, 10.0), "ecoli": (1, 1)} # (주의, 위험) 가정 기준
EVENT_TYPES = { # 오염 사건: 변수별 최대 변화량, 최대치 도달 시간(h)
"sewage": ({"turbidity": 2.8, "chlorine": -0.5, "ecoli": 30, "ph": -0.3}, 4),
"metal": ({"metal": 22.0, "ph": -0.5}, 4),
"disinfect_fail": ({"chlorine": -0.8, "ecoli": 6}, 6),
"chemical": ({"ph": 1.7, "turbidity": 1.0}, 3),
"creeping_metal": ({"metal": 13.0}, 18),
}
FAULT_TYPES = ["stuck", "spike", "drift", "dropout"]
NOISE = {"ph": 0.04, "turbidity": 0.06, "chlorine": 0.03, "metal": 0.35}
DIGITS = {"ph": 2, "turbidity": 2, "chlorine": 2, "metal": 1}
def make_dataset(days=14, seed=7, n_events=6, n_faults=10, events=None, faults=None, sources=None):
"""events: [(급수원 번호, 사건 유형, 시작 시각 h, 지속 h, 강도)], faults: [(급수원 번호, 고장 유형, 항목, 시작 h, 지속 h)]"""
rng = np.random.default_rng(seed)
sources = sources or SOURCES
T = days * 24
tt = np.arange(T)
hour = tt % 24
night = (hour < 6) | (hour >= 20)
rain = np.zeros(T)
for _ in range(max(1, days // 4)):
s0, d = int(rng.integers(0, T - 14)), int(rng.integers(6, 14))
rain[s0:s0 + d] = np.maximum(rain[s0:s0 + d], rng.uniform(0.6, 1.0))
if events is None:
events = [(int(rng.integers(len(sources))), str(rng.choice(list(EVENT_TYPES))), int(rng.integers(min(24, T // 4), max(T // 4 + 1, T - 48))),
int(rng.integers(10, 30)), 1.0) for _ in range(n_events)]
if faults is None:
faults = [(int(rng.integers(len(sources))), str(rng.choice(FAULT_TYPES)), str(rng.choice(["ph", "turbidity", "chlorine", "metal"])),
int(rng.integers(12, max(13, T - 30))), int(rng.integers(6, 24))) for _ in range(n_faults)]
out = []
for si, (sid, stype, x, y) in enumerate(sources):
ph0, tb0, cl0, mt0 = BASE[stype]
tr = {"ph": ph0 + 0.08 * np.sin(2 * np.pi * hour / 24) + rng.normal(0, 0.03, T),
"turbidity": tb0 * (1 + 0.1 * rng.normal(0, 1, T)),
"chlorine": (cl0 * np.exp(-0.022 * hour) if stype == "water_truck" else cl0 + rng.normal(0, 0.03, T)),
"metal": mt0 + rng.normal(0, 0.2, T), "ecoli": np.zeros(T)}
if stype in ("river_intake", "well"):
tr["turbidity"] = tr["turbidity"] + rain * (0.9 if stype == "river_intake" else 0.45)
f_all, etype = np.zeros(T), np.array(["none"] * T, dtype=object)
for (s_i, et, onset, dur, scale) in events:
if s_i != si:
continue
delta, ramp = EVENT_TYPES[et]
f = np.clip((tt - onset + 1) / ramp, 0, 1) * ((tt >= onset) & (tt < onset + dur + ramp))
for k, v in delta.items():
tr[k] = tr[k] + v * f * scale
etype[f > 0] = et
f_all = np.maximum(f_all, f)
tr["chlorine"] = np.clip(tr["chlorine"], 0.02, None)
tr["turbidity"] = np.clip(tr["turbidity"], 0.05, None)
tr["metal"] = np.clip(tr["metal"], 0.1, None)
grade = np.where(f_all >= 0.2, 2, 0)
mild = (tr["turbidity"] > LIMITS["turbidity"][0]) | (tr["chlorine"] < LIMITS["chlorine"][0]) | (tr["metal"] > LIMITS["metal"][0])
grade = np.where((grade == 0) & mild, 1, grade)
ms = {}
for p in ["ph", "turbidity", "chlorine", "metal"]:
sd = NOISE[p] * np.where(night, 1.5, 1.0) * (tr[p] if p == "turbidity" else 1)
ms[p] = np.round(np.clip(tr[p] + rng.normal(0, 1, T) * sd, 0, None), DIGITS[p])
ms["ecoli"] = (rng.poisson(tr["ecoli"]) + (rng.random(T) < 0.004)).astype(float)
fault, fparam = np.array(["none"] * T, dtype=object), np.array(["-"] * T, dtype=object)
for (s_i, ft, p, s0, d) in faults:
if s_i != si:
continue
seg = slice(s0, min(T, s0 + d))
if ft == "stuck":
ms[p][seg] = ms[p][s0]
elif ft == "spike":
seg = slice(s0, s0 + 1)
ms[p][s0] = {"ph": 11.8, "turbidity": 38.0, "chlorine": 0.0, "metal": 75.0}[p]
elif ft == "drift":
ramp = np.linspace(0, 1, seg.stop - s0)
ms[p][seg] = np.clip(ms[p][seg] + ramp * {"ph": -1.6, "turbidity": 2.6, "chlorine": -0.75, "metal": 12.0}[p], 0, None).round(DIGITS[p])
elif ft == "dropout":
for q in PARAMS:
ms[q][seg] = np.nan
p = "all"
fault[seg], fparam[seg] = ft, p
df = pd.DataFrame({"source_id": sid, "source_type": stype, "x_km": x, "y_km": y, "t": tt,
"cond": np.where(rain > 0, "storm", np.where(night, "night", "day")),
**{p: ms[p] for p in PARAMS}, **{f"true_{p}": tr[p] for p in PARAMS},
"event": etype, "fault": fault, "fault_param": fparam, "grade": grade})
out.append(df)
return pd.concat(out, ignore_index=True)
# ---------------------------------------------------------------------------
# ---- 전처리 공통 함수: 물리 범위 점검 -> 고정값(stuck) 표시 -> 단발 스파이크 제거 -> 결측 보정 ----
PHYS = {"ph": (0, 14), "turbidity": (0, 1000), "chlorine": (0, 10), "metal": (0, 1000), "ecoli": (0, 100000)}
SPIKE_K = {"day": 6.0, "night": 7.0, "storm": 9.0} # 상황별(주간/야간/악천후) 스파이크 판정 배수
MIN_JUMP = {"ph": 0.8, "turbidity": 1.5, "chlorine": 0.3, "metal": 6.0}
STUCK_RUN, SHORT_GAP = 6, 3
def clean(df, spike_k=None, min_jump=None):
"""급수원별 시계열을 정제하고 항목별 처리 내역(<항목>_flag: range/stuck/spike/interp/long_gap)을 남긴다."""
# [TODO : 학생 실습 8] 오염물질 대량 유입 가상 시나리오 부하 테스트 및 경보 반응 시간 검증 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
GRADE_NAME = {0: "안전", 1: "주의", 2: "위험"}
def rule_grade(df):
"""기준치 1차 필터: 위험(음용 부적합 의심) = 한 항목이라도 위험 기준 초과, 주의 = 주의 기준 초과."""
# [TODO : 학생 실습 8] 오염물질 대량 유입 가상 시나리오 부하 테스트 및 경보 반응 시간 검증 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def reasons(row):
r = []
if not LIMITS["ph"][0] <= row["ph"] <= LIMITS["ph"][1]:
r.append(f"pH {row['ph']:.2f}")
if row["turbidity"] > LIMITS["turbidity"][0]:
r.append(f"탁도 {row['turbidity']:.2f}NTU")
if row["chlorine"] < LIMITS["chlorine"][0]:
r.append(f"잔류염소 {row['chlorine']:.2f}mg/L")
if row["metal"] > LIMITS["metal"][0]:
r.append(f"중금속 {row['metal']:.1f}ug/L")
if row["ecoli"] >= 1:
r.append(f"대장균 {row['ecoli']:.0f}")
return ", ".join(r) or "-"
FEATS = [f"{p}{s}" for p in ["ph", "turbidity", "chlorine", "metal"] for s in ["", "_d1", "_d3", "_dev", "_sd6"]] + \
["ecoli", "ecoli_3h", "storm", "night", "n_flag", "truck", "purif", "well", "river"]
def add_features(c):
"""시간 순서를 지키는(과거 값만 쓰는) 급수원별 특징: 변화량, 24시간 기준선 대비 편차, 6시간 변동성, 대장균 누적 등."""
c = c.sort_values(["source_id", "t"]).reset_index(drop=True)
g = c.groupby("source_id")
for p in ["ph", "turbidity", "chlorine", "metal"]:
c[f"{p}_d1"] = g[p].diff().fillna(0)
c[f"{p}_d3"] = g[p].diff(3).fillna(0)
base = g[p].transform(lambda s: s.shift(1).rolling(24, min_periods=4).median())
c[f"{p}_dev"] = (c[p] - base).fillna(0)
c[f"{p}_sd6"] = g[p].transform(lambda s: s.rolling(6, min_periods=2).std()).fillna(0)
c["ecoli_3h"] = g["ecoli"].transform(lambda s: s.rolling(3, min_periods=1).sum())
c["storm"], c["night"] = (c["cond"] == "storm").astype(int), (c["cond"] == "night").astype(int)
flags = [f"{p}_flag" for p in PARAMS if f"{p}_flag" in c]
c["n_flag"] = sum(c[f].isin(["stuck", "spike", "long_gap", "range"]).astype(int) for f in flags) if flags else 0
for k, name in [("truck", "water_truck"), ("purif", "purification_unit"), ("well", "well"), ("river", "river_intake")]:
c[k] = (c["source_type"] == name).astype(int)
return c
BUDGET_MS = 300 # 1시간 주기 처리 예산(모의): 직전 주기가 예산을 넘으면 기능 축소 모드로 전환
FULL_EVERY = 12 # 기능 축소 중에도 12시간마다 전체 정제(지연 처리)를 수행
def make_sources(n, seed=0):
rng = np.random.default_rng(seed)
types = [s[1] for s in SOURCES]
return [(f"W{i + 1:02d}", types[i % len(types)], float(round(rng.uniform(0, 30), 1)), float(round(rng.uniform(0, 30), 1))) for i in range(n)]
def lite_check(batch, last):
"""기능 축소 모드: 레코드 단위 범위 점검 -> 직전 정상값 유지 -> 기준치 판정(정제·특징 계산 생략)."""
b = batch.copy()
for p in PARAMS:
v = pd.to_numeric(b[p], errors="coerce")
lo, hi = PHYS[p]
v = v.where(v.between(lo, hi))
b[p] = v.fillna(b["source_id"].map(last.get(p, {}))).fillna(0 if p == "ecoli" else np.nan)
last.setdefault(p, {}).update(b.dropna(subset=[p]).set_index("source_id")[p].to_dict())
b = b.dropna(subset=["ph", "turbidity", "chlorine", "metal"])
b["g"] = rule_grade(b)
return b
def run(n_src, days, events, seed, mode_auto=True, hours=None):
src = make_sources(n_src)
data = make_dataset(days=days, seed=seed, events=events, faults=[], sources=src)
rng = np.random.default_rng(seed)
bad_idx = rng.choice(len(data), int(len(data) * 0.01), replace=False) # 1% 손상 레코드 주입(문자·None)
data[PARAMS] = data[PARAMS].astype(object)
data.loc[bad_idx[: len(bad_idx) // 2], "turbidity"] = "ERR"
data.loc[bad_idx[len(bad_idx) // 2:], "ph"] = None
stored, quarantine, alerts, ticks, down, last = 0, 0, {}, [], 0, {}
buf, mode, switched = [], "full", None
mem = []
for h in range(hours or days * 24):
t0 = time.perf_counter()
try:
batch = data[data["t"] == h]
if mode_auto and mode == "full" and ticks and ticks[-1] > BUDGET_MS:
mode, switched = "lite", h # 한 번 전환하면 부하 해소 확인 전까지 유지
ok_mask = batch[["ph", "turbidity", "chlorine", "metal"]].apply(pd.to_numeric, errors="coerce").notna().all(axis=1)
quarantine += int((~ok_mask).sum())
if mode == "full":
buf.append(batch[ok_mask].astype({p: float for p in PARAMS}))
win = pd.concat(buf[-24:])
c = clean(win)
res = c[c["t"] == h].assign(g=lambda d: rule_grade(d))
else:
res = lite_check(batch, last)
buf.append(batch[ok_mask].astype({p: float for p in PARAMS}))
if h % FULL_EVERY == FULL_EVERY - 1: # 지연 전체 정제(기록 보정용)
clean(pd.concat(buf[-FULL_EVERY:]))
buf = buf[-24:]
stored += len(batch)
for sid in res.loc[res["g"] == 2, "source_id"]:
alerts.setdefault(sid, []).append(h)
except Exception:
down += 1
ticks.append((time.perf_counter() - t0) * 1000)
mem.append(sum(b.memory_usage(deep=True).sum() for b in buf) / 1e6) # 보관 버퍼 메모리(MB)
last = {p: last.get(p, {}) for p in PARAMS}
react = []
for e in events:
hs = [h for h in alerts.get(src[e[0]][0], []) if h >= e[2]]
if hs:
react.append(hs[0] - e[2])
return {"src": n_src, "recs": len(data), "stored": stored, "quarantine": quarantine, "down": down, "mode": mode if switched is None else f"lite({switched}h~)",
"tick_avg": np.mean(ticks), "tick_max": np.max(ticks), "mem_first": np.mean(mem[24:48]) if len(mem) >= 48 else mem[-1], "mem_last": np.mean(mem[-24:]),
"react": react, "detected": len(react), "n_events": len(events), "ticks": ticks}
days = 2
ev8 = [(i, et, 20 + i, 30, 1.0) for i, et in zip(range(5), ["sewage", "metal", "chemical", "disinfect_fail", "sewage"])]
ev48 = [(i, ["sewage", "metal", "chemical", "disinfect_fail"][i % 4], 20 + i % 6, 30, 1.0) for i in range(0, 48, 48 // 30 + 0)][:30]
t_all = time.perf_counter()
R = {"평시(8곳)": run(8, days, ev8, 1), "부하 500%(48곳, 30곳 동시 오염)": run(48, days, ev48, 2)}
print(f"[교육용가상자료] 부하 시험 {days}일 모의, 전체 실행 {time.perf_counter() - t_all:.1f}초")
rows = []
for k, r in R.items():
loss = 1 - r["stored"] / r["recs"]
rows.append({"구분": k, "레코드": r["recs"], "격리(손상)": r["quarantine"], "유실률": f"{loss:.1%}", "중단": r["down"],
"모드": r["mode"], "주기 평균ms": round(r["tick_avg"], 1), "주기 최대ms": round(r["tick_max"], 1),
"메모리MB(초기->말기)": f"{r['mem_first']:.1f}->{r['mem_last']:.1f}",
"경보 탐지": f"{r['detected']}/{r['n_events']}", "반응 평균h": round(float(np.mean(r["react"])), 1)})
print(pd.DataFrame(rows).set_index("구분").T.to_string())
S = R["부하 500%(48곳, 30곳 동시 오염)"]
print(f"\n부하 500%: 시스템 중단 {S['down']}건(목표 0건: {'달성' if S['down'] == 0 else '미달'}), "
f"데이터 유실률 {1 - S['stored'] / S['recs']:.1%}(목표 0%: {'달성' if S['stored'] == S['recs'] else '미달'})")
print(f"메모리 말기/초기 비율 {S['mem_last'] / max(S['mem_first'], 1e-6):.2f} (버퍼 24시간 고정, 1.2 이하면 누수 없음), "
f"경보 반응 최대 {max(S['react'])}시간")
# 기능 축소를 끄고 전체 정제를 강행하면 주기 시간이 얼마나 늘어나는지 (6시간만 측정)
F = run(48, 1, [], 3, mode_auto=False, hours=6)
print(f"기능 축소 미적용(전체 정제 강행) 주기 평균 {np.mean(F['ticks']):.0f}ms vs 기능 축소 {S['tick_avg']:.0f}ms "
f"(예산 {BUDGET_MS}ms) -> 병목: 급수원별 정제 반복(clean 함수)")
print("Fail-safe 원칙: 손상 레코드는 버리지 않고 격리 보관, 판정은 직전 정상값으로 이어가며, 과부하 시 정제·특징 계산을 미루고 경보 판정은 유지")
plt.figure(figsize=(9, 3.5))
plt.plot(R["평시(8곳)"]["ticks"], label="normal (8 sources)")
plt.plot(S["ticks"], label="500% load (48 sources, degraded mode)")
plt.plot(F["ticks"], label="500% load, full mode forced (6 h)")
plt.axhline(BUDGET_MS, color="red", ls="--", lw=1, label="budget")
plt.xlabel("simulated hour"); plt.ylabel("processing ms per tick"); plt.legend(); plt.title("Stress test tick time (synthetic)")
plt.tight_layout(); plt.show()
In [ ]:
"""AAR(사후 검토)로 기준치 Baseline의 수질 이상 오판 사례를 원인별로 분류하고, 단발 대장균 검출·완만한 중금속 증가에 대한 판단 임계치를 튜닝 세트에서 재조정한 뒤 검증 세트·벤치마크 회귀 테스트로 전후를 비교하고 조정 파라미터를 YAML로 출력한다."""
import itertools
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
# ---------------------------------------------------------------------------
# [교육용가상자료] 야전 급수원 가상 수질 센서 데이터 생성기 (실제 부대·위치·검사 결과 없음)
# 가상 급수원 8곳 x 14일 x 1시간 간격. 실제 수질(true_*)·센서 측정값·오염 사건·센서 고장 정답을 함께 만든다.
# 기준치는 교육용 가정값이며, 실제 음용 판정은 공인 검사 결과와 의무(예방의학) 담당 기준에 따른다.
# ---------------------------------------------------------------------------
PARAMS = ["ph", "turbidity", "chlorine", "metal", "ecoli"]
SOURCES = [("W01", "water_truck", 1.0, 2.0), ("W02", "water_truck", 3.5, 1.2), ("W03", "purification_unit", 2.2, 4.1),
("W04", "purification_unit", 5.0, 3.3), ("W05", "well", 0.8, 5.6), ("W06", "well", 4.4, 6.0),
("W07", "river_intake", 6.3, 1.8), ("W08", "river_intake", 6.9, 4.9)]
BASE = {"water_truck": (7.2, 0.25, 0.75, 1.5), "purification_unit": (7.0, 0.20, 0.85, 1.2),
"well": (6.8, 0.45, 0.50, 3.0), "river_intake": (7.3, 0.55, 0.60, 2.0)} # pH, 탁도(NTU), 잔류염소(mg/L), 중금속(ug/L)
LIMITS = {"ph": (5.8, 8.5), "turbidity": (1.0, 2.0), "chlorine": (0.2, 0.1), "metal": (5.0, 10.0), "ecoli": (1, 1)} # (주의, 위험) 가정 기준
EVENT_TYPES = { # 오염 사건: 변수별 최대 변화량, 최대치 도달 시간(h)
"sewage": ({"turbidity": 2.8, "chlorine": -0.5, "ecoli": 30, "ph": -0.3}, 4),
"metal": ({"metal": 22.0, "ph": -0.5}, 4),
"disinfect_fail": ({"chlorine": -0.8, "ecoli": 6}, 6),
"chemical": ({"ph": 1.7, "turbidity": 1.0}, 3),
"creeping_metal": ({"metal": 13.0}, 18),
}
FAULT_TYPES = ["stuck", "spike", "drift", "dropout"]
NOISE = {"ph": 0.04, "turbidity": 0.06, "chlorine": 0.03, "metal": 0.35}
DIGITS = {"ph": 2, "turbidity": 2, "chlorine": 2, "metal": 1}
def make_dataset(days=14, seed=7, n_events=6, n_faults=10, events=None, faults=None, sources=None):
"""events: [(급수원 번호, 사건 유형, 시작 시각 h, 지속 h, 강도)], faults: [(급수원 번호, 고장 유형, 항목, 시작 h, 지속 h)]"""
rng = np.random.default_rng(seed)
sources = sources or SOURCES
T = days * 24
tt = np.arange(T)
hour = tt % 24
night = (hour < 6) | (hour >= 20)
rain = np.zeros(T)
for _ in range(max(1, days // 4)):
s0, d = int(rng.integers(0, T - 14)), int(rng.integers(6, 14))
rain[s0:s0 + d] = np.maximum(rain[s0:s0 + d], rng.uniform(0.6, 1.0))
if events is None:
events = [(int(rng.integers(len(sources))), str(rng.choice(list(EVENT_TYPES))), int(rng.integers(min(24, T // 4), max(T // 4 + 1, T - 48))),
int(rng.integers(10, 30)), 1.0) for _ in range(n_events)]
if faults is None:
faults = [(int(rng.integers(len(sources))), str(rng.choice(FAULT_TYPES)), str(rng.choice(["ph", "turbidity", "chlorine", "metal"])),
int(rng.integers(12, max(13, T - 30))), int(rng.integers(6, 24))) for _ in range(n_faults)]
out = []
for si, (sid, stype, x, y) in enumerate(sources):
ph0, tb0, cl0, mt0 = BASE[stype]
tr = {"ph": ph0 + 0.08 * np.sin(2 * np.pi * hour / 24) + rng.normal(0, 0.03, T),
"turbidity": tb0 * (1 + 0.1 * rng.normal(0, 1, T)),
"chlorine": (cl0 * np.exp(-0.022 * hour) if stype == "water_truck" else cl0 + rng.normal(0, 0.03, T)),
"metal": mt0 + rng.normal(0, 0.2, T), "ecoli": np.zeros(T)}
if stype in ("river_intake", "well"):
tr["turbidity"] = tr["turbidity"] + rain * (0.9 if stype == "river_intake" else 0.45)
f_all, etype = np.zeros(T), np.array(["none"] * T, dtype=object)
for (s_i, et, onset, dur, scale) in events:
if s_i != si:
continue
delta, ramp = EVENT_TYPES[et]
f = np.clip((tt - onset + 1) / ramp, 0, 1) * ((tt >= onset) & (tt < onset + dur + ramp))
for k, v in delta.items():
tr[k] = tr[k] + v * f * scale
etype[f > 0] = et
f_all = np.maximum(f_all, f)
tr["chlorine"] = np.clip(tr["chlorine"], 0.02, None)
tr["turbidity"] = np.clip(tr["turbidity"], 0.05, None)
tr["metal"] = np.clip(tr["metal"], 0.1, None)
grade = np.where(f_all >= 0.2, 2, 0)
mild = (tr["turbidity"] > LIMITS["turbidity"][0]) | (tr["chlorine"] < LIMITS["chlorine"][0]) | (tr["metal"] > LIMITS["metal"][0])
grade = np.where((grade == 0) & mild, 1, grade)
ms = {}
for p in ["ph", "turbidity", "chlorine", "metal"]:
sd = NOISE[p] * np.where(night, 1.5, 1.0) * (tr[p] if p == "turbidity" else 1)
ms[p] = np.round(np.clip(tr[p] + rng.normal(0, 1, T) * sd, 0, None), DIGITS[p])
ms["ecoli"] = (rng.poisson(tr["ecoli"]) + (rng.random(T) < 0.004)).astype(float)
fault, fparam = np.array(["none"] * T, dtype=object), np.array(["-"] * T, dtype=object)
for (s_i, ft, p, s0, d) in faults:
if s_i != si:
continue
seg = slice(s0, min(T, s0 + d))
if ft == "stuck":
ms[p][seg] = ms[p][s0]
elif ft == "spike":
seg = slice(s0, s0 + 1)
ms[p][s0] = {"ph": 11.8, "turbidity": 38.0, "chlorine": 0.0, "metal": 75.0}[p]
elif ft == "drift":
ramp = np.linspace(0, 1, seg.stop - s0)
ms[p][seg] = np.clip(ms[p][seg] + ramp * {"ph": -1.6, "turbidity": 2.6, "chlorine": -0.75, "metal": 12.0}[p], 0, None).round(DIGITS[p])
elif ft == "dropout":
for q in PARAMS:
ms[q][seg] = np.nan
p = "all"
fault[seg], fparam[seg] = ft, p
df = pd.DataFrame({"source_id": sid, "source_type": stype, "x_km": x, "y_km": y, "t": tt,
"cond": np.where(rain > 0, "storm", np.where(night, "night", "day")),
**{p: ms[p] for p in PARAMS}, **{f"true_{p}": tr[p] for p in PARAMS},
"event": etype, "fault": fault, "fault_param": fparam, "grade": grade})
out.append(df)
return pd.concat(out, ignore_index=True)
# ---------------------------------------------------------------------------
# ---- 전처리 공통 함수: 물리 범위 점검 -> 고정값(stuck) 표시 -> 단발 스파이크 제거 -> 결측 보정 ----
PHYS = {"ph": (0, 14), "turbidity": (0, 1000), "chlorine": (0, 10), "metal": (0, 1000), "ecoli": (0, 100000)}
SPIKE_K = {"day": 6.0, "night": 7.0, "storm": 9.0} # 상황별(주간/야간/악천후) 스파이크 판정 배수
MIN_JUMP = {"ph": 0.8, "turbidity": 1.5, "chlorine": 0.3, "metal": 6.0}
STUCK_RUN, SHORT_GAP = 6, 3
def clean(df, spike_k=None, min_jump=None):
"""급수원별 시계열을 정제하고 항목별 처리 내역(<항목>_flag: range/stuck/spike/interp/long_gap)을 남긴다."""
# [TODO : 학생 실습 9] AAR 기반 수질 이상 오판 사례 분석 및 수질 위험 판단 임계치 재조정 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
GRADE_NAME = {0: "안전", 1: "주의", 2: "위험"}
def rule_grade(df):
"""기준치 1차 필터: 위험(음용 부적합 의심) = 한 항목이라도 위험 기준 초과, 주의 = 주의 기준 초과."""
# [TODO : 학생 실습 9] AAR 기반 수질 이상 오판 사례 분석 및 수질 위험 판단 임계치 재조정 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def reasons(row):
r = []
if not LIMITS["ph"][0] <= row["ph"] <= LIMITS["ph"][1]:
r.append(f"pH {row['ph']:.2f}")
if row["turbidity"] > LIMITS["turbidity"][0]:
r.append(f"탁도 {row['turbidity']:.2f}NTU")
if row["chlorine"] < LIMITS["chlorine"][0]:
r.append(f"잔류염소 {row['chlorine']:.2f}mg/L")
if row["metal"] > LIMITS["metal"][0]:
r.append(f"중금속 {row['metal']:.1f}ug/L")
if row["ecoli"] >= 1:
r.append(f"대장균 {row['ecoli']:.0f}")
return ", ".join(r) or "-"
FEATS = [f"{p}{s}" for p in ["ph", "turbidity", "chlorine", "metal"] for s in ["", "_d1", "_d3", "_dev", "_sd6"]] + \
["ecoli", "ecoli_3h", "storm", "night", "n_flag", "truck", "purif", "well", "river"]
def add_features(c):
"""시간 순서를 지키는(과거 값만 쓰는) 급수원별 특징: 변화량, 24시간 기준선 대비 편차, 6시간 변동성, 대장균 누적 등."""
c = c.sort_values(["source_id", "t"]).reset_index(drop=True)
g = c.groupby("source_id")
for p in ["ph", "turbidity", "chlorine", "metal"]:
c[f"{p}_d1"] = g[p].diff().fillna(0)
c[f"{p}_d3"] = g[p].diff(3).fillna(0)
base = g[p].transform(lambda s: s.shift(1).rolling(24, min_periods=4).median())
c[f"{p}_dev"] = (c[p] - base).fillna(0)
c[f"{p}_sd6"] = g[p].transform(lambda s: s.rolling(6, min_periods=2).std()).fillna(0)
c["ecoli_3h"] = g["ecoli"].transform(lambda s: s.rolling(3, min_periods=1).sum())
c["storm"], c["night"] = (c["cond"] == "storm").astype(int), (c["cond"] == "night").astype(int)
flags = [f"{p}_flag" for p in PARAMS if f"{p}_flag" in c]
c["n_flag"] = sum(c[f].isin(["stuck", "spike", "long_gap", "range"]).astype(int) for f in flags) if flags else 0
for k, name in [("truck", "water_truck"), ("purif", "purification_unit"), ("well", "well"), ("river", "river_intake")]:
c[k] = (c["source_type"] == name).astype(int)
return c
BEFORE = {"ecoli_3h_min": 1, "metal_dev": 99.0, "metal_min": 99.0} # 현행: 대장균 1회 검출 즉시 위험, 추세 규칙 없음
def tuned_grade(c, P):
"""재조정 규칙: 대장균은 3시간 누적 기준(또는 탁도·염소 이상 동반 시 즉시), 중금속은 기준선 대비 상승 추세 규칙 추가."""
ph, tb, cl, mt, ec = (c[p].to_numpy(float) for p in PARAMS)
if P["ecoli_3h_min"] <= 1: # 현행 규칙: 그 시각 1회 검출이면 위험
eco = ec >= 1
else:
eco = (c["ecoli_3h"].to_numpy() >= P["ecoli_3h_min"]) | ((ec >= 1) & ((tb > LIMITS["turbidity"][0]) | (cl < LIMITS["chlorine"][0])))
trend = (c["metal_dev"].to_numpy() > P["metal_dev"]) & (mt > P["metal_min"])
danger = (ph < LIMITS["ph"][0]) | (ph > LIMITS["ph"][1]) | (tb > LIMITS["turbidity"][1]) | (cl < LIMITS["chlorine"][1]) \
| (mt > LIMITS["metal"][1]) | eco | trend
caution = (tb > LIMITS["turbidity"][0]) | (cl < LIMITS["chlorine"][0]) | (mt > LIMITS["metal"][0]) | (ec >= 1)
return np.where(danger, 2, np.where(caution, 1, 0))
def cause(r, pred):
if r["grade"] == 2 and pred != 2:
return {"creeping_metal": "놓침-중금속 완만 증가", "chemical": "놓침-화학(초기 pH 변화)"}.get(r["event"], f"놓침-{r['event']}")
if r["grade"] != 2 and pred == 2:
if r["fault"] != "none":
return f"오경보-센서 {r['fault']}"
return "오경보-단발 대장균" if r["ecoli"] >= 1 else "오경보-기타"
return "주의/안전 경계 혼동"
D = {name: add_features(clean(make_dataset(seed=s, n_events=10, n_faults=14))) for name, s in
[("튜닝(seed31)", 31), ("검증(seed41)", 41), ("벤치마크(seed11)", 11)]}
# 1) AAR: 튜닝 세트의 오판 원인 분류
tune = D["튜닝(seed31)"]
p0 = tuned_grade(tune, BEFORE)
err = tune[p0 != tune["grade"]]
causes = pd.Series([cause(r, p) for (_, r), p in zip(err.iterrows(), p0[p0 != tune["grade"]])]).value_counts()
print(f"[교육용가상자료] AAR 대상 오판 {len(err)}건(튜닝 세트 {len(tune):,}건) 원인 분류")
print(causes.to_string())
# 2) 재조정: 고칠 수 있는 원인(단발 대장균, 중금속 완만 증가)에 대한 임계치 격자 탐색
# AAR 결정: 대장균 단발 1회 검출은 '주의 + 재채수'로 낮추고(3시간 누적 2회 이상부터 위험), 그 기준값만 탐색한다
best = None
for e3, md, mm in itertools.product([2, 3], [1.0, 1.5, 2.0, 3.0], [4.0, 5.0, 6.0]):
P = {"ecoli_3h_min": e3, "metal_dev": md, "metal_min": mm}
acc = (tuned_grade(tune, P) == tune["grade"]).mean()
if best is None or acc > best[0]:
best = (acc, P)
AFTER = best[1]
print("\n재조정 파라미터:", AFTER)
# 3) 회귀 테스트: 세 데이터셋에서 전후 비교
rows = []
for name, c in D.items():
b, a = tuned_grade(c, BEFORE), tuned_grade(c, AFTER)
d = c["grade"] == 2
rows.append({"데이터": name, "정확도 전": f"{(b == c['grade']).mean():.1%}", "정확도 후": f"{(a == c['grade']).mean():.1%}",
"향상(%p)": round(((a == c["grade"]).mean() - (b == c["grade"]).mean()) * 100, 1),
"오판 전→후": f"{int((b != c['grade']).sum())}→{int((a != c['grade']).sum())}",
"새로 틀림": int(((b == c["grade"]) & (a != c["grade"])).sum()),
"위험 재현율 후": f"{((a == 2) & d).sum() / d.sum():.1%}"})
reg = pd.DataFrame(rows)
print("\n[회귀 테스트]")
print(reg.to_string(index=False))
gain = reg.loc[1, "향상(%p)"]
print(f"검증 세트 종합 정확도 향상 {gain}%p (목표 10%p 이상: {'달성' if gain >= 10 else '미달'} - Baseline 정확도가 이미 97% 수준이라 10%p 여지가 없음)")
# 4) 재발 오류: 고친 원인 유형이 검증 세트에서 다시 나오는가
val = D["검증(seed41)"]
pa = tuned_grade(val, AFTER)
ev = val[pa != val["grade"]]
vc = pd.Series([cause(r, p) for (_, r), p in zip(ev.iterrows(), pa[pa != val["grade"]])]).value_counts()
fixed_types = ["오경보-단발 대장균", "놓침-중금속 완만 증가"]
rec = {k: int(vc.get(k, 0)) for k in fixed_types}
print(f"\n재발 오류(고친 유형의 검증 세트 재발): {rec} -> 합계 {sum(rec.values())}건(목표 0건: {'달성' if sum(rec.values()) == 0 else '미달'})")
print("남은 원인 상위:", vc.head(3).to_dict(), "-> 센서 고장(드리프트·통신 단절) 오경보는 교차 센서·교정 주기와 미션 5 신뢰도 지수로, 화학 초기 변화는 pH 변화율 규칙 후보로 다음 AAR에 넘긴다")
print("\n# configs/wel03_tuned_params.yaml")
print(f"ecoli:\n danger_3h_count: {AFTER['ecoli_3h_min']} # 단발 1회는 '주의 + 재채수'\nmetal_trend:\n"
f" dev_from_24h_baseline_ugL: {AFTER['metal_dev']}\n min_level_ugL: {AFTER['metal_min']}\nreviewed_by: 예방의학 담당(가상) # 실제 기준은 공식 수질 기준 확인")
fig, ax = plt.subplots(figsize=(8, 3.5))
x = np.arange(len(reg))
ax.bar(x - 0.2, [(tuned_grade(c, BEFORE) != c["grade"]).sum() for c in D.values()], 0.4, label="before", color="tab:gray")
ax.bar(x + 0.2, [(tuned_grade(c, AFTER) != c["grade"]).sum() for c in D.values()], 0.4, label="after", color="tab:blue")
ax.set_xticks(x, ["tune(31)", "valid(41)", "bench(11)"]); ax.set_ylabel("misjudged hours"); ax.legend()
ax.set_title("AAR retuning: misjudgments before/after (synthetic)")
plt.tight_layout(); plt.show()
In [ ]:
"""입력 검증, 전처리, 재조정 기준치 판정과 랜덤 포레스트 위험 분류, 신뢰도 등급, 인간 검토 큐 등록, 상황판 요약까지 가상 수질검사 위험 신호 대시보드를 한 파일로 통합 실행하고, 여러 시나리오 End-to-End 실행·자체 점검(함수 커버리지)·지휘관 브리핑 요약을 출력한다."""
import sys
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.ensemble import RandomForestClassifier
# ---------------------------------------------------------------------------
# [교육용가상자료] 야전 급수원 가상 수질 센서 데이터 생성기 (실제 부대·위치·검사 결과 없음)
# 가상 급수원 8곳 x 14일 x 1시간 간격. 실제 수질(true_*)·센서 측정값·오염 사건·센서 고장 정답을 함께 만든다.
# 기준치는 교육용 가정값이며, 실제 음용 판정은 공인 검사 결과와 의무(예방의학) 담당 기준에 따른다.
# ---------------------------------------------------------------------------
PARAMS = ["ph", "turbidity", "chlorine", "metal", "ecoli"]
SOURCES = [("W01", "water_truck", 1.0, 2.0), ("W02", "water_truck", 3.5, 1.2), ("W03", "purification_unit", 2.2, 4.1),
("W04", "purification_unit", 5.0, 3.3), ("W05", "well", 0.8, 5.6), ("W06", "well", 4.4, 6.0),
("W07", "river_intake", 6.3, 1.8), ("W08", "river_intake", 6.9, 4.9)]
BASE = {"water_truck": (7.2, 0.25, 0.75, 1.5), "purification_unit": (7.0, 0.20, 0.85, 1.2),
"well": (6.8, 0.45, 0.50, 3.0), "river_intake": (7.3, 0.55, 0.60, 2.0)} # pH, 탁도(NTU), 잔류염소(mg/L), 중금속(ug/L)
LIMITS = {"ph": (5.8, 8.5), "turbidity": (1.0, 2.0), "chlorine": (0.2, 0.1), "metal": (5.0, 10.0), "ecoli": (1, 1)} # (주의, 위험) 가정 기준
EVENT_TYPES = { # 오염 사건: 변수별 최대 변화량, 최대치 도달 시간(h)
"sewage": ({"turbidity": 2.8, "chlorine": -0.5, "ecoli": 30, "ph": -0.3}, 4),
"metal": ({"metal": 22.0, "ph": -0.5}, 4),
"disinfect_fail": ({"chlorine": -0.8, "ecoli": 6}, 6),
"chemical": ({"ph": 1.7, "turbidity": 1.0}, 3),
"creeping_metal": ({"metal": 13.0}, 18),
}
FAULT_TYPES = ["stuck", "spike", "drift", "dropout"]
NOISE = {"ph": 0.04, "turbidity": 0.06, "chlorine": 0.03, "metal": 0.35}
DIGITS = {"ph": 2, "turbidity": 2, "chlorine": 2, "metal": 1}
def make_dataset(days=14, seed=7, n_events=6, n_faults=10, events=None, faults=None, sources=None):
"""events: [(급수원 번호, 사건 유형, 시작 시각 h, 지속 h, 강도)], faults: [(급수원 번호, 고장 유형, 항목, 시작 h, 지속 h)]"""
rng = np.random.default_rng(seed)
sources = sources or SOURCES
T = days * 24
tt = np.arange(T)
hour = tt % 24
night = (hour < 6) | (hour >= 20)
rain = np.zeros(T)
for _ in range(max(1, days // 4)):
s0, d = int(rng.integers(0, T - 14)), int(rng.integers(6, 14))
rain[s0:s0 + d] = np.maximum(rain[s0:s0 + d], rng.uniform(0.6, 1.0))
if events is None:
events = [(int(rng.integers(len(sources))), str(rng.choice(list(EVENT_TYPES))), int(rng.integers(min(24, T // 4), max(T // 4 + 1, T - 48))),
int(rng.integers(10, 30)), 1.0) for _ in range(n_events)]
if faults is None:
faults = [(int(rng.integers(len(sources))), str(rng.choice(FAULT_TYPES)), str(rng.choice(["ph", "turbidity", "chlorine", "metal"])),
int(rng.integers(12, max(13, T - 30))), int(rng.integers(6, 24))) for _ in range(n_faults)]
out = []
for si, (sid, stype, x, y) in enumerate(sources):
ph0, tb0, cl0, mt0 = BASE[stype]
tr = {"ph": ph0 + 0.08 * np.sin(2 * np.pi * hour / 24) + rng.normal(0, 0.03, T),
"turbidity": tb0 * (1 + 0.1 * rng.normal(0, 1, T)),
"chlorine": (cl0 * np.exp(-0.022 * hour) if stype == "water_truck" else cl0 + rng.normal(0, 0.03, T)),
"metal": mt0 + rng.normal(0, 0.2, T), "ecoli": np.zeros(T)}
if stype in ("river_intake", "well"):
tr["turbidity"] = tr["turbidity"] + rain * (0.9 if stype == "river_intake" else 0.45)
f_all, etype = np.zeros(T), np.array(["none"] * T, dtype=object)
for (s_i, et, onset, dur, scale) in events:
if s_i != si:
continue
delta, ramp = EVENT_TYPES[et]
f = np.clip((tt - onset + 1) / ramp, 0, 1) * ((tt >= onset) & (tt < onset + dur + ramp))
for k, v in delta.items():
tr[k] = tr[k] + v * f * scale
etype[f > 0] = et
f_all = np.maximum(f_all, f)
tr["chlorine"] = np.clip(tr["chlorine"], 0.02, None)
tr["turbidity"] = np.clip(tr["turbidity"], 0.05, None)
tr["metal"] = np.clip(tr["metal"], 0.1, None)
grade = np.where(f_all >= 0.2, 2, 0)
mild = (tr["turbidity"] > LIMITS["turbidity"][0]) | (tr["chlorine"] < LIMITS["chlorine"][0]) | (tr["metal"] > LIMITS["metal"][0])
grade = np.where((grade == 0) & mild, 1, grade)
ms = {}
for p in ["ph", "turbidity", "chlorine", "metal"]:
sd = NOISE[p] * np.where(night, 1.5, 1.0) * (tr[p] if p == "turbidity" else 1)
ms[p] = np.round(np.clip(tr[p] + rng.normal(0, 1, T) * sd, 0, None), DIGITS[p])
ms["ecoli"] = (rng.poisson(tr["ecoli"]) + (rng.random(T) < 0.004)).astype(float)
fault, fparam = np.array(["none"] * T, dtype=object), np.array(["-"] * T, dtype=object)
for (s_i, ft, p, s0, d) in faults:
if s_i != si:
continue
seg = slice(s0, min(T, s0 + d))
if ft == "stuck":
ms[p][seg] = ms[p][s0]
elif ft == "spike":
seg = slice(s0, s0 + 1)
ms[p][s0] = {"ph": 11.8, "turbidity": 38.0, "chlorine": 0.0, "metal": 75.0}[p]
elif ft == "drift":
ramp = np.linspace(0, 1, seg.stop - s0)
ms[p][seg] = np.clip(ms[p][seg] + ramp * {"ph": -1.6, "turbidity": 2.6, "chlorine": -0.75, "metal": 12.0}[p], 0, None).round(DIGITS[p])
elif ft == "dropout":
for q in PARAMS:
ms[q][seg] = np.nan
p = "all"
fault[seg], fparam[seg] = ft, p
df = pd.DataFrame({"source_id": sid, "source_type": stype, "x_km": x, "y_km": y, "t": tt,
"cond": np.where(rain > 0, "storm", np.where(night, "night", "day")),
**{p: ms[p] for p in PARAMS}, **{f"true_{p}": tr[p] for p in PARAMS},
"event": etype, "fault": fault, "fault_param": fparam, "grade": grade})
out.append(df)
return pd.concat(out, ignore_index=True)
# ---------------------------------------------------------------------------
# ---- 전처리 공통 함수: 물리 범위 점검 -> 고정값(stuck) 표시 -> 단발 스파이크 제거 -> 결측 보정 ----
PHYS = {"ph": (0, 14), "turbidity": (0, 1000), "chlorine": (0, 10), "metal": (0, 1000), "ecoli": (0, 100000)}
SPIKE_K = {"day": 6.0, "night": 7.0, "storm": 9.0} # 상황별(주간/야간/악천후) 스파이크 판정 배수
MIN_JUMP = {"ph": 0.8, "turbidity": 1.5, "chlorine": 0.3, "metal": 6.0}
STUCK_RUN, SHORT_GAP = 6, 3
def clean(df, spike_k=None, min_jump=None):
"""급수원별 시계열을 정제하고 항목별 처리 내역(<항목>_flag: range/stuck/spike/interp/long_gap)을 남긴다."""
# [TODO : 학생 실습 10] 가상 수질검사 데이터 위험 신호 대시보드 시스템 통합 패키지 완성 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
GRADE_NAME = {0: "안전", 1: "주의", 2: "위험"}
def rule_grade(df):
"""기준치 1차 필터: 위험(음용 부적합 의심) = 한 항목이라도 위험 기준 초과, 주의 = 주의 기준 초과."""
# [TODO : 학생 실습 10] 가상 수질검사 데이터 위험 신호 대시보드 시스템 통합 패키지 완성 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def reasons(row):
r = []
if not LIMITS["ph"][0] <= row["ph"] <= LIMITS["ph"][1]:
r.append(f"pH {row['ph']:.2f}")
if row["turbidity"] > LIMITS["turbidity"][0]:
r.append(f"탁도 {row['turbidity']:.2f}NTU")
if row["chlorine"] < LIMITS["chlorine"][0]:
r.append(f"잔류염소 {row['chlorine']:.2f}mg/L")
if row["metal"] > LIMITS["metal"][0]:
r.append(f"중금속 {row['metal']:.1f}ug/L")
if row["ecoli"] >= 1:
r.append(f"대장균 {row['ecoli']:.0f}")
return ", ".join(r) or "-"
FEATS = [f"{p}{s}" for p in ["ph", "turbidity", "chlorine", "metal"] for s in ["", "_d1", "_d3", "_dev", "_sd6"]] + \
["ecoli", "ecoli_3h", "storm", "night", "n_flag", "truck", "purif", "well", "river"]
def add_features(c):
"""시간 순서를 지키는(과거 값만 쓰는) 급수원별 특징: 변화량, 24시간 기준선 대비 편차, 6시간 변동성, 대장균 누적 등."""
c = c.sort_values(["source_id", "t"]).reset_index(drop=True)
g = c.groupby("source_id")
for p in ["ph", "turbidity", "chlorine", "metal"]:
c[f"{p}_d1"] = g[p].diff().fillna(0)
c[f"{p}_d3"] = g[p].diff(3).fillna(0)
base = g[p].transform(lambda s: s.shift(1).rolling(24, min_periods=4).median())
c[f"{p}_dev"] = (c[p] - base).fillna(0)
c[f"{p}_sd6"] = g[p].transform(lambda s: s.rolling(6, min_periods=2).std()).fillna(0)
c["ecoli_3h"] = g["ecoli"].transform(lambda s: s.rolling(3, min_periods=1).sum())
c["storm"], c["night"] = (c["cond"] == "storm").astype(int), (c["cond"] == "night").astype(int)
flags = [f"{p}_flag" for p in PARAMS if f"{p}_flag" in c]
c["n_flag"] = sum(c[f].isin(["stuck", "spike", "long_gap", "range"]).astype(int) for f in flags) if flags else 0
for k, name in [("truck", "water_truck"), ("purif", "purification_unit"), ("well", "well"), ("river", "river_intake")]:
c[k] = (c["source_type"] == name).astype(int)
return c
CONFIG = {"ecoli_3h_min": 2, "metal_dev": 1.0, "metal_min": 5.0, # 미션 9 재조정 값
"conf_high": 60, "scale": {"ph": 0.35, "turbidity": 0.6, "chlorine": 0.2, "metal": 3.0}}
README = {"1. 목적": "가상 수질 센서 자료로 급수원 위험 신호를 조기에 알리는 판단 보조 도구",
"2. 실행 방법": "python main_wel03.py (원클릭 실행, 외부 파일·네트워크 불필요)",
"3. 입력 형식": "급수원 ID·유형·시각·pH·탁도·잔류염소·중금속·대장균(미션 1 스키마)",
"4. 판정 규칙": "가정 기준치 + 재조정 규칙 + 모델, 신뢰도 High만 급수 중단 검토 상신",
"5. 인간 검토": "급수 중단·재개는 예방의학 담당 검토 후 지휘관 승인, 자동 확정 없음",
"6. 한계와 주의": "교육용 가상 자료, 실제 판정은 공식 수질 기준과 의무·환경 담당 확인",
"7. 연동 확장": "인접 부대 급수 정보 교환용 표준 필드(JSON) API를 차기 과제로 둔다"}
def validate(df):
ok = df["source_id"].astype(str).str.match(r"^W\d{2}$") & df["t"].ge(0)
return df[ok].copy(), int((~ok).sum())
def tuned_grade(c):
ph, tb, cl, mt, ec = (c[p].to_numpy(float) for p in PARAMS)
eco = (c["ecoli_3h"].to_numpy() >= CONFIG["ecoli_3h_min"]) | ((ec >= 1) & ((tb > LIMITS["turbidity"][0]) | (cl < LIMITS["chlorine"][0])))
trend = (c["metal_dev"].to_numpy() > CONFIG["metal_dev"]) & (mt > CONFIG["metal_min"])
danger = (ph < LIMITS["ph"][0]) | (ph > LIMITS["ph"][1]) | (tb > LIMITS["turbidity"][1]) | (cl < LIMITS["chlorine"][1]) \
| (mt > LIMITS["metal"][1]) | eco | trend
caution = (tb > LIMITS["turbidity"][0]) | (cl < LIMITS["chlorine"][0]) | (mt > LIMITS["metal"][0]) | (ec >= 1)
return np.where(danger, 2, np.where(caution, 1, 0))
def confidence(c, alert):
S = CONFIG["scale"]
n_dev = sum(((c[f"{p}_dev"].abs() > S[p]) | (c[f"{p}_d3"].abs() > S[p])).astype(int) for p in S) + (c["ecoli_3h"] >= 2)
a = pd.Series(alert.astype(int), index=c.index)
run = a.groupby([c["source_id"], (a != a.groupby(c["source_id"]).shift()).cumsum()]).cumsum()
g = c.groupby("source_id")
flagged = g["n_flag"].transform(lambda s: s.rolling(6, min_periods=1).max()) > 0
# 드리프트 의심: 한 항목만 6시간 중 5시간 이상 같은 방향으로 꾸준히 변함(잡음이 커지는 야간에도 유지되는 기준)
mono = pd.concat([g[f"{p}_d1"].transform(lambda s: np.sign(s).rolling(6, min_periods=3).sum()).abs() >= 4 for p in S], axis=1).any(axis=1)
drift = (n_dev <= 1) & mono & (c["ecoli_3h"] < 2)
return 40 * np.minimum(n_dev, 3) / 3 + 25 * np.minimum(run, 3) / 3 + 20 * (~flagged) + 15 * (~drift)
def train_model():
tr = pd.concat([add_features(clean(make_dataset(seed=s, n_events=14, n_faults=14))) for s in (7, 8)], ignore_index=True)
return RandomForestClassifier(n_estimators=150, max_depth=10, class_weight="balanced_subsample", random_state=0, n_jobs=1).fit(tr[FEATS], tr["grade"])
def run_pipeline(raw, model):
data, n_bad = validate(raw)
c = add_features(clean(data))
c["rule"] = tuned_grade(c)
c["ml"] = model.predict(c[FEATS])
c["final"] = np.maximum(c["rule"], c["ml"]) # 둘 중 하나라도 위험이면 위험(놓침 방지)
alert = (c["final"] == 2).to_numpy()
c["conf"] = np.where(alert, confidence(c, alert), np.nan)
queue = c[alert & (c["conf"] >= CONFIG["conf_high"])].groupby("source_id").agg(first=("t", "min"), hours=("t", "size"))
queue["status"] = "PENDING(지휘관 결정 대기)"
return c, queue, n_bad
def self_tests(model):
res = {}
clean_df = make_dataset(days=2, seed=50, events=[], faults=[])
c, q, _ = run_pipeline(clean_df, model)
res["오염 없는 자료에서 급수 중단 상신 0건"] = len(q) == 0
ev = make_dataset(days=2, seed=50, events=[(0, "sewage", 20, 12, 1.0)], faults=[])
c, q, _ = run_pipeline(ev, model)
res["하수 유입 급수원(W01) 상신"] = "W01" in q.index
res["상신 건은 모두 PENDING(자동 확정 없음)"] = q["status"].str.startswith("PENDING").all()
bad = ev.copy(); bad.loc[3, "source_id"] = "??"
res["손상 레코드 격리"] = run_pipeline(bad, model)[2] == 1
dr = make_dataset(days=2, seed=50, events=[], faults=[(1, "drift", "turbidity", 10, 18)])
c, q, _ = run_pipeline(dr, model)
n_alert = int((c.loc[c["source_id"] == "W02", "final"] == 2).sum()) # 드리프트 구간 위험 판정 시간 중 High 상신 시간이 절반 미만이면 억제 작동
res["센서 드리프트 상신 시간 절반 미만으로 억제"] = ("W02" not in q.index) or q.loc["W02", "hours"] < 0.5 * n_alert
res["기준치 필터·판정 근거 문구 생성"] = len(reasons(c[c["source_id"] == "W02"].iloc[40])) > 0 and rule_grade(c).shape[0] == len(c)
res["결측 보정 후 남은 결측 0"] = int(c[PARAMS].isna().sum().sum()) == 0
res["운영 문서 7개 장 작성"] = all(len(v) > 10 for v in README.values())
return res
# ---- 자체 점검(함수 호출 추적으로 커버리지 측정) ----
t_all = time.perf_counter()
called = set()
own = {n for n, f in list(globals().items()) if callable(f) and getattr(f, "__module__", "") == "__main__" and hasattr(f, "__code__")}
sys.setprofile(lambda fr, ev, arg: called.add(fr.f_code.co_name) if ev == "call" else None)
model = train_model()
tests = self_tests(model)
e2e = []
for s in (11, 21, 31, 41, 51):
try:
c, q, nb = run_pipeline(make_dataset(seed=s, n_events=10, n_faults=14), model)
e2e.append((s, True, (c["final"] == c["grade"]).mean(), len(q)))
except Exception as e:
e2e.append((s, False, np.nan, 0))
sys.setprofile(None)
cov = len(own & called) / len(own)
print(f"[교육용가상자료] 통합 실행 {time.perf_counter() - t_all:.1f}초 (모델 학습 포함)")
print("[End-to-End 시나리오 5종]")
for s, ok, acc, nq in e2e:
print(f" seed {s}: {'성공' if ok else '실패'}, 최종 판정 정확도 {acc:.1%}, 급수 중단 검토 상신 {nq}곳")
rate = np.mean([x[1] for x in e2e])
print(f"E2E 성공률 {rate:.0%} (목표 100%: {'달성' if rate == 1 else '미달'})")
print("[자체 점검]")
for k, v in tests.items():
print(f" [{'PASS' if v else 'FAIL'}] {k}")
print(f"자체 점검 {sum(tests.values())}/{len(tests)} 통과, 함수 커버리지 {cov:.0%}({len(own & called)}/{len(own)}, 목표 85% 이상: {'달성' if cov >= 0.85 else '미달'})")
print(f"운영 문서(README) {len(README)}개 장: " + ", ".join(README))
# ---- 브리핑 요약(벤치마크 seed 11, 운용 시점 기준) ----
c, q, _ = run_pipeline(make_dataset(seed=11, n_events=10, n_faults=14), model)
d = c["grade"] == 2
print("\n[지휘관 브리핑 요약]")
print(f"1. 급수원 8곳 14일 모의: 위험 재현율 {((c['final'] == 2) & d).sum() / d.sum():.1%}, 종합 정확도 {(c['final'] == c['grade']).mean():.1%}")
print(f"2. 급수 중단 검토 상신 {len(q)}곳({', '.join(q.index)}): 모두 예방의학 담당 검토·지휘관 결정 대기")
print("3. 센서 이상(드리프트·스파이크·단절)은 재측정·교정 요청으로 분리해 불필요한 단수를 줄인다")
print("4. 실제 음용 판정은 공식 수질 기준과 의무(예방의학)·환경 담당 확인에 따른다(본 도구는 판단 보조)")
last = c[c["t"] == c["t"].max()].set_index("source_id")
fig, ax = plt.subplots(1, 2, figsize=(11, 4))
ax[0].bar(last.index, last["final"], color=[{0: "tab:green", 1: "gold", 2: "red"}[g] for g in last["final"]])
ax[0].set_yticks([0, 1, 2], ["safe", "caution", "danger"]); ax[0].set_title("Final status by source (last hour)")
ax[1].bar([str(x[0]) for x in e2e], [x[2] for x in e2e], color="tab:blue"); ax[1].set_ylim(0.9, 1.0)
ax[1].set_title("E2E accuracy by scenario seed"); ax[1].set_xlabel("seed")
plt.suptitle("Integrated water risk dashboard package (synthetic)"); plt.tight_layout(); plt.show()