|
|
import requests |
|
|
from geopy.geocoders import Nominatim |
|
|
from datetime import datetime, date, timedelta |
|
|
import pandas as pd |
|
|
import gradio as gr |
|
|
|
|
|
|
|
|
WEATHER_CODES = { |
|
|
0: "Clear sky", 1: "Mainly clear", 2: "Partly cloudy", 3: "Overcast", |
|
|
45: "Fog", 48: "Depositing rime fog", 51: "Light drizzle", 53: "Moderate drizzle", 55: "Dense drizzle", |
|
|
61: "Slight rain", 63: "Moderate rain", 65: "Heavy rain", 71: "Slight snowfall", 73: "Moderate snowfall", |
|
|
75: "Heavy snowfall", 80: "Slight rain showers", 81: "Moderate rain showers", 82: "Violent rain showers", |
|
|
95: "Thunderstorm", 96: "Thunderstorm with hail" |
|
|
} |
|
|
|
|
|
def get_weather_forecast(location_name, target_date, include_hourly=False): |
|
|
try: |
|
|
if isinstance(target_date, str): |
|
|
target_date = datetime.strptime(target_date, "%Y-%m-%d").date() |
|
|
date_str = target_date.isoformat() |
|
|
|
|
|
geolocator = Nominatim(user_agent="weather_api") |
|
|
location = geolocator.geocode(location_name) |
|
|
if not location: |
|
|
return {"error": f"Could not find coordinates for '{location_name}'."} |
|
|
lat, lon = location.latitude, location.longitude |
|
|
|
|
|
url = "https://api.open-meteo.com/v1/forecast" |
|
|
params = { |
|
|
"latitude": lat, |
|
|
"longitude": lon, |
|
|
"daily": "sunrise,sunset,uv_index_max,temperature_2m_max,temperature_2m_min,weather_code", |
|
|
"temperature_unit": "celsius", |
|
|
"windspeed_unit": "kmh", |
|
|
"timeformat": "iso8601", |
|
|
"timezone": "auto", |
|
|
"start_date": date_str, |
|
|
"end_date": date_str |
|
|
} |
|
|
|
|
|
if include_hourly: |
|
|
params["hourly"] = "temperature_2m,weather_code,uv_index,visibility" |
|
|
|
|
|
response = requests.get(url, params=params) |
|
|
if response.status_code != 200: |
|
|
return {"error": f"API error {response.status_code}: {response.text}"} |
|
|
raw = response.json() |
|
|
|
|
|
if "daily" not in raw or date_str not in raw["daily"]["time"]: |
|
|
return {"error": f"Weather data for {date_str} not available."} |
|
|
|
|
|
idx = raw["daily"]["time"].index(date_str) |
|
|
result = { |
|
|
"date": date_str, |
|
|
"sunrise": raw["daily"]["sunrise"][idx].split("T")[1], |
|
|
"sunset": raw["daily"]["sunset"][idx].split("T")[1], |
|
|
"uv_max": round(raw["daily"]["uv_index_max"][idx], 1), |
|
|
"temp_min": round(raw["daily"]["temperature_2m_min"][idx]), |
|
|
"temp_max": round(raw["daily"]["temperature_2m_max"][idx]), |
|
|
"weather": WEATHER_CODES.get(int(raw["daily"]["weather_code"][idx]), "Unknown") |
|
|
} |
|
|
|
|
|
if include_hourly and "hourly" in raw: |
|
|
hourly_df = pd.DataFrame({ |
|
|
"time": raw["hourly"]["time"], |
|
|
"temp": raw["hourly"]["temperature_2m"], |
|
|
"code": raw["hourly"]["weather_code"], |
|
|
"uv": raw["hourly"]["uv_index"], |
|
|
"visibility": [v / 1000 for v in raw["hourly"]["visibility"]] |
|
|
}) |
|
|
hourly_df["time"] = pd.to_datetime(hourly_df["time"]) |
|
|
hourly_df = hourly_df[hourly_df["time"].dt.date == target_date] |
|
|
hourly_df["weather"] = hourly_df["code"].apply(lambda c: WEATHER_CODES.get(int(c), "Unknown")) |
|
|
|
|
|
result["hourly"] = [ |
|
|
{ |
|
|
"time": t.strftime("%Y-%m-%d %H:%M"), |
|
|
"temp": f"{round(temp)}°C", |
|
|
"weather": w, |
|
|
"uv": round(uv, 1), |
|
|
"visibility": f"{round(vis, 1)} km" |
|
|
} |
|
|
for t, temp, w, uv, vis in zip( |
|
|
hourly_df["time"], hourly_df["temp"], hourly_df["weather"], |
|
|
hourly_df["uv"], hourly_df["visibility"] |
|
|
) |
|
|
] |
|
|
elif include_hourly: |
|
|
result["note"] = "Hourly weather data unavailable for this date." |
|
|
|
|
|
return result |
|
|
|
|
|
except Exception as e: |
|
|
return {"error": str(e)} |
|
|
|
|
|
|
|
|
def get_weather_forecast_range(location_name, start_date, end_date): |
|
|
try: |
|
|
if isinstance(start_date, str): |
|
|
start_date = datetime.strptime(start_date, "%Y-%m-%d").date() |
|
|
if isinstance(end_date, str): |
|
|
end_date = datetime.strptime(end_date, "%Y-%m-%d").date() |
|
|
|
|
|
today = date.today() |
|
|
days_ahead = (end_date - today).days |
|
|
|
|
|
if days_ahead > 15: |
|
|
return {"error": "Weather data only available up to 15 days from today."} |
|
|
|
|
|
geolocator = Nominatim(user_agent="weather_api") |
|
|
location = geolocator.geocode(location_name) |
|
|
if not location: |
|
|
return {"error": f"Could not find coordinates for '{location_name}'."} |
|
|
lat, lon = location.latitude, location.longitude |
|
|
|
|
|
include_hourly = days_ahead <= 6 |
|
|
|
|
|
url = "https://api.open-meteo.com/v1/forecast" |
|
|
params = { |
|
|
"latitude": lat, |
|
|
"longitude": lon, |
|
|
"daily": "sunrise,sunset,uv_index_max,temperature_2m_max,temperature_2m_min,weather_code", |
|
|
"temperature_unit": "celsius", |
|
|
"windspeed_unit": "kmh", |
|
|
"timeformat": "iso8601", |
|
|
"timezone": "auto", |
|
|
"start_date": start_date.isoformat(), |
|
|
"end_date": end_date.isoformat() |
|
|
} |
|
|
|
|
|
if include_hourly: |
|
|
params["hourly"] = "temperature_2m,weather_code,uv_index,visibility" |
|
|
|
|
|
response = requests.get(url, params=params) |
|
|
if response.status_code != 200: |
|
|
return {"error": f"API error {response.status_code}: {response.text}"} |
|
|
raw = response.json() |
|
|
|
|
|
forecasts = [] |
|
|
for idx, d in enumerate(raw["daily"]["time"]): |
|
|
day_result = { |
|
|
"date": d, |
|
|
"sunrise": raw["daily"]["sunrise"][idx].split("T")[1], |
|
|
"sunset": raw["daily"]["sunset"][idx].split("T")[1], |
|
|
"uv_max": round(raw["daily"]["uv_index_max"][idx], 1), |
|
|
"temp_min": round(raw["daily"]["temperature_2m_min"][idx]), |
|
|
"temp_max": round(raw["daily"]["temperature_2m_max"][idx]), |
|
|
"weather": WEATHER_CODES.get(int(raw["daily"]["weather_code"][idx]), "Unknown") |
|
|
} |
|
|
|
|
|
if include_hourly and "hourly" in raw: |
|
|
hourly_df = pd.DataFrame({ |
|
|
"time": raw["hourly"]["time"], |
|
|
"temp": raw["hourly"]["temperature_2m"], |
|
|
"code": raw["hourly"]["weather_code"], |
|
|
"uv": raw["hourly"]["uv_index"], |
|
|
"visibility": [v / 1000 for v in raw["hourly"]["visibility"]] |
|
|
}) |
|
|
|
|
|
hourly_df["time"] = pd.to_datetime(hourly_df["time"]) |
|
|
target_date = datetime.strptime(d, "%Y-%m-%d").date() |
|
|
df_day = hourly_df[hourly_df["time"].dt.date == target_date] |
|
|
|
|
|
df_day["weather"] = df_day["code"].apply(lambda c: WEATHER_CODES.get(int(c), "Unknown")) |
|
|
|
|
|
day_result["hourly"] = [ |
|
|
{ |
|
|
"time": t.strftime("%Y-%m-%d %H:%M"), |
|
|
"temp": f"{round(temp)}°C", |
|
|
"weather": w, |
|
|
"uv": round(uv, 1), |
|
|
"visibility": f"{round(vis, 1)} km" |
|
|
} |
|
|
for t, temp, w, uv, vis in zip( |
|
|
df_day["time"], df_day["temp"], df_day["weather"], |
|
|
df_day["uv"], df_day["visibility"] |
|
|
) |
|
|
] |
|
|
else: |
|
|
day_result["note"] = "Hourly weather data is only available for the next 7 days." |
|
|
|
|
|
forecasts.append(day_result) |
|
|
|
|
|
return forecasts |
|
|
|
|
|
except Exception as e: |
|
|
return {"error": str(e)} |
|
|
|
|
|
demo= gr.Interface( |
|
|
fn=get_weather_forecast_range, |
|
|
inputs=[ |
|
|
gr.Textbox(label="Location Name", placeholder="Enter a city or place name"), |
|
|
gr.Textbox(label="Start Date (YYYY-MM-DD)", value=date.today()), |
|
|
gr.Textbox(label="End Date (YYYY-MM-DD)", value=date.today() + timedelta(days=6)) |
|
|
], |
|
|
outputs=gr.JSON(label="Weather Forecast"), |
|
|
title="Weather Forecast Tool", |
|
|
description="Get weather forecasts for a specific location and date range." |
|
|
) |
|
|
|
|
|
demo.launch(mcp_server=True, share=True) |
