Update tools.py

tools.py

@@ -1,17 +1,22 @@
-#Tools
-from smolagents import DuckDuckGoSearchTool
 from smolagents import Tool
 import random
 from huggingface_hub import list_models
-
+import requests
+import os
+import sqlite3
+from googletrans import Translator
+from gtts import gTTS
+import speech_recognition as sr
+import cv2
+import numpy as np
+from textblob import TextBlob
 
 # Initialize the DuckDuckGo search tool
-#search_tool = DuckDuckGoSearchTool()
-
+# search_tool = DuckDuckGoSearchTool()
 
 class WeatherInfoTool(Tool):
     name = "weather_info"
-    description = "Fetches dummy weather information for a given location."
+    description = "Fetches weather information for a given location."
     inputs = {
         "location": {
             "type": "string",
@@ -21,15 +26,20 @@ class WeatherInfoTool(Tool):
     output_type = "string"
 
     def forward(self, location: str):
-        # Dummy weather data
-        weather_conditions = [
-            {"condition": "Rainy", "temp_c": 15},
-            {"condition": "Clear", "temp_c": 25},
-            {"condition": "Windy", "temp_c": 20}
-        ]
-        # Randomly select a weather condition
-        data = random.choice(weather_conditions)
-        return f"Weather in {location}: {data['condition']}, {data['temp_c']}°C"
+        # Use a real weather API here
+        api_key = os.getenv("WEATHER_API_KEY")
+        if not api_key:
+            return "Weather API key not found."
+
+        try:
+            response = requests.get(f"http://api.weatherapi.com/v1/current.json?key={api_key}&q={location}")
+            response.raise_for_status()
+            data = response.json()
+            condition = data["current"]["condition"]["text"]
+            temp_c = data["current"]["temp_c"]
+            return f"Weather in {location}: {condition}, {temp_c}°C"
+        except Exception as e:
+            return f"Error fetching weather for {location}: {str(e)}"
 
 class HubStatsTool(Tool):
     name = "hub_stats"
@@ -46,11 +56,342 @@ class HubStatsTool(Tool):
         try:
             # List models from the specified author, sorted by downloads
             models = list(list_models(author=author, sort="downloads", direction=-1, limit=1))
-            
+
             if models:
                 model = models[0]
                 return f"The most downloaded model by {author} is {model.id} with {model.downloads:,} downloads."
             else:
                 return f"No models found for author {author}."
         except Exception as e:
-            return f"Error fetching models for {author}: {str(e)}"
+            return f"Error fetching models for {author}: {str(e)}"
+
+class CalculatorTool(Tool):
+    name = "calculator"
+    description = "Performs mathematical calculations."
+    inputs = {
+        "expression": {
+            "type": "string",
+            "description": "The mathematical expression to evaluate."
+        }
+    }
+    output_type = "string"
+
+    def forward(self, expression: str):
+        try:
+            result = eval(expression)
+            return f"The result of the expression '{expression}' is {result}."
+        except Exception as e:
+            return f"Error evaluating expression: {str(e)}"
+
+class CalendarTool(Tool):
+    name = "calendar"
+    description = "Manages and retrieves information about dates and events."
+    inputs = {
+        "action": {
+            "type": "string",
+            "description": "The action to perform (e.g., 'add', 'get', 'delete')."
+        },
+        "date": {
+            "type": "string",
+            "description": "The date of the event (format: YYYY-MM-DD)."
+        },
+        "event": {
+            "type": "string",
+            "description": "The event description."
+        }
+    }
+    output_type = "string"
+
+    def __init__(self):
+        self.events = {}
+
+    def forward(self, action: str, date: str, event: str = None):
+        if action == "add":
+            self.events[date] = event
+            return f"Event '{event}' added to {date}."
+        elif action == "get":
+            return f"Event on {date}: {self.events.get(date, 'No event found.')}"
+        elif action == "delete":
+            if date in self.events:
+                del self.events[date]
+                return f"Event on {date} deleted."
+            else:
+                return f"No event found on {date}."
+        else:
+            return "Invalid action."
+
+class EmailTool(Tool):
+    name = "email"
+    description = "Sends and receives emails."
+    inputs = {
+        "action": {
+            "type": "string",
+            "description": "The action to perform (e.g., 'send')."
+        },
+        "to": {
+            "type": "string",
+            "description": "The recipient's email address."
+        },
+        "subject": {
+            "type": "string",
+            "description": "The subject of the email."
+        },
+        "body": {
+            "type": "string",
+            "description": "The body of the email."
+        }
+    }
+    output_type = "string"
+
+    def __init__(self, smtp_server, smtp_port, email, password):
+        self.smtp_server = smtp_server
+        self.smtp_port = smtp_port
+        self.email = email
+        self.password = password
+
+    def forward(self, action: str, to: str, subject: str, body: str):
+        if action == "send":
+            try:
+                msg = MIMEText(body)
+                msg['Subject'] = subject
+                msg['From'] = self.email
+                msg['To'] = to
+
+                with smtplib.SMTP(self.smtp_server, self.smtp_port) as server:
+                    server.starttls()
+                    server.login(self.email, self.password)
+                    server.sendmail(self.email, [to], msg.as_string())
+
+                return f"Email sent to {to}."
+            except Exception as e:
+                return f"Error sending email: {str(e)}"
+        else:
+            return "Invalid action."
+
+class FileManagementTool(Tool):
+    name = "file_management"
+    description = "Handles file operations like reading, writing, and managing files."
+    inputs = {
+        "action": {
+            "type": "string",
+            "description": "The action to perform (e.g., 'read', 'write', 'delete')."
+        },
+        "file_path": {
+            "type": "string",
+            "description": "The path of the file."
+        },
+        "content": {
+            "type": "string",
+            "description": "The content to write to the file."
+        }
+    }
+    output_type = "string"
+
+    def forward(self, action: str, file_path: str, content: str = None):
+        if action == "read":
+            try:
+                with open(file_path, 'r') as file:
+                    content = file.read()
+                return f"Content of {file_path}: {content}"
+            except Exception as e:
+                return f"Error reading file: {str(e)}"
+        elif action == "write":
+            try:
+                with open(file_path, 'w') as file:
+                    file.write(content)
+                return f"Content written to {file_path}."
+            except Exception as e:
+                return f"Error writing to file: {str(e)}"
+        elif action == "delete":
+            try:
+                os.remove(file_path)
+                return f"File {file_path} deleted."
+            except Exception as e:
+                return f"Error deleting file: {str(e)}"
+        else:
+            return "Invalid action."
+
+class DatabaseQueryTool(Tool):
+    name = "database_query"
+    description = "Interacts with databases for storing and retrieving information."
+    inputs = {
+        "action": {
+            "type": "string",
+            "description": "The action to perform (e.g., 'query', 'insert')."
+        },
+        "query": {
+            "type": "string",
+            "description": "The SQL query to execute."
+        }
+    }
+    output_type = "string"
+
+    def __init__(self, db_path):
+        self.db_path = db_path
+
+    def forward(self, action: str, query: str):
+        try:
+            conn = sqlite3.connect(self.db_path)
+            cursor = conn.cursor()
+
+            if action == "query":
+                cursor.execute(query)
+                results = cursor.fetchall()
+                return f"Query results: {results}"
+            elif action == "insert":
+                cursor.execute(query)
+                conn.commit()
+                return "Data inserted successfully."
+            else:
+                return "Invalid action."
+
+        except Exception as e:
+            return f"Error executing query: {str(e)}"
+        finally:
+            conn.close()
+
+class TranslationTool(Tool):
+    name = "translation"
+    description = "Translates text between different languages."
+    inputs = {
+        "text": {
+            "type": "string",
+            "description": "The text to translate."
+        },
+        "src_lang": {
+            "type": "string",
+            "description": "The source language code."
+        },
+        "dest_lang": {
+            "type": "string",
+            "description": "The destination language code."
+        }
+    }
+    output_type = "string"
+
+    def forward(self, text: str, src_lang: str, dest_lang: str):
+        try:
+            translator = Translator()
+            translation = translator.translate(text, src=src_lang, dest=dest_lang)
+            return f"Translated text: {translation.text}"
+        except Exception as e:
+            return f"Error translating text: {str(e)}"
+
+class TextToSpeechTool(Tool):
+    name = "text_to_speech"
+    description = "Converts text to speech."
+    inputs = {
+        "text": {
+            "type": "string",
+            "description": "The text to convert to speech."
+        }
+    }
+    output_type = "string"
+
+    def forward(self, text: str):
+        try:
+            tts = gTTS(text=text, lang='en')
+            tts.save("output.mp3")
+            return "Text converted to speech and saved as output.mp3."
+        except Exception as e:
+            return f"Error converting text to speech: {str(e)}"
+
+class SpeechToTextTool(Tool):
+    name = "speech_to_text"
+    description = "Converts speech to text."
+    inputs = {
+        "audio_file": {
+            "type": "string",
+            "description": "The path to the audio file to convert to text."
+        }
+    }
+    output_type = "string"
+
+    def forward(self, audio_file: str):
+        try:
+            recognizer = sr.Recognizer()
+            with sr.AudioFile(audio_file) as source:
+                audio = recognizer.record(source)
+                text = recognizer.recognize_google(audio)
+                return f"Converted speech to text: {text}"
+        except Exception as e:
+            return f"Error converting speech to text: {str(e)}"
+
+class ImageRecognitionTool(Tool):
+    name = "image_recognition"
+    description = "Analyzes and interprets images."
+    inputs = {
+        "image_path": {
+            "type": "string",
+            "description": "The path to the image to analyze."
+        }
+    }
+    output_type = "string"
+
+    def forward(self, image_path: str):
+        try:
+            image = cv2.imread(image_path)
+            gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
+            faces = cv2.CascadeClassifier(cv2.data.haarcascades + 'haarcascade_frontalface_default.xml').detectMultiScale(gray, 1.3, 5)
+            return f"Found {len(faces)} faces in the image."
+        except Exception as e:
+            return f"Error analyzing image: {str(e)}"
+
+class NLPTool(Tool):
+    name = "nlp"
+    description = "Performs advanced text processing tasks like sentiment analysis, named entity recognition, etc."
+    inputs = {
+        "text": {
+            "type": "string",
+            "description": "The text to analyze."
+        },
+        "task": {
+            "type": "string",
+            "description": "The NLP task to perform (e.g., 'sentiment', 'entities')."
+        }
+    }
+    output_type = "string"
+
+    def forward(self, text: str, task: str):
+        blob = TextBlob(text)
+        if task == "sentiment":
+            sentiment = blob.sentiment
+            return f"Sentiment analysis: Polarity={sentiment.polarity}, Subjectivity={sentiment.subjectivity}"
+        elif task == "entities":
+            entities = blob.noun_phrases
+            return f"Named entities: {entities}"
+        else:
+            return "Invalid task."
+
+class APIIntegrationTool(Tool):
+    name = "api_integration"
+    description = "Interacts with various external APIs for fetching or sending data."
+    inputs = {
+        "api_url": {
+            "type": "string",
+            "description": "The URL of the API endpoint."
+        },
+        "method": {
+            "type": "string",
+            "description": "The HTTP method to use (e.g., 'GET', 'POST')."
+        },
+        "data": {
+            "type": "string",
+            "description": "The data to send with the request."
+        }
+    }
+    output_type = "string"
+
+    def forward(self, api_url: str, method: str, data: str = None):
+        try:
+            if method == "GET":
+                response = requests.get(api_url)
+            elif method == "POST":
+                response = requests.post(api_url, json=data)
+            else:
+                return "Invalid method."
+
+            response.raise_for_status()
+            return f"API response: {response.json()}"
+        except Exception as e:
+            return f"Error interacting with API: {str(e)}"