Files
simva-home/backend/app/weather.py
T

196 lines
8.8 KiB
Python

import time
import math
import datetime
from typing import Dict, Any, Optional
import httpx
# 주요 한국 도시 좌표 목록
KOREAN_CITIES = [
{"name": "서울", "lat": 37.5665, "lon": 126.9780},
{"name": "인천", "lat": 37.4563, "lon": 126.7052},
{"name": "수원", "lat": 37.2636, "lon": 127.0286},
{"name": "춘천", "lat": 37.8813, "lon": 127.7298},
{"name": "강릉", "lat": 37.7519, "lon": 128.8761},
{"name": "대전", "lat": 36.3504, "lon": 127.3845},
{"name": "세종", "lat": 36.4800, "lon": 127.2890},
{"name": "청주", "lat": 36.6424, "lon": 127.4890},
{"name": "대구", "lat": 35.8714, "lon": 128.6014},
{"name": "포항", "lat": 36.0190, "lon": 129.3435},
{"name": "전주", "lat": 35.8242, "lon": 127.1480},
{"name": "광주", "lat": 35.1595, "lon": 126.8526},
{"name": "부산", "lat": 35.1796, "lon": 129.0756},
{"name": "울산", "lat": 35.5384, "lon": 129.3114},
{"name": "창원", "lat": 35.2280, "lon": 128.6811},
{"name": "제주", "lat": 33.4996, "lon": 126.5312},
]
def find_nearest_city(lat: float, lon: float) -> str:
closest = min(
KOREAN_CITIES,
key=lambda c: (c["lat"] - lat) ** 2 + (c["lon"] - lon) ** 2
)
dist_sq = (closest["lat"] - lat) ** 2 + (closest["lon"] - lon) ** 2
# 약 50km 이내이면 해당 도시명 반환
if dist_sq < 0.25:
return closest["name"]
return "현재 위치"
# WMO Weather interpretation code (WW)
# https://open-meteo.com/en/docs
WMO_MAP = {
0: ("맑음", "mdi-weather-sunny", "mdi-weather-night", "amber-darken-1", "indigo-lighten-2"),
1: ("대체로 맑음", "mdi-weather-partly-cloudy", "mdi-weather-night-partly-cloudy", "amber-darken-2", "indigo-lighten-3"),
2: ("구름 조금", "mdi-weather-partly-cloudy", "mdi-weather-partly-cloudy", "blue-grey-lighten-1", "blue-grey-lighten-1"),
3: ("흐림", "mdi-weather-cloudy", "mdi-weather-cloudy", "grey-darken-1", "grey-darken-1"),
45: ("안개", "mdi-weather-fog", "mdi-weather-fog", "blue-grey", "blue-grey"),
48: ("서리 안개", "mdi-weather-fog", "mdi-weather-fog", "blue-grey", "blue-grey"),
51: ("약한 이슬비", "mdi-weather-rainy", "mdi-weather-rainy", "light-blue", "light-blue"),
53: ("이슬비", "mdi-weather-rainy", "mdi-weather-rainy", "light-blue-darken-1", "light-blue-darken-1"),
55: ("강한 이슬비", "mdi-weather-rainy", "mdi-weather-rainy", "blue-darken-1", "blue-darken-1"),
56: ("약한 어는 이슬비", "mdi-weather-snowy-rainy", "mdi-weather-snowy-rainy", "cyan", "cyan"),
57: ("강한 어는 이슬비", "mdi-weather-snowy-rainy", "mdi-weather-snowy-rainy", "cyan-darken-1", "cyan-darken-1"),
61: ("약한 비", "mdi-weather-rainy", "mdi-weather-rainy", "blue", "blue"),
63: ("비", "mdi-weather-rainy", "mdi-weather-rainy", "blue-darken-1", "blue-darken-1"),
65: ("강한 비", "mdi-weather-pouring", "mdi-weather-pouring", "blue-darken-3", "blue-darken-3"),
66: ("약한 어는 비", "mdi-weather-snowy-rainy", "mdi-weather-snowy-rainy", "cyan", "cyan"),
67: ("강한 어는 비", "mdi-weather-snowy-rainy", "mdi-weather-snowy-rainy", "cyan-darken-2", "cyan-darken-2"),
71: ("약한 눈", "mdi-weather-snowy", "mdi-weather-snowy", "light-blue-lighten-3", "light-blue-lighten-3"),
73: ("눈", "mdi-weather-snowy", "mdi-weather-snowy", "light-blue-lighten-2", "light-blue-lighten-2"),
75: ("강한 눈", "mdi-weather-snowy-heavy", "mdi-weather-snowy-heavy", "blue-lighten-4", "blue-lighten-4"),
77: ("싸락눈", "mdi-weather-snowy", "mdi-weather-snowy", "light-blue-lighten-1", "light-blue-lighten-1"),
80: ("약한 소나기", "mdi-weather-partly-rainy", "mdi-weather-partly-rainy", "blue", "blue"),
81: ("소나기", "mdi-weather-partly-rainy", "mdi-weather-partly-rainy", "blue-darken-1", "blue-darken-1"),
82: ("강한 소나기", "mdi-weather-pouring", "mdi-weather-pouring", "blue-darken-2", "blue-darken-2"),
85: ("약한 눈 소나기", "mdi-weather-snowy", "mdi-weather-snowy", "light-blue", "light-blue"),
86: ("강한 눈 소나기", "mdi-weather-snowy-heavy", "mdi-weather-snowy-heavy", "blue-lighten-3", "blue-lighten-3"),
95: ("뇌우", "mdi-weather-lightning", "mdi-weather-lightning", "deep-purple", "deep-purple"),
96: ("뇌우 및 우박", "mdi-weather-lightning-rainy", "mdi-weather-lightning-rainy", "deep-purple-darken-2", "deep-purple-darken-2"),
99: ("강한 뇌우 및 우박", "mdi-weather-lightning-rainy", "mdi-weather-lightning-rainy", "deep-purple-darken-3", "deep-purple-darken-3"),
}
def resolve_weather_style(code: int, is_day: int = 1):
info = WMO_MAP.get(code)
if not info:
return {
"text": "흐림",
"icon": "mdi-weather-cloudy",
"color": "grey"
}
text, day_icon, night_icon, day_color, night_color = info
return {
"text": text,
"icon": day_icon if is_day else night_icon,
"color": day_color if is_day else night_color,
}
# 캐시 저장소: (lat, lon) -> (timestamp, data)
_cache: Dict[str, Any] = {}
CACHE_TTL = 900 # 15분
KOREAN_WEEKDAYS = ["월", "화", "수", "목", "금", "토", "일"]
async def get_weather_forecast(lat: float = 37.5665, lon: float = 126.9780) -> Dict[str, Any]:
cache_key = f"{round(lat, 2)}_{round(lon, 2)}"
now = time.time()
if cache_key in _cache:
cached_time, cached_data = _cache[cache_key]
if now - cached_time < CACHE_TTL:
return cached_data
url = (
f"https://api.open-meteo.com/v1/forecast"
f"?latitude={lat}&longitude={lon}"
f"&current=temperature_2m,relative_humidity_2m,apparent_temperature,is_day,precipitation,weather_code,wind_speed_10m"
f"&daily=weather_code,temperature_2m_max,temperature_2m_min,precipitation_probability_max"
f"&timezone=auto"
)
try:
async with httpx.AsyncClient(timeout=10.0) as client:
resp = await client.get(url)
resp.raise_for_status()
data = resp.json()
except Exception as e:
# 캐시가 만료되었더라도 기존 캐시가 있으면 반환
if cache_key in _cache:
return _cache[cache_key][1]
raise e
current_raw = data.get("current", {})
daily_raw = data.get("daily", {})
is_day = current_raw.get("is_day", 1)
current_code = current_raw.get("weather_code", 0)
current_style = resolve_weather_style(current_code, is_day)
current_formatted = {
"temperature": round(current_raw.get("temperature_2m", 0.0), 1),
"apparent_temperature": round(current_raw.get("apparent_temperature", 0.0), 1),
"humidity": current_raw.get("relative_humidity_2m", 0),
"wind_speed": round(current_raw.get("wind_speed_10m", 0.0), 1),
"precipitation": current_raw.get("precipitation", 0.0),
"weather_code": current_code,
"weather_text": current_style["text"],
"icon": current_style["icon"],
"color": current_style["color"],
"is_day": bool(is_day),
"time": current_raw.get("time"),
}
daily_list = []
dates = daily_raw.get("time", [])
weather_codes = daily_raw.get("weather_code", [])
temps_max = daily_raw.get("temperature_2m_max", [])
temps_min = daily_raw.get("temperature_2m_min", [])
precip_probs = daily_raw.get("precipitation_probability_max", [])
for i in range(len(dates)):
date_str = dates[i]
try:
dt = datetime.date.fromisoformat(date_str)
weekday_kr = KOREAN_WEEKDAYS[dt.weekday()]
if i == 0:
day_label = "오늘"
elif i == 1:
day_label = "내일"
else:
day_label = f"{weekday_kr}요일"
formatted_date = f"{dt.month}.{dt.day} ({weekday_kr})"
except Exception:
day_label = date_str
formatted_date = date_str
code = weather_codes[i] if i < len(weather_codes) else 0
style = resolve_weather_style(code, 1)
daily_list.append({
"date": date_str,
"day_label": day_label,
"formatted_date": formatted_date,
"weather_code": code,
"weather_text": style["text"],
"icon": style["icon"],
"color": style["color"],
"temp_max": round(temps_max[i], 1) if i < len(temps_max) else 0.0,
"temp_min": round(temps_min[i], 1) if i < len(temps_min) else 0.0,
"precipitation_prob": precip_probs[i] if i < len(precip_probs) else 0,
})
city_name = find_nearest_city(lat, lon)
result = {
"location": {
"name": city_name,
"lat": lat,
"lon": lon,
},
"current": current_formatted,
"daily": daily_list,
"updated_at": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
}
_cache[cache_key] = (now, result)
return result