import streamlit as st import pandas as pd import matplotlib.pyplot as plt import google.generativeai as genai from datetime import datetime, timedelta import time import xml.etree.ElementTree as ET import requests import matplotlib.font_manager as fm import os # 폰트 설정 plt.rcParams['font.family'] = 'DejaVu Sans' plt.rcParams['axes.unicode_minus'] = False # API 키 설정 GEMINI_API_KEY = os.environ.get('GEMINI_API_KEY') WEATHER_API_KEY = os.environ.get('WEATHER_API_KEY') # Gemini 모델 설정 genai.configure(api_key=GEMINI_API_KEY) model = genai.GenerativeModel('gemini-2.0-flash-exp') def get_weather_data(base_date): url = "https://apihub.kma.go.kr/api/typ02/openApi/VilageFcstInfoService_2.0/getVilageFcst" params = { 'pageNo': '1', 'numOfRows': '1000', 'dataType': 'XML', 'base_date': base_date, 'base_time': '0500', 'nx': '59', 'ny': '125', 'authKey': WEATHER_API_KEY } response = requests.get(url, params=params) response.raise_for_status() return response.text def parse_weather_data(xml_data): root = ET.fromstring(xml_data) all_data = [] for item in root.findall('.//item'): all_data.append({ 'base_date': item.find('baseDate').text, 'category': item.find('category').text, 'fcstTime': item.find('fcstTime').text, 'fcstValue': item.find('fcstValue').text, 'fcstDate': item.find('fcstDate').text }) return pd.DataFrame(all_data) def get_precipitation_times(pty_data): if pty_data.empty: return "없음" # 강수가 있는 시간대 찾기 precipitation_times = pty_data[pty_data['fcstValue'] != '0'] if precipitation_times.empty: return "없음" # 연속된 시간대 그룹화 times = precipitation_times['fcst_datetime'].dt.strftime('%H시') return f"{times.iloc[0]}부터 {times.iloc[-1]}까지" def analyze_weather_trends(df, current_time): tomorrow = (current_time + timedelta(days=1)).strftime("%Y%m%d") # 오늘과 내일 날짜의 데이터 분리 today_data = df[df['fcstDate'] == current_time.strftime("%Y%m%d")] tomorrow_data = df[df['fcstDate'] == tomorrow] # 오늘 기온 데이터 처리 temp_data = today_data[today_data['category'] == 'TMP'].copy() if not temp_data.empty: temp_data['fcst_datetime'] = pd.to_datetime(temp_data['fcstDate'] + temp_data['fcstTime'], format='%Y%m%d%H00') closest_time_data = temp_data.iloc[(temp_data['fcst_datetime'] - current_time).abs().argsort()[:1]] current_temp = float(closest_time_data['fcstValue'].iloc[0]) else: current_temp = 0 # 오늘 시간대별 기온 계산 temp_data['hour'] = temp_data['fcstTime'].str[:2].astype(int) morning_temps = temp_data[temp_data['hour'].between(6, 11)]['fcstValue'].astype(float) afternoon_temps = temp_data[temp_data['hour'].between(12, 17)]['fcstValue'].astype(float) evening_temps = temp_data[temp_data['hour'].between(18, 23)]['fcstValue'].astype(float) # 오늘 하늘상태 데이터 sky_data = today_data[today_data['category'] == 'SKY'].copy() if not sky_data.empty: sky_data['fcst_datetime'] = pd.to_datetime(sky_data['fcstDate'] + sky_data['fcstTime'], format='%Y%m%d%H00') morning_sky_data = sky_data[sky_data['fcst_datetime'].dt.hour.between(6, 11)].sort_values(by='fcst_datetime').iloc[0] if len(sky_data[sky_data['fcst_datetime'].dt.hour.between(6, 11)]) > 0 else None afternoon_sky_data = sky_data[sky_data['fcst_datetime'].dt.hour.between(12, 17)].sort_values(by='fcst_datetime').iloc[0] if len(sky_data[sky_data['fcst_datetime'].dt.hour.between(12, 17)]) > 0 else None evening_sky_data = sky_data[sky_data['fcst_datetime'].dt.hour.between(18, 23)].sort_values(by='fcst_datetime').iloc[0] if len(sky_data[sky_data['fcst_datetime'].dt.hour.between(18, 23)]) > 0 else None morning_sky = morning_sky_data['fcstValue'] if morning_sky_data is not None else '1' afternoon_sky = afternoon_sky_data['fcstValue'] if afternoon_sky_data is not None else '1' evening_sky = evening_sky_data['fcstValue'] if evening_sky_data is not None else '1' else: morning_sky = afternoon_sky = evening_sky = '1' # 오늘 강수형태 데이터 pty_data = today_data[today_data['category'] == 'PTY'].copy() if not pty_data.empty: pty_data['fcst_datetime'] = pd.to_datetime(pty_data['fcstDate'] + pty_data['fcstTime'], format='%Y%m%d%H00') current_pty_data = pty_data.iloc[(pty_data['fcst_datetime'] - current_time).abs().argsort()[:1]] current_pty = current_pty_data['fcstValue'].iloc[0] if not current_pty_data.empty else '0' today_precip_times = get_precipitation_times(pty_data) else: current_pty = '0' today_precip_times = "없음" # 내일 데이터 분석 tomorrow_temp_data = tomorrow_data[tomorrow_data['category'] == 'TMP'].copy() tomorrow_sky_data = tomorrow_data[tomorrow_data['category'] == 'SKY'].copy() tomorrow_pty_data = tomorrow_data[tomorrow_data['category'] == 'PTY'].copy() if not tomorrow_pty_data.empty: tomorrow_pty_data['fcst_datetime'] = pd.to_datetime(tomorrow_pty_data['fcstDate'] + tomorrow_pty_data['fcstTime'], format='%Y%m%d%H00') tomorrow_precip_times = get_precipitation_times(tomorrow_pty_data) else: tomorrow_precip_times = "없음" # 내일 시간대별 기온 계산 tomorrow_temp_data['hour'] = tomorrow_temp_data['fcstTime'].str[:2].astype(int) tomorrow_morning_temps = tomorrow_temp_data[tomorrow_temp_data['hour'].between(6, 11)]['fcstValue'].astype(float) tomorrow_afternoon_temps = tomorrow_temp_data[tomorrow_temp_data['hour'].between(12, 17)]['fcstValue'].astype(float) tomorrow_evening_temps = tomorrow_temp_data[tomorrow_temp_data['hour'].between(18, 23)]['fcstValue'].astype(float) # 내일 하늘상태 if not tomorrow_sky_data.empty: tomorrow_sky_data['fcst_datetime'] = pd.to_datetime(tomorrow_sky_data['fcstDate'] + tomorrow_sky_data['fcstTime'], format='%Y%m%d%H00') tomorrow_morning_sky = tomorrow_sky_data[tomorrow_sky_data['fcst_datetime'].dt.hour.between(6, 11)].iloc[0]['fcstValue'] if len(tomorrow_sky_data[tomorrow_sky_data['fcst_datetime'].dt.hour.between(6, 11)]) > 0 else '1' tomorrow_afternoon_sky = tomorrow_sky_data[tomorrow_sky_data['fcst_datetime'].dt.hour.between(12, 17)].iloc[0]['fcstValue'] if len(tomorrow_sky_data[tomorrow_sky_data['fcst_datetime'].dt.hour.between(12, 17)]) > 0 else '1' tomorrow_evening_sky = tomorrow_sky_data[tomorrow_sky_data['fcst_datetime'].dt.hour.between(18, 23)].iloc[0]['fcstValue'] if len(tomorrow_sky_data[tomorrow_sky_data['fcst_datetime'].dt.hour.between(18, 23)]) > 0 else '1' else: tomorrow_morning_sky = tomorrow_afternoon_sky = tomorrow_evening_sky = '1' # 내일 강수상태 if not tomorrow_pty_data.empty: tomorrow_morning_pty = tomorrow_pty_data[tomorrow_pty_data['fcst_datetime'].dt.hour.between(6, 11)].iloc[0]['fcstValue'] if len(tomorrow_pty_data[tomorrow_pty_data['fcst_datetime'].dt.hour.between(6, 11)]) > 0 else '0' tomorrow_afternoon_pty = tomorrow_pty_data[tomorrow_pty_data['fcst_datetime'].dt.hour.between(12, 17)].iloc[0]['fcstValue'] if len(tomorrow_pty_data[tomorrow_pty_data['fcst_datetime'].dt.hour.between(12, 17)]) > 0 else '0' tomorrow_evening_pty = tomorrow_pty_data[tomorrow_pty_data['fcst_datetime'].dt.hour.between(18, 23)].iloc[0]['fcstValue'] if len(tomorrow_pty_data[tomorrow_pty_data['fcst_datetime'].dt.hour.between(18, 23)]) > 0 else '0' else: tomorrow_morning_pty = tomorrow_afternoon_pty = tomorrow_evening_pty = '0' # 날씨 상태 매핑 sky_status = {'1': '맑음', '2': '구름조금', '3': '구름많음', '4': '흐림'} pty_status = {'0': '없음', '1': '비', '2': '비/눈', '3': '눈', '4': '소나기'} prompt = f""" 현재 시각: {current_time.strftime('%Y년 %m월 %d일 %H시 %M분')} 기준 [오늘 날씨 정보] 1. 기온 정보: - 현재 기온: {current_temp}도 - 최고 기온: {max(temp_data['fcstValue'].astype(float))}도 - 최저 기온: {min(temp_data['fcstValue'].astype(float))}도 - 아침 평균 기온: {morning_temps.mean():.1f}도 - 오후 평균 기온: {afternoon_temps.mean():.1f}도 - 저녁 평균 기온: {evening_temps.mean():.1f}도 2. 하늘상태: - 아침: {sky_status.get(morning_sky, '알수없음')} - 오후: {sky_status.get(afternoon_sky, '알수없음')} - 저녁: {sky_status.get(evening_sky, '알수없음')} 3. 강수 상태: {pty_status.get(current_pty, '알수없음')} 강수 시간: {today_precip_times} [내일 날씨 정보] 1. 기온 정보: - 최고 기온: {max(tomorrow_temp_data['fcstValue'].astype(float))}도 - 최저 기온: {min(tomorrow_temp_data['fcstValue'].astype(float))}도 - 아침 평균 기온: {tomorrow_morning_temps.mean():.1f}도 - 오후 평균 기온: {tomorrow_afternoon_temps.mean():.1f}도 - 저녁 평균 기온: {tomorrow_evening_temps.mean():.1f}도 2. 하늘상태: - 아침: {sky_status.get(tomorrow_morning_sky, '알수없음')} - 오후: {sky_status.get(tomorrow_afternoon_sky, '알수없음')} - 저녁: {sky_status.get(tomorrow_evening_sky, '알수없음')} 3. 강수 상태: - 아침: {pty_status.get(tomorrow_morning_pty, '알수없음')} - 오후: {pty_status.get(tomorrow_afternoon_pty, '알수없음')} - 저녁: {pty_status.get(tomorrow_evening_pty, '알수없음')} 강수 시간: {tomorrow_precip_times} 위 정보를 바탕으로 오늘과 내일의 날씨를 총 100자 정도로 전문 기상캐스터처럼 자연스럽게 설명해주세요. "네","설명드리겠습니다."등과 같은 인사말은 제외하고, 날씨정보만 오늘과 내일 날씨를 이어서 설명해주세요. 현재시각 기온을 먼저 알려주고, 오늘과 내일 아침, 오후, 저녁의 최고/최저온도와 눈/비 여부를 반드시 포함하고, 눈이나 비가 오면 몇시부터 몇시까지 오는지 반드시 알려주세요. """ response = model.generate_content(prompt) return response.text def get_current_temperature(temp_data, current_time): # fcstTime을 datetime 객체로 변환 temp_data['fcst_datetime'] = pd.to_datetime(temp_data['fcstDate'] + temp_data['fcstTime'], format='%Y%m%d%H00') # 현재 시간과 가장 가까운 예보 시간 찾기 closest_time_data = temp_data.iloc[(temp_data['fcst_datetime'] - current_time).abs().argsort()[:1]] return float(closest_time_data['fcstValue']) def main(): # 페이지 여백 줄이기 위한 CSS 추가 st.markdown(""" """, unsafe_allow_html=True) while True: try: current_time = datetime.now() + timedelta(hours=8) print(f"[DEBUG] Current time: {current_time}") base_date = current_time.strftime("%Y%m%d") print(f"[DEBUG] Base date: {base_date}") # 05시 이전이면 전날 데이터 사용 if current_time.hour < 5: base_date = (current_time - timedelta(days=1)).strftime("%Y%m%d") else: base_date = current_time.strftime("%Y%m%d") print(f"[DEBUG] Base date for API request: {base_date}") xml_data = get_weather_data(base_date) print(f"[DEBUG] XML data received: {len(xml_data)} bytes") print(f"[DEBUG] XML content: {xml_data[:500]}") # 처음 500자만 출력 # XML 파싱 전에 root 요소 확인 root = ET.fromstring(xml_data) print(f"[DEBUG] XML root tag: {root.tag}") print(f"[DEBUG] XML root children: {[child.tag for child in root]}") df = parse_weather_data(xml_data) print(f"[DEBUG] Parsed DataFrame shape: {df.shape}") print(f"[DEBUG] DataFrame columns: {df.columns}") # 온도 데이터 처리 temp_data = df[df['category'] == 'TMP'].copy() print(f"[DEBUG] Temperature data shape: {temp_data.shape}") temp_data['datetime'] = pd.to_datetime(temp_data['fcstDate'] + temp_data['fcstTime'], format='%Y%m%d%H%M') print(f"[DEBUG] Datetime range: {temp_data['datetime'].min()} to {temp_data['datetime'].max()}") # 현재 기온 계산 current_temp = get_current_temperature(temp_data, current_time) print(f"[DEBUG] Current temperature: {current_temp}") # 날씨 예보 생성 weather_forecast = analyze_weather_trends(df, current_time) print(f"[DEBUG] Weather forecast generated: {len(weather_forecast)} characters") # 내일 아침 기온 계산 tomorrow_data = df[df['fcstDate'] == (current_time + timedelta(days=1)).strftime("%Y%m%d")] tomorrow_temp_data = tomorrow_data[tomorrow_data['category'] == 'TMP'].copy() tomorrow_temp_data['hour'] = tomorrow_temp_data['fcstTime'].str[:2].astype(int) tomorrow_morning_temps = tomorrow_temp_data[tomorrow_temp_data['hour'].between(6, 11)]['fcstValue'].astype(float) tomorrow_morning_temp = tomorrow_temp_data[tomorrow_temp_data['hour'] == 6]['fcstValue'].iloc[0] if not tomorrow_temp_data[tomorrow_temp_data['hour'] == 6].empty else 0 st.markdown(f"""실시간 날씨 대시보드 (현재온도: = 0 else 'blue'}'>{current_temp:.1f}도, 내일 아침 6시 온도: = 0 else 'blue'}'>{float(tomorrow_morning_temp):.1f}도)""", unsafe_allow_html=True) # 그래프 그리기 fig, ax = plt.subplots(figsize=(25, 10)) # 색상 팔레트 설정 colors = ['#FF0000', '#00AA00', '#0000FF', '#FFA500', '#800080'] # 빨강, 초록, 파랑, 주황, 보라 for idx, date in enumerate(temp_data['fcstDate'].unique()): date_data = temp_data[temp_data['fcstDate'] == date] temps = date_data['fcstValue'].astype(float) times = date_data['datetime'] color = colors[idx % len(colors)] # 기본 온도 선 그래프 - 선 스타일 개선 line = ax.plot(times, temps, marker='o', label=f'{date[:4]}-{date[4:6]}-{date[6:]}', markersize=12, # 마커 크기 증가 linewidth=5, # 선 굵기 증가 color=color, markeredgewidth=2.5, markerfacecolor=color, markeredgecolor='white', linestyle='-', alpha=0.8) # 약간의 투명도 추가 # 최고/최저 온도 찾기 max_temp = temps.max() min_temp = temps.min() max_temp_time = times[temps.idxmax()] min_temp_time = times[temps.idxmin()] # 최고 온도 표시 - 더 눈에 띄게 수정 ax.annotate(f'Max: {max_temp}°C', xy=(max_temp_time, max_temp), xytext=(10, 15), textcoords='offset points', ha='left', va='bottom', bbox=dict(boxstyle='round,pad=0.5', fc='yellow', alpha=0.7, edgecolor=color), fontsize=27, fontweight='bold', color=color) # 최저 온도 표시 - 더 눈에 띄게 수정 ax.annotate(f'Min: {min_temp}°C', xy=(min_temp_time, min_temp), xytext=(10, -15), textcoords='offset points', ha='left', va='top', bbox=dict(boxstyle='round,pad=0.5', fc='lightblue', alpha=0.7, edgecolor=color), fontsize=27, fontweight='bold', color=color) # 그래프 스타일 개선 ax.set_title('', fontsize=18, pad=20, fontweight='bold') ax.set_xlabel('날짜 및 시간', fontsize=16, fontweight='bold') ax.set_ylabel('기온 (°C)', fontsize=16, fontweight='bold') ax.tick_params(axis='both', labelsize=14) ax.grid(True, linestyle='--', alpha=0.4) # 밤 시간대 음영 처리 dates = temp_data['fcstDate'].unique() for date in dates: evening = pd.to_datetime(f"{date[:4]}-{date[4:6]}-{date[6:]} 18:00:00") next_morning = evening + timedelta(hours=12) ax.axvspan(evening, next_morning, alpha=0.2, color='gray', label='_nolegend_') # 날짜 변경선 추가 dates = temp_data['fcstDate'].unique() for date in dates[1:]: midnight = pd.to_datetime(f"{date[:4]}-{date[4:6]}-{date[6:]} 00:00:00") ax.axvline(x=midnight, color='black', linestyle=':', linewidth=2, alpha=0.5, label='_nolegend_') ax.annotate(f'{date[4:6]}/{date[6:]}', xy=(midnight, ax.get_ylim()[1]), xytext=(0, 10), textcoords='offset points', ha='center', fontsize=25, fontweight='bold', color='black') ax.set_facecolor('#f8f9fa') min_time = temp_data['datetime'].min() max_time = temp_data['datetime'].max() # 현재 시각 세로선 추가 current_datetime = pd.to_datetime(current_time) if min_time <= current_datetime <= max_time: ax.axvline(x=current_datetime, color='red', linestyle='--', linewidth=3, alpha=0.7) ax.annotate('Now', xy=(current_datetime, ax.get_ylim()[1]), xytext=(0, 10), textcoords='offset points', ha='center', va='bottom', bbox=dict(boxstyle='round,pad=0.5', fc='white', ec='red', alpha=0.8), fontsize=25, fontweight='bold', color='red') ax.set_xlim(min_time, max_time) # 범례 스타일 설정 개선 legend = ax.legend(fontsize=25, frameon=True, facecolor='white', edgecolor='gray', loc='upper right', bbox_to_anchor=(1.13, 1.0)) for text in legend.get_texts(): text.set_fontweight('bold') plt.xticks(rotation=45) plt.tight_layout() st.pyplot(fig, use_container_width=True) plt.close() # 날씨 예보 텍스트를 스크롤 효과와 함께 표시 st.markdown( '' '
' f'
' f' {weather_forecast.replace(chr(10), " ")}' f' {weather_forecast.replace(chr(10), " ")}' # 동일한 텍스트 한번 더 추가 '
' '
', unsafe_allow_html=True ) time.sleep(300) st.rerun() except Exception as e: st.error(f"오류 발생: {str(e)}") time.sleep(300) if __name__ == "__main__": main()