Spaces:
Sleeping
Sleeping
File size: 3,337 Bytes
5194558 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 | """Open-Meteo data fetching and local CSV/JSON cache helpers."""
from __future__ import annotations
import json
from datetime import date, datetime, timedelta, timezone
from typing import Any
import numpy as np
import pandas as pd
import requests
from weatherpred.config import (
DAILY_COLUMNS,
DATA_DIR,
FORECAST_DAYS,
HISTORY_CSV_PATH,
HISTORY_JSON_PATH,
HISTORY_YEARS,
LATEST_JSON_PATH,
TOKYO_LATITUDE,
TOKYO_LONGITUDE,
TOKYO_TIMEZONE,
)
def fetch_json(url: str, params: dict[str, str | float | int]) -> dict[str, Any]:
response = requests.get(url, params=params, timeout=40)
response.raise_for_status()
return response.json()
def normalize_daily_weather(raw: dict[str, Any], observed: bool) -> pd.DataFrame:
daily = raw["daily"]
frame = pd.DataFrame(
{
"date": pd.to_datetime(daily["time"]),
"high_c": daily["temperature_2m_max"],
"low_c": daily["temperature_2m_min"],
"rain_mm": daily["precipitation_sum"],
"wind_kmh": daily["wind_speed_10m_max"],
}
)
frame["confidence"] = 1.0 if observed else np.nan
return frame.dropna(subset=["high_c", "low_c", "rain_mm", "wind_kmh"])
def fetch_historical_weather(
years: int = HISTORY_YEARS,
latitude: float = TOKYO_LATITUDE,
longitude: float = TOKYO_LONGITUDE,
) -> pd.DataFrame:
end = date.today() - timedelta(days=1)
start = end.replace(year=end.year - years)
params = {
"latitude": latitude,
"longitude": longitude,
"timezone": TOKYO_TIMEZONE,
"start_date": start.isoformat(),
"end_date": end.isoformat(),
"daily": ",".join(DAILY_COLUMNS),
}
raw = fetch_json("https://archive-api.open-meteo.com/v1/archive", params)
return normalize_daily_weather(raw, observed=True)
def fetch_api_forecast(
latitude: float = TOKYO_LATITUDE,
longitude: float = TOKYO_LONGITUDE,
days: int = FORECAST_DAYS,
) -> pd.DataFrame:
params = {
"latitude": latitude,
"longitude": longitude,
"timezone": TOKYO_TIMEZONE,
"forecast_days": days,
"current": "temperature_2m,precipitation,wind_speed_10m",
"daily": ",".join(DAILY_COLUMNS),
}
raw = fetch_json("https://api.open-meteo.com/v1/forecast", params)
latest = {
"fetched_at": datetime.now(timezone.utc).isoformat(),
"current": raw.get("current", {}),
"daily": raw.get("daily", {}),
}
DATA_DIR.mkdir(parents=True, exist_ok=True)
LATEST_JSON_PATH.write_text(json.dumps(latest, indent=2) + "\n", encoding="utf-8")
return normalize_daily_weather(raw, observed=False)
def save_history(history: pd.DataFrame) -> None:
DATA_DIR.mkdir(parents=True, exist_ok=True)
history.to_csv(HISTORY_CSV_PATH, index=False)
HISTORY_JSON_PATH.write_text(
history.assign(date=history["date"].dt.strftime("%Y-%m-%d")).to_json(
orient="records", indent=2
)
+ "\n",
encoding="utf-8",
)
def load_or_fetch_history(force_refresh: bool = False) -> pd.DataFrame:
if HISTORY_CSV_PATH.exists() and not force_refresh:
return pd.read_csv(HISTORY_CSV_PATH, parse_dates=["date"])
history = fetch_historical_weather()
save_history(history)
return history
|