"""気象庁の無料JSON APIから天気予報を取得し、洗濯指数を判定する。""" import logging from datetime import datetime, timedelta import httpx logger = logging.getLogger(__name__) JMA_FORECAST_URL = "https://www.jma.go.jp/bosai/forecast/data/forecast/{area_code}.json" async def fetch_tomorrow_weather(area_code: str, area_name_keyword: str) -> dict: """指定エリアの明日の天気予報を取得する。 area_code: 気象庁の府県予報区コード(例: 千葉県 = 120000) area_name_keyword: 府県内の地域区分名の一部(例: 市川市が属する「北西部」) """ async with httpx.AsyncClient(timeout=10.0) as client: response = await client.get(JMA_FORECAST_URL.format(area_code=area_code)) response.raise_for_status() data = response.json() # data[0] が短期予報(今日・明日・明後日) short_term = data[0] time_series = short_term["timeSeries"] weather_series = time_series[0] pop_series = time_series[1] if len(time_series) > 1 else None tomorrow_index = _find_tomorrow_index(weather_series["timeDefines"]) weather_area = _find_area(weather_series["areas"], area_name_keyword) weather_text = weather_area["weathers"][tomorrow_index] if weather_area else "不明" pop_value = None if pop_series: pop_tomorrow_index = _find_tomorrow_index(pop_series["timeDefines"]) pop_area = _find_area(pop_series["areas"], area_name_keyword) if pop_area and pop_area.get("pops"): pops = pop_area["pops"] # pops は3時間ごとの配列のことが多いため、明日に該当する値の最大値を取る relevant_pops = [ int(p) for p, t in zip(pops, pop_series["timeDefines"]) if _is_same_date(t, pop_series["timeDefines"][pop_tomorrow_index]) and p ] if relevant_pops: pop_value = max(relevant_pops) return { "weather": weather_text.strip(), "pop": pop_value, } def _find_tomorrow_index(time_defines: list[str]) -> int: tomorrow = (datetime.now() + timedelta(days=1)).date() for i, t in enumerate(time_defines): dt = datetime.fromisoformat(t) if dt.date() == tomorrow: return i return min(1, len(time_defines) - 1) def _is_same_date(t1: str, t2: str) -> bool: return datetime.fromisoformat(t1).date() == datetime.fromisoformat(t2).date() def _find_area(areas: list[dict], keyword: str) -> dict | None: for area in areas: if keyword in area["area"]["name"]: return area return areas[0] if areas else None def judge_laundry(weather: str, pop: int | None) -> tuple[str, str]: """天気と降水確率から洗濯指数を判定する。戻り値: (判定, 理由)""" is_rainy = "雨" in weather or "雪" in weather if is_rainy: return "室内干し推奨", f"明日は「{weather}」のため" if pop is None: return "外干しOK(降水確率不明)", "降水確率データが取得できませんでした" if pop <= 30: return "外干しOK", f"降水確率{pop}%と低めです" if pop <= 50: return "室内干し推奨", f"降水確率{pop}%とやや高めです" return "室内干し推奨", f"降水確率{pop}%と高いです"