201 lines
9.2 KiB
Python
201 lines
9.2 KiB
Python
# --- START OF FILE weather_tool.py ---
|
|
|
|
from typing import Dict, Any
|
|
import os
|
|
import logging
|
|
import requests
|
|
import json
|
|
import slack
|
|
from slack import WebClient
|
|
from slack.errors import SlackApiError
|
|
|
|
# Initialize Slack client
|
|
slack_client = slack.WebClient(token=os.environ['SLACK_TOKEN'])
|
|
|
|
# --- Tool Definition (for LLM) ---
|
|
TOOL_DEFINITION = {
|
|
"name": "get_weather",
|
|
"description": "Retrieve current weather information for a given location so that you can provide that info to the user",
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"location": {
|
|
"type": "string",
|
|
"description": "The city name or city,country code to get weather for. Input MUST be formatted as {city name},{state code},{country code}. eg. Lewiston,ID,US. if state or country aren't provided by the USER, guess between WA and ID for state, and assume US for country"
|
|
},
|
|
"channel": {
|
|
"type": "string",
|
|
"description": "The Slack channel ID to post a webcam image to, if a webcam is available for the location"
|
|
}
|
|
},
|
|
"required": ["location", "channel"] # Both required as per original schema
|
|
}
|
|
}
|
|
# --- End Tool Definition ---
|
|
|
|
# Define a dictionary of placenames and their corresponding webcam URLs
|
|
WEBCAM_URLS = {
|
|
"Elk City": {"url": "https://511.idaho.gov/map/Cctv/205.C1--2", "info": "SH-14 Eastbound View"},
|
|
"Grangeville": {"url": "https://www.deq.idaho.gov/wp-content/uploads/cameras/Grangeville.jpg", "info": "North off High Camp"},
|
|
"Lewiston": {"url": "https://www.deq.idaho.gov/wp-content/uploads/cameras/Lewiston.jpg", "info": "South off Lewiston Rim"},
|
|
"Lapwai": {"url": "https://www.deq.idaho.gov/wp-content/uploads/cameras/Lapwai.jpg", "info": "East off Lewiston Rim"},
|
|
"Potlatch": {"url": "https://www.deq.idaho.gov/wp-content/uploads/cameras/Potlatch.jpg", "info": "North off West Twin"},
|
|
"Teakean Butte": {"url": "https://www.deq.idaho.gov/wp-content/uploads/cameras/Teakean.jpg", "info": "West off Teakean Butte"},
|
|
"Moscow": {"url": "https://media.kuoi.org/camera/latest.jpg", "info": "NNE off Morill Hall"}
|
|
# Add more locations as needed
|
|
}
|
|
|
|
# --- Tool Implementation ---
|
|
def get_weather(**kwargs: Any) -> Dict[str, Any]:
|
|
"""
|
|
Retrieve current weather information for a given location using OpenWeatherMap API,
|
|
validating input arguments internally.
|
|
|
|
Args:
|
|
**kwargs (Any): Keyword arguments matching the tool's input_schema properties
|
|
(expects 'location' and 'channel').
|
|
|
|
Returns:
|
|
Dict[str, Any]: A dictionary containing weather information or an error message.
|
|
"""
|
|
location = kwargs.get('location')
|
|
channel = kwargs.get('channel')
|
|
|
|
# --- Input Validation Moved Here ---
|
|
if not location or not isinstance(location, str):
|
|
logging.error("get_weather validation failed: Missing or invalid 'location' argument.")
|
|
return {"error": "Missing or invalid required argument: 'location'."}
|
|
if not channel or not isinstance(channel, str):
|
|
# This check is important because the function needs the channel to post webcams
|
|
logging.error("get_weather validation failed: Missing or invalid 'channel' argument.")
|
|
# Although the schema requires it, the LLM might still fail. Provide clear error.
|
|
return {"error": "Missing or invalid required argument: 'channel'."}
|
|
# --- End Input Validation ---
|
|
|
|
try:
|
|
# Log the incoming location request
|
|
logging.info(f"Attempting to fetch weather for location: {location}")
|
|
|
|
# Retrieve API key from environment variables
|
|
api_key = os.environ.get('OPENWEATHERMAP_API_KEY')
|
|
if not api_key:
|
|
logging.error("OpenWeatherMap API key not found in environment variables.")
|
|
return {
|
|
"error": "OpenWeatherMap API key not found. Please set OPENWEATHERMAP_API_KEY in your environment variables."
|
|
}
|
|
|
|
# Construct the API URL
|
|
base_url = "http://api.openweathermap.org/data/2.5/weather"
|
|
params = {
|
|
"q": location,
|
|
"appid": api_key,
|
|
"units": "imperial" # don't use metric units (Celsius) but rather, imperial (Fahrenheit)
|
|
}
|
|
|
|
# Log the API request details
|
|
logging.info(f"API Request URL: {base_url}")
|
|
logging.info(f"API Request Params: {params}")
|
|
|
|
# check if the location is in the webcam URLs dictionary, and if so, get the webcam URL. we should see if any part of the location input
|
|
# matches any of the locations in the dictionary, regardless of case, and if so, send the the webcam URL and info to the slack channel, and
|
|
# proceed on
|
|
webcam_posted = False # Flag to track if webcam was posted
|
|
for loc, webcam in WEBCAM_URLS.items():
|
|
if loc.lower() in location.lower():
|
|
webcam_url = webcam['url']
|
|
webcam_info = webcam['info']
|
|
logging.info(f"Webcam URL found for location: {loc}")
|
|
logging.info(f"Webcam URL: {webcam_url}, Info: {webcam_info}")
|
|
|
|
# Send the webcam URL and info to the Slack channel
|
|
try:
|
|
slack_client.chat_postMessage(
|
|
channel=channel,
|
|
text=f"Webcam for {loc} ({webcam_info}): {webcam_url}" # Added context
|
|
)
|
|
webcam_posted = True
|
|
except SlackApiError as e:
|
|
logging.error(f"Error sending message to Slack channel {channel}: {e.response['error']}")
|
|
# Decide if this should be returned as part of the tool result error
|
|
# For now, just log it and continue with weather lookup
|
|
|
|
break # Stop checking after the first match
|
|
|
|
|
|
# Make the API request
|
|
try:
|
|
response = requests.get(base_url, params=params, timeout=10)
|
|
|
|
# Log the response status
|
|
logging.info(f"API Response Status Code: {response.status_code}")
|
|
logging.debug(f"API Response Content: {response.text}")
|
|
|
|
# Check if the request was successful
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
weather_info = {
|
|
"location": data['name'],
|
|
"country": data['sys']['country'],
|
|
"temperature": data['main']['temp'],
|
|
"feels_like": data['main']['feels_like'],
|
|
"description": data['weather'][0]['description'],
|
|
"humidity": data['main']['humidity'],
|
|
"wind_speed": data['wind']['speed'],
|
|
"webcam_posted": webcam_posted # Include status of webcam post
|
|
}
|
|
|
|
# Log successful weather retrieval
|
|
logging.info(f"Successfully retrieved weather for {location}")
|
|
logging.info(f"Weather details: {weather_info}")
|
|
|
|
return weather_info
|
|
else:
|
|
# Log unsuccessful API response
|
|
logging.error(f"Failed to retrieve weather. Status code: {response.status_code}")
|
|
logging.error(f"Response content: {response.text}")
|
|
|
|
return {
|
|
"error": f"Failed to retrieve weather. Status code: {response.status_code}, Response: {response.text}",
|
|
"webcam_posted": webcam_posted # Include status even on error
|
|
}
|
|
|
|
except requests.exceptions.RequestException as req_err:
|
|
# Log network-related errors
|
|
logging.error(f"Request error occurred: {req_err}")
|
|
return {
|
|
"error": f"Network error occurred: {str(req_err)}",
|
|
"webcam_posted": webcam_posted
|
|
}
|
|
|
|
except Exception as e:
|
|
# Log any unexpected errors
|
|
logging.error(f"Unexpected error occurred while fetching weather: {str(e)}", exc_info=True) # Added exc_info
|
|
# Attempt to get webcam_posted status if it was set before the error
|
|
wc_status = 'unknown'
|
|
if 'webcam_posted' in locals():
|
|
wc_status = webcam_posted
|
|
|
|
return {
|
|
"error": f"An unexpected error occurred while fetching weather: {str(e)}",
|
|
"webcam_posted": wc_status
|
|
}
|
|
# --- End Tool Implementation ---
|
|
|
|
# Example usage remains the same
|
|
if __name__ == "__main__":
|
|
from dotenv import load_dotenv
|
|
load_dotenv()
|
|
test_channel = os.environ.get("TEST_SLACK_CHANNEL_ID", "C08B9A6RPN1")
|
|
print("--- Testing get_weather ---")
|
|
result1 = get_weather(location="Lewiston,ID,US", channel=test_channel)
|
|
print(f"Result (Lewiston): {result1}")
|
|
result2 = get_weather(location="London", channel=test_channel)
|
|
print(f"Result (London): {result2}")
|
|
result3 = get_weather(location="", channel=test_channel) # Test validation
|
|
print(f"Result (Empty Location): {result3}")
|
|
result4 = get_weather(location="Paris,FR") # Missing channel - kwargs will be missing 'channel'
|
|
print(f"Result (Missing Channel): {result4}")
|
|
result5 = get_weather(location="Grangeville", channel=test_channel) # With webcam
|
|
print(f"Result (Grangeville with Webcam): {result5}")
|
|
|
|
# --- END OF FILE weather_tool.py --- |