92 lines
3.3 KiB
Python
92 lines
3.3 KiB
Python
import os
|
|
import requests
|
|
from typing import Literal, List, Optional
|
|
from datetime import datetime
|
|
|
|
|
|
from blueprints.function_calling_blueprint import Pipeline as FunctionCallingBlueprint
|
|
|
|
|
|
class Pipeline(FunctionCallingBlueprint):
|
|
class Tools:
|
|
def __init__(self, pipeline) -> None:
|
|
self.pipeline = pipeline
|
|
|
|
def get_current_time(
|
|
self,
|
|
) -> str:
|
|
"""
|
|
Get the current time.
|
|
|
|
:return: The current time.
|
|
"""
|
|
|
|
now = datetime.now()
|
|
current_time = now.strftime("%H:%M:%S")
|
|
return f"Current Time = {current_time}"
|
|
|
|
def get_current_weather(
|
|
self,
|
|
location: str,
|
|
unit: Literal["metric", "fahrenheit"] = "fahrenheit",
|
|
) -> str:
|
|
"""
|
|
Get the current weather for a location. If the location is not found, return an empty string.
|
|
|
|
:param location: The location to get the weather for.
|
|
:param unit: The unit to get the weather in. Default is fahrenheit.
|
|
:return: The current weather for the location.
|
|
"""
|
|
|
|
# https://openweathermap.org/api
|
|
|
|
if self.pipeline.valves.OPENWEATHERMAP_API_KEY == "":
|
|
return "OpenWeatherMap API Key not set, ask the user to set it up."
|
|
else:
|
|
units = "imperial" if unit == "fahrenheit" else "metric"
|
|
params = {
|
|
"q": location,
|
|
"appid": self.pipeline.valves.OPENWEATHERMAP_API_KEY,
|
|
"units": units,
|
|
}
|
|
|
|
response = requests.get(
|
|
"http://api.openweathermap.org/data/2.5/weather", params=params
|
|
)
|
|
response.raise_for_status() # Raises an HTTPError for bad responses
|
|
data = response.json()
|
|
|
|
weather_description = data["weather"][0]["description"]
|
|
temperature = data["main"]["temp"]
|
|
|
|
return f"{location}: {weather_description.capitalize()}, {temperature}°{unit.capitalize()[0]}"
|
|
|
|
def calculator(self, equation: str) -> str:
|
|
"""
|
|
Calculate the result of an equation.
|
|
|
|
:param equation: The equation to calculate.
|
|
"""
|
|
|
|
# Avoid using eval in production code
|
|
# https://nedbatchelder.com/blog/201206/eval_really_is_dangerous.html
|
|
try:
|
|
result = eval(equation)
|
|
return f"{equation} = {result}"
|
|
except Exception as e:
|
|
print(e)
|
|
return "Invalid equation"
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.valves = self.Valves(
|
|
**{
|
|
"pipelines": ["pipeline-testmodel"], # Connect to all pipelines
|
|
}
|
|
)
|
|
self.tools = self.Tools(self)
|
|
# Optionally, you can set the id and name of the pipeline.
|
|
# Best practice is to not specify the id so that it can be automatically inferred from the filename, so that users can install multiple versions of the same pipeline.
|
|
# The identifier must be unique across all pipelines.
|
|
# The identifier must be an alphanumeric string that can include underscores or hyphens. It cannot contain spaces, special characters, slashes, or backslashes.
|
|
# self.id = "my_tools_pipeline" |