| 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) |
