app.py

import tensorflow
import torch
import gradio as gr
from transformers import pipeline, AutoModelForCausalLM, AutoTokenizer
import gradio as gr
import importlib
import subprocess
import sys
import io
import threading
import time
from flask import Flask

def install_and_import(package_name):
    """Installs a package using pip and imports it."""
    subprocess.check_call(["pip", "install", package_name])
    return importlib.import_module(package_name)

def extract_package_name(input_str):
    """Extracts the package name from a PyPI URL or pip command."""
    if input_str.startswith("https://pypi.org/project/"):
        return input_str.split("/")[-2]
    elif input_str.startswith("pip install "):
        return input_str.split(" ")[2]
    else:
        return input_str

def create_interface_from_input(input_str):
    """
    Creates a Gradio interface with buttons for functions from a package.
    """
    try:
        package_name = extract_package_name(input_str)
        module = install_and_import(package_name)

        # Handle Flask application context if needed
        if 'flask' in sys.modules or 'flask_restful' in sys.modules:
            app = Flask(__name__)
            with app.app_context():
                functions = [getattr(module, name) for name in dir(module) if callable(getattr(module, name))]
        else:
            functions = [getattr(module, name) for name in dir(module) if callable(getattr(module, name))]

        function_list = [(func.__name__, func) for func in functions if not func.__name__.startswith("_")]
        return function_list, f"Interface for `{package_name}`"

    except Exception as e:
        return [], str(e)

def execute_pip_command(command, add_message):
    """Executes a pip command and streams the output."""
    process = subprocess.Popen(command.split(), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
    while True:
        output = process.stdout.readline()
        if output == '' and process.poll() is not None:
            break
        if output:
            add_message("System", f"```\n{output.strip()}\n```")
            time.sleep(0.1)  # Simulate delay for more realistic streaming
    rc = process.poll()
    return rc

# --- Load the NLP pipeline for text classification ---
classifier = pipeline("text-classification")

# --- Define the function to generate mini-apps based on user input ---
def generate_mini_apps(theme):
    # Use the NLP pipeline to classify the input theme
    classification = classifier(theme)

    # Generate a set of mini-apps based on the classification
    if classification[0]['label'] == 'Productivity':
        mini_apps = [
            'Idea-to-Codebase Generator',
            'Automated GitHub Repo Guardian Angel',
            'AI-Powered IDE'
        ]
    elif classification[0]['label'] == 'Creativity':
        mini_apps = [
            'Brainstorming Assistant',
            'Mood Board Generator',
            'Writing Assistant'
        ]
    elif classification[0]['label'] == 'Well-being':
        mini_apps = [
            'Meditation Guide',
            'Mood Tracker',
            'Sleep Tracker'
        ]
    else:
        mini_apps = ["No matching mini-apps found. Try a different theme."]

    # Return the generated mini-apps
    return mini_apps

# --- Load the model and tokenizer from the provided files ---
model = AutoModelForCausalLM.from_pretrained("./", trust_remote_code=True)  # Load from the current directory
tokenizer = AutoTokenizer.from_pretrained("./")

# --- Define a function to generate text using the model ---
def generate_text(input_text):
    inputs = tokenizer(input_text, return_tensors="pt")
    output = model.generate(**inputs, max_length=50, num_return_sequences=1)
    return tokenizer.decode(output[0], skip_special_tokens=True)

# --- Create the Gradio interface ---
demo = gr.Interface(
    fn=generate_mini_apps,
    inputs=gr.Textbox(label="Enter a theme for your life"),
    outputs=gr.Textbox(label="Generated Mini-Apps"),
    title="AI4ME: Personalized AI Tools",
    description="Enter your hobby/interest/job and we'll generate a set of AI-powered mini-apps tailored to your specific needs."
)

# --- Add a text generation tab ---
with gr.Blocks() as demo_text:
    gr.Markdown("## Text Generation")
    input_text = gr.Textbox(label="Enter your text")
    output_text = gr.Textbox(label="Generated Text")
    input_text.submit(generate_text, inputs=input_text, outputs=output_text)

def handle_chat(input_text, history):
    """Handles the chat input and updates the chat history."""
    def add_message(sender, message):
        history.append((sender, message))

    add_message("User", input_text)

    if input_text.startswith("pip install ") or input_text.startswith("https://pypi.org/project/"):
        package_name = extract_package_name(input_text)
        add_message("System", f"Installing `{package_name}`...")

        execute_pip_command(input_text, add_message)
        
        function_list, message = create_interface_from_input(input_text)
        add_message("System", message)
        
        if function_list:
            functions_str = "\n".join([f" - {name}()" for name, _ in function_list])
            add_message("System", f"Available functions:\n{functions_str}")

        return history, function_list
    else:
        # Check if the input is to call a function
        if '(' in input_text and ')' in input_text:
            func_name = input_text.split('(')[0].strip()
            func_args = input_text.split('(')[1].split(')')[0].strip()
            
            if func_args:
                func_args = [arg.strip() for arg in func_args.split(',')]

            # Find the function in the current dynamic interface
            for name, func in dynamic_functions:
                if func_name == name:
                    try:
                        result = func(*func_args)
                        add_message("System", f"Result of {func_name}({', '.join(func_args)}): {result}")
                    except Exception as e:
                        add_message("System", f"Error: {str(e)}")
                    break
            else:
                add_message("System", f"Function '{func_name}' not found.")
        else:
            add_message("System", "Invalid function call. Please use the format 'function_name(arg1, arg2, ...)'")

        return history, dynamic_functions

# Initialize dynamic functions list to store available functions
dynamic_functions = []

# Gradio app
with gr.Blocks() as demo:
    with gr.Row():
        chat_interface = gr.Chatbot()
    
    with gr.Row():
        chat_input = gr.Textbox(placeholder="Enter pip command or package URL", show_label=False)
        submit_button = gr.Button("Submit")

    def chat_handler(input_text, history):
        global dynamic_functions
        history, new_functions = handle_chat(input_text, history)
        dynamic_functions = new_functions
        return history

    submit_button.click(fn=chat_handler, inputs=[chat_input, chat_interface], outputs=[chat_interface])
    chat_input.submit(fn=chat_handler, inputs=[chat_input, chat_interface], outputs=[chat_interface])

demo.launch()