import gradio as gr
import os
from datetime import datetime
from config import STATIONS, STATION_NAMES
from supabase_utils import get_supabase_client
from api_utils import (
api_get_tide_level,
api_get_tide_series,
api_get_extremes_info,
api_check_tide_alert,
api_compare_stations,
api_health_check
)
# API 문서 모듈 import
from api_docs import generate_api_docs
# 노이즈 시나리오 모듈 import
from noise_scenarios import apply_noise_scenario
def noise_test_handler(scenario, intensity, csv_file, station_id):
"""노이즈 테스트 실행 핸들러"""
if csv_file is None:
return None, "❌ CSV 파일을 업로드해주세요.", {}
try:
import pandas as pd
# 원본 데이터 로드
df_original = pd.read_csv(csv_file.name)
# 노이즈 시나리오 적용
df_noisy, comparison_plot = apply_noise_scenario(df_original, scenario, intensity)
# 안전한 비교를 위해 공통 컬럼과 인덱스만 사용
common_cols = set(df_original.columns) & set(df_noisy.columns)
numeric_cols = [col for col in common_cols if df_original[col].dtype in ['float64', 'int64']]
# 변화된 값 계산 (인덱스 리셋하여 안전하게 비교)
df_orig_reset = df_original.tail(len(df_noisy)).reset_index(drop=True)
df_noisy_reset = df_noisy.reset_index(drop=True)
changed_count = 0
for col in numeric_cols:
if col in df_orig_reset.columns and col in df_noisy_reset.columns:
orig_vals = df_orig_reset[col]
noisy_vals = df_noisy_reset[col]
# 길이 맞추기
min_len = min(len(orig_vals), len(noisy_vals))
orig_vals = orig_vals[:min_len]
noisy_vals = noisy_vals[:min_len]
# NaN이 아닌 값들 중에서 변화된 것 계산
mask = orig_vals.notna() & noisy_vals.notna()
if mask.any():
changed = (orig_vals[mask] != noisy_vals[mask]).sum()
changed_count += changed
# 로그 생성
log_message = f"""✅ 노이즈 테스트 완료
📁 파일: {csv_file.name}
🌪️ 시나리오: {scenario} (강도: {intensity})
📊 원본 데이터: {len(df_original)}행
🔧 노이즈 데이터: {len(df_noisy)}행
⚡ 변화된 값: {changed_count}개"""
# 성능 지표 계산
metrics = calculate_noise_metrics(df_original, df_noisy)
return comparison_plot, log_message, metrics
except Exception as e:
return None, f"❌ 노이즈 테스트 실패: {str(e)}", {}
def calculate_noise_metrics(df_original, df_noisy):
"""노이즈 영향 지표 계산"""
import pandas as pd
import numpy as np
metrics = {}
# 안전한 비교를 위해 인덱스 정렬 및 길이 맞추기
df_orig_tail = df_original.tail(len(df_noisy)).reset_index(drop=True)
df_noisy_reset = df_noisy.reset_index(drop=True)
# 공통 컬럼만 처리
common_cols = set(df_orig_tail.columns) & set(df_noisy_reset.columns)
numeric_cols = [col for col in common_cols if df_orig_tail[col].dtype in ['float64', 'int64'] and col != 'date']
for col in numeric_cols:
try:
original_vals = df_orig_tail[col].dropna()
noisy_vals = df_noisy_reset[col].dropna()
if len(original_vals) > 0 and len(noisy_vals) > 0:
# 길이 맞추기
min_len = min(len(original_vals), len(noisy_vals))
orig_subset = df_orig_tail[col][:min_len]
noisy_subset = df_noisy_reset[col][:min_len]
# NaN이 아닌 값들만 비교
valid_mask = orig_subset.notna() & noisy_subset.notna()
if valid_mask.any():
orig_valid = orig_subset[valid_mask]
noisy_valid = noisy_subset[valid_mask]
# 평균 절대 오차
mae = np.mean(np.abs(orig_valid - noisy_valid))
# 상대 오차 (%)
mean_abs_orig = np.mean(np.abs(orig_valid))
relative_error = (mae / mean_abs_orig * 100) if mean_abs_orig > 0 else 0
# 결측치 증가율
missing_increase = df_noisy_reset[col].isna().sum() - df_orig_tail[col].isna().sum()
# 최대/최소값 변화 (안전하게 계산)
orig_max = df_orig_tail[col].max() if df_orig_tail[col].notna().any() else 0
noisy_max = df_noisy_reset[col].max() if df_noisy_reset[col].notna().any() else 0
orig_min = df_orig_tail[col].min() if df_orig_tail[col].notna().any() else 0
noisy_min = df_noisy_reset[col].min() if df_noisy_reset[col].notna().any() else 0
metrics[col] = {
"평균_절대_오차": round(mae, 2),
"상대_오차_퍼센트": round(relative_error, 2),
"결측치_증가": missing_increase,
"최대값_변화": round(noisy_max - orig_max, 2),
"최소값_변화": round(noisy_min - orig_min, 2)
}
except Exception as e:
print(f"컬럼 {col} 메트릭 계산 오류: {e}")
continue
return metrics
def noise_prediction_handler(scenario, intensity, csv_file, station_id, prediction_handler):
"""노이즈 데이터로 예측 실행 핸들러"""
if csv_file is None:
return None, "❌ CSV 파일을 업로드해주세요.", {}
try:
import pandas as pd
import tempfile
import os
# 원본 데이터 로드
df_original = pd.read_csv(csv_file.name)
# 노이즈 시나리오 적용
df_noisy, _ = apply_noise_scenario(df_original, scenario, intensity)
# 노이즈 데이터를 임시 파일로 저장
import tempfile
temp_fd, temp_path = tempfile.mkstemp(suffix='.csv', text=True)
try:
# 파일 디스크립터를 닫고 경로만 사용
os.close(temp_fd)
df_noisy.to_csv(temp_path, index=False)
# Gradio File 형식에 맞는 객체 생성
class TempFile:
def __init__(self, path):
self.name = path
temp_file_obj = TempFile(temp_path)
print(f"📁 노이즈 임시 파일 생성: {temp_path}")
except Exception as e:
print(f"❌ 임시 파일 생성 실패: {e}")
if os.path.exists(temp_path):
os.unlink(temp_path)
raise
try:
# 파일 존재 및 유효성 확인
if not os.path.exists(temp_path):
raise FileNotFoundError(f"임시 파일이 존재하지 않습니다: {temp_path}")
file_size = os.path.getsize(temp_path)
if file_size == 0:
raise ValueError("임시 파일이 비어있습니다")
print(f"📊 임시 파일 크기: {file_size} bytes")
# 1. 원본 데이터로 예측
print("🔵 원본 데이터로 예측 중...")
original_plot, original_df, original_log = prediction_handler(station_id, csv_file)
# 2. 노이즈 데이터로 예측
print("🔴 노이즈 데이터로 예측 중...")
print(f"📁 노이즈 파일 경로: {temp_file_obj.name}")
noise_plot, noise_df, noise_log = prediction_handler(station_id, temp_file_obj)
# 디버깅: 예측 결과 구조 확인
print(f"📊 원본 예측 결과 타입: {type(original_df)}")
if original_df is not None:
if hasattr(original_df, 'columns'):
print(f"📊 원본 예측 컬럼: {list(original_df.columns)}")
else:
print(f"📊 원본 예측 내용: {original_df}")
print(f"📊 노이즈 예측 결과 타입: {type(noise_df)}")
if noise_df is not None:
if hasattr(noise_df, 'columns'):
print(f"📊 노이즈 예측 컬럼: {list(noise_df.columns)}")
else:
print(f"📊 노이즈 예측 내용: {noise_df}")
# 3. 예측 결과 비교 시각화 생성
comparison_plot = create_prediction_comparison_plot(original_df, noise_df, scenario)
# 4. 성능 비교 메트릭 계산
performance_metrics = calculate_prediction_metrics(original_df, noise_df)
# 5. 로그 생성
log_message = f"""✅ 노이즈 예측 비교 완료
🌪️ 시나리오: {scenario} (강도: {intensity})
🔵 원본 예측: {len(original_df) if original_df is not None else 0}개 포인트
🔴 노이즈 예측: {len(noise_df) if noise_df is not None else 0}개 포인트
📊 견고성 평가: {performance_metrics.get('robustness_score', 'N/A')}"""
return comparison_plot, log_message, performance_metrics
finally:
# 임시 파일 정리
if os.path.exists(temp_path):
os.unlink(temp_path)
except Exception as e:
return None, f"❌ 노이즈 예측 실패: {str(e)}", {}
def create_prediction_comparison_plot(original_df, noise_df, scenario_name):
"""원본 예측 vs 노이즈 예측 비교 시각화"""
import plotly.graph_objects as go
from plotly.subplots import make_subplots
import pandas as pd
import numpy as np
if original_df is None or noise_df is None:
return go.Figure().add_annotation(text="예측 데이터가 없습니다", xref="paper", yref="paper", x=0.5, y=0.5)
# DataFrame이 아닌 경우 처리
if not isinstance(original_df, pd.DataFrame):
return go.Figure().add_annotation(text="예측 데이터 형식 오류", xref="paper", yref="paper", x=0.5, y=0.5)
fig = make_subplots(
rows=2, cols=1,
subplot_titles=['🔮 조위 예측 비교', '📊 예측 차이'],
vertical_spacing=0.12
)
# 시간축 (예측 결과 인덱스)
time_axis = list(range(len(original_df)))
# 예측값 컬럼 찾기 및 숫자 변환
def get_prediction_values(df):
# 한국어 컬럼명도 포함한 가능한 예측값 컬럼 이름들
possible_cols = [
'final_tide', 'predicted', 'prediction', 'tide_level', 'residual',
'최종 조위 (cm)', '잔차 예측 (cm)', '조화 예측 (cm)'
]
for col in possible_cols:
if col in df.columns:
try:
# 괄호가 있는 경우 괄호 제거 후 숫자 변환 시도
if '(' in col:
values = df[col]
if values.dtype == 'object':
# 문자열에서 숫자 추출 시도
values = pd.to_numeric(values.astype(str).str.replace(r'[^\d.-]', '', regex=True), errors='coerce')
else:
values = pd.to_numeric(values, errors='coerce')
else:
values = pd.to_numeric(df[col], errors='coerce')
if not values.isna().all():
return values
except:
continue
# 마지막 숫자 컬럼 시도
numeric_cols = df.select_dtypes(include=[np.number]).columns
if len(numeric_cols) > 0:
return df[numeric_cols[-1]]
# 마지막 컬럼을 숫자로 변환 시도
try:
return pd.to_numeric(df.iloc[:, -1], errors='coerce')
except:
return pd.Series([0] * len(df))
original_values = get_prediction_values(original_df)
noise_values = get_prediction_values(noise_df)
# 원본 예측 결과
fig.add_trace(
go.Scatter(
x=time_axis,
y=original_values,
name='🔵 원본 예측',
line=dict(color='#2E86AB', width=3),
hovertemplate='원본 예측: %{y:.1f}cm
시점: %{x}'
),
row=1, col=1
)
# 노이즈 예측 결과
fig.add_trace(
go.Scatter(
x=time_axis[:len(noise_values)],
y=noise_values,
name='🔴 노이즈 예측',
line=dict(color='#F24236', width=3, dash='dash'),
hovertemplate='노이즈 예측: %{y:.1f}cm
시점: %{x}'
),
row=1, col=1
)
# 예측 차이 계산 및 시각화
if len(original_values) == len(noise_values):
try:
difference = noise_values - original_values
fig.add_trace(
go.Scatter(
x=time_axis,
y=difference,
name='📊 예측 차이',
line=dict(color='orange', width=2),
fill='tonexty',
hovertemplate='예측 차이: %{y:.1f}cm
시점: %{x}'
),
row=2, col=1
)
# 0선 추가
fig.add_hline(y=0, line_dash="dot", line_color="gray", row=2, col=1)
except Exception as e:
print(f"차이 계산 오류: {e}")
fig.update_layout(
title={
'text': f"🔮 예측 견고성 테스트: {scenario_name}",
'x': 0.5,
'font': {'size': 18, 'color': '#2E86AB'}
},
height=700,
showlegend=True,
legend=dict(x=0.02, y=0.98, bgcolor='rgba(255,255,255,0.8)')
)
fig.update_xaxes(title_text="미래 시점", showgrid=True)
fig.update_yaxes(title_text="조위 (cm)", showgrid=True)
return fig
def calculate_prediction_metrics(original_df, noise_df):
"""예측 성능 비교 메트릭 계산"""
if original_df is None or noise_df is None:
return {"error": "예측 데이터 없음"}
import numpy as np
import pandas as pd
try:
# 예측값 추출 함수 (동일한 로직)
def get_prediction_values(df):
# 한국어 컬럼명도 포함
possible_cols = [
'final_tide', 'predicted', 'prediction', 'tide_level', 'residual',
'최종 조위 (cm)', '잔차 예측 (cm)', '조화 예측 (cm)', '예측 시간'
]
for col in possible_cols:
if col in df.columns:
try:
# 시간 컬럼은 건너뛰기
if '시간' in col or 'time' in col.lower():
continue
# 괄호가 있는 경우 괄호 제거 후 숫자 변환 시도
if '(' in col:
values = df[col]
if values.dtype == 'object':
# 문자열에서 숫자 추출 시도
values = pd.to_numeric(values.astype(str).str.replace(r'[^\d.-]', '', regex=True), errors='coerce')
else:
values = pd.to_numeric(values, errors='coerce')
else:
values = pd.to_numeric(df[col], errors='coerce')
if not values.isna().all():
return values
except:
continue
numeric_cols = df.select_dtypes(include=[np.number]).columns
if len(numeric_cols) > 0:
return df[numeric_cols[-1]]
try:
return pd.to_numeric(df.iloc[:, -1], errors='coerce')
except:
return pd.Series([0] * len(df))
original_values = get_prediction_values(original_df)
noise_values = get_prediction_values(noise_df)
# 길이 맞추기
min_len = min(len(original_values), len(noise_values))
original_values = original_values[:min_len]
noise_values = noise_values[:min_len]
# NaN 제거
mask = ~(original_values.isna() | noise_values.isna())
original_clean = original_values[mask]
noise_clean = noise_values[mask]
if len(original_clean) == 0:
return {"error": "유효한 예측 데이터 없음"}
# 메트릭 계산
mae = np.mean(np.abs(noise_clean - original_clean)) # 평균 절대 오차
rmse = np.sqrt(np.mean((noise_clean - original_clean) ** 2)) # RMSE
max_diff = np.max(np.abs(noise_clean - original_clean)) # 최대 차이
# 견고성 점수 (0-100, 100이 가장 견고함)
mean_abs_original = np.mean(np.abs(original_clean))
if mean_abs_original > 0:
relative_error = mae / mean_abs_original * 100
else:
relative_error = 0
robustness_score = max(0, 100 - relative_error)
return {
"평균_절대_오차_cm": round(mae, 2),
"RMSE_cm": round(rmse, 2),
"최대_차이_cm": round(max_diff, 2),
"상대_오차_퍼센트": round(relative_error, 2),
"견고성_점수": round(robustness_score, 1),
"평가": "견고함" if robustness_score > 80 else "보통" if robustness_score > 60 else "취약함",
"비교_데이터_수": len(original_clean)
}
except Exception as e:
return {"error": f"메트릭 계산 실패: {str(e)}"}
def create_ui(prediction_handler, chatbot_handler, api_handlers: dict):
"""Gradio UI를 생성하고 반환합니다."""
with gr.Blocks(title="통합 조위 예측 시스템", theme=gr.themes.Soft()) as demo:
gr.Markdown("# 🌊 통합 조위 예측 시스템 with Gemini")
# 연결 상태 표시
client = get_supabase_client()
supabase_status = "🟢 연결됨" if client else "🔴 연결 안됨 (환경변수 확인 필요)"
gemini_status = "🟢 연결됨" if os.getenv("GEMINI_API_KEY") else "🔴 연결 안됨 (환경변수 확인 필요)"
gr.Markdown(f"**Supabase 상태**: {supabase_status} | **Gemini 상태**: {gemini_status}")
with gr.Tabs():
# 1. 예측 탭
with gr.TabItem("🔮 통합 조위 예측"):
gr.Markdown("""
### TimeXer 모델을 활용한 조위 예측
- 과거 기상 데이터를 업로드하여 미래 72시간 조위 예측
- 잔차 예측 + 조화 예측 = 최종 조위
""")
with gr.Row():
with gr.Column(scale=1):
station_id_input = gr.Dropdown(
choices=[(f"{STATION_NAMES[s]} ({s})", s) for s in STATIONS],
label="관측소 선택",
value="DT_0001"
)
input_csv = gr.File(label="과거 데이터 업로드 (.csv)")
predict_btn = gr.Button("🚀 예측 실행", variant="primary")
with gr.Column(scale=3):
output_plot = gr.Plot(label="예측 결과 시각화")
output_df = gr.DataFrame(label="예측 결과 데이터")
output_log = gr.Textbox(label="실행 로그", lines=5, interactive=False)
# 2. 챗봇 탭
with gr.TabItem("💬 AI 조위 챗봇"):
gr.Markdown("### 🤖 AI 조위 전문가")
gr.Markdown("조위에 대해 궁금한 점을 물어보세요.")
with gr.Row():
with gr.Column():
chatbot_display = gr.Chatbot(
label="대화창",
height=400,
show_label=True
)
with gr.Row():
user_input = gr.Textbox(
label="질문을 입력하세요",
placeholder="예: 인천 현재 조위 알려줘",
lines=1,
scale=4
)
send_btn = gr.Button("전송", variant="primary", scale=1)
gr.Examples(
examples=[
"인천 현재 조위 알려줘",
"오늘 만조 시간은?",
"부산과 인천 조위 비교해줘"
],
inputs=user_input
)
# 챗봇 이벤트 연결
def chat_response(message, history):
if not message.strip():
return history, ""
# 사용자 메시지 추가
history = history or []
history.append([message, None])
# AI 응답 생성
try:
response = chatbot_handler(message, history)
history[-1][1] = response
except Exception as e:
history[-1][1] = f"죄송합니다. 오류가 발생했습니다: {str(e)}"
return history, ""
send_btn.click(
fn=chat_response,
inputs=[user_input, chatbot_display],
outputs=[chatbot_display, user_input]
)
user_input.submit(
fn=chat_response,
inputs=[user_input, chatbot_display],
outputs=[chatbot_display, user_input]
)
# 3. 항구 혼잡도 분석 탭
with gr.TabItem("🚢 항구 혼잡도 분석"):
gr.Markdown("### 🎯 YOLO 기반 선박 탐지")
gr.Markdown("미리 업로드된 샘플 영상을 선택하여 항구 내 선박을 자동 탐지하고 혼잡도를 분석합니다.")
# 미리 업로드된 영상 목록 가져오기
import glob
video_files = glob.glob("videos/*.mp4") + glob.glob("videos/*.avi") + glob.glob("videos/*.mov")
video_choices = [("샘플 영상 없음", None)] if not video_files else [(f"📹 {os.path.basename(f)}", f) for f in video_files]
with gr.Row():
with gr.Column(scale=1):
video_dropdown = gr.Dropdown(
label="분석할 영상 선택",
choices=video_choices,
value=None
)
video_upload = gr.Video(
label="또는 새 영상 업로드",
sources=["upload"],
visible=True
)
analyze_btn = gr.Button("🔍 영상 분석 시작", variant="primary", size="lg")
gr.Markdown("""
**사용법**:
1. 드롭다운에서 미리 업로드된 영상 선택 또는
2. 새로운 영상 직접 업로드
**분석 내용**:
- 선박 개수 탐지
- 선박 위치 추적
- 항구 혼잡도 점수
""")
with gr.Column(scale=2):
video_output = gr.Video(
label="분석 결과 영상",
interactive=False
)
analysis_results = gr.Textbox(
label="분석 결과",
lines=8,
interactive=False,
placeholder="영상 분석 결과가 여기에 표시됩니다..."
)
def process_harbor_video(selected_video, uploaded_video):
"""항구 영상 분석 함수 (YOLO 기반)"""
# 선택된 영상이나 업로드된 영상 중 하나 선택
video_path = selected_video if selected_video else uploaded_video
if video_path is None:
return None, "드롭다운에서 영상을 선택하거나 새 영상을 업로드해주세요."
try:
# TODO: YOLO 모델을 사용한 선박 탐지 로직 구현
# 현재는 placeholder 결과 반환
video_name = os.path.basename(video_path) if video_path else "업로드된 영상"
results_text = f"""
📊 영상 분석 완료: {video_name}
🚢 탐지된 선박 수: 2-3척
📍 평균 선박 위치: 항구 중앙부
🟡 혼잡도 점수: 2/10 (낮음)
⏱️ 분석 시간: 80초
"""
return video_path, results_text
except Exception as e:
return None, f"영상 분석 중 오류가 발생했습니다: {str(e)}"
analyze_btn.click(
fn=process_harbor_video,
inputs=[video_dropdown, video_upload],
outputs=[video_output, analysis_results]
)
# 4. API 탭
with gr.TabItem("🔌 API"):
gr.Markdown("## RESTful API 엔드포인트\n실무에서 바로 사용 가능한 API 기능을 제공합니다.")
with gr.Tabs():
# 3-1. 특정 시간 조위
with gr.TabItem("조위 조회"):
gr.Markdown("#### 특정 관측소, 특정 시간의 조위 정보를 조회합니다.")
with gr.Row():
with gr.Column(scale=1):
api1_station = gr.Dropdown(choices=[(f"{STATION_NAMES[s]} ({s})", s) for s in STATIONS], label="관측소", value="DT_0001")
api1_time = gr.Textbox(label="조회 시간 (비워두면 현재)", placeholder="2025-08-10T15:00:00")
api1_btn = gr.Button("UI에서 테스트", variant="primary")
with gr.Column(scale=2):
api1_output = gr.JSON(label="API 응답")
api1_btn.click(
fn=lambda s, t: api_handlers["tide_level"](s, t if t else None),
inputs=[api1_station, api1_time],
outputs=api1_output
)
generate_api_docs("tide_level")
# 3-2. 시계열 데이터
with gr.TabItem("시계열"):
gr.Markdown("#### 지정된 기간의 시계열 조위 데이터를 조회합니다.")
with gr.Row():
with gr.Column(scale=1):
api2_station = gr.Dropdown(choices=[(f"{STATION_NAMES[s]} ({s})", s) for s in STATIONS], label="관측소", value="DT_0001")
api2_start = gr.Textbox(label="시작 시간", placeholder="2025-08-10T00:00:00")
api2_end = gr.Textbox(label="종료 시간", placeholder="2025-08-11T00:00:00")
api2_interval = gr.Number(label="간격(분)", value=60, minimum=5)
api2_btn = gr.Button("UI에서 테스트", variant="primary")
with gr.Column(scale=2):
api2_output = gr.JSON(label="API 응답 (공공 API 형식)")
api2_btn.click(
fn=lambda s, st, et, i: api_handlers["tide_series"](s, st if st else None, et if et else None, int(i)),
inputs=[api2_station, api2_start, api2_end, api2_interval],
outputs=api2_output
)
generate_api_docs("tide_series")
# 3-3. 만조/간조
with gr.TabItem("만조/간조"):
gr.Markdown("#### 특정 날짜의 만조/간조 정보를 조회합니다.")
with gr.Row():
with gr.Column(scale=1):
api3_station = gr.Dropdown(choices=[(f"{STATION_NAMES[s]} ({s})", s) for s in STATIONS], label="관측소", value="DT_0001")
api3_date = gr.Textbox(label="날짜 (YYYY-MM-DD)", placeholder=datetime.now().strftime("%Y-%m-%d"))
api3_secondary = gr.Checkbox(label="부차 만조/간조 포함", value=False)
api3_btn = gr.Button("UI에서 테스트", variant="primary")
with gr.Column(scale=2):
api3_output = gr.JSON(label="만조/간조 정보")
api3_btn.click(
fn=lambda s, d, sec: api_handlers["extremes"](s, d if d else None, sec),
inputs=[api3_station, api3_date, api3_secondary],
outputs=api3_output
)
generate_api_docs("extremes")
# 3-4. 위험 알림
with gr.TabItem("위험 알림"):
gr.Markdown("#### 향후 설정된 시간 동안 위험 수위 도달 여부를 체크합니다.")
with gr.Row():
with gr.Column(scale=1):
api4_station = gr.Dropdown(choices=[(f"{STATION_NAMES[s]} ({s})", s) for s in STATIONS], label="관측소", value="DT_0001")
api4_hours = gr.Number(label="확인 시간(시간)", value=24, minimum=1, maximum=72)
api4_warning = gr.Number(label="주의 수위(cm)", value=700)
api4_danger = gr.Number(label="경고 수위(cm)", value=750)
api4_btn = gr.Button("UI에서 테스트", variant="primary")
with gr.Column(scale=2):
api4_output = gr.JSON(label="위험 수위 정보")
api4_btn.click(
fn=lambda s, h, w, d: api_handlers["alert"](s, int(h), w, d),
inputs=[api4_station, api4_hours, api4_warning, api4_danger],
outputs=api4_output
)
generate_api_docs("alert")
# 3-5. 관측소 비교
with gr.TabItem("비교"):
gr.Markdown("#### 여러 관측소의 조위를 동시에 비교합니다.")
with gr.Row():
with gr.Column(scale=1):
api5_stations = gr.CheckboxGroup(choices=[(f"{STATION_NAMES[s]}", s) for s in STATIONS], label="비교할 관측소 선택", value=["DT_0001", "DT_0002", "DT_0003"])
api5_time = gr.Textbox(label="비교 시간 (비워두면 현재)", placeholder="2025-08-10T15:00:00")
api5_btn = gr.Button("UI에서 테스트", variant="primary")
with gr.Column(scale=2):
api5_output = gr.JSON(label="비교 결과")
api5_btn.click(
fn=lambda s, t: api_handlers["compare"](s, t if t else None),
inputs=[api5_stations, api5_time],
outputs=api5_output
)
generate_api_docs("compare")
# 3-6. 상태 체크
with gr.TabItem("상태"):
gr.Markdown("#### API 및 시스템의 현재 상태를 확인합니다.")
with gr.Row():
api6_btn = gr.Button("🔍 상태 확인", variant="secondary", scale=1)
api6_output = gr.JSON(label="시스템 상태", scale=2)
api6_btn.click(
fn=api_handlers["health"],
inputs=[],
outputs=api6_output
)
generate_api_docs("health")
# 4. 노이즈 테스트 탭
with gr.TabItem("🌪️ 노이즈 테스트"):
gr.Markdown("""
### 🧪 시스템 견고성 테스트
- 극한 기상 상황에서 예측 성능 평가
- 센서 오작동 및 데이터 결측 시뮬레이션
- 원본 vs 노이즈 데이터 비교 시각화
""")
with gr.Row():
with gr.Column(scale=1):
# 노이즈 시나리오 선택
noise_scenario = gr.Dropdown(
choices=[
("🌀 태풍 (폭풍해일)", "typhoon"),
("📡 센서 오작동", "sensor_malfunction"),
("❌ 연속 결측치", "burst_missing"),
("🌡️ 극한 기상", "extreme_weather")
],
label="노이즈 시나리오 선택",
value="typhoon"
)
# 노이즈 강도 조절
noise_intensity = gr.Slider(
minimum=0.5, maximum=2.0, step=0.1, value=1.0,
label="노이즈 강도 (0.5=약함, 2.0=극심)"
)
# 테스트 데이터 업로드
noise_test_csv = gr.File(label="테스트 데이터 업로드 (.csv)")
# 관측소 선택
noise_station = gr.Dropdown(
choices=[(f"{STATION_NAMES[s]} ({s})", s) for s in STATIONS],
label="관측소 선택",
value="DT_0001"
)
# 실행 버튼
noise_test_btn = gr.Button("🧪 노이즈 테스트 실행", variant="primary")
noise_predict_btn = gr.Button("🔮 노이즈 데이터로 예측", variant="secondary")
with gr.Column(scale=3):
# 노이즈 비교 시각화
noise_comparison_plot = gr.Plot(label="원본 vs 노이즈 데이터 비교")
# 예측 결과 비교 (원본 예측 vs 노이즈 예측)
noise_prediction_plot = gr.Plot(label="예측 결과 비교")
# 로그 및 성능 지표
with gr.Row():
noise_log = gr.Textbox(label="노이즈 테스트 로그", lines=4, interactive=False)
noise_metrics = gr.JSON(label="성능 지표 (정확도 비교)")
# 노이즈 테스트 이벤트 핸들러
noise_test_btn.click(
fn=noise_test_handler,
inputs=[noise_scenario, noise_intensity, noise_test_csv, noise_station],
outputs=[noise_comparison_plot, noise_log, noise_metrics]
)
# 노이즈 예측 이벤트 핸들러
noise_predict_btn.click(
fn=lambda scenario, intensity, csv_file, station_id: noise_prediction_handler(
scenario, intensity, csv_file, station_id, prediction_handler
),
inputs=[noise_scenario, noise_intensity, noise_test_csv, noise_station],
outputs=[noise_prediction_plot, noise_log, noise_metrics]
)
# 이벤트 핸들러 연결
predict_btn.click(
fn=prediction_handler,
inputs=[station_id_input, input_csv],
outputs=[output_plot, output_df, output_log],
api_name="predict"
)
return demo