기상청단기예보 ((구)동네예보) 조회서비스
https://www.data.go.kr/data/15084084/openapi.do
import requests
from datetime import datetime
import xmltodict
def get_current_date():
current_date = datetime.now().date()
return current_date.strftime("%Y%m%d")
def get_current_hour():
now = datetime.now()
return datetime.now().strftime("%H%M")
int_to_weather = {
"0": "맑음",
"1": "비",
"2": "비/눈",
"3": "눈",
"5": "빗방울",
"6": "빗방울눈날림",
"7": "눈날림"
}
def forecast(params):
url = 'http://apis.data.go.kr/1360000/VilageFcstInfoService_2.0/getUltraSrtNcst' # 초단기예보
# 값 요청 (웹 브라우저 서버에서 요청 - url주소와 파라미터)
res = requests.get(url, params)
#XML -> 딕셔너리
xml_data = res.text
dict_data = xmltodict.parse(xml_data)
for item in dict_data['response']['body']['items']['item']:
if item['category'] == 'T1H':
temp = item['obsrValue']
# 강수형태: 없음(0), 비(1), 비/눈(2), 눈(3), 빗방울(5), 빗방울눈날림(6), 눈날림(7)
if item['category'] == 'PTY':
sky = item['obsrValue']
sky = int_to_weather[sky]
return temp, sky
공공데이터 포털 회원가입 -> 활용 신청 -> API키 발급 -> 복사
keys = '발급 받은 Key'
params ={'serviceKey' : keys,
'pageNo' : '1',
'numOfRows' : '10',
'dataType' : 'XML',
'base_date' : get_current_date(),
'base_time' : get_current_hour(),
'nx' : '55',
'ny' : '127' }
forecast(params)
회원가입 -> 서비스 신청 Current weather and forecast -> API 키 발급
*회원가입만 하면 임의로 API키가 발급됨
무료 플랜 사용
import requests
import json
city = "Seoul" #도시
apiKey = "발급받은Key"
lang = 'kr' #언어
units = 'metric' #화씨 온도를 섭씨 온도로 변경
api = f"https://api.openweathermap.org/data/2.5/weather?q={city}&APPID={apiKey}&lang={lang}&units={units}"
result = requests.get(api)
result = json.loads(result.text)
print(result['main']['temp'])
print(result['weather'][0]['description'])
# -11.04
# 맑음
import aiohttp
import asyncio
import json
async def fetch_weather(session, city, api_key, lang, units):
api_url = f"https://api.openweathermap.org/data/2.5/weather?q={city}&APPID={api_key}&lang={lang}&units={units}"
async with session.get(api_url) as response:
return await response.text()
def convert_temperature(temp_celsius):
if temp_celsius < 0:
return f"영하 -{int(abs(temp_celsius))}(C)"
else:
return f"{temp_celsius}(C)"
async def forecast():
city = "Seoul"
api_key = "발급받은Key"
lang = 'kr'
units = 'metric'
async with aiohttp.ClientSession() as session:
response_text = await fetch_weather(session, city, api_key, lang, units)
result = json.loads(response_text)
sky = result['weather'][0]['description']
temp = convert_temperature(result['main']['temp'])
return sky, temp
현재 2023-12-22 : 오후 12:02 시간 기준 서울 날씨
import json
import aiohttp
from fastapi import APIRouter, Depends
from fastapi.responses import JSONResponse
from app.config import settings, paths
router = APIRouter(
prefix="/weather",
tags=["weather"],
responses={404: {"description": "Not found"}},
)
async def fetch_weather(session, city, api_key, lang, units):
api_url = (
f"https://api.openweathermap.org/data/2.5/weather?"
f"q={city}&"
f"APPID={api_key}&"
f"lang={lang}&"
f"units={units}"
)
async with session.get(api_url) as response:
return await response.text()
async def forecast():
def convert_temperature(temp_celsius):
if temp_celsius < 0:
return f"영하 -{int(abs(temp_celsius))}(C)"
else:
return f"{temp_celsius}(C)"
city = "Seoul"
api_key = settings.WEATHER_API_KEY
lang = "kr"
units = "metric"
async with aiohttp.ClientSession() as session:
response_text = await fetch_weather(session, city, api_key, lang, units)
result = json.loads(response_text)
sky = result["weather"][0]["description"]
temp = convert_temperature(result["main"]["temp"])
return sky, temp
async def save_curr_weather():
sky, temp = await forecast()
with open(f"{paths.LOGS_DIR}/weather.txt", "w", encoding="utf-8") as file:
file.write(f"{sky}\n{temp}")
@router.post("/current")
async def forecast_endpoint(weather=Depends(forecast)):
return JSONResponse(content={"weather": weather})
https://iridescent-zeal.tistory.com/177
https://dalseobi.tistory.com/130