From 3967bc8264ab5f95bc1c17ed50758959acb34347 Mon Sep 17 00:00:00 2001 From: KWON Date: Mon, 21 Jul 2025 17:39:08 +0900 Subject: [PATCH] =?UTF-8?q?=EC=B4=88=EB=8B=A8=EA=B8=B0,=20=EB=8B=A8?= =?UTF-8?q?=EA=B8=B0,=20=EC=A4=91=EA=B8=B0=EC=98=88=EB=B3=B4=EB=A5=BC=20?= =?UTF-8?q?=EA=B0=80=EC=A0=B8=EC=98=A4=EA=B8=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- lib/weather_forecast.py | 525 +++++++++++++++++++++++----------------- 1 file changed, 302 insertions(+), 223 deletions(-) diff --git a/lib/weather_forecast.py b/lib/weather_forecast.py index 200c3c3..f4ac5bd 100644 --- a/lib/weather_forecast.py +++ b/lib/weather_forecast.py @@ -1,6 +1,12 @@ import requests +import os +import json from datetime import datetime, timedelta +def valid_until_hours(cached, hours=2): + ts = datetime.fromisoformat(cached['ts']) + return datetime.now() - ts < timedelta(hours=hours) + def parse_precip(value): if value == '강수없음': return 0.0 @@ -12,6 +18,27 @@ def parse_precip(value): except: return 0.0 +def ensure_cache_dir(): + cache_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'data', 'cache')) + os.makedirs(cache_dir, exist_ok=True) + return cache_dir + +def get_cache_or_request(name, valid_until_fn, request_fn): + cache_dir = ensure_cache_dir() + today = datetime.now().strftime("%Y%m%d") + cache_file = os.path.join(cache_dir, f"{name}_{today}.json") + + if os.path.exists(cache_file): + with open(cache_file, 'r', encoding='utf-8') as f: + cached = json.load(f) + if valid_until_fn(cached): + return cached['data'] + + data = request_fn() + with open(cache_file, 'w', encoding='utf-8') as f: + json.dump({'ts': datetime.now().isoformat(), 'data': data}, f, ensure_ascii=False) + return data + def get_latest_base_date_time(now=None): if now is None: now = datetime.now() @@ -30,219 +57,228 @@ def get_latest_base_date_time(now=None): 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 + def request(): + 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' } - 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}") + 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 {} - item = items[0] + 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 - 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)) + 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': round(sumRn, 1), + 'minTa': round(minTa, 1), + 'maxTa': round(maxTa, 1), + 'avgRhm': round(avgRhm, 1) } - except: - temps[day] = {'min': 0, 'max': 0} - return temps + return result + + return get_cache_or_request('ultra_forecast', lambda cached: valid_until_hours(cached, 2), request) + +def get_daily_vilage_forecast(serviceKey): + def request(): + 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': round(sumRn, 1), + 'minTa': round(minTa, 1), + 'maxTa': round(maxTa, 1), + 'avgRhm': round(avgRhm, 1) + } + + return result + + return get_cache_or_request('vilage_forecast', lambda cached: valid_until_hours(cached, 6), request) + +def get_midterm_forecast(serviceKey, regId='11B20305'): + def request(): + url = "http://apis.data.go.kr/1360000/MidFcstInfoService/getMidLandFcst" + 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 {} + + 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 + + return get_cache_or_request('midterm_precip', lambda cached: valid_until_hours(cached, 12), request) + +def get_midterm_temperature_forecast(serviceKey, regId='11B20305'): + def request(): + url = "http://apis.data.go.kr/1360000/MidFcstInfoService/getMidTa" + 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}' + min_val = item.get(min_key) + max_val = item.get(max_key) + try: + temps[day] = { + 'min': int(min_val) if min_val is not None else None, + 'max': int(max_val) if max_val is not None else None + } + except Exception: + temps[day] = {'min': None, 'max': None} + + + return temps + + return get_cache_or_request('midterm_temp', lambda cached: valid_until_hours(cached, 12), request) def get_weekly_precip(serviceKey): from datetime import date @@ -251,7 +287,7 @@ def get_weekly_precip(serviceKey): ultra = get_daily_ultra_forecast(serviceKey) short = get_daily_vilage_forecast(serviceKey) - mid_precip, _ = get_midterm_forecast(serviceKey) + mid_precip = get_midterm_forecast(serviceKey) mid_temp = get_midterm_temperature_forecast(serviceKey) results = {} @@ -267,34 +303,77 @@ def get_weekly_precip(serviceKey): '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'] + day_offset = (dt - today).days # 0부터 시작 + if day_offset >= 3: + # 중기예보 강수 우선 적용 + if day_offset in mid_precip: + mid_rain = float(mid_precip[day_offset]) / 100 * 5.0 + if mid_rain > results[dt_str]['sumRn']: + results[dt_str]['sumRn'] = mid_rain + + # 중기예보 기온 적용: 단, None이거나 0이면 단기예보로 대체 + key = str(day_offset) + if key in mid_temp: + mid_min = mid_temp[key]['min'] + mid_max = mid_temp[key]['max'] + + if mid_min not in (None, 0): + results[dt_str]['minTa'] = mid_min + elif dt_str in short and short[dt_str]['minTa'] != 0: + results[dt_str]['minTa'] = short[dt_str]['minTa'] + + if mid_max not in (None, 0): + results[dt_str]['maxTa'] = mid_max + elif dt_str in short and short[dt_str]['maxTa'] != 0: + results[dt_str]['maxTa'] = short[dt_str]['maxTa'] + + # 중기 기온 적용 이후, 습도 보완 + if results[dt_str]['avgRhm'] == 0 and dt_str in short and short[dt_str]['avgRhm'] != 0: + results[dt_str]['avgRhm'] = short[dt_str]['avgRhm'] + + results[dt_str] = { + 'sumRn': round(results[dt_str]['sumRn'], 1), + 'minTa': round(results[dt_str]['minTa'], 1), + 'maxTa': round(results[dt_str]['maxTa'], 1), + 'avgRhm': round(results[dt_str]['avgRhm'], 1), + } return results + + +def print_weekly_precip_table(data_dict): + # 헤더 출력 + header = f"{'날짜':<10} {'강수량(mm)':>10} {'최저기온(℃)':>12} {'최고기온(℃)':>12} {'평균습도(%)':>12}" + print(header) + print('-' * len(header)) + + # 날짜 순서대로 출력 + for dt in sorted(data_dict.keys()): + vals = data_dict[dt] + print(f"{dt:<10} {vals['sumRn']:10.1f} {vals['minTa']:12.1f} {vals['maxTa']:12.1f} {vals['avgRhm']:12.1f}") + + + if __name__ == '__main__': - serviceKey = "mHrZoSnzVc+2S4dpCe3A1CgI9cAu1BRttqRdoEy9RGbnKAKyQT4sqcESDqqY3grgBGQMuLeEgWIS3Qxi8rcDVA==" + import os, sys + sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) + from lib.common import load_config + serviceKey = load_config()['DATA_API']['serviceKey'] + + data = get_weekly_precip(serviceKey) print(get_weekly_precip(serviceKey)) + print_weekly_precip_table(data) + + print(get_daily_vilage_forecast(serviceKey)) + print(get_midterm_temperature_forecast(serviceKey)) \ No newline at end of file