import asyncio
import httpx
import os
import datetime
import json
import sys
from dotenv import load_dotenv
# Add src to path to import utils
sys.path.append(os.path.join(os.path.dirname(__file__), "src"))
try:
from utils import convert_latlon_to_grid
except ImportError:
# If running from different CWD
sys.path.append(os.path.join(os.path.dirname(__file__), "kma-weather-mcp", "src"))
from utils import convert_latlon_to_grid
load_dotenv()
# Prefer Decoded key for requests
API_KEY = os.getenv("KMA_API_KEY_DECODED") or os.getenv("KMA_API_KEY")
if not API_KEY:
print("Error: No API Key found in .env")
sys.exit(1)
API_BASE_URL = "http://apis.data.go.kr/1360000/VilageFcstInfoService_2.0"
async def get_current_weather(lat, lon):
nx, ny = convert_latlon_to_grid(lat, lon)
# Calculate base_time (approx logic)
now = datetime.datetime.now()
if now.minute < 40:
target = now - datetime.timedelta(hours=1)
else:
target = now
base_date = target.strftime("%Y%m%d")
base_time = target.strftime("%H00")
# Use standard params with DECODED key, but use HTTPS and User-Agent
API_KEY = os.getenv("KMA_API_KEY_DECODED") or os.getenv("KMA_API_KEY")
# HTTPS
url = "https://apis.data.go.kr/1360000/VilageFcstInfoService_2.0/getUltraSrtNcst"
print(f"Requesting weather (HTTPS) for Lat:{lat}, Lon:{lon} (Grid: {nx}, {ny})")
print(f"BaseTime: {base_date} {base_time}")
params = {
"serviceKey": API_KEY,
"pageNo": "1",
"numOfRows": "1000",
"dataType": "JSON",
"base_date": base_date,
"base_time": base_time,
"nx": nx,
"ny": ny
}
headers = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
}
async with httpx.AsyncClient(verify=False) as client: # Verify=False for some gov cert issues
try:
response = await client.get(url, params=params, headers=headers)
if response.status_code != 200:
print(f"Error: HTTP {response.status_code}")
print(response.text)
return
try:
data = response.json()
if data['response']['header']['resultCode'] != '00':
print(f"API Error: {data['response']['header']['resultMsg']}")
return
items = data['response']['body']['items']['item']
weather_map = {
"T1H": "기온",
"RN1": "1시간 강수량",
"REH": "습도",
"PTY": "강수형태",
"VEC": "풍향",
"WSD": "풍속"
}
# PTY Code: 0: 없음, 1: 비, 2: 비/눈, 3: 눈, 5: 빗방울, 6: 빗방울눈날림, 7: 눈날림
pty_map = {
"0": "없음", "1": "비", "2": "비/눈", "3": "눈",
"5": "빗방울", "6": "빗방울+눈날림", "7": "눈날림"
}
results = {}
for item in items:
results[item['category']] = item['obsrValue']
print("\n=== 서울 현재 날씨 ===")
t1h = results.get('T1H', '-')
reh = results.get('REH', '-')
wsd = results.get('WSD', '-')
rn1 = results.get('RN1', '-')
pty = results.get('PTY', '0')
pty_str = pty_map.get(pty, pty)
print(f"🌡️ 기온: {t1h}℃")
print(f"💧 습도: {reh}%")
print(f"🌬️ 풍속: {wsd}m/s")
print(f"☔️ 강수량(1h): {rn1}mm")
print(f"☁️ 강수형태: {pty_str}")
except (KeyError, json.JSONDecodeError) as e:
print("Error parsing JSON response")
print(response.text[:500])
except Exception as e:
print(f"Request failed: {e}")
if __name__ == "__main__":
# Seoul City Hall coordinates
asyncio.run(get_current_weather(37.5665, 126.9780))
