diff --git a/lib/weather_forecast.py b/lib/weather_forecast.py new file mode 100644 index 0000000..200c3c3 --- /dev/null +++ b/lib/weather_forecast.py @@ -0,0 +1,300 @@ +import requests +from datetime import datetime, timedelta + +def parse_precip(value): + if value == '강수없음': + return 0.0 + elif '1mm 미만' in str(value): + return 0.5 + else: + try: + return float(value) + except: + return 0.0 + +def get_latest_base_date_time(now=None): + if now is None: + now = datetime.now() + base_times = ["2330", "0230", "0530", "0830", "1130", "1430", "1730", "2030"] + candidate = None + for bt in reversed(base_times): + hour = int(bt[:2]) + minute = int(bt[2:]) + if (now.hour > hour) or (now.hour == hour and now.minute >= minute): + candidate = bt + break + if candidate is None: + candidate = "2330" + now -= timedelta(days=1) + base_date = now.strftime("%Y%m%d") + return base_date, candidate + +def get_daily_ultra_forecast(serviceKey): + base_date, base_time = get_latest_base_date_time() + url = "http://apis.data.go.kr/1360000/VilageFcstInfoService_2.0/getUltraSrtFcst" + params = { + 'serviceKey': serviceKey, + 'numOfRows': '1000', + 'pageNo': '1', + 'dataType': 'JSON', + 'base_date': base_date, + 'base_time': base_time, + 'nx': '57', + 'ny': '130' + } + try: + resp = requests.get(url, params=params, timeout=10) + resp.raise_for_status() + items = resp.json()['response']['body']['items']['item'] + except Exception as e: + print(f"[ERROR] 초단기예보 호출 실패: {e}") + return {} + + daily_data = {} + for item in items: + dt = item['fcstDate'] + cat = item['category'] + val = item['fcstValue'] + if dt not in daily_data: + daily_data[dt] = {'sumRn': 0, 'minTa': [], 'maxTa': [], 'rhm': []} + if cat == 'RN1': + daily_data[dt]['sumRn'] += parse_precip(val) + elif cat == 'T3H': + try: + t = float(val) + daily_data[dt]['minTa'].append(t) + daily_data[dt]['maxTa'].append(t) + except: + pass + elif cat == 'REH': + try: + daily_data[dt]['rhm'].append(float(val)) + except: + pass + + result = {} + for dt, vals in daily_data.items(): + minTa = min(vals['minTa']) if vals['minTa'] else 0 + maxTa = max(vals['maxTa']) if vals['maxTa'] else 0 + avgRhm = sum(vals['rhm']) / len(vals['rhm']) if vals['rhm'] else 0 + sumRn = round(vals['sumRn'], 2) + result[dt] = {'sumRn': sumRn, 'minTa': minTa, 'maxTa': maxTa, 'avgRhm': avgRhm} + return result + +def get_daily_vilage_forecast(serviceKey): + base_date, _ = get_latest_base_date_time() + url = "http://apis.data.go.kr/1360000/VilageFcstInfoService_2.0/getVilageFcst" + params = { + 'serviceKey': serviceKey, + 'numOfRows': '1000', + 'pageNo': '1', + 'dataType': 'JSON', + 'base_date': base_date, + 'base_time': '0200', + 'nx': '57', + 'ny': '130' + } + try: + resp = requests.get(url, params=params, timeout=10) + resp.raise_for_status() + items = resp.json()['response']['body']['items']['item'] + except Exception as e: + print(f"[ERROR] 단기예보 호출 실패: {e}") + return {} + + daily_data = {} + for item in items: + dt = item['fcstDate'] + cat = item['category'] + val = item['fcstValue'] + if dt not in daily_data: + daily_data[dt] = {'sumRn': 0, 'minTa': [], 'maxTa': [], 'rhm': []} + + if cat == 'RN1': + daily_data[dt]['sumRn'] += parse_precip(val) + elif cat == 'TMN': + try: + daily_data[dt]['minTa'].append(float(val)) + except: + pass + elif cat == 'TMX': + try: + daily_data[dt]['maxTa'].append(float(val)) + except: + pass + elif cat == 'REH': + try: + daily_data[dt]['rhm'].append(float(val)) + except: + pass + + result = {} + for dt, vals in daily_data.items(): + minTa = min(vals['minTa']) if vals['minTa'] else 0 + maxTa = max(vals['maxTa']) if vals['maxTa'] else 0 + avgRhm = sum(vals['rhm']) / len(vals['rhm']) if vals['rhm'] else 0 + sumRn = round(vals['sumRn'], 2) + result[dt] = { + 'sumRn': sumRn, + 'minTa': minTa, + 'maxTa': maxTa, + 'avgRhm': avgRhm + } + return result + + +def get_midterm_forecast(serviceKey, regId='11B20305'): + # 중기 강수확률 예보 + url = "http://apis.data.go.kr/1360000/MidFcstInfoService/getMidLandFcst" + + # 발표 시각 계산: 06시 또는 18시만 존재 + now = datetime.now() + if now.hour < 6: + tmFc = (now - timedelta(days=1)).strftime("%Y%m%d") + "1800" + elif now.hour < 18: + tmFc = now.strftime("%Y%m%d") + "0600" + else: + tmFc = now.strftime("%Y%m%d") + "1800" + + params = { + 'serviceKey': serviceKey, + 'regId': regId, + 'tmFc': tmFc, + 'numOfRows': 10, + 'pageNo': 1, + 'dataType': 'JSON', + } + + try: + resp = requests.get(url, params=params, timeout=10) + resp.raise_for_status() + data = resp.json() + items = data.get('response', {}).get('body', {}).get('items', {}).get('item', []) + + if not items: + print(f"[ERROR] 중기예보 응답 item 없음. tmFc={tmFc}, regId={regId}") + return {}, {} + + item = items[0] # 실제 예보 데이터 + + except Exception as e: + print(f"[ERROR] 중기예보 호출 실패: {e}") + return {}, {} + + # 3~10일 후 강수확률 추출 + precip_probs = {} + for day in range(3, 11): + key = f'rnSt{day}' + try: + precip_probs[day] = int(item.get(key, 0)) + except: + precip_probs[day] = 0 + + return precip_probs, item + +def get_midterm_temperature_forecast(serviceKey, regId='11B20305'): # 파주 코드 + url = "http://apis.data.go.kr/1360000/MidFcstInfoService/getMidTa" + + # 발표시각은 06:00 또는 18:00 + now = datetime.now() + if now.hour < 6: + tmFc = (now - timedelta(days=1)).strftime("%Y%m%d") + "1800" + elif now.hour < 18: + tmFc = now.strftime("%Y%m%d") + "0600" + else: + tmFc = now.strftime("%Y%m%d") + "1800" + + params = { + 'serviceKey': serviceKey, + 'regId': regId, + 'tmFc': tmFc, + 'pageNo': '1', + 'numOfRows': '10', + 'dataType': 'JSON' + } + + try: + resp = requests.get(url, params=params, timeout=10) + resp.raise_for_status() + data = resp.json() + + # 응답 검증 + items = data.get("response", {}).get("body", {}).get("items", {}).get("item", []) + if not items: + print(f"[ERROR] 응답에 item 없음. tmFc={tmFc}, regId={regId}") + return {} + + item = items[0] + + except Exception as e: + print(f"[ERROR] 중기기온예보 호출 실패: {e}") + return {} + + temps = {} + for day in range(3, 11): + min_key = f'taMin{day}' + max_key = f'taMax{day}' + try: + temps[day] = { + 'min': int(item.get(min_key, 0)), + 'max': int(item.get(max_key, 0)) + } + except: + temps[day] = {'min': 0, 'max': 0} + + return temps + +def get_weekly_precip(serviceKey): + from datetime import date + today = date.today() + sunday = today + timedelta(days=(6 - today.weekday())) + + ultra = get_daily_ultra_forecast(serviceKey) + short = get_daily_vilage_forecast(serviceKey) + mid_precip, _ = get_midterm_forecast(serviceKey) + mid_temp = get_midterm_temperature_forecast(serviceKey) + + results = {} + + for i in range((sunday - today).days + 1): + dt = today + timedelta(days=i) + dt_str = dt.strftime("%Y%m%d") + + results[dt_str] = { + 'sumRn': 0, + 'minTa': 0, + 'maxTa': 0, + 'avgRhm': 0 + } + + # 강수량과 습도는 초단기예보 우선 반영 + if dt_str in ultra: + results[dt_str]['sumRn'] = ultra[dt_str]['sumRn'] + results[dt_str]['avgRhm'] = ultra[dt_str]['avgRhm'] + + # 최고/최저기온은 단기예보로만 덮어쓰기 (0이 아니면 덮어쓰기) + if dt_str in short: + if short[dt_str]['minTa'] != 0: + results[dt_str]['minTa'] = short[dt_str]['minTa'] + if short[dt_str]['maxTa'] != 0: + results[dt_str]['maxTa'] = short[dt_str]['maxTa'] + + # 중기예보 보정 (3일 이후부터) + day_idx = (dt - today).days + 1 + if day_idx >= 3: + if day_idx in mid_precip: + mid_rain = mid_precip[day_idx] / 100 * 5.0 + if results[dt_str]['sumRn'] < mid_rain: + results[dt_str]['sumRn'] = mid_rain + if day_idx in mid_temp: + # 단기예보로 이미 값이 있으면 건너뛰기 + if results[dt_str]['minTa'] == 0: + results[dt_str]['minTa'] = mid_temp[day_idx]['min'] + if results[dt_str]['maxTa'] == 0: + results[dt_str]['maxTa'] = mid_temp[day_idx]['max'] + + return results + +if __name__ == '__main__': + serviceKey = "mHrZoSnzVc+2S4dpCe3A1CgI9cAu1BRttqRdoEy9RGbnKAKyQT4sqcESDqqY3grgBGQMuLeEgWIS3Qxi8rcDVA==" + print(get_weekly_precip(serviceKey))