"""
title: Weather
original author: bwoodruff2021 (updated and expanded by lxwagno)
description: Tool for obtaining the current weather, extended forecast, or historical precip totals from OpenWeatherMap API OneCall for a location specified via user valves.
version: 0.2.0
licence: MIT
requirements: pydantic, dateparser, pytz
"""
import json
import re
import requests
from pydantic import BaseModel, Field
from datetime import datetime, timedelta
import dateparser
import pytz
async def emit_status(emitter, message: str, done: bool = False):
"""Helper function to emit status messages."""
if emitter is not None:
await emitter(
{"type": "status", "data": {"description": message, "done": done}}
)
def get_coordinates(api_key, city, state_code, country_code) -> tuple:
"""
Get the latitude and longitude for the specified city, state, and country.
There are no parameters for this function as it is using the valves.
:return: A tuple containing latitude and longitude or None if an error occurs.
"""
if not api_key:
print("API key is not set.")
return None
base_url = "https://api.openweathermap.org/geo/1.0/direct"
params = {
"q": f"{city},{state_code},{country_code}",
"limit": 1,
"appid": api_key,
}
try:
response = requests.get(base_url, params=params)
response.raise_for_status()
data = response.json()
if not data:
print("No coordinates found for the given location.")
return None
lat = data[0]["lat"]
lon = data[0]["lon"]
return lat, lon
except requests.RequestException as e:
print(f"Error fetching coordinates: {str(e)}")
return None
class Tools:
class Valves(BaseModel):
api_key: str = Field(
default="",
description="OpenWeather API Key is required",
)
class UserValves(BaseModel):
units: str = Field(
default="imperial",
description="imperial or metric units of measure",
)
city: str = Field(
default="Arlington",
description="City (Arlington)",
)
state_code: str = Field(
default="TX",
description="State Code (TX)",
)
country_code: str = Field(
default="US",
description="Country code (US)",
)
time_zone: str = Field(
default="America/Chicago",
description="Time Zone (America/Chicago or America/Denver, etc.)",
)
def __init__(self):
self.valves = self.Valves()
self.user_valves = self.UserValves()
async def get_current_weather(self, __event_emitter__=None) -> str:
"""
Get the current weather.
There are no parameters for this function as it is using the valves.
:return: A string containing current weather information or an error message.
"""
if not self.valves.api_key:
return "API key is not set."
await emit_status(__event_emitter__, "Processing Location...", done=True)
coords = get_coordinates(
self.valves.api_key,
self.user_valves.city,
self.user_valves.state_code,
self.user_valves.country_code,
)
lat, lon = coords
base_url = "https://api.openweathermap.org/data/3.0/onecall"
params = {
"lat": lat,
"lon": lon,
"exclude": "hourly,daily,minutely", # Exclude unnecessary parts
"appid": self.valves.api_key,
"units": self.user_valves.units,
}
try:
await emit_status(__event_emitter__, "Fetching Weather data...", done=True)
headers = {"User-Agent": "OpenWebUI-WeatherScript"}
response = requests.get(base_url, params=params, headers=headers)
response.raise_for_status()
data = response.json()
weather_description = data["current"]["weather"][0]["description"]
temperature = data["current"]["temp"]
feels_like = data["current"]["feels_like"]
humidity = data["current"]["humidity"]
wind_speed = data["current"]["wind_speed"]
unit_symbol = "°F" if self.user_valves.units == "imperial" else "°C"
speed_unit = (
"Mile Per Hour"
if self.user_valves.units == "imperial"
else "Meters Per Second"
)
return (
f"Weather in {self.user_valves.city}: {temperature}{unit_symbol}, feels like {feels_like}{unit_symbol}, {weather_description}, "
f"Humidity: {humidity}%, Wind Speed: {wind_speed} {speed_unit}"
)
except requests.RequestException as e:
return f"Error fetching weather data: {str(e)}"
async def get_extended_forecast(self, __event_emitter__=None) -> str:
"""
Get an extended forecast (8-day forecast).
There are no parameters for this function as it is using the valves.
:return: A string containing extended forecast information with days separated by \n, or an error message.
"""
if not self.valves.api_key:
return "API key is not set."
await emit_status(__event_emitter__, "Processing Location...", done=True)
coords = get_coordinates(
self.valves.api_key,
self.user_valves.city,
self.user_valves.state_code,
self.user_valves.country_code,
)
if not coords:
return "Failed to retrieve coordinates."
lat, lon = coords
base_url = "https://api.openweathermap.org/data/3.0/onecall"
params = {
"lat": lat,
"lon": lon,
"exclude": "hourly,minutely,current,alerts", # Exclude unnecessary parts
"appid": self.valves.api_key,
"units": self.user_valves.units,
}
try:
await emit_status(__event_emitter__, "Fetching Weather data...", done=True)
headers = {"User-Agent": "OpenWebUI-WeatherScript"}
response = requests.get(base_url, params=params, headers=headers)
response.raise_for_status()
data = response.json()
forecast_items = data.get("daily", [])
# forecast_items = data["daily"]
if not forecast_items:
return "No forecast data available."
forecast_lines = []
unit_symbol = "°C" if self.user_valves.units == "metric" else "°F"
wind_unit = "m/s" if self.user_valves.units == "metric" else "mph"
target_timezone = pytz.timezone(self.user_valves.time_zone)
daily_forecast = {}
for item in forecast_items:
# Convert the Unix timestamp to a human-readable date/time.
dt = (
datetime.fromtimestamp(item["dt"])
.replace(tzinfo=pytz.utc)
.astimezone(target_timezone)
)
dt_str = dt.strftime("%y-%m-%d")
sunrise = (
datetime.fromtimestamp(item["sunrise"])
.replace(tzinfo=pytz.utc)
.astimezone(target_timezone)
)
sunrise_str = sunrise.strftime("%H:%M")
sunset = (
datetime.fromtimestamp(item["sunset"])
.replace(tzinfo=pytz.utc)
.astimezone(target_timezone)
)
sunset_str = sunset.strftime("%H:%M")
mintemp = item["temp"]["min"]
maxtemp = item["temp"]["max"]
humidity = item["humidity"]
summary = item["summary"]
forecast_lines.append(
f"{dt_str}: {summary}, High: {maxtemp:.1f}{unit_symbol}, Low: {mintemp:.1f}{unit_symbol}, Humidity: {humidity}, Sunrise: {sunrise_str},Sunset: {sunset_str}"
)
return "\n".join(forecast_lines)
except requests.RequestException as e:
return f"Error fetching forecast data: {str(e)}."
async def get_precipitation_totals(self, __event_emitter__=None) -> str:
"""
Get the rainfall/precipitation totals for today, yesterday, past 7 days, and past 30 days.
There are no parameters for this function as it is using the valves.
:return: A string containing precipitation totals on one line per period.
"""
if not self.valves.api_key:
return "API key is not set."
await emit_status(__event_emitter__, "Processing Location...", done=True)
coords = get_coordinates(
self.valves.api_key,
self.user_valves.city,
self.user_valves.state_code,
self.user_valves.country_code,
)
lat, lon = coords
base_url = "https://api.openweathermap.org/data/3.0/onecall/day_summary"
headers = {"User-Agent": "OpenWebUI-WeatherScript"}
unit_symbol = "in" if self.user_valves.units == "imperial" else "mm"
unit_div = 25.4 if self.user_valves.units == "imperial" else 1
precip = 0.0
tody_precip = 0
ysdy_precip = 0
wkly_precip = 0
mly_precip = 0
day = 0
while day <= 30:
try:
await emit_status(
__event_emitter__,
f"Fetching Precipitation data for today - {str(day)}...",
done=True,
)
today_dt = datetime.now()
data_dt = today_dt - timedelta(days=day)
fmt_dt = data_dt.strftime("%Y-%m-%d")
params = {
"lat": lat,
"lon": lon,
"date": fmt_dt,
"appid": self.valves.api_key,
"units": self.user_valves.units,
}
response = requests.get(base_url, params=params, headers=headers)
response.raise_for_status()
data = response.json()
precip = float(data["precipitation"]["total"]) / unit_div
except requests.RequestException as e:
return f"Error fetching weather data for today - {str(day)}: {str(e)}"
if day == 0:
tody_precip = precip
else:
if day == 1:
ysdy_precip = precip
if 1 <= day <= 7:
wkly_precip += precip
mly_precip += precip
day += 1
return (
f"Precipitation Today in {self.user_valves.city}: {round(tody_precip,2)}{unit_symbol}, Yesterday: {round(ysdy_precip,2)}{unit_symbol}, "
f"Last 7 days: {round(wkly_precip,2)}{unit_symbol}, Last 30 days: {round(mly_precip,2)}{unit_symbol}"
)
