tools.py

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()

class WeatherInfoTool(Tool):
    name = "weather_info"
    description = "Fetches weather information for a given location."
    inputs = {
        "location": {
            "type": "string",
            "description": "The location to get weather information for."
        }
    }
    output_type = "string"

    def forward(self, location: str):
        # 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"
    description = "Fetches the most downloaded model from a specific author on the Hugging Face Hub."
    inputs = {
        "author": {
            "type": "string",
            "description": "The username of the model author/organization to find models from."
        }
    }
    output_type = "string"

    def forward(self, author: str):
        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)}"

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.",
            "nullable": True  # Add this line to specify that 'event' is nullable
        }
    }
    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 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 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)}"