"""
OpenMeteo API連携による天気情報取得サービス
"""
import asyncio
from datetime import datetime, timedelta
from typing import Dict, List, Any, Optional
import httpx
from pydantic import BaseModel
class WeatherData(BaseModel):
date: str
weather: str
temperature: float
humidity: int
precipitation_probability: int
class WeatherService:
"""OpenMeteo APIを使用した天気情報取得サービス"""
def __init__(self):
self.base_url = "https://api.open-meteo.com/v1/forecast"
self.timeout = 30.0
async def get_weather(self, latitude: float, longitude: float, days: int = 1) -> List[Dict[str, Any]]:
"""
指定された緯度経度の天気情報を取得
Args:
latitude: 緯度
longitude: 経度
days: 取得する日数（1-7日）
Returns:
天気情報のリスト
"""
try:
# パラメータの検証
if not (-90 <= latitude <= 90):
raise ValueError(f"緯度が範囲外です: {latitude}")
if not (-180 <= longitude <= 180):
raise ValueError(f"経度が範囲外です: {longitude}")
if not (1 <= days <= 7):
raise ValueError(f"日数が範囲外です: {days}")
# APIリクエストパラメータ
params = {
"latitude": latitude,
"longitude": longitude,
"daily": [
"temperature_2m_max",
"temperature_2m_min",
"precipitation_probability_max",
"weathercode",
"relative_humidity_2m_max"
],
"timezone": "Asia/Tokyo",
"forecast_days": days
}
async with httpx.AsyncClient(timeout=self.timeout) as client:
response = await client.get(self.base_url, params=params)
response.raise_for_status()
data = response.json()
return self._format_weather_data(data)
except httpx.RequestError as e:
raise Exception(f"API通信エラー: {str(e)}")
except httpx.HTTPStatusError as e:
raise Exception(f"APIエラー (HTTP {e.response.status_code}): {str(e)}")
except Exception as e:
raise Exception(f"天気情報取得エラー: {str(e)}")
def _format_weather_data(self, api_data: Dict[str, Any]) -> List[Dict[str, Any]]:
"""
OpenMeteo APIのレスポンスを整形
Args:
api_data: OpenMeteo APIのレスポンス
Returns:
整形された天気データのリスト
"""
try:
daily = api_data["daily"]
dates = daily["time"]
max_temps = daily["temperature_2m_max"]
min_temps = daily["temperature_2m_min"]
precip_probs = daily["precipitation_probability_max"]
weather_codes = daily["weathercode"]
humidity = daily["relative_humidity_2m_max"]
weather_data = []
for i in range(len(dates)):
# 平均気温を計算
avg_temp = (max_temps[i] + min_temps[i]) / 2
# 天気コードを日本語に変換
weather_description = self._weather_code_to_japanese(weather_codes[i])
weather_data.append({
"date": dates[i],
"weather": weather_description,
"temperature": round(avg_temp, 1),
"humidity": humidity[i] if humidity[i] is not None else 0,
"precipitation_probability": precip_probs[i] if precip_probs[i] is not None else 0,
"max_temperature": max_temps[i],
"min_temperature": min_temps[i]
})
return weather_data
except (KeyError, IndexError, TypeError) as e:
raise Exception(f"API応答の解析エラー: {str(e)}")
def _weather_code_to_japanese(self, weather_code: int) -> str:
"""
WMO天気コードを日本語に変換
Args:
weather_code: WMO天気コード
Returns:
日本語の天気表現
"""
weather_mapping = {
0: "快晴",
1: "晴れ",
2: "一部曇り",
3: "曇り",
45: "霧",
48: "着氷霧",
51: "弱い霧雨",
53: "霧雨",
55: "強い霧雨",
56: "弱い着氷霧雨",
57: "強い着氷霧雨",
61: "弱い雨",
63: "雨",
65: "強い雨",
66: "弱い着氷雨",
67: "強い着氷雨",
71: "弱い雪",
73: "雪",
75: "強い雪",
77: "霰",
80: "弱いにわか雨",
81: "にわか雨",
82: "強いにわか雨",
85: "弱いにわか雪",
86: "にわか雪",
95: "雷雨",
96: "雹を伴う雷雨",
99: "強い雹を伴う雷雨"
}
return weather_mapping.get(weather_code, f"不明な天気({weather_code})")
async def get_current_weather(self, latitude: float, longitude: float) -> Dict[str, Any]:
"""
現在の天気情報を取得
Args:
latitude: 緯度
longitude: 経度
Returns:
現在の天気情報
"""
try:
params = {
"latitude": latitude,
"longitude": longitude,
"current": [
"temperature_2m",
"relative_humidity_2m",
"weathercode",
"precipitation"
],
"timezone": "Asia/Tokyo"
}
async with httpx.AsyncClient(timeout=self.timeout) as client:
response = await client.get(self.base_url, params=params)
response.raise_for_status()
data = response.json()
current = data["current"]
return {
"temperature": current["temperature_2m"],
"humidity": current["relative_humidity_2m"],
"weather": self._weather_code_to_japanese(current["weathercode"]),
"precipitation": current["precipitation"],
"time": current["time"]
}
except Exception as e:
raise Exception(f"現在の天気情報取得エラー: {str(e)}")
