import os import ast import io import sys import numpy as np import pandas as pd import scipy from pathlib import Path import mimetypes import base64 from google import genai import requests ALLOWED_MODULES = {"numpy", "pandas", "scipy"} GEMINI_API_KEY = os.getenv("GEMINI_TOKEN") GEMINI_MODEL_NAME = "gemini-2.0-flash" def interpret_python_math_code(python_code: str) -> str: """ Interprets a string of Python code to perform math calculations. Security Note: This function uses exec(). While it attempts to restrict imports to numpy, pandas, and scipy, and runs with a restricted global scope, executing arbitrary code always carries risks. Ensure that input code is from a trusted source or properly sanitized. The code must only import modules from the allowed list: numpy, pandas, scipy. Submodules of these (e.g., numpy.linalg, scipy.stats) are permitted. For example: 'import numpy as np' is allowed. 'from scipy.stats import norm' is allowed. 'import os' is NOT allowed. To return a result, the code should either: 1. End with an expression (e.g., '1 + 1' or 'np.array([1,2,3]).sum()'). 2. Assign the result to a variable named '_result' (e.g., '_result = my_calculation'). Print statements will also be captured and returned along with the result. """ # 1. Validate imports using AST try: tree = ast.parse(python_code) for node in tree.body: if isinstance(node, ast.Import): for alias in node.names: root_module = alias.name.split('.')[0] if root_module not in ALLOWED_MODULES: return (f"Error: Import of '{alias.name}' is not allowed. " f"Only modules from {list(ALLOWED_MODULES)} are permitted.") elif isinstance(node, ast.ImportFrom): if node.module: # Handles cases like 'from . import something' where module is None root_module = node.module.split('.')[0] if root_module not in ALLOWED_MODULES: return (f"Error: Import from '{node.module}' is not allowed. " f"Only modules from {list(ALLOWED_MODULES)} are permitted.") except SyntaxError as e: return f"Syntax Error in input code: {e}" # 2. Prepare execution environment restricted_globals = { "__builtins__": { "print": print, "abs": abs, "round": round, "min": min, "max": max, "sum": sum, "len": len, "range": range, "zip": zip, "enumerate": enumerate, "int": int, "float": float, "str": str, "list": list, "dict": dict, "tuple": tuple, "set": set, "True": True, "False": False, "None": None, "__import__": __import__, # Add this line } # numpy, pandas, scipy are NOT pre-loaded here. # The user's code `import numpy` will use Python's import mechanism. # The AST check above is the primary guard. } local_vars = {} # 3. Capture stdout old_stdout = sys.stdout redirected_output = io.StringIO() sys.stdout = redirected_output # 4. Execute code and retrieve result calculated_value = None result_source = "" output_str = "" try: compiled_code = compile(python_code, '', 'exec') exec(compiled_code, restricted_globals, local_vars) # Priority 1: Check for '_result' variable if "_result" in local_vars: calculated_value = local_vars["_result"] result_source = "variable '_result'" # Priority 2: If no _result, and the last AST node was an expression, evaluate it. elif tree.body and isinstance(tree.body[-1], ast.Expr): # Ensure the expression node's value is a valid AST object for ast.Expression if isinstance(tree.body[-1].value, ast.AST): last_expr_ast = ast.Expression(body=tree.body[-1].value) # Compile the expression in 'eval' mode compiled_expr = compile(last_expr_ast, '', 'eval') # Evaluate in the context of restricted_globals and local_vars (which holds state from exec) calculated_value = eval(compiled_expr, restricted_globals, local_vars) result_source = "last expression" sys.stdout = old_stdout # Restore stdout before getting its value output_str = redirected_output.getvalue() if calculated_value is not None: return f"Result (from {result_source}):\n{calculated_value}\n\nCaptured Output:\n{output_str}".strip() else: return f"Executed successfully.\n\nCaptured Output:\n{output_str}\n(No specific result value found via '_result' variable or last expression evaluation.)".strip() except Exception as e: if sys.stdout == redirected_output: # Ensure stdout is restored on error too sys.stdout = old_stdout output_str = redirected_output.getvalue() # Get any output captured before the error return f"Execution Error: {type(e).__name__}: {e}\n\nCaptured Output:\n{output_str}".strip() finally: # Ensure stdout is always restored if sys.stdout == redirected_output: sys.stdout = old_stdout # STT tool def convert_audio_to_text(path_to_audio: str) -> str: """ Converts speech from an audio file into text. Args: path_to_audio (str): The path to the audio file to be transcribed. An URL can also be used. Returns: str: The transcribed text content of the audio file. """ client = genai.Client(api_key=GEMINI_API_KEY) myfile = client.files.upload(file=path_to_audio) transcription = client.models.generate_content( model=GEMINI_MODEL_NAME, contents=["Provide a transcription of this audio file.", myfile] ) return transcription.text # Analyze image tool def image_understanding(url_to_image: str, question: str) -> str: """ Analyzes an image and generates a response to a given question based on the image's content. An URL needs to be used. Args: path_to_image (str): The URL to the image file to be analyzed. question (str): The question to be answered, based on the contents of the image. Returns: str: The response from a VLM, typically a textual analysis or description based on the image. """ client = genai.Client(api_key=GEMINI_API_KEY) image_bytes = requests.get(url_to_image).content image = genai.types.Part.from_bytes(data=image_bytes, mime_type="image/jpeg") response = client.models.generate_content( model=GEMINI_MODEL_NAME, contents=[question, image], ) return response.text # Analyze video tool def video_understanding(url_to_video: str, question: str) -> str: """ Analyzes a video and generates a response to a given question based on the video's content. Args: url_to_video (str): The URL to the video file to be analyzed (example:YouTube). question (str): The question to be answered, based on the contents of the video. Returns: str: The response from a VLM, typically a textual analysis or description based on the video. """ client = genai.Client(api_key=GEMINI_API_KEY) response = client.models.generate_content( model=GEMINI_MODEL_NAME, contents=genai.types.Content( parts=[ genai.types.Part( file_data=genai.types.FileData(file_uri=url_to_video) ), genai.types.Part(text=question) ] ) ) return response.text ## Read .csv file tool def read_csv_file(path_to_csv: str) -> str: """ Reads a CSV file from the specified path and returns its content as plain text. Args: path_to_csv (str): The file path to the CSV file. Returns: str: Content of the CSV file as plain text. """ try: # Read the CSV file using pandas df = pd.read_csv(path_to_csv) # Return df as plain tect return df.to_string(index=False) except Exception as e: return f"Error reading the CSV file: {e}" ## Read .xlsx file tool def read_xlsx_file(path_to_xlsx: str) -> str: """ Reads a XLSX file from the specified path and returns its content as plain text. Args: path_to_xlsx (str): The file path to the XLSX file. Returns: str: Content of the XLSX file as plain text. """ try: # Read the XLSX file using pandas df = pd.read_excel(path_to_xlsx) # Return df as plain tect return df.to_string(index=False) except Exception as e: return f"Error reading the XLSX file: {e}" # Example usage of the tools if __name__ == "__main__": # Example usage of the tools # print(video_understanding("https://www.youtube.com/watch?v=L1vXCYZAYYM", "What is happening in this video?")) print(image_understanding("https://i.etsystatic.com/28810262/r/il/2fc5e0/5785166966/il_1140xN.5785166966_nvy4.jpg", "What does this image represent?"))