176 KiB
176 KiB
🎯 [TRN01] 드론 시뮬레이터 비행로그 기반 숙련도 분석기
Google Colab 실습 환경 (교육·훈련 / Level 3)
📌 1. 프로젝트 개요 및 배경
| 항목 | 내용 |
|---|---|
| 과정 분류 | 교육·훈련 (Level 3) |
| 학습 소요 시간 | 10개 미션 (5일 / 총 27.5시간) |
| 환경 사양 | Python 3.10+, numpy, pandas, matplotlib, scipy, scikit-learn |
📖 프로젝트 목표
러시아-우크라이나 전쟁에서는 소형 드론 조종 인력을 짧은 기간에 대량으로 양성해야 했고, 시뮬레이터 훈련에서 누가 어떤 조작 약점을 가지고 있는지 빠르게 찾아 개인별로 보완하는 일이 전투력 유지의 관건이 되었다. 본 프로젝트에서는 교육용 가상 드론 시뮬레이터 비행로그(경로 비행, 호버링, 돌발 장애물 회피, 비상 착륙)를 코드로 직접 만들어, 로그 스키마 정의, 잡음·결측 전처리, 경로 이탈도·호버링 안정성 기반 Baseline 채점, 장애물 반응·착륙 속도를 반영한 정밀 평가 엔진, 조작 습관 분석과 신뢰도 등급, 3D 재생 대시보드, 교관 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 [ ]:
"""드론 조종 시뮬레이터 비행로그(스틱 조작, 자세각, 고도, 위치·목표 경로)와 비행 세션 메타데이터(가명 조종자, 비행 조건, 전술 파라미터)의 스키마를 정의하고 가상 샘플 데이터를 검증한다."""
import json
import re
import numpy as np
import pandas as pd
# ---------------------------------------------------------------------------
# [교육용가상자료] 드론 조종 시뮬레이터 비행로그 생성기 (실제 훈련생·장비 정보 없음, 조종자는 가명 P001~)
# 비행 1회 = 120초(10Hz): 경로 비행(0~60초) -> 호버링(60~85초) -> 돌발 장애물 회피(85초) -> 비상 착륙(100초 지시)
# 조종자마다 숨은 숙련도(skill)와 조작 습관(급조작 jerky, 과보정 overcorr)이 있고, 교관 점수는 이를 반영해 만든다.
# ---------------------------------------------------------------------------
DT, STEPS = 0.1, 1200
T = np.arange(STEPS) * DT
COND = {"day": dict(wind=0.4, gust=0.0, gps=0.25, delay=0.0), "night": dict(wind=0.4, gust=0.0, gps=0.5, delay=0.15),
"gust": dict(wind=1.2, gust=2.5, gps=0.25, delay=0.0)}
OBST_T, LAND_T = 85.0, 100.0
def make_pilots(n=40, seed=7):
rng = np.random.default_rng(seed)
skill = rng.beta(2.2, 2.0, n)
return pd.DataFrame({"pilot": [f"P{i + 1:03d}" for i in range(n)], "skill": skill,
"jerky": rng.random(n) < 0.15 + 0.35 * (1 - skill), "overcorr": rng.random(n) < 0.15 + 0.35 * (1 - skill)})
def target_path(t):
"""사각형 경로(40x30m, 고도 20m)를 3m/s로 비행 -> 60초 지점에서 호버링 -> 착륙 지시 후 고도 0."""
# [TODO : 학생 실습 1] 드론 조종 시뮬레이터 비행로그(스틱 조작, 자세각, 고도 유지, 경로 오차) 데이터 스키마 정의 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def fly(pilots, cond="day", seed=0, delay_extra=0.0, obstacle_side=None):
"""여러 조종자의 비행 1회를 동시에 모의한다(벡터화). 반환: 로그 배열 dict, 비행별 정답 정보 DataFrame."""
# [TODO : 학생 실습 1] 드론 조종 시뮬레이터 비행로그(스틱 조작, 자세각, 고도 유지, 경로 오차) 데이터 스키마 정의 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def one(log, j):
"""여러 조종자 로그에서 j번째 조종자 1명의 로그(DataFrame)를 꺼낸다."""
return pd.DataFrame({k: (v if v.ndim == 1 else v[:, j]) for k, v in log.items()})
# ---------------------------------------------------------------------------
SCHEMA = {
"session": {
"session_id": {"type": str, "pattern": r"^S\d{4}$"},
"pilot": {"type": str, "pattern": r"^P\d{3}$"}, # 가명 ID만 허용(실명·군번 금지)
"sim_course": {"type": str, "enum": ["RECT40x30_HOVER_OBS_LAND"]},
"rate_hz": {"type": int, "enum": [10, 20, 50]},
"duration_s": {"type": (int, float), "min": 10, "max": 1800},
},
"tactical": {
"condition": {"type": str, "enum": list(COND)},
"wind_mps": {"type": (int, float), "min": 0, "max": 20},
"gps_error_m": {"type": (int, float), "min": 0, "max": 50},
"video_delay_ms": {"type": (int, float), "min": 0, "max": 2000},
},
"record": {
"t": {"type": (int, float), "min": 0},
"stick_roll": {"type": (int, float), "min": -1, "max": 1},
"stick_pitch": {"type": (int, float), "min": -1, "max": 1},
"throttle": {"type": (int, float), "min": -1, "max": 1},
"roll_deg": {"type": (int, float), "min": -90, "max": 90},
"pitch_deg": {"type": (int, float), "min": -90, "max": 90},
"alt": {"type": (int, float), "min": -1, "max": 500},
"x": {"type": (int, float)}, "y": {"type": (int, float)},
"tx": {"type": (int, float)}, "ty": {"type": (int, float)}, "talt": {"type": (int, float), "min": 0, "max": 500},
},
}
def check(name, value, rule):
if not isinstance(value, rule["type"]) or isinstance(value, bool):
return [f"{name}: 자료형 오류({type(value).__name__})"]
errs = []
if "enum" in rule and value not in rule["enum"]:
errs.append(f"{name}: 허용되지 않은 값 '{value}'")
if "min" in rule and value < rule["min"]:
errs.append(f"{name}: 최소값 {rule['min']} 미만")
if "max" in rule and value > rule["max"]:
errs.append(f"{name}: 최대값 {rule['max']} 초과")
if "pattern" in rule and not re.match(rule["pattern"], value):
errs.append(f"{name}: 형식 불일치 '{value}'")
return errs
def validate(sess):
errs = []
for g in ("session", "tactical"):
for k, rule in SCHEMA[g].items():
errs += [f"{g}.{k}: 필수 필드 누락"] if k not in sess.get(g, {}) else check(f"{g}.{k}", sess[g][k], rule)
prev = -1.0
for i, rec in enumerate(sess.get("records", [])):
for k, rule in SCHEMA["record"].items():
errs += [f"records[{i}].{k}: 필수 필드 누락"] if k not in rec else check(f"records[{i}].{k}", rec[k], rule)
if isinstance(rec.get("t"), (int, float)) and rec["t"] <= prev:
errs.append(f"records[{i}].t: 시각 역행/중복")
prev = rec.get("t", prev)
return errs
# [교육용가상자료] 조종자 20명 x 비행 1회 -> 세션 20건(각 세션에서 레코드 30개만 표본 검증)
pilots = make_pilots(20, seed=7)
log, truth = fly(pilots, "day", seed=1)
rng = np.random.default_rng(2)
sessions = []
for j in range(20):
d = one(log, j).iloc[::40][list(SCHEMA["record"])].round(3)
sessions.append({"session": {"session_id": f"S{j + 1:04d}", "pilot": pilots["pilot"][j], "sim_course": "RECT40x30_HOVER_OBS_LAND",
"rate_hz": 10, "duration_s": 120.0},
"tactical": {"condition": "day", "wind_mps": float(round(rng.uniform(0, 3), 1)), "gps_error_m": 0.25,
"video_delay_ms": int(rng.integers(50, 250))},
"records": [{k: float(v) for k, v in r.items()} for r in d.to_dict("records")]})
ok = sum(1 for s in sessions if not validate(s))
print(f"정상 세션 {len(sessions)}건(레코드 {sum(len(s['records']) for s in sessions)}개) 중 검증 통과 {ok}건, 오류 {len(sessions) - ok}건")
bad = json.loads(json.dumps(sessions[0]))
bad["session"]["pilot"] = "KIM-01" # 실명형 ID 입력
bad["tactical"]["condition"] = "storm"
del bad["tactical"]["video_delay_ms"]
bad["records"][3]["stick_roll"] = 1.8
bad["records"][5]["t"] = bad["records"][4]["t"]
print("[오류 주입 결과]")
for e in validate(bad):
print(" -", e)
rows = []
for g in SCHEMA:
for k, r in SCHEMA[g].items():
t = r["type"].__name__ if isinstance(r["type"], type) else "number"
rows.append((g, k, t, str({x: y for x, y in r.items() if x != "type"})[:45]))
print("\n[데이터 명세 요약]")
print(pd.DataFrame(rows, columns=["그룹", "필드", "자료형", "규칙"]).to_string(index=False))
print("\n비행 단계: 경로 0~60초 / 호버링 60~85초 / 장애물 회피 85초 / 비상 착륙 지시 100초, 10Hz 기록")
print("개인정보: 조종자는 가명 ID(P001~)로만 저장하고, 실명·군번·소속은 별도 대장에서 교관만 관리한다.")
In [ ]:
"""시뮬레이터 원시 비행로그의 시각 중복·역행 정리, 스틱 PWM 값(1000~2000us) 정규화, GPS 튐(이상치) 제거(Hampel 필터), 센서 단절 구간 보간을 수행하는 전처리 모듈을 구현하고, 비행 조건(주간/야간/돌풍)별 임계값을 달리 적용해 처리 속도와 위치 오차 개선을 확인한다."""
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
# ---------------------------------------------------------------------------
# [교육용가상자료] 드론 조종 시뮬레이터 비행로그 생성기 (실제 훈련생·장비 정보 없음, 조종자는 가명 P001~)
# 비행 1회 = 120초(10Hz): 경로 비행(0~60초) -> 호버링(60~85초) -> 돌발 장애물 회피(85초) -> 비상 착륙(100초 지시)
# 조종자마다 숨은 숙련도(skill)와 조작 습관(급조작 jerky, 과보정 overcorr)이 있고, 교관 점수는 이를 반영해 만든다.
# ---------------------------------------------------------------------------
DT, STEPS = 0.1, 1200
T = np.arange(STEPS) * DT
COND = {"day": dict(wind=0.4, gust=0.0, gps=0.25, delay=0.0), "night": dict(wind=0.4, gust=0.0, gps=0.5, delay=0.15),
"gust": dict(wind=1.2, gust=2.5, gps=0.25, delay=0.0)}
OBST_T, LAND_T = 85.0, 100.0
def make_pilots(n=40, seed=7):
rng = np.random.default_rng(seed)
skill = rng.beta(2.2, 2.0, n)
return pd.DataFrame({"pilot": [f"P{i + 1:03d}" for i in range(n)], "skill": skill,
"jerky": rng.random(n) < 0.15 + 0.35 * (1 - skill), "overcorr": rng.random(n) < 0.15 + 0.35 * (1 - skill)})
def target_path(t):
"""사각형 경로(40x30m, 고도 20m)를 3m/s로 비행 -> 60초 지점에서 호버링 -> 착륙 지시 후 고도 0."""
# [TODO : 학생 실습 2] 시계열 비행 궤적 데이터 정제 및 이상 조작 노이즈 필터링 전처리 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def fly(pilots, cond="day", seed=0, delay_extra=0.0, obstacle_side=None):
"""여러 조종자의 비행 1회를 동시에 모의한다(벡터화). 반환: 로그 배열 dict, 비행별 정답 정보 DataFrame."""
# [TODO : 학생 실습 2] 시계열 비행 궤적 데이터 정제 및 이상 조작 노이즈 필터링 전처리 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def one(log, j):
"""여러 조종자 로그에서 j번째 조종자 1명의 로그(DataFrame)를 꺼낸다."""
return pd.DataFrame({k: (v if v.ndim == 1 else v[:, j]) for k, v in log.items()})
# ---------------------------------------------------------------------------
HAMPEL_K = {"day": 3.0, "night": 3.5, "gust": 4.5} # 조건별 이상치 판정 임계값(돌풍은 실제 큰 움직임이 많아 느슨하게)
def make_raw(log, j, cond, rng):
"""[교육용가상자료] 원시 로그: PWM 스틱값, GPS 튐 1%, 센서 단절(0.5~2초) 2회, 시각 중복 2%."""
d = one(log, j)
raw = d[["t", "x", "y", "alt", "roll_deg", "pitch_deg", "true_x", "true_y", "true_alt"]].copy()
for s in ("stick_roll", "stick_pitch", "throttle"):
raw[s + "_pwm"] = (1500 + 500 * d[s] + rng.normal(0, 4, len(d))).round().astype(int)
k = rng.random(len(raw)) < 0.01
for c in ("x", "y"):
raw.loc[k, c] += rng.choice([-1, 1], k.sum()) * rng.uniform(6, 15, k.sum())
for _ in range(2):
a = int(rng.integers(50, len(raw) - 30)); raw.loc[a:a + int(rng.integers(5, 20)), ["x", "y", "alt"]] = np.nan
dup = raw.sample(frac=0.02, random_state=int(rng.integers(1e6)))
raw = pd.concat([raw, dup]).sample(frac=1.0, random_state=1).reset_index(drop=True) # 순서 섞임 + 중복
raw["t"] = raw["t"] + rng.normal(0, 0.003, len(raw)) # 시각 흔들림
return raw
def pwm_to_stick(p, dead=8):
v = (np.asarray(p, float) - 1500) / 500
v[np.abs(np.asarray(p) - 1500) < dead] = 0.0 # 중립 부근 데드밴드
return np.clip(v, -1, 1)
def hampel(x, win=7, k=3.0):
s = pd.Series(x)
med = s.rolling(win, center=True, min_periods=1).median()
mad = (s - med).abs().rolling(win, center=True, min_periods=1).median() * 1.4826
out = (s - med).abs() > k * (mad + 0.3) # 0.3m: GPS 잡음 수준 이하 흔들림은 이상치로 보지 않음
return np.where(out, np.nan, s), int(out.sum())
def preprocess(raw, cond):
d = raw.copy()
d["k"] = (d["t"] / DT).round().astype(int) # 10Hz 격자 번호로 시각 정렬
d = d.sort_values("k").drop_duplicates("k").set_index("k")
d = d.reindex(np.arange(0, d.index.max() + 1))
d["t"] = np.round(d.index * DT, 1)
n_missing = int(d[["x", "y", "alt"]].isna().any(axis=1).sum())
n_out = 0
for c in ("x", "y", "alt"):
d[c], n = hampel(d[c].values, k=HAMPEL_K[cond]); n_out += n
d[["x", "y", "alt"]] = d[["x", "y", "alt"]].interpolate(limit_direction="both")
for s in ("stick_roll", "stick_pitch", "throttle"):
d[s] = pwm_to_stick(d[s + "_pwm"].interpolate(limit_direction="both").round())
return d.reset_index(drop=True), n_missing, n_out
pilots = make_pilots(30, seed=7)
rng = np.random.default_rng(3)
rows, n_rec, t_total = [], 0, 0.0
for cond in COND:
log, _ = fly(pilots, cond, seed=1)
for j in range(len(pilots)):
raw = make_raw(log, j, cond, rng)
t0 = time.perf_counter()
clean, miss, out = preprocess(raw, cond)
t_total += time.perf_counter() - t0; n_rec += len(raw)
e0 = np.sqrt((raw["x"] - raw["true_x"]) ** 2 + (raw["y"] - raw["true_y"]) ** 2).mean()
e1 = np.sqrt((clean["x"] - clean["true_x"]) ** 2 + (clean["y"] - clean["true_y"]) ** 2).mean()
rows.append({"조건": cond, "결측행": miss, "이상치": out, "남은결측": int(clean[["x", "y", "alt"]].isna().sum().sum()),
"위치오차 전(m)": e0, "위치오차 후(m)": e1})
df = pd.DataFrame(rows)
print(f"원시 레코드 {n_rec:,}개(로그 {len(df)}개) 처리, {n_rec / t_total:,.0f} 레코드/초 (목표 1,000건/초 이상), 처리 중 오류 0건")
print(f"결측 보정율 {1 - df['남은결측'].sum() / max(df['결측행'].sum(), 1):.0%} (결측행 {df['결측행'].sum()}개 모두 보간)")
print("\n[조건별 전처리 효과(로그 평균)]")
print(df.groupby("조건")[["결측행", "이상치", "위치오차 전(m)", "위치오차 후(m)"]].mean().round(2).to_string())
print("조건별 Hampel 임계값:", HAMPEL_K, "-> 돌풍 조건은 실제 급기동을 이상치로 지우지 않도록 느슨하게 둔다")
# 단위 테스트 5종
raw = make_raw(fly(pilots.iloc[:1], "day", seed=9)[0], 0, "day", np.random.default_rng(0))
clean, _, _ = preprocess(raw, "day")
spike = np.r_[np.zeros(10), 50.0, np.zeros(10)]
tests = {
"PWM 1000/1500/2000 -> -1/0/1": np.allclose(pwm_to_stick([1000, 1500, 2000]), [-1, 0, 1]),
"데드밴드(1504 -> 0)": pwm_to_stick([1504])[0] == 0.0,
"단일 튐 값 제거": np.isnan(hampel(spike)[0][10]),
"시각 중복 제거·단조 증가": clean["t"].is_monotonic_increasing and clean["t"].duplicated().sum() == 0,
"결측 없음 + 10Hz 1,200행": clean[["x", "y", "alt"]].isna().sum().sum() == 0 and len(clean) == STEPS,
}
for k, v in tests.items():
print(f"[{'PASS' if v else 'FAIL'}] {k}")
print(f"단위 테스트 {sum(tests.values())}/{len(tests)} 통과")
seg = slice(400, 700)
r2 = raw.sort_values("t").iloc[seg]
plt.figure(figsize=(9, 3.5))
plt.plot(r2["t"], r2["y"], ".", ms=3, c="gray", label="raw (spikes, gaps)")
plt.plot(clean["t"].iloc[seg], clean["y"].iloc[seg], c="tab:blue", label="preprocessed")
plt.plot(clean["t"].iloc[seg], clean["true_y"].iloc[seg], "k:", label="true")
plt.xlabel("time (s)"); plt.ylabel("y (m)"); plt.legend(); plt.title("Flight log preprocessing (synthetic)")
plt.tight_layout(); plt.show()
In [ ]:
"""비행 경로 이탈도(경로 RMSE)와 호버링 안정성(평균 이탈·흔들림), 고도 유지 오차로 1차 숙련도 점수(0~100)와 등급(A/B/C)을 계산하는 Baseline 채점 알고리즘을 구현하고, 교관 평가 등급과의 일치율(처리 정확도)과 판단 지연을 측정한다."""
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.linear_model import LinearRegression
# ---------------------------------------------------------------------------
# [교육용가상자료] 드론 조종 시뮬레이터 비행로그 생성기 (실제 훈련생·장비 정보 없음, 조종자는 가명 P001~)
# 비행 1회 = 120초(10Hz): 경로 비행(0~60초) -> 호버링(60~85초) -> 돌발 장애물 회피(85초) -> 비상 착륙(100초 지시)
# 조종자마다 숨은 숙련도(skill)와 조작 습관(급조작 jerky, 과보정 overcorr)이 있고, 교관 점수는 이를 반영해 만든다.
# ---------------------------------------------------------------------------
DT, STEPS = 0.1, 1200
T = np.arange(STEPS) * DT
COND = {"day": dict(wind=0.4, gust=0.0, gps=0.25, delay=0.0), "night": dict(wind=0.4, gust=0.0, gps=0.5, delay=0.15),
"gust": dict(wind=1.2, gust=2.5, gps=0.25, delay=0.0)}
OBST_T, LAND_T = 85.0, 100.0
def make_pilots(n=40, seed=7):
rng = np.random.default_rng(seed)
skill = rng.beta(2.2, 2.0, n)
return pd.DataFrame({"pilot": [f"P{i + 1:03d}" for i in range(n)], "skill": skill,
"jerky": rng.random(n) < 0.15 + 0.35 * (1 - skill), "overcorr": rng.random(n) < 0.15 + 0.35 * (1 - skill)})
def target_path(t):
"""사각형 경로(40x30m, 고도 20m)를 3m/s로 비행 -> 60초 지점에서 호버링 -> 착륙 지시 후 고도 0."""
# [TODO : 학생 실습 3] 비행 경로 이탈도 및 호버링 안정성 계산 1차 숙련도 채점 Baseline 알고리즘 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def fly(pilots, cond="day", seed=0, delay_extra=0.0, obstacle_side=None):
"""여러 조종자의 비행 1회를 동시에 모의한다(벡터화). 반환: 로그 배열 dict, 비행별 정답 정보 DataFrame."""
# [TODO : 학생 실습 3] 비행 경로 이탈도 및 호버링 안정성 계산 1차 숙련도 채점 Baseline 알고리즘 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def one(log, j):
"""여러 조종자 로그에서 j번째 조종자 1명의 로그(DataFrame)를 꺼낸다."""
return pd.DataFrame({k: (v if v.ndim == 1 else v[:, j]) for k, v in log.items()})
# ---------------------------------------------------------------------------
# ---- 비행로그 특징 추출 공통 함수(여러 조종자 로그를 한 번에 계산) ----
from scipy.ndimage import uniform_filter1d
BASE_F = ["path_rmse", "hover_drift", "hover_std", "alt_err"]
ADV_F = BASE_F + ["obs_rt", "obs_lateral", "touch_vz", "att_exceed", "jerk", "overshoot", "n_risk"]
RULES = {"near_collision": ("obs_lateral", "<", 2.5), "hard_landing": ("touch_vz", ">", 2.0), "attitude_exceed": ("att_exceed", ">=", 10)}
def detect(df):
"""위험 조작 사건 탐지 규칙: 장애물 2.5초 후 회피 거리 2.5m 미만, 접지 속도 2.0m/s 초과, 자세 35도 초과 1초 이상."""
out = {}
for ev, (f, op, th) in RULES.items():
out[ev] = df[f] < th if op == "<" else (df[f] > th if op == ">" else df[f] >= th)
return pd.DataFrame(out)
def features(log):
"""경로 이탈도·호버링 안정성(기본) + 장애물 반응·착륙 속도·자세 한계·급조작·과보정(고도화) 특징."""
X, Y, A = (uniform_filter1d(log[k], 9, axis=0) for k in ("x", "y", "alt")) # GPS 잡음 완화(이동평균 0.9초)
h = np.hypot(X - log["tx"], Y - log["ty"])
cr, hv = slice(0, 600), slice(620, 850)
k0, k1 = int(OBST_T / DT), int((OBST_T + 2.5) / DT)
side = np.sign(log["ty"][k0 + 1] - log["ty"][k0 - 1])
moved = (log["stick_roll"][k0:k0 + 40] * side) > 0.3
rt = np.where(moved.any(0), moved.argmax(0) * DT, 4.0)
vz = -np.gradient(A, DT, axis=0)
land = np.argmax((A < 0.4) & (T[:, None] >= LAND_T), axis=0)
touch = np.array([vz[max(i - 6, 0):i - 2, j].mean() if i > 6 else 3.0 for j, i in enumerate(land)])
ds = np.abs(np.diff(np.stack([log["stick_roll"], log["stick_pitch"]]), axis=1)) / DT
return pd.DataFrame({
"path_rmse": np.sqrt((h[cr] ** 2).mean(0)), "hover_drift": h[hv].mean(0), "hover_std": np.hypot(X[hv].std(0), Y[hv].std(0)),
"alt_err": np.abs(A[:850] - log["talt"][:850]).mean(0),
"obs_rt": rt, "obs_lateral": np.abs(Y[k1] - Y[k0]), "touch_vz": touch,
"att_exceed": (np.maximum(np.abs(log["roll_deg"]), np.abs(log["pitch_deg"])) > 35).sum(0),
"jerk": np.percentile(ds, 95, axis=(0, 1)),
"overshoot": np.clip(np.abs(Y[k0:k0 + 120] - Y[k0]).max(0) - 6.0, 0, None)}).pipe(
lambda d: d.assign(n_risk=detect(d).sum(axis=1))) # overshoot: 회피 후 목표 6m를 넘어선 거리(과보정)
def make_dataset(n_pilots=40, seed=7, conds=("day", "night", "gust"), flights=1):
"""조종자 n명 x 조건 x 반복 비행 -> 비행별 특징 + 정답(교관 점수·등급·위험 사건) 표."""
pilots = make_pilots(n_pilots, seed)
out = []
for c in conds:
for f in range(flights):
log, tr = fly(pilots, c, seed=seed * 100 + list(COND).index(c) * 10 + f)
out.append(pd.concat([tr.reset_index(drop=True), features(log)], axis=1).assign(flight=f))
return pd.concat(out, ignore_index=True)
def cond_z(df, ref, cols):
"""조건별 기준 집단(ref)의 평균·표준편차로 표준화: 돌풍처럼 어려운 조건에서 받는 불이익을 줄인다."""
z = df.copy()
z[cols] = z[cols].astype(float)
for c in df["cond"].unique():
r = ref[ref["cond"] == c][cols]
z.loc[df["cond"] == c, cols] = (df.loc[df["cond"] == c, cols] - r.mean()) / (r.std() + 1e-9)
return z
def to_grade(score):
return np.where(score >= 75, "A", np.where(score >= 55, "B", "C"))
# [교육용가상자료] 기준(튜닝)용 조종자 40명, 평가용 조종자 40명 x 조건 3종(주간·야간·돌풍)
train = make_dataset(40, seed=7)
test = make_dataset(40, seed=8)
# 1) 휴리스틱 가중치 1차 튜닝: 조건별 표준화 특징 -> 교관 점수 선형 근사
ztr, zte = cond_z(train, train, BASE_F), cond_z(test, train, BASE_F)
lr = LinearRegression().fit(ztr[BASE_F], train["instructor"])
print("채점식: 점수 = {:.1f}".format(lr.intercept_) + "".join(f" {w:+.1f}x{f}" for f, w in zip(BASE_F, lr.coef_)) + " (특징은 조건별 표준화 값)")
t0 = time.perf_counter()
test["base_score"] = np.clip(lr.predict(zte[BASE_F]), 0, 100)
test["base_grade"] = to_grade(test["base_score"])
ms = (time.perf_counter() - t0) * 1000 / len(test)
acc = (test["base_grade"] == test["grade"]).mean()
print(f"평가 비행 {len(test)}회: 등급 일치율(처리 정확도) {acc:.1%} (목표 75% 이상: {'달성' if acc >= 0.75 else '미달'}), "
f"점수 평균 오차 {np.abs(test['base_score'] - test['instructor']).mean():.1f}점, 채점 지연 {ms:.3f} ms/비행")
# 2) 조건별 정확도 + 조건 표준화를 하지 않았을 때 비교
raw = LinearRegression().fit(train[BASE_F], train["instructor"])
test["raw_grade"] = to_grade(raw.predict(test[BASE_F]))
print("\n조건 표준화 O 표준화 X")
for c in COND:
s = test[test["cond"] == c]
print(f"{c:6s} {(s['base_grade'] == s['grade']).mean():8.1%} {(s['raw_grade'] == s['grade']).mean():8.1%}")
print(f"전체 {acc:8.1%} {(test['raw_grade'] == test['grade']).mean():8.1%}")
gust_raw = raw.predict(test[test["cond"] == "gust"][BASE_F]).mean() - raw.predict(test[test["cond"] == "day"][BASE_F]).mean()
print(f"-> 표준화하지 않으면 돌풍 비행 점수가 주간보다 평균 {gust_raw:+.1f}점 달라진다(조건 효과가 점수에 섞임). 표준화는 같은 조건 동료 대비 비교다.")
# 3) 혼동표와 한계 분석
print("\n[교관 등급 x Baseline 등급]")
print(pd.crosstab(test["grade"].rename("교관"), test["base_grade"].rename("Baseline")).to_string())
over = test[(test["grade"] == "C") & (test["base_grade"] != "C")]
ev = ["near_collision", "hard_landing"]
print(f"\n가장 많은 오류: 교관 C -> Baseline B {len(over)}건. 교관 C 비행의 {test[test['grade'] == 'C'][ev].any(axis=1).mean():.0%}가"
f" 장애물 근접·경착륙을 포함하지만 Baseline 특징(경로·호버링·고도)에는 이 정보가 없다.")
print("-> 경로·호버링만 보는 Baseline은 돌발 상황 대응(장애물 회피, 비상 착륙)을 반영하지 못한다. 미션 4에서 반응 특징을 추가한다.")
plt.figure(figsize=(6, 5))
for c, col in zip(COND, ["tab:blue", "tab:purple", "tab:orange"]):
s = test[test["cond"] == c]
plt.scatter(s["instructor"], s["base_score"], s=15, c=col, label=c)
plt.plot([20, 100], [20, 100], "k--"); plt.axhline(75, c="gray", lw=0.5); plt.axhline(55, c="gray", lw=0.5)
plt.xlabel("instructor score"); plt.ylabel("baseline AI score"); plt.legend(); plt.title("Baseline skill score vs instructor (synthetic)")
plt.tight_layout(); plt.show()
In [ ]:
"""돌발 장애물 회피 반응시간·회피 거리, 비상 착륙 접지 속도, 자세 한계 초과, 급조작·과보정 특징을 추가한 정밀 숙련도 평가 엔진으로 고도화하고, 복합 시나리오 10종에서 위험 조작 사건(장애물 근접·경착륙·자세 한계) 탐지 재현율, 등급 일치율, 로그당 처리 지연을 검증한다."""
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.linear_model import LinearRegression, Ridge
# ---------------------------------------------------------------------------
# [교육용가상자료] 드론 조종 시뮬레이터 비행로그 생성기 (실제 훈련생·장비 정보 없음, 조종자는 가명 P001~)
# 비행 1회 = 120초(10Hz): 경로 비행(0~60초) -> 호버링(60~85초) -> 돌발 장애물 회피(85초) -> 비상 착륙(100초 지시)
# 조종자마다 숨은 숙련도(skill)와 조작 습관(급조작 jerky, 과보정 overcorr)이 있고, 교관 점수는 이를 반영해 만든다.
# ---------------------------------------------------------------------------
DT, STEPS = 0.1, 1200
T = np.arange(STEPS) * DT
COND = {"day": dict(wind=0.4, gust=0.0, gps=0.25, delay=0.0), "night": dict(wind=0.4, gust=0.0, gps=0.5, delay=0.15),
"gust": dict(wind=1.2, gust=2.5, gps=0.25, delay=0.0)}
OBST_T, LAND_T = 85.0, 100.0
def make_pilots(n=40, seed=7):
rng = np.random.default_rng(seed)
skill = rng.beta(2.2, 2.0, n)
return pd.DataFrame({"pilot": [f"P{i + 1:03d}" for i in range(n)], "skill": skill,
"jerky": rng.random(n) < 0.15 + 0.35 * (1 - skill), "overcorr": rng.random(n) < 0.15 + 0.35 * (1 - skill)})
def target_path(t):
"""사각형 경로(40x30m, 고도 20m)를 3m/s로 비행 -> 60초 지점에서 호버링 -> 착륙 지시 후 고도 0."""
# [TODO : 학생 실습 4] 돌발 장애물 회피 및 비상 착륙 반응 속도 기반 정밀 숙련도 평가 엔진 고도화 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def fly(pilots, cond="day", seed=0, delay_extra=0.0, obstacle_side=None):
"""여러 조종자의 비행 1회를 동시에 모의한다(벡터화). 반환: 로그 배열 dict, 비행별 정답 정보 DataFrame."""
# [TODO : 학생 실습 4] 돌발 장애물 회피 및 비상 착륙 반응 속도 기반 정밀 숙련도 평가 엔진 고도화 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def one(log, j):
"""여러 조종자 로그에서 j번째 조종자 1명의 로그(DataFrame)를 꺼낸다."""
return pd.DataFrame({k: (v if v.ndim == 1 else v[:, j]) for k, v in log.items()})
# ---------------------------------------------------------------------------
# ---- 비행로그 특징 추출 공통 함수(여러 조종자 로그를 한 번에 계산) ----
from scipy.ndimage import uniform_filter1d
BASE_F = ["path_rmse", "hover_drift", "hover_std", "alt_err"]
ADV_F = BASE_F + ["obs_rt", "obs_lateral", "touch_vz", "att_exceed", "jerk", "overshoot", "n_risk"]
RULES = {"near_collision": ("obs_lateral", "<", 2.5), "hard_landing": ("touch_vz", ">", 2.0), "attitude_exceed": ("att_exceed", ">=", 10)}
def detect(df):
"""위험 조작 사건 탐지 규칙: 장애물 2.5초 후 회피 거리 2.5m 미만, 접지 속도 2.0m/s 초과, 자세 35도 초과 1초 이상."""
out = {}
for ev, (f, op, th) in RULES.items():
out[ev] = df[f] < th if op == "<" else (df[f] > th if op == ">" else df[f] >= th)
return pd.DataFrame(out)
def features(log):
"""경로 이탈도·호버링 안정성(기본) + 장애물 반응·착륙 속도·자세 한계·급조작·과보정(고도화) 특징."""
X, Y, A = (uniform_filter1d(log[k], 9, axis=0) for k in ("x", "y", "alt")) # GPS 잡음 완화(이동평균 0.9초)
h = np.hypot(X - log["tx"], Y - log["ty"])
cr, hv = slice(0, 600), slice(620, 850)
k0, k1 = int(OBST_T / DT), int((OBST_T + 2.5) / DT)
side = np.sign(log["ty"][k0 + 1] - log["ty"][k0 - 1])
moved = (log["stick_roll"][k0:k0 + 40] * side) > 0.3
rt = np.where(moved.any(0), moved.argmax(0) * DT, 4.0)
vz = -np.gradient(A, DT, axis=0)
land = np.argmax((A < 0.4) & (T[:, None] >= LAND_T), axis=0)
touch = np.array([vz[max(i - 6, 0):i - 2, j].mean() if i > 6 else 3.0 for j, i in enumerate(land)])
ds = np.abs(np.diff(np.stack([log["stick_roll"], log["stick_pitch"]]), axis=1)) / DT
return pd.DataFrame({
"path_rmse": np.sqrt((h[cr] ** 2).mean(0)), "hover_drift": h[hv].mean(0), "hover_std": np.hypot(X[hv].std(0), Y[hv].std(0)),
"alt_err": np.abs(A[:850] - log["talt"][:850]).mean(0),
"obs_rt": rt, "obs_lateral": np.abs(Y[k1] - Y[k0]), "touch_vz": touch,
"att_exceed": (np.maximum(np.abs(log["roll_deg"]), np.abs(log["pitch_deg"])) > 35).sum(0),
"jerk": np.percentile(ds, 95, axis=(0, 1)),
"overshoot": np.clip(np.abs(Y[k0:k0 + 120] - Y[k0]).max(0) - 6.0, 0, None)}).pipe(
lambda d: d.assign(n_risk=detect(d).sum(axis=1))) # overshoot: 회피 후 목표 6m를 넘어선 거리(과보정)
def make_dataset(n_pilots=40, seed=7, conds=("day", "night", "gust"), flights=1):
"""조종자 n명 x 조건 x 반복 비행 -> 비행별 특징 + 정답(교관 점수·등급·위험 사건) 표."""
pilots = make_pilots(n_pilots, seed)
out = []
for c in conds:
for f in range(flights):
log, tr = fly(pilots, c, seed=seed * 100 + list(COND).index(c) * 10 + f)
out.append(pd.concat([tr.reset_index(drop=True), features(log)], axis=1).assign(flight=f))
return pd.concat(out, ignore_index=True)
def cond_z(df, ref, cols):
"""조건별 기준 집단(ref)의 평균·표준편차로 표준화: 돌풍처럼 어려운 조건에서 받는 불이익을 줄인다."""
z = df.copy()
z[cols] = z[cols].astype(float)
for c in df["cond"].unique():
r = ref[ref["cond"] == c][cols]
z.loc[df["cond"] == c, cols] = (df.loc[df["cond"] == c, cols] - r.mean()) / (r.std() + 1e-9)
return z
def to_grade(score):
return np.where(score >= 75, "A", np.where(score >= 55, "B", "C"))
train = make_dataset(40, seed=7)
zt = cond_z(train, train, ADV_F)
base = LinearRegression().fit(cond_z(train, train, BASE_F)[BASE_F], train["instructor"])
eng = Ridge(alpha=1.0).fit(zt[ADV_F], train["instructor"])
# [교육용가상자료] 복합 시나리오 10종: (조건, 영상 지연 추가 초, 장애물 방향)
SCEN = [("1 주간 기본", "day", 0.0, None), ("2 야간", "night", 0.0, None), ("3 돌풍", "gust", 0.0, None), ("4 장애물 좌측", "day", 0.0, -1),
("5 장애물 우측", "day", 0.0, 1), ("6 영상 지연 0.3초", "day", 0.3, None), ("7 돌풍+장애물 좌측", "gust", 0.0, -1),
("8 야간+영상 지연", "night", 0.3, None), ("9 돌풍+영상 지연", "gust", 0.3, None), ("10 야간+장애물 우측", "night", 0.0, 1)]
pilots = make_pilots(40, seed=8)
rows, all_det, ms = [], [], []
for i, (name, cond, dly, side) in enumerate(SCEN):
log, tr = fly(pilots, cond, seed=500 + i, delay_extra=dly, obstacle_side=side)
t0 = time.perf_counter()
f = features(log)
df = pd.concat([tr.reset_index(drop=True), f], axis=1)
df["score"] = np.clip(eng.predict(cond_z(df, train, ADV_F)[ADV_F]), 0, 100)
det = detect(df)
ms.append((time.perf_counter() - t0) * 1000 / len(df))
df["base"] = np.clip(base.predict(cond_z(df, train, BASE_F)[BASE_F]), 0, 100)
truth = df[list(RULES)].astype(bool)
tp, pos = int((det & truth).values.sum()), int(truth.values.sum())
all_det.append((det, truth))
rows.append({"시나리오": name, "위험사건": pos, "탐지": tp, "재현율": round(tp / max(pos, 1), 3), "오탐": int((det & ~truth).values.sum()),
"Baseline 일치": round((to_grade(df["base"]) == df["grade"]).mean(), 3), "고도화 일치": round((to_grade(df["score"]) == df["grade"]).mean(), 3)})
R = pd.DataFrame(rows)
print(R.to_string(index=False))
rec = R["탐지"].sum() / R["위험사건"].sum()
print(f"\n위험 조작 사건 탐지 재현율 {rec:.1%} (목표 85% 이상: {'달성' if rec >= 0.85 else '미달'}), 로그당 처리 {np.mean(ms):.1f} ms (최대 {np.max(ms):.1f}, 목표 200 ms 이하)")
print(f"등급 일치율 Baseline {R['Baseline 일치'].mean():.1%} -> 고도화 엔진 {R['고도화 일치'].mean():.1%}, 10종 중 재현율 85% 이상 {(R['재현율'] >= 0.85).sum()}개")
# 사건별 재현율·오탐(오경보 유발 패턴)
det = pd.concat([d for d, _ in all_det]); tru = pd.concat([t for _, t in all_det])
print("\n사건 실제 재현율 오탐")
for ev in RULES:
print(f"{ev:15s} {tru[ev].sum():4d} {(det[ev] & tru[ev]).sum() / max(tru[ev].sum(), 1):.2f} {(det[ev] & ~tru[ev]).sum():4d}")
print("오탐 패턴: 접지 속도는 고도 로그를 미분해 추정하므로 고도 잡음이 큰 비행에서 과대 추정된다 -> 접지 직전 0.4초 평균 사용, 임계값 2.0m/s로 여유를 둠")
w = pd.Series(eng.coef_, index=ADV_F).sort_values()
print("\n엔진 가중치(조건별 표준화 특징 1단위당 점수):", ", ".join(f"{k} {v:+.1f}" for k, v in w.items()))
plt.figure(figsize=(8, 4))
plt.barh(w.index, w.values, color=["tab:red" if v < 0 else "tab:blue" for v in w.values])
plt.xlabel("score change per 1 SD"); plt.title("Advanced skill engine weights (synthetic)"); plt.tight_layout(); plt.show()
In [ ]:
"""조종자별 여러 비행의 조작 습관(급조작=스틱 변화율, 과보정=회피 후 초과 이동)을 분석하고, 비행 점수를 묶어 숙련도 등급과 신뢰도(부트스트랩 등급 확률)를 산출한다. 비행 1회로 '재교육 대상'을 지정하던 방식 대비 오지정률(FAR) 감축, ROC-AUC, 임계값 민감도, 등급 일관성을 검증한다."""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.linear_model import Ridge
from sklearn.metrics import roc_auc_score
# ---------------------------------------------------------------------------
# [교육용가상자료] 드론 조종 시뮬레이터 비행로그 생성기 (실제 훈련생·장비 정보 없음, 조종자는 가명 P001~)
# 비행 1회 = 120초(10Hz): 경로 비행(0~60초) -> 호버링(60~85초) -> 돌발 장애물 회피(85초) -> 비상 착륙(100초 지시)
# 조종자마다 숨은 숙련도(skill)와 조작 습관(급조작 jerky, 과보정 overcorr)이 있고, 교관 점수는 이를 반영해 만든다.
# ---------------------------------------------------------------------------
DT, STEPS = 0.1, 1200
T = np.arange(STEPS) * DT
COND = {"day": dict(wind=0.4, gust=0.0, gps=0.25, delay=0.0), "night": dict(wind=0.4, gust=0.0, gps=0.5, delay=0.15),
"gust": dict(wind=1.2, gust=2.5, gps=0.25, delay=0.0)}
OBST_T, LAND_T = 85.0, 100.0
def make_pilots(n=40, seed=7):
rng = np.random.default_rng(seed)
skill = rng.beta(2.2, 2.0, n)
return pd.DataFrame({"pilot": [f"P{i + 1:03d}" for i in range(n)], "skill": skill,
"jerky": rng.random(n) < 0.15 + 0.35 * (1 - skill), "overcorr": rng.random(n) < 0.15 + 0.35 * (1 - skill)})
def target_path(t):
"""사각형 경로(40x30m, 고도 20m)를 3m/s로 비행 -> 60초 지점에서 호버링 -> 착륙 지시 후 고도 0."""
# [TODO : 학생 실습 5] 조종자별 조작 습관(급조작, 과보정) 분석 및 신뢰도 등급 산출 모듈 개발 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def fly(pilots, cond="day", seed=0, delay_extra=0.0, obstacle_side=None):
"""여러 조종자의 비행 1회를 동시에 모의한다(벡터화). 반환: 로그 배열 dict, 비행별 정답 정보 DataFrame."""
# [TODO : 학생 실습 5] 조종자별 조작 습관(급조작, 과보정) 분석 및 신뢰도 등급 산출 모듈 개발 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def one(log, j):
"""여러 조종자 로그에서 j번째 조종자 1명의 로그(DataFrame)를 꺼낸다."""
return pd.DataFrame({k: (v if v.ndim == 1 else v[:, j]) for k, v in log.items()})
# ---------------------------------------------------------------------------
# ---- 비행로그 특징 추출 공통 함수(여러 조종자 로그를 한 번에 계산) ----
from scipy.ndimage import uniform_filter1d
BASE_F = ["path_rmse", "hover_drift", "hover_std", "alt_err"]
ADV_F = BASE_F + ["obs_rt", "obs_lateral", "touch_vz", "att_exceed", "jerk", "overshoot", "n_risk"]
RULES = {"near_collision": ("obs_lateral", "<", 2.5), "hard_landing": ("touch_vz", ">", 2.0), "attitude_exceed": ("att_exceed", ">=", 10)}
def detect(df):
"""위험 조작 사건 탐지 규칙: 장애물 2.5초 후 회피 거리 2.5m 미만, 접지 속도 2.0m/s 초과, 자세 35도 초과 1초 이상."""
out = {}
for ev, (f, op, th) in RULES.items():
out[ev] = df[f] < th if op == "<" else (df[f] > th if op == ">" else df[f] >= th)
return pd.DataFrame(out)
def features(log):
"""경로 이탈도·호버링 안정성(기본) + 장애물 반응·착륙 속도·자세 한계·급조작·과보정(고도화) 특징."""
X, Y, A = (uniform_filter1d(log[k], 9, axis=0) for k in ("x", "y", "alt")) # GPS 잡음 완화(이동평균 0.9초)
h = np.hypot(X - log["tx"], Y - log["ty"])
cr, hv = slice(0, 600), slice(620, 850)
k0, k1 = int(OBST_T / DT), int((OBST_T + 2.5) / DT)
side = np.sign(log["ty"][k0 + 1] - log["ty"][k0 - 1])
moved = (log["stick_roll"][k0:k0 + 40] * side) > 0.3
rt = np.where(moved.any(0), moved.argmax(0) * DT, 4.0)
vz = -np.gradient(A, DT, axis=0)
land = np.argmax((A < 0.4) & (T[:, None] >= LAND_T), axis=0)
touch = np.array([vz[max(i - 6, 0):i - 2, j].mean() if i > 6 else 3.0 for j, i in enumerate(land)])
ds = np.abs(np.diff(np.stack([log["stick_roll"], log["stick_pitch"]]), axis=1)) / DT
return pd.DataFrame({
"path_rmse": np.sqrt((h[cr] ** 2).mean(0)), "hover_drift": h[hv].mean(0), "hover_std": np.hypot(X[hv].std(0), Y[hv].std(0)),
"alt_err": np.abs(A[:850] - log["talt"][:850]).mean(0),
"obs_rt": rt, "obs_lateral": np.abs(Y[k1] - Y[k0]), "touch_vz": touch,
"att_exceed": (np.maximum(np.abs(log["roll_deg"]), np.abs(log["pitch_deg"])) > 35).sum(0),
"jerk": np.percentile(ds, 95, axis=(0, 1)),
"overshoot": np.clip(np.abs(Y[k0:k0 + 120] - Y[k0]).max(0) - 6.0, 0, None)}).pipe(
lambda d: d.assign(n_risk=detect(d).sum(axis=1))) # overshoot: 회피 후 목표 6m를 넘어선 거리(과보정)
def make_dataset(n_pilots=40, seed=7, conds=("day", "night", "gust"), flights=1):
"""조종자 n명 x 조건 x 반복 비행 -> 비행별 특징 + 정답(교관 점수·등급·위험 사건) 표."""
pilots = make_pilots(n_pilots, seed)
out = []
for c in conds:
for f in range(flights):
log, tr = fly(pilots, c, seed=seed * 100 + list(COND).index(c) * 10 + f)
out.append(pd.concat([tr.reset_index(drop=True), features(log)], axis=1).assign(flight=f))
return pd.concat(out, ignore_index=True)
def cond_z(df, ref, cols):
"""조건별 기준 집단(ref)의 평균·표준편차로 표준화: 돌풍처럼 어려운 조건에서 받는 불이익을 줄인다."""
z = df.copy()
z[cols] = z[cols].astype(float)
for c in df["cond"].unique():
r = ref[ref["cond"] == c][cols]
z.loc[df["cond"] == c, cols] = (df.loc[df["cond"] == c, cols] - r.mean()) / (r.std() + 1e-9)
return z
def to_grade(score):
return np.where(score >= 75, "A", np.where(score >= 55, "B", "C"))
train = make_dataset(40, seed=7)
eng = Ridge(alpha=1.0).fit(cond_z(train, train, ADV_F)[ADV_F], train["instructor"])
base = Ridge(alpha=1.0).fit(cond_z(train, train, BASE_F)[BASE_F], train["instructor"]) # 미션 3 Baseline 채점식
zt = cond_z(train, train, ADV_F) # 습관 판정 기준: 같은 조건에서 습관 없는 조종자의 상위 10%
jerk_th = zt.loc[~train["jerky"], "jerk"].quantile(0.9)
over_th = zt.loc[~train["overcorr"], "overshoot"].quantile(0.9)
# [교육용가상자료] 평가 조종자 80명 x 비행 6회(주간·야간·돌풍 각 2회)
test = make_dataset(80, seed=9, flights=2)
test["score"] = np.clip(eng.predict(cond_z(test, train, ADV_F)[ADV_F]), 0, 100)
ze = cond_z(test, train, ADV_F)
test["jerk_z"], test["over_z"] = ze["jerk"], ze["overshoot"]
test["base"] = np.clip(base.predict(cond_z(test, train, BASE_F)[BASE_F]), 0, 100)
rng = np.random.default_rng(0)
rows = []
for pid, g in test.groupby("pilot"):
boot = [to_grade(rng.choice(g["score"].values, len(g)).mean()).item() for _ in range(200)]
grade = to_grade(g["score"].mean()).item()
conf = boot.count(grade) / len(boot)
rows.append({"pilot": pid, "점수": g["score"].mean(), "점수SD": g["score"].std(), "등급": grade, "신뢰도": conf,
"신뢰등급": "High" if conf >= 0.9 else ("Medium" if conf >= 0.7 else "Low"),
"급조작": (g["jerk_z"] > jerk_th).mean() >= 0.5, "과보정": (g["over_z"] > over_th).mean() >= 0.5,
"실제급조작": g["jerky"].iloc[0], "실제과보정": g["overcorr"].iloc[0],
"교관등급": to_grade(g["instructor"].mean()).item(), "단일비행점수": g["base"].iloc[int(rng.integers(len(g)))]})
P = pd.DataFrame(rows)
print(f"습관 판정 정확도: 급조작 {(P['급조작'] == P['실제급조작']).mean():.0%}, 과보정 {(P['과보정'] == P['실제과보정']).mean():.0%} (조종자 {len(P)}명)")
# 재교육 대상(C) 지정: 기존(무작위 비행 1회의 Baseline 점수 55 미만) vs 엔진(6회 평균 C + 신뢰도 Medium 이상, Low는 교관 확인으로 보냄)
need = P["교관등급"] == "C"
old = P["단일비행점수"] < 55
new = (P["등급"] == "C") & (P["신뢰등급"] != "Low")
review = (P["등급"] == "C") & (P["신뢰등급"] == "Low")
def far_rec(flag):
return (flag & ~need).sum() / max(flag.sum(), 1), (flag & need).sum() / max(need.sum(), 1), int(flag.sum())
f0, r0, n0 = far_rec(old)
f1, r1, n1 = far_rec(new)
print(pd.DataFrame([{"방식": "기존(비행 1회 Baseline<55)", "지정": n0, "오지정률": round(f0, 3), "재현율": round(r0, 3)},
{"방식": "엔진(6회 평균 C, 신뢰 Medium 이상)", "지정": n1, "오지정률": round(f1, 3), "재현율": round(r1, 3)}]).to_string(index=False))
cut = 1 - f1 / max(f0, 1e-9)
print(f"오지정률(FAR) 감축 {cut:.1%} (목표 40% 이상: {'달성' if cut >= 0.4 else '미달'}), 신뢰도 Low로 교관 확인에 넘긴 조종자 {int(review.sum())}명")
print(f"ROC-AUC(재교육 필요 판별, 평균 점수 기준) {roc_auc_score(need, -P['점수']):.3f}")
print("\n기준점수 지정 오지정률 재현율")
for th in [50, 55, 60, 65]:
f, r, n = far_rec(P["점수"] < th)
print(f" {th} {n:3d} {f:.3f} {r:.3f}")
# 등급 일관성: 비행 6회 중 5회만 무작위로 써도 같은 등급이 나오는 비율
cons = []
for pid, g in test.groupby("pilot"):
full = to_grade(g["score"].mean()).item()
cons += [to_grade(g["score"].sample(5, random_state=k).mean()).item() == full for k in range(20)]
print(f"\n등급 일관성 {np.mean(cons):.1%} (목표 90% 이상: {'달성' if np.mean(cons) >= 0.9 else '미달'})")
print("[신뢰등급 분포]", P["신뢰등급"].value_counts().to_dict(), "/ 등급 일치(교관 대비)", f"{(P['등급'] == P['교관등급']).mean():.0%}")
print("신뢰등급 기준: High(부트스트랩 동일 등급 90% 이상)=결과 통보, Medium(70~90%)=교관 검토 후 통보, Low=추가 비행 후 재평가")
P = P.sort_values("점수")
plt.figure(figsize=(10, 4))
col = P["신뢰등급"].map({"High": "tab:green", "Medium": "tab:orange", "Low": "tab:red"})
plt.errorbar(range(len(P)), P["점수"], yerr=P["점수SD"] / np.sqrt(6) * 2, fmt="none", ecolor="gray")
plt.scatter(range(len(P)), P["점수"], c=col)
plt.axhline(55, ls="--", c="k"); plt.axhline(75, ls="--", c="k")
plt.xticks(range(len(P)), P["pilot"], rotation=90, fontsize=6); plt.ylabel("mean skill score (+-2 SE)")
plt.title("Pilot skill score with uncertainty (color = confidence High/Medium/Low)"); plt.tight_layout(); plt.show()
In [ ]:
"""조종자 1명의 비행 궤적을 3D로 재생하고 경로 이탈·자세 한계·급조작 구간을 취약 구간으로 강조하는 분석 대시보드(야간 모드)를 구현한다. 재생 1초마다 지표를 갱신하는 처리시간을 측정하고 취약 구간 목록을 출력한다."""
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
# ---------------------------------------------------------------------------
# [교육용가상자료] 드론 조종 시뮬레이터 비행로그 생성기 (실제 훈련생·장비 정보 없음, 조종자는 가명 P001~)
# 비행 1회 = 120초(10Hz): 경로 비행(0~60초) -> 호버링(60~85초) -> 돌발 장애물 회피(85초) -> 비상 착륙(100초 지시)
# 조종자마다 숨은 숙련도(skill)와 조작 습관(급조작 jerky, 과보정 overcorr)이 있고, 교관 점수는 이를 반영해 만든다.
# ---------------------------------------------------------------------------
DT, STEPS = 0.1, 1200
T = np.arange(STEPS) * DT
COND = {"day": dict(wind=0.4, gust=0.0, gps=0.25, delay=0.0), "night": dict(wind=0.4, gust=0.0, gps=0.5, delay=0.15),
"gust": dict(wind=1.2, gust=2.5, gps=0.25, delay=0.0)}
OBST_T, LAND_T = 85.0, 100.0
def make_pilots(n=40, seed=7):
rng = np.random.default_rng(seed)
skill = rng.beta(2.2, 2.0, n)
return pd.DataFrame({"pilot": [f"P{i + 1:03d}" for i in range(n)], "skill": skill,
"jerky": rng.random(n) < 0.15 + 0.35 * (1 - skill), "overcorr": rng.random(n) < 0.15 + 0.35 * (1 - skill)})
def target_path(t):
"""사각형 경로(40x30m, 고도 20m)를 3m/s로 비행 -> 60초 지점에서 호버링 -> 착륙 지시 후 고도 0."""
# [TODO : 학생 실습 6] 조종자 비행 궤적 3D 재생 및 취약 구간 하이라이트 분석 대시보드 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def fly(pilots, cond="day", seed=0, delay_extra=0.0, obstacle_side=None):
"""여러 조종자의 비행 1회를 동시에 모의한다(벡터화). 반환: 로그 배열 dict, 비행별 정답 정보 DataFrame."""
# [TODO : 학생 실습 6] 조종자 비행 궤적 3D 재생 및 취약 구간 하이라이트 분석 대시보드 구현 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def one(log, j):
"""여러 조종자 로그에서 j번째 조종자 1명의 로그(DataFrame)를 꺼낸다."""
return pd.DataFrame({k: (v if v.ndim == 1 else v[:, j]) for k, v in log.items()})
# ---------------------------------------------------------------------------
# ---- 비행로그 특징 추출 공통 함수(여러 조종자 로그를 한 번에 계산) ----
from scipy.ndimage import uniform_filter1d
BASE_F = ["path_rmse", "hover_drift", "hover_std", "alt_err"]
ADV_F = BASE_F + ["obs_rt", "obs_lateral", "touch_vz", "att_exceed", "jerk", "overshoot", "n_risk"]
RULES = {"near_collision": ("obs_lateral", "<", 2.5), "hard_landing": ("touch_vz", ">", 2.0), "attitude_exceed": ("att_exceed", ">=", 10)}
def detect(df):
"""위험 조작 사건 탐지 규칙: 장애물 2.5초 후 회피 거리 2.5m 미만, 접지 속도 2.0m/s 초과, 자세 35도 초과 1초 이상."""
out = {}
for ev, (f, op, th) in RULES.items():
out[ev] = df[f] < th if op == "<" else (df[f] > th if op == ">" else df[f] >= th)
return pd.DataFrame(out)
def features(log):
"""경로 이탈도·호버링 안정성(기본) + 장애물 반응·착륙 속도·자세 한계·급조작·과보정(고도화) 특징."""
X, Y, A = (uniform_filter1d(log[k], 9, axis=0) for k in ("x", "y", "alt")) # GPS 잡음 완화(이동평균 0.9초)
h = np.hypot(X - log["tx"], Y - log["ty"])
cr, hv = slice(0, 600), slice(620, 850)
k0, k1 = int(OBST_T / DT), int((OBST_T + 2.5) / DT)
side = np.sign(log["ty"][k0 + 1] - log["ty"][k0 - 1])
moved = (log["stick_roll"][k0:k0 + 40] * side) > 0.3
rt = np.where(moved.any(0), moved.argmax(0) * DT, 4.0)
vz = -np.gradient(A, DT, axis=0)
land = np.argmax((A < 0.4) & (T[:, None] >= LAND_T), axis=0)
touch = np.array([vz[max(i - 6, 0):i - 2, j].mean() if i > 6 else 3.0 for j, i in enumerate(land)])
ds = np.abs(np.diff(np.stack([log["stick_roll"], log["stick_pitch"]]), axis=1)) / DT
return pd.DataFrame({
"path_rmse": np.sqrt((h[cr] ** 2).mean(0)), "hover_drift": h[hv].mean(0), "hover_std": np.hypot(X[hv].std(0), Y[hv].std(0)),
"alt_err": np.abs(A[:850] - log["talt"][:850]).mean(0),
"obs_rt": rt, "obs_lateral": np.abs(Y[k1] - Y[k0]), "touch_vz": touch,
"att_exceed": (np.maximum(np.abs(log["roll_deg"]), np.abs(log["pitch_deg"])) > 35).sum(0),
"jerk": np.percentile(ds, 95, axis=(0, 1)),
"overshoot": np.clip(np.abs(Y[k0:k0 + 120] - Y[k0]).max(0) - 6.0, 0, None)}).pipe(
lambda d: d.assign(n_risk=detect(d).sum(axis=1))) # overshoot: 회피 후 목표 6m를 넘어선 거리(과보정)
def make_dataset(n_pilots=40, seed=7, conds=("day", "night", "gust"), flights=1):
"""조종자 n명 x 조건 x 반복 비행 -> 비행별 특징 + 정답(교관 점수·등급·위험 사건) 표."""
pilots = make_pilots(n_pilots, seed)
out = []
for c in conds:
for f in range(flights):
log, tr = fly(pilots, c, seed=seed * 100 + list(COND).index(c) * 10 + f)
out.append(pd.concat([tr.reset_index(drop=True), features(log)], axis=1).assign(flight=f))
return pd.concat(out, ignore_index=True)
def cond_z(df, ref, cols):
"""조건별 기준 집단(ref)의 평균·표준편차로 표준화: 돌풍처럼 어려운 조건에서 받는 불이익을 줄인다."""
z = df.copy()
z[cols] = z[cols].astype(float)
for c in df["cond"].unique():
r = ref[ref["cond"] == c][cols]
z.loc[df["cond"] == c, cols] = (df.loc[df["cond"] == c, cols] - r.mean()) / (r.std() + 1e-9)
return z
def to_grade(score):
return np.where(score >= 75, "A", np.where(score >= 55, "B", "C"))
PHASES = [(0, 60, "path"), (60, 85, "hover"), (85, 100, "obstacle"), (100, 120, "landing")]
# [교육용가상자료] 조종자 12명 중 가장 점수가 낮은 조종자의 돌풍 비행을 재생
pilots = make_pilots(12, seed=4)
log, truth = fly(pilots, "gust", seed=21)
j = int(truth["instructor"].idxmin())
d = one(log, j)
FE = features(log).assign(cond="gust")
f = FE.iloc[j]
d["err"] = np.hypot(d["x"] - d["tx"], d["y"] - d["ty"])
d["att"] = np.maximum(d["roll_deg"].abs(), d["pitch_deg"].abs())
d["stick_rate"] = np.r_[0, np.abs(np.diff(d["stick_roll"])) / DT]
def weak_flags(df):
return {"경로이탈>3m": df["err"] > 3.0, "자세>30도": df["att"] > 30, "급조작(변화율>6/s)": df["stick_rate"] > 6}
# 1) 재생 모의: 1초(10프레임)마다 누적 지표 갱신 + 처리시간 측정
ms, kpis = [], []
for k in range(10, len(d) + 1, 10):
t0 = time.perf_counter()
cur = d.iloc[:k]
flags = weak_flags(cur)
kpis.append({"t": cur["t"].iloc[-1], "err_now": cur["err"].iloc[-1], "weak_ratio": np.mean(np.any(np.c_[tuple(flags.values())], axis=1))})
ms.append((time.perf_counter() - t0) * 1000)
print(f"조종자 {truth['pilot'][j]}(교관 {truth['instructor'][j]}점, 등급 {truth['grade'][j]}) 돌풍 비행 재생: 갱신 {len(ms)}회, "
f"처리 평균 {np.mean(ms):.2f} ms, 최대 {np.max(ms):.2f} ms (목표 1초 주기·100 ms 이내 -> {'달성' if np.max(ms) < 100 else '미달'})")
# 2) 취약 구간 목록(연속 0.5초 이상)
segs = []
for name, m in weak_flags(d).items():
m = m.values.astype(int)
edges = np.flatnonzero(np.diff(np.r_[0, m, 0]))
for a, b in zip(edges[::2], edges[1::2]):
if b - a >= 5:
ph = next(p for s, e, p in PHASES if s <= d["t"][a] < e)
segs.append({"구분": name, "시작(s)": round(d["t"][a], 1), "길이(s)": round((b - a) * DT, 1), "단계": ph})
S = pd.DataFrame(segs).sort_values("시작(s)")
print(f"\n[취약 구간 {len(S)}개 - 시간순 상위 10개]")
print(S.head(10).to_string(index=False))
print("\n단계별 취약 시간(s):", S.groupby("단계")["길이(s)"].sum().round(1).to_dict())
print(f"돌발 대응: 장애물 반응 {f['obs_rt']:.1f}s, 2.5초 후 회피거리 {f['obs_lateral']:.1f}m, 접지 속도 {f['touch_vz']:.1f}m/s -> "
f"위험 사건 {', '.join(k for k, v in detect(FE.iloc[[j]]).iloc[0].items() if v) or '없음'}")
# 3) 대시보드(야간 모드, 구성요소 4종)
plt.style.use("dark_background")
fig = plt.figure(figsize=(13, 8))
ax = fig.add_subplot(2, 2, 1, projection="3d")
bad = np.any(np.c_[tuple(weak_flags(d).values())], axis=1)
ax.plot(d["tx"], d["ty"], d["talt"], "--", c="gray", lw=1, label="course")
ax.scatter(d["x"][~bad], d["y"][~bad], d["alt"][~bad], s=2, c="deepskyblue", label="normal")
ax.scatter(d["x"][bad], d["y"][bad], d["alt"][bad], s=6, c="red", label="weak segment")
ax.set_xlabel("x (m)"); ax.set_ylabel("y (m)"); ax.set_zlabel("alt (m)"); ax.legend(fontsize=7); ax.set_title("3D replay (red = weak)")
a2 = fig.add_subplot(2, 2, 2)
a2.plot(d["t"], d["err"], c="lime", lw=1); a2.axhline(3, c="red", ls="--")
for s, e, p in PHASES:
a2.axvspan(s, e, alpha=0.08 if p in ("path", "obstacle") else 0.16, color="white"); a2.text(s + 1, a2.get_ylim()[1] * 0.9, p, fontsize=8)
a2.set_title("Path deviation (m) by phase")
a3 = fig.add_subplot(2, 2, 3)
a3.plot(d["t"], d["stick_roll"], lw=0.8, label="roll"); a3.plot(d["t"], d["stick_pitch"], lw=0.8, label="pitch"); a3.plot(d["t"], d["throttle"], lw=0.8, label="throttle")
a3.scatter(d["t"][d["stick_rate"] > 6], d["stick_roll"][d["stick_rate"] > 6], c="red", s=10, label="abrupt")
a3.legend(fontsize=7, ncol=4); a3.set_title("Stick inputs"); a3.set_xlabel("time (s)")
a4 = fig.add_subplot(2, 2, 4)
z = cond_z(FE, FE, ADV_F).iloc[j][["path_rmse", "hover_std", "obs_rt", "touch_vz", "jerk", "overshoot"]]
a4.barh(z.index, z.values, color=["red" if v > 1 else "orange" if v > 0 else "deepskyblue" for v in z.values])
a4.axvline(0, c="white", lw=0.5); a4.set_title("Weakness profile (z-score vs peers, right = worse)")
fig.suptitle(f"Pilot {truth['pilot'][j]} flight replay dashboard (night mode) - training data only")
plt.tight_layout(); plt.show()
In [ ]:
# [TODO : 학생 실습 7] 교관이 직접 조종자 성향을 진단하고 피드백을 입력하는 HITL 평가 인터페이스 설계
# 미션 지침 및 검증 기준을 확인하고 코드를 실행 및 튜닝하세요.
"""AI가 제안한 숙련도 등급·조작 성향을 교관이 직접 확인·수정하고 피드백을 입력하는 HITL 평가 인터페이스를 구현한다. AI는 등급을 확정하지 않으며, 불리한 결정(재교육 지정·자격 보류)은 상세 근거와 두 번째 교관 확인을 요구하고, 모든 결정은 위·변조 확인이 가능한 감사 로그에 남긴다."""
import hashlib
import json
import pandas as pd
ACTIONS = {"CONFIRM": "AI 제안 등급 확정", "MODIFY": "등급 수정", "REFLIGHT": "추가 비행 후 재평가", "RETRAIN": "재교육 지정", "QUAL_HOLD": "자격 부여 보류"}
CONF_ORDER = {"Low": 0, "Medium": 1, "High": 2}
class ReviewError(ValueError):
pass
class FakeClock:
t = 0.0
def __call__(self):
return self.t
class ReviewQueue:
def __init__(self, clock):
self.items, self.log, self.clock = {}, [], clock
def _record(self, body):
prev = self.log[-1]["hash"] if self.log else "GENESIS"
body = {**body, "ts": round(self.clock(), 1)}
self.log.append({**body, "prev": prev, "hash": hashlib.sha256((prev + json.dumps(body, ensure_ascii=False, sort_keys=True)).encode()).hexdigest()})
def verify_log(self):
prev = "GENESIS"
for n, rec in enumerate(self.log, start=1):
body = {k: v for k, v in rec.items() if k not in ("prev", "hash")}
if rec["prev"] != prev or rec["hash"] != hashlib.sha256((prev + json.dumps(body, ensure_ascii=False, sort_keys=True)).encode()).hexdigest():
return False, n
prev = rec["hash"]
return True, -1
def submit(self, pilot, grade, conf, habits, evidence):
"""AI 제안 등록: 항상 PENDING으로 시작하며 AI가 확정하지 않는다."""
self.items[pilot] = {"pilot": pilot, "ai_grade": grade, "conf": conf, "habits": habits, "evidence": evidence,
"status": "PENDING", "final": None, "submitted": self.clock(), "approvals": []}
self._record({"event": "AI_SUGGEST", "pilot": pilot, "grade": grade, "conf": conf, "by": "AI"})
def order(self):
"""검토 순서: 신뢰도 낮은 제안 -> C 등급 -> 나머지(교관이 판단이 필요한 건부터 본다)."""
todo = [i for i in self.items.values() if i["status"] in ("PENDING", "AWAIT_2ND")]
return sorted(todo, key=lambda i: (CONF_ORDER[i["conf"]], i["ai_grade"] != "C", i["pilot"]))
def decide(self, pilot, action, instructor, reason, new_grade=None):
if action not in ACTIONS:
raise ReviewError(f"허용되지 않은 조치: {action}")
if not instructor:
raise ReviewError("교관 ID 필요")
it = self.items.get(pilot)
if it is None or it["status"] == "FINAL":
raise ReviewError("검토 대기 건이 아님")
if len(reason.strip()) < 10:
raise ReviewError("진단·피드백 10자 이상 필요")
if action == "MODIFY" and new_grade not in ("A", "B", "C"):
raise ReviewError("수정 등급(A/B/C) 필요")
if action in ("RETRAIN", "QUAL_HOLD") and len(reason.strip()) < 20:
raise ReviewError("불리한 결정은 근거 20자 이상 필요")
if action == "QUAL_HOLD":
if instructor in it["approvals"]:
raise ReviewError("두 번째 확인은 다른 교관이 해야 함")
it["approvals"].append(instructor)
self._record({"event": "DECIDE", "pilot": pilot, "action": action, "by": instructor, "reason": reason, "step": len(it["approvals"])})
if len(it["approvals"]) < 2:
it["status"] = "AWAIT_2ND"
return "두 번째 교관 확인 대기"
else:
self._record({"event": "DECIDE", "pilot": pilot, "action": action, "by": instructor, "reason": reason, "new_grade": new_grade})
it["status"], it["action"], it["decided"] = "FINAL", action, self.clock()
it["final"] = new_grade if action == "MODIFY" else (None if action == "REFLIGHT" else it["ai_grade"])
return "확정"
clock = FakeClock()
rq = ReviewQueue(clock)
# [교육용가상자료] 미션 5 결과에서 넘어온 AI 제안 6건(가명)
for p, g, c, h, e in [("P003", "A", "High", [], "경로 RMSE 하위 10%, 접지 0.4m/s"), ("P011", "C", "High", ["급조작"], "경착륙 2회, 스틱 변화율 상위 5%"),
("P017", "B", "Low", ["과보정"], "비행별 점수 편차 큼(48~79)"), ("P024", "C", "Medium", ["과보정"], "장애물 근접 3회"),
("P030", "B", "High", [], "호버링 안정, 장애물 반응 1.2s"), ("P038", "C", "Low", [], "돌풍 비행만 낮음")]:
clock.t += 1; rq.submit(p, g, c, h, e)
print("검토 순서:", [f"{i['pilot']}({i['ai_grade']}/{i['conf']})" for i in rq.order()])
print("AI가 확정한 건수:", sum(i["status"] == "FINAL" for i in rq.items.values()))
for args in [("P011", "RETRAIN", "", "경착륙 반복, 착륙 절차 재교육"), ("P003", "CONFIRM", "inst01", "좋음"), ("P017", "MODIFY", "inst01", "편차가 커서 등급 조정함"),
("P024", "RETRAIN", "inst02", "근접 3회 재교육"), ("P003", "AUTO", "inst01", "자동 확정 시도합니다")]:
try:
rq.decide(*args)
except ReviewError as e:
print(f"차단: {args[0]} {args[1]} -> {e}")
clock.t += 120; print("P003:", rq.decide("P003", "CONFIRM", "inst01", "경로·호버링 안정, 돌발 대응 양호 확인"))
clock.t += 200; print("P017:", rq.decide("P017", "REFLIGHT", "inst01", "점수 편차 큼, 주간 비행 2회 추가 후 재평가"))
clock.t += 150; print("P011:", rq.decide("P011", "RETRAIN", "inst02", "경착륙 2회와 급조작 확인, 착륙 감속 절차 재교육 지정"))
clock.t += 90; print("P024 1차:", rq.decide("P024", "QUAL_HOLD", "inst02", "장애물 근접 3회 반복, 회피 기동 숙달 전 자격 보류"))
clock.t += 10
try:
rq.decide("P024", "QUAL_HOLD", "inst02", "같은 교관이 두 번째 확인까지 직접 하려고 시도함")
except ReviewError as e:
print("차단: P024 QUAL_HOLD ->", e)
clock.t += 300; print("P024 2차:", rq.decide("P024", "QUAL_HOLD", "inst05", "영상 재확인 결과 회피 지연 동의, 보류 유지"))
clock.t += 60; print("P038:", rq.decide("P038", "MODIFY", "inst03", "돌풍 외 비행은 B 수준, 돌풍 대응만 보완", new_grade="B"))
df = pd.DataFrame(rq.items.values())
df["대기(s)"] = (df["decided"] - df["submitted"]).round(0)
print("\n" + df[["pilot", "ai_grade", "conf", "action", "final", "대기(s)"]].fillna("-").to_string(index=False))
fin = df[df["status"] == "FINAL"]
print(f"교관 결정 {len(fin)}건 중 AI 제안 등급과 다른 결정 {int((fin['final'] != fin['ai_grade']).sum())}건(추가 비행 포함), 미검토 {int((df['status'] != 'FINAL').sum())}건은 대기 유지")
print(f"\n감사 로그 {len(rq.log)}건 기록(제안 {len(df)}건 + 결정 {len(rq.log) - len(df)}건, 로깅 성공률 100%), 해시 체인 검증: {rq.verify_log()}")
rq.log[7]["reason"] = "근거 수정"
print("8번째 기록 변조 후 검증(False, 깨진 위치):", rq.verify_log())
form = {"pilot": "P011", "strength": "(교관 입력)", "weakness": "착륙 감속 늦음, 급조작", "next_training": "착륙 절차 반복 3회", "instructor": "inst02", "pilot_ack": "(서명)"}
print("\n[조종자 피드백 양식]\n" + json.dumps(form, ensure_ascii=False))
print("AI 점수는 교관 판단 보조 자료이며 인사·자격 결정의 단독 근거로 쓰지 않는다. 최종 등급의 책임은 결정한 교관에게 있다.")
In [ ]:
"""조종자 100명의 대량 비행로그를 배치 분석하는 스트레스 테스트를 수행한다. 평시 100건에서 500% 증가(600건), 손상 로그 혼입, 묶음 처리 중 예외 상황을 주입해 시스템 다운 0건·로그 유실 0건(채점+격리=전체)과 처리량·메모리·병목 함수를 측정하고, 예외 시 1건씩 처리로 기능을 축소하는 안전 실패 동작을 확인한다."""
import cProfile
import io
import pstats
import time
import tracemalloc
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.linear_model import Ridge
# ---------------------------------------------------------------------------
# [교육용가상자료] 드론 조종 시뮬레이터 비행로그 생성기 (실제 훈련생·장비 정보 없음, 조종자는 가명 P001~)
# 비행 1회 = 120초(10Hz): 경로 비행(0~60초) -> 호버링(60~85초) -> 돌발 장애물 회피(85초) -> 비상 착륙(100초 지시)
# 조종자마다 숨은 숙련도(skill)와 조작 습관(급조작 jerky, 과보정 overcorr)이 있고, 교관 점수는 이를 반영해 만든다.
# ---------------------------------------------------------------------------
DT, STEPS = 0.1, 1200
T = np.arange(STEPS) * DT
COND = {"day": dict(wind=0.4, gust=0.0, gps=0.25, delay=0.0), "night": dict(wind=0.4, gust=0.0, gps=0.5, delay=0.15),
"gust": dict(wind=1.2, gust=2.5, gps=0.25, delay=0.0)}
OBST_T, LAND_T = 85.0, 100.0
def make_pilots(n=40, seed=7):
rng = np.random.default_rng(seed)
skill = rng.beta(2.2, 2.0, n)
return pd.DataFrame({"pilot": [f"P{i + 1:03d}" for i in range(n)], "skill": skill,
"jerky": rng.random(n) < 0.15 + 0.35 * (1 - skill), "overcorr": rng.random(n) < 0.15 + 0.35 * (1 - skill)})
def target_path(t):
"""사각형 경로(40x30m, 고도 20m)를 3m/s로 비행 -> 60초 지점에서 호버링 -> 착륙 지시 후 고도 0."""
# [TODO : 학생 실습 8] 100명 조종자 대량 비행로그 배치 분석 스트레스 테스트 및 성능 프로파일링 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def fly(pilots, cond="day", seed=0, delay_extra=0.0, obstacle_side=None):
"""여러 조종자의 비행 1회를 동시에 모의한다(벡터화). 반환: 로그 배열 dict, 비행별 정답 정보 DataFrame."""
# [TODO : 학생 실습 8] 100명 조종자 대량 비행로그 배치 분석 스트레스 테스트 및 성능 프로파일링 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def one(log, j):
"""여러 조종자 로그에서 j번째 조종자 1명의 로그(DataFrame)를 꺼낸다."""
return pd.DataFrame({k: (v if v.ndim == 1 else v[:, j]) for k, v in log.items()})
# ---------------------------------------------------------------------------
# ---- 비행로그 특징 추출 공통 함수(여러 조종자 로그를 한 번에 계산) ----
from scipy.ndimage import uniform_filter1d
BASE_F = ["path_rmse", "hover_drift", "hover_std", "alt_err"]
ADV_F = BASE_F + ["obs_rt", "obs_lateral", "touch_vz", "att_exceed", "jerk", "overshoot", "n_risk"]
RULES = {"near_collision": ("obs_lateral", "<", 2.5), "hard_landing": ("touch_vz", ">", 2.0), "attitude_exceed": ("att_exceed", ">=", 10)}
def detect(df):
"""위험 조작 사건 탐지 규칙: 장애물 2.5초 후 회피 거리 2.5m 미만, 접지 속도 2.0m/s 초과, 자세 35도 초과 1초 이상."""
out = {}
for ev, (f, op, th) in RULES.items():
out[ev] = df[f] < th if op == "<" else (df[f] > th if op == ">" else df[f] >= th)
return pd.DataFrame(out)
def features(log):
"""경로 이탈도·호버링 안정성(기본) + 장애물 반응·착륙 속도·자세 한계·급조작·과보정(고도화) 특징."""
X, Y, A = (uniform_filter1d(log[k], 9, axis=0) for k in ("x", "y", "alt")) # GPS 잡음 완화(이동평균 0.9초)
h = np.hypot(X - log["tx"], Y - log["ty"])
cr, hv = slice(0, 600), slice(620, 850)
k0, k1 = int(OBST_T / DT), int((OBST_T + 2.5) / DT)
side = np.sign(log["ty"][k0 + 1] - log["ty"][k0 - 1])
moved = (log["stick_roll"][k0:k0 + 40] * side) > 0.3
rt = np.where(moved.any(0), moved.argmax(0) * DT, 4.0)
vz = -np.gradient(A, DT, axis=0)
land = np.argmax((A < 0.4) & (T[:, None] >= LAND_T), axis=0)
touch = np.array([vz[max(i - 6, 0):i - 2, j].mean() if i > 6 else 3.0 for j, i in enumerate(land)])
ds = np.abs(np.diff(np.stack([log["stick_roll"], log["stick_pitch"]]), axis=1)) / DT
return pd.DataFrame({
"path_rmse": np.sqrt((h[cr] ** 2).mean(0)), "hover_drift": h[hv].mean(0), "hover_std": np.hypot(X[hv].std(0), Y[hv].std(0)),
"alt_err": np.abs(A[:850] - log["talt"][:850]).mean(0),
"obs_rt": rt, "obs_lateral": np.abs(Y[k1] - Y[k0]), "touch_vz": touch,
"att_exceed": (np.maximum(np.abs(log["roll_deg"]), np.abs(log["pitch_deg"])) > 35).sum(0),
"jerk": np.percentile(ds, 95, axis=(0, 1)),
"overshoot": np.clip(np.abs(Y[k0:k0 + 120] - Y[k0]).max(0) - 6.0, 0, None)}).pipe(
lambda d: d.assign(n_risk=detect(d).sum(axis=1))) # overshoot: 회피 후 목표 6m를 넘어선 거리(과보정)
def make_dataset(n_pilots=40, seed=7, conds=("day", "night", "gust"), flights=1):
"""조종자 n명 x 조건 x 반복 비행 -> 비행별 특징 + 정답(교관 점수·등급·위험 사건) 표."""
pilots = make_pilots(n_pilots, seed)
out = []
for c in conds:
for f in range(flights):
log, tr = fly(pilots, c, seed=seed * 100 + list(COND).index(c) * 10 + f)
out.append(pd.concat([tr.reset_index(drop=True), features(log)], axis=1).assign(flight=f))
return pd.concat(out, ignore_index=True)
def cond_z(df, ref, cols):
"""조건별 기준 집단(ref)의 평균·표준편차로 표준화: 돌풍처럼 어려운 조건에서 받는 불이익을 줄인다."""
z = df.copy()
z[cols] = z[cols].astype(float)
for c in df["cond"].unique():
r = ref[ref["cond"] == c][cols]
z.loc[df["cond"] == c, cols] = (df.loc[df["cond"] == c, cols] - r.mean()) / (r.std() + 1e-9)
return z
def to_grade(score):
return np.where(score >= 75, "A", np.where(score >= 55, "B", "C"))
KEYS = ["t", "x", "y", "alt", "stick_pitch", "stick_roll", "throttle", "roll_deg", "pitch_deg", "tx", "ty", "talt"]
def check_log(L):
"""로그 1건 검증. 문제가 있으면 사유 문자열, 없으면 None."""
if any(k not in L for k in KEYS):
return "필드 누락"
if len(L["t"]) != STEPS:
return f"길이 오류({len(L['t'])})"
if np.isnan(L["x"]).mean() > 0.2:
return "결측 20% 초과"
if np.any(np.diff(L["t"]) <= 0):
return "시각 역행"
if np.abs(L["stick_roll"]).max() > 1.0001:
return "스틱 범위 초과"
return None
def score_chunk(logs, conds, eng, ref):
stacked = {k: np.stack([L[k] for L in logs], axis=1) if k != "t" else T for k in KEYS}
stacked = {k: (pd.DataFrame(v).interpolate(limit_direction="both").values if v.ndim == 2 else v) for k, v in stacked.items()}
f = features(stacked).assign(cond=conds)
return np.clip(eng.predict(cond_z(f, ref, ADV_F)[ADV_F]), 0, 100)
def run_batch(logs, conds, eng, ref, chunk=100, fail_chunk=None):
scored, quarantine, degraded, down = {}, {}, 0, 0
try:
ok = []
for i, L in enumerate(logs):
why = check_log(L)
(quarantine.__setitem__(i, why) if why else ok.append(i))
for c0 in range(0, len(ok), chunk):
idx = ok[c0:c0 + chunk]
try:
if fail_chunk is not None and c0 // chunk == fail_chunk:
raise MemoryError("모의 자원 부족")
for i, s in zip(idx, score_chunk([logs[i] for i in idx], [conds[i] for i in idx], eng, ref)):
scored[i] = s
except Exception: # 기능 축소: 1건씩 처리, 그래도 실패하면 격리
degraded += 1
for i in idx:
try:
scored[i] = score_chunk([logs[i]], [conds[i]], eng, ref)[0]
except Exception as e:
quarantine[i] = f"처리 실패({type(e).__name__})"
except Exception as e:
down = 1; print("다운:", e)
return scored, quarantine, degraded, down
# 엔진 준비(미션 4 방식)
train = make_dataset(40, seed=7)
eng = Ridge(alpha=1.0).fit(cond_z(train, train, ADV_F)[ADV_F], train["instructor"])
# [교육용가상자료] 조종자 100명 x 비행 6회 = 로그 600건 생성
pilots = make_pilots(100, seed=12)
LOGS, CONDS = [], []
for f in range(6):
cond = list(COND)[f % 3]
log, _ = fly(pilots, cond, seed=900 + f)
for j in range(len(pilots)):
LOGS.append({k: (v if v.ndim == 1 else v[:, j].copy()) for k, v in log.items() if k in KEYS}); CONDS.append(cond)
rng = np.random.default_rng(5)
BAD = []
for i in rng.choice(len(LOGS), 30, replace=False): # 손상 로그 5% 주입
L = {k: v.copy() for k, v in LOGS[i].items()}
kind = i % 4
if kind == 0:
del L["throttle"]
elif kind == 1:
L["x"][:400] = np.nan
elif kind == 2:
L = {k: v[:500] for k, v in L.items()}
else:
L["stick_roll"][100] = 7.0
BAD.append((i, L))
CORRUPT = list(LOGS)
for i, L in BAD:
CORRUPT[i] = L
CASES = [("평시 100건", LOGS[:100], CONDS[:100], None), ("100명 x 5회 500건", LOGS[:500], CONDS[:500], None),
("500% 증가 600건", LOGS, CONDS, None), ("600건 + 손상 5%", CORRUPT, CONDS, None),
("600건 + 손상 + 묶음 예외", CORRUPT, CONDS, 2)]
rows = []
for name, logs, conds, fail in CASES:
t0 = time.perf_counter()
sc, qu, dg, dn = run_batch(logs, conds, eng, train, fail_chunk=fail)
dt = time.perf_counter() - t0
rows.append({"상황": name, "로그": len(logs), "채점": len(sc), "격리": len(qu), "유실": len(logs) - len(sc) - len(qu),
"다운": dn, "기능축소 묶음": dg, "처리량(건/s)": round(len(logs) / dt), "소요(s)": round(dt, 2)})
R = pd.DataFrame(rows)
print(R.to_string(index=False))
print(f"\n시스템 다운 {R['다운'].sum()}건, 로그 유실 {R['유실'].sum()}건(채점+격리=전체), 최저 처리량 {R['처리량(건/s)'].min()}건/s")
_, qu, _, _ = run_batch(CORRUPT, CONDS, eng, train)
print("격리 사유:", pd.Series(list(qu.values())).value_counts().to_dict(), f"/ 주입 {len(BAD)}건")
tracemalloc.start()
mem = []
for k in range(3):
run_batch(LOGS[:200], CONDS[:200], eng, train)
mem.append(tracemalloc.get_traced_memory()[0] / 1e6)
peak = tracemalloc.get_traced_memory()[1] / 1e6
tracemalloc.stop()
print(f"200건 처리 최대 메모리 {peak:.0f} MB, 반복 후 잔류 {[round(m, 1) for m in mem]} MB -> {'누수 의심' if mem[-1] > mem[0] * 1.5 + 1 else '누수 징후 없음'}")
pr = cProfile.Profile(); pr.enable()
run_batch(LOGS[:300], CONDS[:300], eng, train)
pr.disable()
s = io.StringIO(); pstats.Stats(pr, stream=s).sort_stats("cumulative").print_stats(30)
top = [ln.split() for ln in s.getvalue().splitlines() if ln.strip()[:1].isdigit() and ".py" in ln and "m08" in ln.split()[-1]]
print("[누적 시간 상위 함수(이 파일)]")
for f in top[:4]:
print(f" {f[0]:>6}회 {f[3]}s {f[-1].split(':')[-1]}")
print("개선 방향: 비행별 보간(DataFrame.interpolate)을 numpy 일괄 보간으로 바꾸고, 묶음 크기를 메모리에 맞춰 조정한다.")
plt.figure(figsize=(8, 4))
plt.bar([f"case{i + 1}" for i in range(len(R))], R["처리량(건/s)"], color=["tab:blue"] * 3 + ["tab:orange", "tab:red"])
plt.ylabel("logs per second"); plt.title("Batch analysis stress test throughput (synthetic)")
plt.tight_layout(); plt.show()
In [ ]:
"""AAR(사후검토)로 교관 평가 점수와 AI 숙련도 점수의 편차를 비행 조건·위험 사건·조작 습관별로 분석해 원인을 규명하고, 특징 구성·조건 표준화·가중치(규제 강도)를 격자 탐색으로 재튜닝한 뒤 다른 조종자 집단으로 회귀 테스트해 개선 전후를 비교한다."""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.linear_model import Ridge
# ---------------------------------------------------------------------------
# [교육용가상자료] 드론 조종 시뮬레이터 비행로그 생성기 (실제 훈련생·장비 정보 없음, 조종자는 가명 P001~)
# 비행 1회 = 120초(10Hz): 경로 비행(0~60초) -> 호버링(60~85초) -> 돌발 장애물 회피(85초) -> 비상 착륙(100초 지시)
# 조종자마다 숨은 숙련도(skill)와 조작 습관(급조작 jerky, 과보정 overcorr)이 있고, 교관 점수는 이를 반영해 만든다.
# ---------------------------------------------------------------------------
DT, STEPS = 0.1, 1200
T = np.arange(STEPS) * DT
COND = {"day": dict(wind=0.4, gust=0.0, gps=0.25, delay=0.0), "night": dict(wind=0.4, gust=0.0, gps=0.5, delay=0.15),
"gust": dict(wind=1.2, gust=2.5, gps=0.25, delay=0.0)}
OBST_T, LAND_T = 85.0, 100.0
def make_pilots(n=40, seed=7):
rng = np.random.default_rng(seed)
skill = rng.beta(2.2, 2.0, n)
return pd.DataFrame({"pilot": [f"P{i + 1:03d}" for i in range(n)], "skill": skill,
"jerky": rng.random(n) < 0.15 + 0.35 * (1 - skill), "overcorr": rng.random(n) < 0.15 + 0.35 * (1 - skill)})
def target_path(t):
"""사각형 경로(40x30m, 고도 20m)를 3m/s로 비행 -> 60초 지점에서 호버링 -> 착륙 지시 후 고도 0."""
# [TODO : 학생 실습 9] AAR 기반 교관 평가 점수와 AI 숙련도 점수 간 편차 분석 및 가중치 재튜닝 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def fly(pilots, cond="day", seed=0, delay_extra=0.0, obstacle_side=None):
"""여러 조종자의 비행 1회를 동시에 모의한다(벡터화). 반환: 로그 배열 dict, 비행별 정답 정보 DataFrame."""
# [TODO : 학생 실습 9] AAR 기반 교관 평가 점수와 AI 숙련도 점수 간 편차 분석 및 가중치 재튜닝 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def one(log, j):
"""여러 조종자 로그에서 j번째 조종자 1명의 로그(DataFrame)를 꺼낸다."""
return pd.DataFrame({k: (v if v.ndim == 1 else v[:, j]) for k, v in log.items()})
# ---------------------------------------------------------------------------
# ---- 비행로그 특징 추출 공통 함수(여러 조종자 로그를 한 번에 계산) ----
from scipy.ndimage import uniform_filter1d
BASE_F = ["path_rmse", "hover_drift", "hover_std", "alt_err"]
ADV_F = BASE_F + ["obs_rt", "obs_lateral", "touch_vz", "att_exceed", "jerk", "overshoot", "n_risk"]
RULES = {"near_collision": ("obs_lateral", "<", 2.5), "hard_landing": ("touch_vz", ">", 2.0), "attitude_exceed": ("att_exceed", ">=", 10)}
def detect(df):
"""위험 조작 사건 탐지 규칙: 장애물 2.5초 후 회피 거리 2.5m 미만, 접지 속도 2.0m/s 초과, 자세 35도 초과 1초 이상."""
out = {}
for ev, (f, op, th) in RULES.items():
out[ev] = df[f] < th if op == "<" else (df[f] > th if op == ">" else df[f] >= th)
return pd.DataFrame(out)
def features(log):
"""경로 이탈도·호버링 안정성(기본) + 장애물 반응·착륙 속도·자세 한계·급조작·과보정(고도화) 특징."""
X, Y, A = (uniform_filter1d(log[k], 9, axis=0) for k in ("x", "y", "alt")) # GPS 잡음 완화(이동평균 0.9초)
h = np.hypot(X - log["tx"], Y - log["ty"])
cr, hv = slice(0, 600), slice(620, 850)
k0, k1 = int(OBST_T / DT), int((OBST_T + 2.5) / DT)
side = np.sign(log["ty"][k0 + 1] - log["ty"][k0 - 1])
moved = (log["stick_roll"][k0:k0 + 40] * side) > 0.3
rt = np.where(moved.any(0), moved.argmax(0) * DT, 4.0)
vz = -np.gradient(A, DT, axis=0)
land = np.argmax((A < 0.4) & (T[:, None] >= LAND_T), axis=0)
touch = np.array([vz[max(i - 6, 0):i - 2, j].mean() if i > 6 else 3.0 for j, i in enumerate(land)])
ds = np.abs(np.diff(np.stack([log["stick_roll"], log["stick_pitch"]]), axis=1)) / DT
return pd.DataFrame({
"path_rmse": np.sqrt((h[cr] ** 2).mean(0)), "hover_drift": h[hv].mean(0), "hover_std": np.hypot(X[hv].std(0), Y[hv].std(0)),
"alt_err": np.abs(A[:850] - log["talt"][:850]).mean(0),
"obs_rt": rt, "obs_lateral": np.abs(Y[k1] - Y[k0]), "touch_vz": touch,
"att_exceed": (np.maximum(np.abs(log["roll_deg"]), np.abs(log["pitch_deg"])) > 35).sum(0),
"jerk": np.percentile(ds, 95, axis=(0, 1)),
"overshoot": np.clip(np.abs(Y[k0:k0 + 120] - Y[k0]).max(0) - 6.0, 0, None)}).pipe(
lambda d: d.assign(n_risk=detect(d).sum(axis=1))) # overshoot: 회피 후 목표 6m를 넘어선 거리(과보정)
def make_dataset(n_pilots=40, seed=7, conds=("day", "night", "gust"), flights=1):
"""조종자 n명 x 조건 x 반복 비행 -> 비행별 특징 + 정답(교관 점수·등급·위험 사건) 표."""
pilots = make_pilots(n_pilots, seed)
out = []
for c in conds:
for f in range(flights):
log, tr = fly(pilots, c, seed=seed * 100 + list(COND).index(c) * 10 + f)
out.append(pd.concat([tr.reset_index(drop=True), features(log)], axis=1).assign(flight=f))
return pd.concat(out, ignore_index=True)
def cond_z(df, ref, cols):
"""조건별 기준 집단(ref)의 평균·표준편차로 표준화: 돌풍처럼 어려운 조건에서 받는 불이익을 줄인다."""
z = df.copy()
z[cols] = z[cols].astype(float)
for c in df["cond"].unique():
r = ref[ref["cond"] == c][cols]
z.loc[df["cond"] == c, cols] = (df.loc[df["cond"] == c, cols] - r.mean()) / (r.std() + 1e-9)
return z
def to_grade(score):
return np.where(score >= 75, "A", np.where(score >= 55, "B", "C"))
def fit(df, feats, norm, alpha):
X = cond_z(df, df, feats)[feats] if norm else df[feats]
return Ridge(alpha=alpha).fit(X, df["instructor"])
def predict(m, df, ref, feats, norm):
X = cond_z(df, ref, feats)[feats] if norm else df[feats]
return np.clip(m.predict(X), 0, 100)
# 초기 운용 AI: 주간 비행만으로 맞춘 Baseline(경로·호버링·고도, 조건 표준화 없음)
day_train = make_dataset(40, seed=7, conds=("day",))
INIT = dict(feats=BASE_F, norm=False, alpha=1.0)
init_m = fit(day_train, **INIT)
# 1) 무엇이 일어났나: 훈련 1개 기수(조종자 40명 x 3조건)의 교관 점수와 AI 점수 비교
aar = make_dataset(40, seed=8)
aar["ai"] = predict(init_m, aar, day_train, INIT["feats"], INIT["norm"])
aar["dev"] = aar["ai"] - aar["instructor"]
acc0 = (to_grade(aar["ai"]) == aar["grade"]).mean()
print(f"[1. 무엇이 일어났나] 초기 AI 등급 일치율 {acc0:.1%}, 점수 편차(AI-교관) 평균 {aar['dev'].mean():+.1f}점, 절대 편차 {aar['dev'].abs().mean():.1f}점")
# 2) 왜 일어났나: 편차 원인 분해
aar["위험사건"] = aar[["near_collision", "hard_landing", "attitude_exceed"]].sum(axis=1).clip(0, 2).astype(str).str.replace("2", "2+")
print("\n[2. 왜 일어났나] 조건별 평균 편차:", aar.groupby("cond")["dev"].mean().round(1).to_dict())
ng = aar[aar["cond"] != "gust"] # 조건 효과를 뺀 주간·야간 비행으로 비교
ev = ng.groupby("위험사건")["dev"].mean().round(1)
print("주간·야간 비행의 위험 사건 수별 평균 편차:", ev.to_dict())
print("주간·야간 비행의 습관별 평균 편차: 급조작", round(ng[ng["jerky"]]["dev"].mean(), 1), "/ 과보정", round(ng[ng["overcorr"]]["dev"].mean(), 1),
"/ 습관 없음", round(ng[~ng["jerky"] & ~ng["overcorr"]]["dev"].mean(), 1))
print(f"-> 원인: (가) 주간 자료로만 맞춰 돌풍 비행을 평균 {aar[aar['cond'] == 'gust']['dev'].mean():.0f}점 과소평가(경로 이탈을 실력 부족으로 오인), "
f"(나) 위험 사건 2건 이상 비행을 0건 대비 {ev.get('2+', np.nan) - ev.get('0', np.nan):+.1f}점 과대평가(장애물·착륙 대응 미반영), (다) 급조작 습관 미반영")
# 3) 무엇을 바꾸나: AAR 자료(3조건)로 격자 탐색, 기수 내 2겹 교차검증으로 선택
pid = aar["pilot"].str[1:].astype(int) % 2
grid = []
for name, feats in [("기본", BASE_F), ("고도화", ADV_F)]:
for norm in (False, True):
for alpha in (0.1, 1.0, 10.0):
accs = []
for k in (0, 1):
tr, va = aar[pid != k], aar[pid == k]
m = fit(tr, feats, norm, alpha)
accs.append((to_grade(predict(m, va, tr, feats, norm)) == va["grade"]).mean())
grid.append({"특징": name, "조건표준화": norm, "alpha": alpha, "교차검증 일치율": round(np.mean(accs), 3)})
g = pd.DataFrame(grid).sort_values("교차검증 일치율", ascending=False)
print("\n[3. 무엇을 바꾸나] 격자 12개 중 상위 4개")
print(g.head(4).to_string(index=False))
best = g.iloc[0]
TUNED = dict(feats=ADV_F if best["특징"] == "고도화" else BASE_F, norm=bool(best["조건표준화"]), alpha=float(best["alpha"]))
tuned_m = fit(aar, **TUNED)
# 4) 회귀 테스트: 새 기수 2개(학습에 쓰지 않은 조종자)
print("\n[회귀 테스트] 기수 초기 AI 재튜닝 변화")
reg, rep, near = [], [], []
for sd in (21, 22):
new = make_dataset(40, seed=sd)
a0 = to_grade(predict(init_m, new, day_train, INIT["feats"], INIT["norm"]))
a1 = to_grade(predict(tuned_m, new, aar, TUNED["feats"], TUNED["norm"]))
r0, r1 = (a0 == new["grade"]).mean(), (a1 == new["grade"]).mean()
both = (a0 != new["grade"]) & (a1 != new["grade"])
reg.append((r0, r1)); rep.append(int(both.sum()))
near += list((np.minimum(np.abs(new["instructor"] - 55), np.abs(new["instructor"] - 75)) <= 5)[both])
print(f" #{sd} {r0:6.1%} {r1:6.1%} {(r1 - r0) * 100:+5.1f}%p")
gain = np.mean([b - a for a, b in reg])
print(f"평균 향상 {gain * 100:+.1f}%p (목표 +10%p 이상: {'달성' if gain >= 0.10 else '미달'}), 회귀 테스트 {sum(b >= a for a, b in reg)}/{len(reg)} 통과")
print(f"재튜닝 후에도 반복된 불일치 {sum(rep)}건(전체 {2 * len(new)}회 중), 그중 교관 점수가 등급 경계(55·75점) ±5점 이내인 비행 {np.mean(near):.0%}")
w = pd.Series(tuned_m.coef_, index=TUNED["feats"]).round(2)
yaml = (f"features: [{', '.join(TUNED['feats'])}]\ncondition_normalization: {str(TUNED['norm']).lower()}\nridge_alpha: {TUNED['alpha']}\n"
f"grade_cut: {{A: 75, B: 55}}\nweights: {{{', '.join(f'{k}: {v}' for k, v in w.items())}}}")
print("\n[재튜닝 파라미터(trn01_tuned_params.yaml 내용)]\n" + yaml)
print("지속 개선: 기수마다 교관 점수를 AAR 자료로 누적해 같은 절차로 재튜닝하고, 회귀 테스트 통과 시에만 교체한다.")
plt.figure(figsize=(7, 4))
for c, col in zip(COND, ["tab:blue", "tab:purple", "tab:orange"]):
s = aar[aar["cond"] == c]
plt.scatter(s["instructor"], s["dev"], s=12, c=col, label=c)
plt.axhline(0, c="k"); plt.xlabel("instructor score"); plt.ylabel("AI - instructor (initial AI)")
plt.legend(); plt.title("AAR: deviation of initial AI score (synthetic)"); plt.tight_layout(); plt.show()
In [ ]:
"""비행로그 검증·전처리부터 특징 추출, 재튜닝된 숙련도 엔진, 위험 조작 탐지, 조종자별 등급·신뢰도, 교관 HITL 검토 대기열, 분석 대시보드까지 전체 파이프라인을 한 파일로 통합 실행하고, 함수 커버리지를 포함한 자체 점검 결과와 교육훈련 브리핑 요약을 출력한다."""
import hashlib
import inspect
import json
import sys
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.linear_model import Ridge
# ---------------------------------------------------------------------------
# [교육용가상자료] 드론 조종 시뮬레이터 비행로그 생성기 (실제 훈련생·장비 정보 없음, 조종자는 가명 P001~)
# 비행 1회 = 120초(10Hz): 경로 비행(0~60초) -> 호버링(60~85초) -> 돌발 장애물 회피(85초) -> 비상 착륙(100초 지시)
# 조종자마다 숨은 숙련도(skill)와 조작 습관(급조작 jerky, 과보정 overcorr)이 있고, 교관 점수는 이를 반영해 만든다.
# ---------------------------------------------------------------------------
DT, STEPS = 0.1, 1200
T = np.arange(STEPS) * DT
COND = {"day": dict(wind=0.4, gust=0.0, gps=0.25, delay=0.0), "night": dict(wind=0.4, gust=0.0, gps=0.5, delay=0.15),
"gust": dict(wind=1.2, gust=2.5, gps=0.25, delay=0.0)}
OBST_T, LAND_T = 85.0, 100.0
def make_pilots(n=40, seed=7):
rng = np.random.default_rng(seed)
skill = rng.beta(2.2, 2.0, n)
return pd.DataFrame({"pilot": [f"P{i + 1:03d}" for i in range(n)], "skill": skill,
"jerky": rng.random(n) < 0.15 + 0.35 * (1 - skill), "overcorr": rng.random(n) < 0.15 + 0.35 * (1 - skill)})
def target_path(t):
"""사각형 경로(40x30m, 고도 20m)를 3m/s로 비행 -> 60초 지점에서 호버링 -> 착륙 지시 후 고도 0."""
# [TODO : 학생 실습 10] 드론 비행로그 숙련도 분석 시스템 통합 및 교육훈련 브리핑 패키지 완성 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def fly(pilots, cond="day", seed=0, delay_extra=0.0, obstacle_side=None):
"""여러 조종자의 비행 1회를 동시에 모의한다(벡터화). 반환: 로그 배열 dict, 비행별 정답 정보 DataFrame."""
# [TODO : 학생 실습 10] 드론 비행로그 숙련도 분석 시스템 통합 및 교육훈련 브리핑 패키지 완성 관련 핵심 로직을 구현하세요.
# 힌트: 상단 주석 및 데이터 스키마/지침을 참고하여 작성합니다.
# <<< 여기에 코드를 작성하세요 >>>
pass # 작성 후 pass를 제거하세요.
def one(log, j):
"""여러 조종자 로그에서 j번째 조종자 1명의 로그(DataFrame)를 꺼낸다."""
return pd.DataFrame({k: (v if v.ndim == 1 else v[:, j]) for k, v in log.items()})
# ---------------------------------------------------------------------------
# ---- 비행로그 특징 추출 공통 함수(여러 조종자 로그를 한 번에 계산) ----
from scipy.ndimage import uniform_filter1d
BASE_F = ["path_rmse", "hover_drift", "hover_std", "alt_err"]
ADV_F = BASE_F + ["obs_rt", "obs_lateral", "touch_vz", "att_exceed", "jerk", "overshoot", "n_risk"]
RULES = {"near_collision": ("obs_lateral", "<", 2.5), "hard_landing": ("touch_vz", ">", 2.0), "attitude_exceed": ("att_exceed", ">=", 10)}
def detect(df):
"""위험 조작 사건 탐지 규칙: 장애물 2.5초 후 회피 거리 2.5m 미만, 접지 속도 2.0m/s 초과, 자세 35도 초과 1초 이상."""
out = {}
for ev, (f, op, th) in RULES.items():
out[ev] = df[f] < th if op == "<" else (df[f] > th if op == ">" else df[f] >= th)
return pd.DataFrame(out)
def features(log):
"""경로 이탈도·호버링 안정성(기본) + 장애물 반응·착륙 속도·자세 한계·급조작·과보정(고도화) 특징."""
X, Y, A = (uniform_filter1d(log[k], 9, axis=0) for k in ("x", "y", "alt")) # GPS 잡음 완화(이동평균 0.9초)
h = np.hypot(X - log["tx"], Y - log["ty"])
cr, hv = slice(0, 600), slice(620, 850)
k0, k1 = int(OBST_T / DT), int((OBST_T + 2.5) / DT)
side = np.sign(log["ty"][k0 + 1] - log["ty"][k0 - 1])
moved = (log["stick_roll"][k0:k0 + 40] * side) > 0.3
rt = np.where(moved.any(0), moved.argmax(0) * DT, 4.0)
vz = -np.gradient(A, DT, axis=0)
land = np.argmax((A < 0.4) & (T[:, None] >= LAND_T), axis=0)
touch = np.array([vz[max(i - 6, 0):i - 2, j].mean() if i > 6 else 3.0 for j, i in enumerate(land)])
ds = np.abs(np.diff(np.stack([log["stick_roll"], log["stick_pitch"]]), axis=1)) / DT
return pd.DataFrame({
"path_rmse": np.sqrt((h[cr] ** 2).mean(0)), "hover_drift": h[hv].mean(0), "hover_std": np.hypot(X[hv].std(0), Y[hv].std(0)),
"alt_err": np.abs(A[:850] - log["talt"][:850]).mean(0),
"obs_rt": rt, "obs_lateral": np.abs(Y[k1] - Y[k0]), "touch_vz": touch,
"att_exceed": (np.maximum(np.abs(log["roll_deg"]), np.abs(log["pitch_deg"])) > 35).sum(0),
"jerk": np.percentile(ds, 95, axis=(0, 1)),
"overshoot": np.clip(np.abs(Y[k0:k0 + 120] - Y[k0]).max(0) - 6.0, 0, None)}).pipe(
lambda d: d.assign(n_risk=detect(d).sum(axis=1))) # overshoot: 회피 후 목표 6m를 넘어선 거리(과보정)
def make_dataset(n_pilots=40, seed=7, conds=("day", "night", "gust"), flights=1):
"""조종자 n명 x 조건 x 반복 비행 -> 비행별 특징 + 정답(교관 점수·등급·위험 사건) 표."""
pilots = make_pilots(n_pilots, seed)
out = []
for c in conds:
for f in range(flights):
log, tr = fly(pilots, c, seed=seed * 100 + list(COND).index(c) * 10 + f)
out.append(pd.concat([tr.reset_index(drop=True), features(log)], axis=1).assign(flight=f))
return pd.concat(out, ignore_index=True)
def cond_z(df, ref, cols):
"""조건별 기준 집단(ref)의 평균·표준편차로 표준화: 돌풍처럼 어려운 조건에서 받는 불이익을 줄인다."""
z = df.copy()
z[cols] = z[cols].astype(float)
for c in df["cond"].unique():
r = ref[ref["cond"] == c][cols]
z.loc[df["cond"] == c, cols] = (df.loc[df["cond"] == c, cols] - r.mean()) / (r.std() + 1e-9)
return z
def to_grade(score):
return np.where(score >= 75, "A", np.where(score >= 55, "B", "C"))
CONFIG = {"features": ADV_F, "condition_normalization": True, "ridge_alpha": 10.0} # 미션 9 재튜닝값
KEYS = ["x", "y", "alt", "stick_pitch", "stick_roll", "throttle", "roll_deg", "pitch_deg", "tx", "ty", "talt"]
def validate(L):
if any(k not in L for k in KEYS) or len(L["x"]) != STEPS:
return "필드·길이 오류"
if np.isnan(L["x"]).mean() > 0.2:
return "결측 20% 초과"
return None
def clean(L):
out = {}
for k in KEYS:
s = pd.Series(L[k])
if k in ("x", "y", "alt"):
med = s.rolling(7, center=True, min_periods=1).median()
s = s.mask((s - med).abs() > 3 * ((s - med).abs().rolling(7, center=True, min_periods=1).median() * 1.4826 + 0.3))
out[k] = s.interpolate(limit_direction="both").values
return out
class Review:
def __init__(self):
self.items, self.log = [], []
def submit(self, rec):
prev = self.log[-1]["hash"] if self.log else "GENESIS"
body = {"event": "AI_SUGGEST", "pilot": rec["pilot"], "grade": rec["grade"], "conf": rec["conf"]}
self.log.append({**body, "prev": prev, "hash": hashlib.sha256((prev + json.dumps(body, sort_keys=True)).encode()).hexdigest()})
self.items.append({**body, "status": "PENDING"})
called = set()
sys.setprofile(lambda f, e, a: called.add(f.f_code.co_name) if e == "call" and f.f_globals is globals() else None)
t_start = time.perf_counter()
# ---------------- 1. 엔진 준비(AAR 자료로 재튜닝된 설정) ----------------
ref = make_dataset(40, seed=8)
eng = Ridge(alpha=CONFIG["ridge_alpha"]).fit(cond_z(ref, ref, ADV_F)[ADV_F], ref["instructor"])
# ---------------- 2. 신규 기수 로그 수집(조종자 30명 x 3조건) + 손상·잡음 ----------------
pilots = make_pilots(30, seed=31)
rng = np.random.default_rng(4)
logs, conds, truth = [], [], []
for c in COND:
log, tr = fly(pilots, c, seed=700 + len(logs))
for j in range(len(pilots)):
L = {k: log[k][:, j].copy() for k in KEYS}
sp = rng.random(STEPS) < 0.01
L["x"][sp] += 10.0 # GPS 튐
L["y"][300:310] = np.nan # 센서 단절 1초
logs.append(L); conds.append(c)
truth.append(tr)
truth = pd.concat(truth, ignore_index=True)
logs[7] = {k: v for k, v in logs[7].items() if k != "throttle"} # 손상 로그 2건
logs[40]["x"][:500] = np.nan
# ---------------- 3. 검증·전처리·특징·채점 ----------------
rejected = {i: r for i, L in enumerate(logs) if (r := validate(L))}
ok = [i for i in range(len(logs)) if i not in rejected]
t0 = time.perf_counter()
cleaned = [clean(logs[i]) for i in ok]
stack = {k: np.stack([c[k] for c in cleaned], axis=1) for k in KEYS}
stack["t"] = T
F = features(stack).assign(cond=[conds[i] for i in ok])
D = pd.concat([truth.iloc[ok].reset_index(drop=True), F.drop(columns="cond")], axis=1)
D["score"] = np.clip(eng.predict(cond_z(F, ref, ADV_F)[ADV_F]), 0, 100)
ev = detect(F)
ms = (time.perf_counter() - t0) * 1000 / len(ok)
# ---------------- 4. 조종자 집계·신뢰도·교관 검토 대기열 ----------------
rows = []
for pid, g in D.groupby("pilot"):
boot = [to_grade(rng.choice(g["score"].values, len(g)).mean()).item() for _ in range(100)]
gr = to_grade(g["score"].mean()).item(); cf = boot.count(gr) / 100
rows.append({"pilot": pid, "score": g["score"].mean(), "grade": gr, "conf": "High" if cf >= 0.9 else ("Medium" if cf >= 0.7 else "Low"),
"inst_grade": to_grade(g["instructor"].mean()).item(), "risk": int(ev.loc[g.index].values.sum())})
P = pd.DataFrame(rows)
rv = Review()
for r in P.to_dict("records"):
rv.submit(r)
sys.setprofile(None)
# ---------------- 5. 자체 점검 ----------------
mine = [n for n, o in list(globals().items()) if inspect.isfunction(o) and o.__module__ == "__main__"] + ["submit", "__init__"]
cov = len(set(mine) & called) / len(set(mine))
fl_acc = (to_grade(D["score"]) == D["grade"]).mean()
pl_acc = (P["grade"] == P["inst_grade"]).mean()
truth_ev = D[["near_collision", "hard_landing", "attitude_exceed"]].astype(bool).values
rec = (ev.values & truth_ev).sum() / max(truth_ev.sum(), 1)
chk = {
"손상 로그 2건 격리": len(rejected) == 2,
"로그 유실 0건(채점+격리=전체)": len(ok) + len(rejected) == len(logs),
"전처리 후 결측 0": not any(np.isnan(v).any() for v in stack.values()),
"비행별 등급 일치율 75% 이상": fl_acc >= 0.75,
"조종자 등급 일치율 85% 이상": pl_acc >= 0.85,
"위험 사건 탐지 재현율 85% 이상": rec >= 0.85,
"로그당 처리 200ms 이내": ms < 200,
"AI 확정 0건(전부 교관 대기)": all(i["status"] == "PENDING" for i in rv.items),
"감사 로그 건수 일치": len(rv.log) == len(rv.items),
"함수 커버리지 85% 이상": cov >= 0.85,
}
passed = sum(chk.values())
for k, v in chk.items():
print(f"[{'PASS' if v else 'FAIL'}] {k}")
print(f"자체 점검 {passed}/{len(chk)} 통과, 함수 커버리지 {cov:.0%}, 파이프라인 실행 성공 {'예' if passed == len(chk) else '아니오'}")
print("조종자 등급 분포:", P["grade"].value_counts().reindex(["A", "B", "C"]).fillna(0).astype(int).to_dict(),
"/ 신뢰도:", P["conf"].value_counts().to_dict())
# ---------------- 6. 대시보드 ----------------
fig, ax = plt.subplots(2, 2, figsize=(12, 7))
ax[0, 0].scatter(D["instructor"], D["score"], s=10, c=D["cond"].map({"day": "tab:blue", "night": "tab:purple", "gust": "tab:orange"}))
ax[0, 0].plot([20, 100], [20, 100], "k--"); ax[0, 0].set_xlabel("instructor"); ax[0, 0].set_ylabel("AI"); ax[0, 0].set_title("Flight score: AI vs instructor")
ax[0, 1].bar(["near_collision", "hard_landing", "attitude"], ev.sum().values, color="tab:red"); ax[0, 1].set_title("Detected risk events")
Ps = P.sort_values("score")
ax[1, 0].bar(Ps["pilot"], Ps["score"], color=Ps["conf"].map({"High": "tab:green", "Medium": "tab:orange", "Low": "tab:red"}))
ax[1, 0].axhline(55, c="k", ls="--"); ax[1, 0].axhline(75, c="k", ls="--"); ax[1, 0].tick_params(axis="x", rotation=90, labelsize=6)
ax[1, 0].set_title("Pilot mean score (color = confidence)")
ax[1, 1].axis("off")
ax[1, 1].text(0, 0.9, f"Logs {len(logs)} ({len(rejected)} rejected)\nFlight acc {fl_acc:.1%}\nPilot acc {pl_acc:.1%}\n"
f"Risk recall {rec:.1%}\nPending {len(rv.items)} reviews\nSelf-check {passed}/{len(chk)}", family="monospace", fontsize=13, va="top")
plt.suptitle("TRN01 integrated skill analyzer - training data only"); plt.tight_layout(); plt.show()
# ---------------- 7. 교육훈련 브리핑 요약 ----------------
c_list = P[P["grade"] == "C"]["pilot"].tolist()
print(f"""
[교육훈련 브리핑 요약 - 교육용가상자료]
1. 처리: 비행로그 {len(logs)}건 중 {len(ok)}건 분석, 손상 {len(rejected)}건 격리, 로그당 {ms:.1f} ms, 전체 {time.perf_counter() - t_start:.1f}s
2. 성능: 비행별 등급 일치율 {fl_acc:.1%}, 조종자 등급 일치율 {pl_acc:.1%}, 위험 사건 탐지 재현율 {rec:.1%}
3. 결과: A {int((P['grade'] == 'A').sum())}명 / B {int((P['grade'] == 'B').sum())}명 / C {len(c_list)}명(재교육 검토 대상 {', '.join(c_list[:5])}{' 외' if len(c_list) > 5 else ''}), 전원 교관 검토 대기
4. 운용 유의: AI 등급은 교관 판단 보조 자료이며, 재교육·자격 보류는 교관 근거 기록과 두 번째 교관 확인 후 확정한다.
5. 한계·확장: 가상 시뮬레이터 로그 기준 결과다. 실제 시뮬레이터 로그 형식에 맞춘 재검증이 필요하며, 훈련 관리 체계와는 결과 조회 API로 연동한다.""")