327 lines
11 KiB
Python
327 lines
11 KiB
Python
# %%
|
|
# Imports [standard]
|
|
# Standard library modules (no try-except needed)
|
|
import argparse
|
|
import asyncio
|
|
import anyio
|
|
import json
|
|
import logging
|
|
import os
|
|
import queue
|
|
import re
|
|
import time
|
|
from datetime import datetime
|
|
|
|
def try_import(module_name, pip_name=None):
|
|
try:
|
|
__import__(module_name)
|
|
except ImportError:
|
|
print(f"Module '{module_name}' not found. Install it using:")
|
|
print(f" pip install {pip_name or module_name}")
|
|
|
|
# Third-party modules with import checks
|
|
try_import('gradio')
|
|
try_import('ollama')
|
|
try_import('openai')
|
|
try_import('pydle')
|
|
try_import('pytz')
|
|
try_import('requests')
|
|
try_import('yfinance', 'yfinance')
|
|
try_import('dotenv', 'python-dotenv')
|
|
try_import('geopy', 'geopy')
|
|
|
|
from dotenv import load_dotenv
|
|
from geopy.geocoders import Nominatim
|
|
import gradio as gr
|
|
import ollama
|
|
import openai
|
|
import pydle
|
|
import pytz
|
|
import requests
|
|
import yfinance as yf
|
|
|
|
# %%
|
|
def get_weather_by_location(city, state, country="USA"):
|
|
"""
|
|
Get weather information from weather.gov based on city, state, and country.
|
|
|
|
Args:
|
|
city (str): City name
|
|
state (str): State name or abbreviation
|
|
country (str): Country name (defaults to "USA" as weather.gov is for US locations)
|
|
|
|
Returns:
|
|
dict: Weather forecast information
|
|
"""
|
|
# Step 1: Get coordinates for the location using geocoding
|
|
location = f"{city}, {state}, {country}"
|
|
coordinates = get_coordinates(location)
|
|
|
|
if not coordinates:
|
|
return {"error": f"Could not find coordinates for {location}"}
|
|
|
|
# Step 2: Get the forecast grid endpoint for the coordinates
|
|
grid_endpoint = get_grid_endpoint(coordinates)
|
|
|
|
if not grid_endpoint:
|
|
return {"error": f"Could not find weather grid for coordinates {coordinates}"}
|
|
|
|
# Step 3: Get the forecast data from the grid endpoint
|
|
forecast = get_forecast(grid_endpoint)
|
|
|
|
return forecast
|
|
|
|
def get_coordinates(location):
|
|
"""Convert a location string to latitude and longitude using Nominatim geocoder."""
|
|
try:
|
|
# Create a geocoder with a meaningful user agent
|
|
geolocator = Nominatim(user_agent="weather_app_example")
|
|
|
|
# Get the location
|
|
location_data = geolocator.geocode(location)
|
|
|
|
if location_data:
|
|
return {
|
|
"latitude": location_data.latitude,
|
|
"longitude": location_data.longitude
|
|
}
|
|
else:
|
|
print(f"Location not found: {location}")
|
|
return None
|
|
except Exception as e:
|
|
print(f"Error getting coordinates: {e}")
|
|
return None
|
|
|
|
def get_grid_endpoint(coordinates):
|
|
"""Get the grid endpoint from weather.gov based on coordinates."""
|
|
try:
|
|
lat = coordinates["latitude"]
|
|
lon = coordinates["longitude"]
|
|
|
|
# Define headers for the API request
|
|
headers = {
|
|
"User-Agent": "WeatherAppExample/1.0 (your_email@example.com)",
|
|
"Accept": "application/geo+json"
|
|
}
|
|
|
|
# Make the request to get the grid endpoint
|
|
url = f"https://api.weather.gov/points/{lat},{lon}"
|
|
response = requests.get(url, headers=headers)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
return data["properties"]["forecast"]
|
|
else:
|
|
print(f"Error getting grid: {response.status_code} - {response.text}")
|
|
return None
|
|
except Exception as e:
|
|
print(f"Error in get_grid_endpoint: {e}")
|
|
return None
|
|
|
|
# Weather related function
|
|
|
|
def get_forecast(grid_endpoint):
|
|
"""Get the forecast data from the grid endpoint."""
|
|
try:
|
|
# Define headers for the API request
|
|
headers = {
|
|
"User-Agent": "WeatherAppExample/1.0 (your_email@example.com)",
|
|
"Accept": "application/geo+json"
|
|
}
|
|
|
|
# Make the request to get the forecast
|
|
response = requests.get(grid_endpoint, headers=headers)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
|
|
# Extract the relevant forecast information
|
|
periods = data["properties"]["periods"]
|
|
|
|
# Process the forecast data into a simpler format
|
|
forecast = {
|
|
"location": data["properties"].get("relativeLocation", {}).get("properties", {}),
|
|
"updated": data["properties"].get("updated", ""),
|
|
"periods": []
|
|
}
|
|
|
|
for period in periods:
|
|
forecast["periods"].append({
|
|
"name": period.get("name", ""),
|
|
"temperature": period.get("temperature", ""),
|
|
"temperatureUnit": period.get("temperatureUnit", ""),
|
|
"windSpeed": period.get("windSpeed", ""),
|
|
"windDirection": period.get("windDirection", ""),
|
|
"shortForecast": period.get("shortForecast", ""),
|
|
"detailedForecast": period.get("detailedForecast", "")
|
|
})
|
|
|
|
return forecast
|
|
else:
|
|
print(f"Error getting forecast: {response.status_code} - {response.text}")
|
|
return {"error": f"API Error: {response.status_code}"}
|
|
except Exception as e:
|
|
print(f"Error in get_forecast: {e}")
|
|
return {"error": f"Exception: {str(e)}"}
|
|
|
|
# Example usage
|
|
def do_weather():
|
|
city = input("Enter city: ")
|
|
state = input("Enter state: ")
|
|
country = input("Enter country (default USA): ") or "USA"
|
|
|
|
print(f"Getting weather for {city}, {state}, {country}...")
|
|
weather_data = get_weather_by_location(city, state, country)
|
|
|
|
if "error" in weather_data:
|
|
print(f"Error: {weather_data['error']}")
|
|
else:
|
|
print("\nWeather Forecast:")
|
|
print(f"Location: {weather_data.get('location', {}).get('city', city)}, {weather_data.get('location', {}).get('state', state)}")
|
|
print(f"Last Updated: {weather_data.get('updated', 'N/A')}")
|
|
print("\nForecast Periods:")
|
|
|
|
for period in weather_data.get("periods", []):
|
|
print(f"\n{period['name']}:")
|
|
print(f" Temperature: {period['temperature']}{period['temperatureUnit']}")
|
|
print(f" Wind: {period['windSpeed']} {period['windDirection']}")
|
|
print(f" Forecast: {period['shortForecast']}")
|
|
print(f" Details: {period['detailedForecast']}")
|
|
|
|
# %%
|
|
|
|
# Stock related function
|
|
def get_ticker_price(ticker_symbols):
|
|
"""
|
|
Look up the current price of a stock using its ticker symbol.
|
|
|
|
Args:
|
|
ticker_symbol (str): The stock ticker symbol (e.g., 'AAPL' for Apple)
|
|
|
|
Returns:
|
|
dict: Current stock information including price
|
|
"""
|
|
results = []
|
|
print(f"get_ticker_price('{ticker_symbols}')")
|
|
for ticker_symbol in ticker_symbols.split(','):
|
|
ticker_symbol = ticker_symbol.strip()
|
|
if ticker_symbol == "":
|
|
continue
|
|
# Create a Ticker object
|
|
try:
|
|
ticker = yf.Ticker(ticker_symbol)
|
|
|
|
# Get the latest market data
|
|
ticker_data = ticker.history(period="1d")
|
|
|
|
if ticker_data.empty:
|
|
results.append({"error": f"No data found for ticker {ticker_symbol}"})
|
|
continue
|
|
|
|
# Get the latest closing price
|
|
latest_price = ticker_data['Close'].iloc[-1]
|
|
|
|
# Get some additional info
|
|
info = ticker.info
|
|
results.append({ 'symbol': ticker_symbol, 'price': latest_price })
|
|
|
|
except Exception as e:
|
|
results.append({"error": f"Error fetching data for {ticker_symbol}: {str(e)}"})
|
|
|
|
return results[0] if len(results) == 1 else results
|
|
#{
|
|
# "symbol": ticker_symbol,
|
|
# "price": latest_price,
|
|
# "currency": info.get("currency", "Unknown"),
|
|
# "company_name": info.get("shortName", "Unknown"),
|
|
# "previous_close": info.get("previousClose", "Unknown"),
|
|
# "market_cap": info.get("marketCap", "Unknown"),
|
|
#}
|
|
|
|
|
|
# %%
|
|
def get_current_datetime(timezone="America/Los_Angeles"):
|
|
"""
|
|
Returns the current date and time in the specified timezone in ISO 8601 format.
|
|
|
|
Args:
|
|
timezone (str): Timezone name (e.g., "UTC", "America/New_York", "Europe/London")
|
|
Default is "America/Los_Angeles"
|
|
|
|
Returns:
|
|
str: Current date and time with timezone in the format YYYY-MM-DDTHH:MM:SS+HH:MM
|
|
"""
|
|
try:
|
|
if timezone == 'system' or timezone == '' or not timezone:
|
|
timezone = 'America/Los_Angeles'
|
|
# Get current UTC time (timezone-aware)
|
|
local_tz = pytz.timezone("America/Los_Angeles")
|
|
local_now = datetime.now(tz=local_tz)
|
|
|
|
# Convert to target timezone
|
|
target_tz = pytz.timezone(timezone)
|
|
target_time = local_now.astimezone(target_tz)
|
|
|
|
return target_time.isoformat()
|
|
except Exception as e:
|
|
return {'error': f"Invalid timezone {timezone}: {str(e)}"}
|
|
|
|
|
|
# %%
|
|
tools = [{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "get_ticker_price",
|
|
"description": "Get the current stock price of one or more ticker symbols. Returns an array of objects with 'symbol' and 'price' fields. Call this whenever you need to know the latest value of stock ticker symbols, for example when a user asks 'How much is Intel trading at?' or 'What are the prices of AAPL and MSFT?'",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"ticker": {
|
|
"type": "string",
|
|
"description": "The company stock ticker symbol. For multiple tickers, provide a comma-separated list (e.g., 'AAPL,MSFT,GOOGL').",
|
|
},
|
|
},
|
|
"required": ["ticker"],
|
|
"additionalProperties": False
|
|
}
|
|
}
|
|
}, {
|
|
"type": "function",
|
|
"function": {
|
|
"name": "get_current_datetime",
|
|
"description": "Get the current date and time in a specified timezone",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"timezone": {
|
|
"type": "string",
|
|
"description": "Timezone name (e.g., 'UTC', 'America/New_York', 'Europe/London', 'America/Los_Angeles'). Default is 'America/Los_Angeles'."
|
|
}
|
|
},
|
|
"required": []
|
|
}
|
|
}
|
|
}, {
|
|
"type": "function",
|
|
"function": {
|
|
"name": "get_weather_by_location",
|
|
"description": "Get the current and future weather for a given CITY and STATE location in the United States. For example, if the user asks 'What is the weather in Portland?' or 'What is the forecast for tomorrow?'",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"city": {
|
|
"type": "string",
|
|
"description": "City to find the weather (e.g., 'Portland', 'Seattle')."
|
|
},
|
|
"state": {
|
|
"type": "string",
|
|
"description": "State to find the weather (e.g., 'OR', 'WA')."
|
|
}
|
|
},
|
|
"required": [ "city", "state" ],
|
|
"additionalProperties": False
|
|
}
|
|
}
|
|
}]
|
|
|
|
__all__ = [ 'tools', 'get_current_datetime', 'get_weather_by_location', 'get_ticker_price' ] |