firstgarden/static
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

초단기, 단기, 중기예보를 가져오기

Browse Source
This commit is contained in:
KWON
2025-07-21 17:39:08 +09:00
parent d831f62426
commit 3967bc8264
1 changed files with 302 additions and 223 deletions
Download Patch File Download Diff File Expand all files Collapse all files

165
lib/weather_forecast.py
View File

@ -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,6 +57,7 @@ def get_latest_base_date_time(now=None):
return base_date, candidate
def get_daily_ultra_forecast(serviceKey):
def request():
base_date, base_time = get_latest_base_date_time()
url = "http://apis.data.go.kr/1360000/VilageFcstInfoService_2.0/getUltraSrtFcst"
params = {
@ -78,10 +106,19 @@ def get_daily_ultra_forecast(serviceKey):
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}
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('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 = {
@ -135,19 +172,19 @@ def get_daily_vilage_forecast(serviceKey):
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
'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"
# 발표 시각 계산: 06시 또는 18시만 존재
now = datetime.now()
if now.hour < 6:
tmFc = (now - timedelta(days=1)).strftime("%Y%m%d") + "1800"
@ -170,18 +207,14 @@ def get_midterm_forecast(serviceKey, regId='11B20305'):
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] # 실제 예보 데이터
return {}
item = items[0]
except Exception as e:
print(f"[ERROR] 중기예보 호출 실패: {e}")
return {}, {}
return {}
# 3~10일 후 강수확률 추출
precip_probs = {}
for day in range(3, 11):
key = f'rnSt{day}'
@ -190,12 +223,13 @@ def get_midterm_forecast(serviceKey, regId='11B20305'):
except:
precip_probs[day] = 0
return precip_probs, item
return precip_probs
def get_midterm_temperature_forecast(serviceKey, regId='11B20305'): # 파주 코드
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"
# 발표시각은 06:00 또는 18:00
now = datetime.now()
if now.hour < 6:
tmFc = (now - timedelta(days=1)).strftime("%Y%m%d") + "1800"
@ -217,15 +251,12 @@ def get_midterm_temperature_forecast(serviceKey, regId='11B20305'): # 파주
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 {}
@ -234,16 +265,21 @@ def get_midterm_temperature_forecast(serviceKey, regId='11B20305'): # 파주
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(item.get(min_key, 0)),
'max': int(item.get(max_key, 0))
'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:
temps[day] = {'min': 0, 'max': 0}
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
today = date.today()
@ -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))