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)