toolbox.py

from langchain_core.tools.simple import Tool

from duckduckgo_search import DDGS
import pint
import sympy as sp


class _Math:

    @staticmethod
    def symbolic_calc(expression: str) -> str:
        """
        Evaluates complex mathematical expressions using SymPy.

        Args:
            expression (str): Mathematical expression as string

        Returns:
            Result of the calculation
        """
        print(f"-> symbolic_math_calculator tool used (input: {expression}) !")
        try:
            result = sp.sympify(expression)
            print(f"-> (output: {result}) !")
            return str(result)
        except Exception as e:
            return f"Error in calculation: {str(e)}"

    @staticmethod
    def unit_converter(value: float, from_unit: str, to_unit: str) -> str:
        """
        Converts values between different units of measurement.

        Args:
            value (float): The numerical value to convert
            from_unit (str): The source unit (e.g., 'meter', 'kg', 'celsius')
            to_unit (str): The target unit (e.g., 'feet', 'pound', 'fahrenheit')

        Returns:
            The converted value with appropriate units
        """
        print(f"-> unit_converter tool used (inputs: [value: {value}, from_unit: {from_unit}, to_unit: {to_unit}]) !")
        try:
            # Create unit registry
            ureg = pint.UnitRegistry()
            
            # Create quantity with source unit
            quantity = value * ureg(from_unit)
            
            # Convert to target unit
            result = quantity.to(to_unit)
            
            answer = f"{value} {from_unit} = {result.magnitude} {to_unit}"
            print(f"-> (output: {result}) !")
            return answer
        except Exception as e:
            return f"Error in unit conversion: {str(e)}"


class MathToolbox:
    symbolic_calc = Tool(
        name="symbolic_calc",
        func=_Math.symbolic_calc,
        description=_Math.symbolic_calc. __doc__ or ""
    )
    unit_converter = Tool(
        name="unit_converter",
        func=_Math.unit_converter,
        description=_Math.unit_converter. __doc__ or ""
    )


class _WebSearch:
    @staticmethod
    def duckduckgo_text_search(keywords, max_results=5) -> list[dict[str, str]]:
        """DuckDuckGo text search.

        Args:
            keywords (str): keywords for query.
            max_results (int): max number of results. If None, returns results only from the first response. Defaults to 5.

        Returns:
            List of dictionaries with search results, or None if there was an error.

        Raises:
            DuckDuckGoSearchException: Base exception for duckduckgo_search errors.
            RatelimitException: Inherits from DuckDuckGoSearchException, raised for exceeding API request rate limits.
            TimeoutException: Inherits from DuckDuckGoSearchException, raised for API request timeouts.
        """
        return DDGS().text(keywords, max_results=max_results)

    @staticmethod
    def duckduckgo_images_search(keywords, license = None, max_results=5) -> list[dict[str, str]]:
        """DuckDuckGo images search.

        Args:
            keywords (str): keywords for query.
            license (str|None): any (All Creative Commons), Public (PublicDomain),
                Share (Free to Share and Use), ShareCommercially (Free to Share and Use Commercially),
                Modify (Free to Modify, Share, and Use), ModifyCommercially (Free to Modify, Share, and
                Use Commercially). Defaults to None.
            max_results (int): max number of results. If None, returns results only from the first response. Defaults to 5.

        Returns:
            List of dictionaries with images search results.

        Raises:
            DuckDuckGoSearchException: Base exception for duckduckgo_search errors.
            RatelimitException: Inherits from DuckDuckGoSearchException, raised for exceeding API request rate limits.
            TimeoutException: Inherits from DuckDuckGoSearchException, raised for API request timeouts.
        """
        return DDGS().images(keywords, license_image=license, max_results=max_results)

    @staticmethod
    def duckduckgo_videos_search(keywords, license = None, max_results=5) -> list[dict[str, str]]:
        """DuckDuckGo videos search.

        Args:
            keywords (str): keywords for query.
            license (str|None): creativeCommon, youtube. Defaults to None.
            max_results (int): max number of results. If None, returns results only from the first response. Defaults to 5.

        Returns:
            List of dictionaries with videos search results.

        Raises:
            DuckDuckGoSearchException: Base exception for duckduckgo_search errors.
            RatelimitException: Inherits from DuckDuckGoSearchException, raised for exceeding API request rate limits.
            TimeoutException: Inherits from DuckDuckGoSearchException, raised for API request timeouts.
        """
        return DDGS().videos(keywords, license_videos=license, max_results=max_results)


class WebSearchToolbox:
    duckduckgo_text_search = Tool(
        name="duckduckgo_text_search",
        func=_WebSearch.duckduckgo_text_search,
        description=_WebSearch.duckduckgo_text_search. __doc__ or ""
    )
    duckduckgo_images_search = Tool(
        name="duckduckgo_images_search",
        func=_WebSearch.duckduckgo_images_search,
        description=_WebSearch.duckduckgo_images_search. __doc__ or ""
    )
    duckduckgo_videos_search = Tool(
        name="duckduckgo_videos_search",
        func=_WebSearch.duckduckgo_videos_search,
        description=_WebSearch.duckduckgo_videos_search. __doc__ or ""
    )


class _Encryption:

    @staticmethod
    def ascii_encode(text: str) -> str:
        """
        Converts each character in a string to its ASCII value.

        Args:
            text (str): The text to encode

        Returns:
            Space-separated ASCII values
        """
        print(f"-> ascii_encode tool used (input: {text[:30]}...) !")
        try:
            ascii_values = [str(ord(char)) for char in text]
            result = " ".join(ascii_values)
            return result
        except Exception as e:
            return f"Error in ASCII encoding: {str(e)}"

    @staticmethod
    def ascii_decode(text: str) -> str:
        """
        Converts space-separated ASCII values back to characters.

        Args:
            text (str): Space-separated ASCII values

        Returns:
            Decoded string
        """
        print(f"-> ascii_decode tool used (input: {text[:30]}...) !")
        try:
            ascii_values = text.split()
            result = "".join([chr(int(value)) for value in ascii_values])
            return result
        except Exception as e:
            return f"Error in ASCII decoding: {str(e)}"

    @staticmethod
    def base64_encode(text: str) -> str:
        """
        Encodes a string to base64.

        Args:
            text (str): The text to encode

        Returns:
            Base64 encoded string
        """
        print(f"-> base64_encode tool used (input: {text[:30]}...) !")
        import base64
        try:
            encoded_bytes = base64.b64encode(text.encode('utf-8'))
            encoded_text = encoded_bytes.decode('utf-8')
            return encoded_text
        except Exception as e:
            return f"Error in base64 encoding: {str(e)}"

    @staticmethod
    def base64_decode(encoded_text: str) -> str:
        """
        Decodes a base64 string to plain text.

        Args:
            encoded_text (str): The base64 encoded text

        Returns:
            Decoded string
        """
        print(f"-> base64_decode tool used (input: {encoded_text[:30]}...) !")
        import base64
        try:
            decoded_bytes = base64.b64decode(encoded_text)
            decoded_text = decoded_bytes.decode('utf-8')
            return decoded_text
        except Exception as e:
            return f"Error in base64 decoding: {str(e)}"

    @staticmethod
    def caesar_cipher_encode(text: str, shift: int) -> str:
        """
        Encodes text using Caesar cipher with specified shift.

        Args:
            text (str): The text to encode
            shift (int): Number of positions to shift each character

        Returns:
            Caesar cipher encoded string
        """
        print(f"-> caesar_cipher_encode tool used (input: {text[:30]}..., shift: {shift}) !")
        result = ""
        try:
            for char in text:
                if char.isalpha():
                    ascii_offset = ord('a') if char.islower() else ord('A')
                    encoded_char = chr((ord(char) - ascii_offset + shift) % 26 + ascii_offset)
                    result += encoded_char
                else:
                    result += char
            return result
        except Exception as e:
            return f"Error in Caesar cipher encoding: {str(e)}"

    @classmethod
    def caesar_cipher_decode(cls, encoded_text: str, shift: int) -> str:
        """
        Decodes Caesar cipher text with specified shift.

        Args:
            encoded_text (str): The encoded text
            shift (int): Number of positions the text was shifted

        Returns:
            Decoded string
        """
        print(f"-> caesar_cipher_decode tool used (input: {encoded_text[:30]}..., shift: {shift}) !")
        # To decode, we shift in the opposite direction
        return cls.caesar_cipher_encode(encoded_text, -shift)

    @classmethod
    def caesar_cipher_brute_force(cls, text: str) -> str:
        """
        Performs a brute force attack on a Caesar cipher by trying all 26 shifts.

        Args:
            text (str): The Caesar cipher encoded text

        Returns:
            All possible decoding results with their respective shifts
        """
        print(f"-> caesar_cipher_brute_force tool used (input: {text[:30]}...) !")
        results = []

        # Try all 26 possible shifts for English alphabet
        for shift in range(26):
            decoded = cls.caesar_cipher_decode(text, shift)
            results.append(f"Shift {shift}: {decoded}")

        output = "\n".join(results)
        return output

    @staticmethod
    def reverse_string(text: str) -> str:
        """
        Reverses a string.

        Args:
            text (str): The text to reverse

        Returns:
            Reversed string
        """
        print(f"-> reverse_string tool used (input: {text[:30]}...) !")
        reversed_text = text[::-1]
        return reversed_text


class EncryptionToolbox:
    ascii_encode = Tool(
        name="ascii_encode",
        func=_Encryption.ascii_encode,
        description=_Encryption.ascii_encode. __doc__ or ""
    )
    ascii_decode = Tool(
        name="ascii_decode",
        func=_Encryption.ascii_decode,
        description=_Encryption.ascii_decode. __doc__ or ""
    )
    base64_encode = Tool(
        name="base64_encode",
        func=_Encryption.base64_encode,
        description=_Encryption.base64_encode. __doc__ or ""
    )
    base64_decode = Tool(
        name="base64_decode",
        func=_Encryption.base64_decode,
        description=_Encryption.base64_decode. __doc__ or ""
    )
    caesar_cipher_encode = Tool(
        name="caesar_cipher_encode",
        func=_Encryption.caesar_cipher_encode,
        description=_Encryption.caesar_cipher_encode. __doc__ or ""
    )
    caesar_cipher_decode = Tool(
        name="caesar_cipher_decode",
        func=_Encryption.caesar_cipher_decode,
        description=_Encryption.caesar_cipher_decode. __doc__ or ""
    )
    caesar_cipher_brute_force = Tool(
        name="caesar_cipher_brute_force",
        func=_Encryption.caesar_cipher_brute_force,
        description=_Encryption.caesar_cipher_brute_force. __doc__ or ""
    )
    reverse_string = Tool(
        name="reverse_string",
        func=_Encryption.reverse_string,
        description=_Encryption.reverse_string. __doc__ or ""
    )


class Toolbox:
    math = MathToolbox()
    web_search = WebSearchToolbox()
    encryption = EncryptionToolbox()