331 lines
8.6 KiB
Python
331 lines
8.6 KiB
Python
import json
|
|
from datetime import datetime
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import urlencode
|
|
from urllib.request import Request, urlopen
|
|
|
|
from flask import Flask, render_template, request
|
|
|
|
|
|
app = Flask(__name__)
|
|
|
|
DEFAULT_LOCATION = "Seattle"
|
|
GEOCODE_URL = "https://geocoding-api.open-meteo.com/v1/search"
|
|
FORECAST_URL = "https://api.open-meteo.com/v1/forecast"
|
|
USER_AGENT = "SETI-Flask-Weather-Demo/1.0"
|
|
|
|
CURRENT_FIELDS = [
|
|
"temperature_2m",
|
|
"relative_humidity_2m",
|
|
"apparent_temperature",
|
|
"is_day",
|
|
"precipitation",
|
|
"weather_code",
|
|
"cloud_cover",
|
|
"wind_speed_10m",
|
|
"wind_direction_10m",
|
|
"wind_gusts_10m",
|
|
]
|
|
|
|
DAILY_FIELDS = [
|
|
"weather_code",
|
|
"temperature_2m_max",
|
|
"temperature_2m_min",
|
|
"precipitation_probability_max",
|
|
"precipitation_sum",
|
|
"wind_speed_10m_max",
|
|
]
|
|
|
|
WEATHER_CODES = {
|
|
0: "Clear sky",
|
|
1: "Mainly clear",
|
|
2: "Partly cloudy",
|
|
3: "Overcast",
|
|
45: "Fog",
|
|
48: "Depositing rime fog",
|
|
51: "Light drizzle",
|
|
53: "Drizzle",
|
|
55: "Heavy drizzle",
|
|
56: "Light freezing drizzle",
|
|
57: "Freezing drizzle",
|
|
61: "Light rain",
|
|
63: "Rain",
|
|
65: "Heavy rain",
|
|
66: "Light freezing rain",
|
|
67: "Freezing rain",
|
|
71: "Light snow",
|
|
73: "Snow",
|
|
75: "Heavy snow",
|
|
77: "Snow grains",
|
|
80: "Light rain showers",
|
|
81: "Rain showers",
|
|
82: "Heavy rain showers",
|
|
85: "Light snow showers",
|
|
86: "Snow showers",
|
|
95: "Thunderstorm",
|
|
96: "Thunderstorm with hail",
|
|
99: "Severe thunderstorm with hail",
|
|
}
|
|
|
|
|
|
class WeatherServiceError(Exception):
|
|
"""Raised when the weather or geocoding API cannot return usable data."""
|
|
|
|
|
|
def fetch_json(url, params):
|
|
request_url = f"{url}?{urlencode(params)}"
|
|
api_request = Request(request_url, headers={"User-Agent": USER_AGENT})
|
|
|
|
try:
|
|
with urlopen(api_request, timeout=10) as response:
|
|
return json.loads(response.read().decode("utf-8"))
|
|
except HTTPError as exc:
|
|
raise WeatherServiceError(f"Weather API returned HTTP {exc.code}.") from exc
|
|
except (TimeoutError, URLError) as exc:
|
|
raise WeatherServiceError("Weather API is unreachable right now.") from exc
|
|
except json.JSONDecodeError as exc:
|
|
raise WeatherServiceError("Weather API returned invalid JSON.") from exc
|
|
|
|
|
|
def geocode_location(query):
|
|
cleaned_query = (query or DEFAULT_LOCATION).strip() or DEFAULT_LOCATION
|
|
search_terms = [cleaned_query]
|
|
if "," in cleaned_query:
|
|
search_terms.append(cleaned_query.split(",", 1)[0].strip())
|
|
|
|
results = []
|
|
for search_term in search_terms:
|
|
if not search_term:
|
|
continue
|
|
|
|
payload = fetch_json(
|
|
GEOCODE_URL,
|
|
{
|
|
"name": search_term,
|
|
"count": 1,
|
|
"language": "en",
|
|
"format": "json",
|
|
},
|
|
)
|
|
results = payload.get("results") or []
|
|
if results:
|
|
break
|
|
|
|
if not results:
|
|
raise WeatherServiceError(f"No location found for '{cleaned_query}'.")
|
|
|
|
result = results[0]
|
|
display_parts = [
|
|
result.get("name"),
|
|
result.get("admin1"),
|
|
result.get("country"),
|
|
]
|
|
|
|
return {
|
|
"name": result["name"],
|
|
"display": ", ".join(part for part in display_parts if part),
|
|
"latitude": result["latitude"],
|
|
"longitude": result["longitude"],
|
|
"timezone": result.get("timezone", "auto"),
|
|
}
|
|
|
|
|
|
def get_forecast(location):
|
|
payload = fetch_json(
|
|
FORECAST_URL,
|
|
{
|
|
"latitude": location["latitude"],
|
|
"longitude": location["longitude"],
|
|
"current": ",".join(CURRENT_FIELDS),
|
|
"daily": ",".join(DAILY_FIELDS),
|
|
"temperature_unit": "fahrenheit",
|
|
"wind_speed_unit": "mph",
|
|
"precipitation_unit": "inch",
|
|
"forecast_days": 5,
|
|
"timezone": "auto",
|
|
},
|
|
)
|
|
|
|
return normalize_forecast(location, payload)
|
|
|
|
|
|
def normalize_forecast(location, payload):
|
|
current = payload.get("current") or {}
|
|
daily = payload.get("daily") or {}
|
|
current_code = current.get("weather_code")
|
|
|
|
return {
|
|
"location": location,
|
|
"current": {
|
|
"time": format_timestamp(current.get("time")),
|
|
"temperature": format_number(current.get("temperature_2m"), "F"),
|
|
"apparent_temperature": format_number(
|
|
current.get("apparent_temperature"),
|
|
"F",
|
|
),
|
|
"humidity": format_number(current.get("relative_humidity_2m"), "%"),
|
|
"precipitation": format_number(
|
|
current.get("precipitation"),
|
|
" in",
|
|
digits=2,
|
|
),
|
|
"cloud_cover": format_number(current.get("cloud_cover"), "%"),
|
|
"wind": format_wind(
|
|
current.get("wind_speed_10m"),
|
|
current.get("wind_direction_10m"),
|
|
),
|
|
"wind_gusts": format_number(current.get("wind_gusts_10m"), " mph"),
|
|
"label": weather_label(current_code),
|
|
"icon": weather_icon(current_code),
|
|
},
|
|
"days": normalize_daily(daily),
|
|
"generation_ms": round(payload.get("generationtime_ms", 0), 2),
|
|
}
|
|
|
|
|
|
def normalize_daily(daily):
|
|
days = []
|
|
dates = daily.get("time") or []
|
|
|
|
for index, date_value in enumerate(dates):
|
|
weather_code = daily_value(daily, "weather_code", index)
|
|
days.append(
|
|
{
|
|
"date": format_date(date_value),
|
|
"label": weather_label(weather_code),
|
|
"icon": weather_icon(weather_code),
|
|
"high": format_number(
|
|
daily_value(daily, "temperature_2m_max", index),
|
|
"F",
|
|
),
|
|
"low": format_number(
|
|
daily_value(daily, "temperature_2m_min", index),
|
|
"F",
|
|
),
|
|
"precip_probability": format_number(
|
|
daily_value(daily, "precipitation_probability_max", index),
|
|
"%",
|
|
),
|
|
"precip_sum": format_number(
|
|
daily_value(daily, "precipitation_sum", index),
|
|
" in",
|
|
digits=2,
|
|
),
|
|
"wind": format_number(
|
|
daily_value(daily, "wind_speed_10m_max", index),
|
|
" mph",
|
|
),
|
|
}
|
|
)
|
|
|
|
return days
|
|
|
|
|
|
def daily_value(daily, field, index):
|
|
values = daily.get(field) or []
|
|
if index >= len(values):
|
|
return None
|
|
return values[index]
|
|
|
|
|
|
def weather_label(code):
|
|
return WEATHER_CODES.get(code, "Forecast unavailable")
|
|
|
|
|
|
def weather_icon(code):
|
|
if code in {0, 1}:
|
|
return "sun"
|
|
if code in {2, 3, 45, 48}:
|
|
return "cloud"
|
|
if code in {51, 53, 55, 56, 57, 61, 63, 65, 66, 67, 80, 81, 82}:
|
|
return "rain"
|
|
if code in {71, 73, 75, 77, 85, 86}:
|
|
return "snow"
|
|
if code in {95, 96, 99}:
|
|
return "storm"
|
|
return "cloud"
|
|
|
|
|
|
def format_number(value, suffix="", digits=0):
|
|
if value is None:
|
|
return "--"
|
|
rounded = round(value, digits)
|
|
if digits == 0:
|
|
rounded = int(rounded)
|
|
return f"{rounded}{suffix}"
|
|
|
|
|
|
def format_wind(speed, direction):
|
|
if speed is None:
|
|
return "--"
|
|
return f"{format_number(speed, ' mph')} {compass_direction(direction)}"
|
|
|
|
|
|
def compass_direction(degrees):
|
|
if degrees is None:
|
|
return ""
|
|
|
|
directions = [
|
|
"N",
|
|
"NNE",
|
|
"NE",
|
|
"ENE",
|
|
"E",
|
|
"ESE",
|
|
"SE",
|
|
"SSE",
|
|
"S",
|
|
"SSW",
|
|
"SW",
|
|
"WSW",
|
|
"W",
|
|
"WNW",
|
|
"NW",
|
|
"NNW",
|
|
]
|
|
index = round(degrees / 22.5) % 16
|
|
return directions[index]
|
|
|
|
|
|
def format_timestamp(value):
|
|
if not value:
|
|
return "Latest model run"
|
|
try:
|
|
parsed = datetime.fromisoformat(value)
|
|
except ValueError:
|
|
return value
|
|
return parsed.strftime("%b %-d, %-I:%M %p")
|
|
|
|
|
|
def format_date(value):
|
|
try:
|
|
parsed = datetime.fromisoformat(value)
|
|
except ValueError:
|
|
return value
|
|
return parsed.strftime("%a, %b %-d")
|
|
|
|
|
|
@app.route("/")
|
|
def index():
|
|
query = request.args.get("q", DEFAULT_LOCATION)
|
|
forecast = None
|
|
error = None
|
|
|
|
try:
|
|
location = geocode_location(query)
|
|
forecast = get_forecast(location)
|
|
except WeatherServiceError as exc:
|
|
error = str(exc)
|
|
|
|
return render_template(
|
|
"index.html",
|
|
default_location=DEFAULT_LOCATION,
|
|
error=error,
|
|
forecast=forecast,
|
|
query=query,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="127.0.0.1", port=5000, debug=True)
|