app.py

import gradio as gr
import json
import pandas as pd
import random
import spaces
import io
from PIL import Image, ImageDraw

# ==========================================
# 1. FALLBACK / MOCK ENGINE (When no API Key is provided)
# ==========================================
MOCK_SCHEMAS = {
    "retail": {
        "problem_domain": "Tabular Demand Forecasting",
        "technical_summary": "Uses regression techniques to estimate future weekly store demand metrics based on weather and marketing variables.",
        "recommended_architecture": "XGBoost Regressor",
        "confidence_score": 0.95,
        "fallback_assumptions_made": "",
        "inputs": [
            {"name": "weekly_marketing_spend", "type": "numeric", "range": [1000.0, 50000.0], "description": "Total ad spend in USD."},
            {"name": "is_holiday_week", "type": "categorical", "categories": ["Yes", "No"], "description": "Whether the week contains a national holiday."},
            {"name": "average_temperature_f", "type": "numeric", "range": [-10.0, 110.0], "description": "Average regional temperature in Fahrenheit."}
        ],
        "outputs": [
            {"name": "predicted_store_sales_usd", "type": "numeric", "range": [5000.0, 150000.0], "description": "Forecasted revenue."}
        ],
        "clarifications_needed": []
    },
    "default": {
        "problem_domain": "NLP Sentiment Analysis",
        "technical_summary": "Classifies incoming text reviews to determine operational urgency levels.",
        "recommended_architecture": "DistilBERT Sequence Classifier",
        "confidence_score": 0.8,
        "fallback_assumptions_made": "Assumed user wants a classification engine.",
        "inputs": [
            {"name": "customer_review", "type": "text", "description": "Raw text of the review."}
        ],
        "outputs": [
            {"name": "sentiment_label", "type": "categorical", "categories": ["Positive", "Neutral", "Negative"], "description": "Underlying emotional charge."},
            {"name": "urgency_score", "type": "numeric", "range": [0.0, 1.0], "description": "Required response speed."}
        ],
        "clarifications_needed": ["Would you benefit from named entity extraction as well?"]
    }
}

# ==========================================
# 2. PROMPT TEMPLATES & GEMINI API HANDLER
# ==========================================
DEFAULT_SYSTEM_PROMPT = """You are an expert AI Solutions Architect. Your job is to parse unstructured, chaotic, or vague AI problem statements and translate them into a rigorous, production-ready JSON data contract. This contract will directly drive automated mock data generation and UI scaffolding.



### OUTPUT FORMAT CONSTRAINT

You must output exactly one JSON object. Do not include introductory text, conversational pleasantries, or concluding notes.



### CRITICAL DATA DICTIONARY CONSTRAINTS

To prevent breaking downstream scripts, values for the "type" fields must strictly be one of these exact string literals:

- "text" (for unstructured text, reviews, descriptions)

- "numeric" (for continuous integers or floats, like prices, age, coordinates)

- "categorical" (for discrete classes, labels, choices, or binary classifications)

- "image" (for visual files, bounding box arrays, pixels)



### TARGET JSON SCHEMA

{

  "problem_domain": "string",

  "technical_summary": "string",

  "recommended_architecture": "string",

  "confidence_score": float,

  "fallback_assumptions_made": "string",

  "inputs": [

    {

      "name": "string (snake_case column name)",

      "type": "string (exactly 'text', 'numeric', 'categorical', or 'image')",

      "description": "string",

      "categories": ["string"], // REQUIRED ONLY IF type is 'categorical'

      "range": [float, float] // REQUIRED ONLY IF type is 'numeric'

    }

  ],

  "outputs": [

    {

      "name": "string (snake_case column name)",

      "type": "string (exactly 'text', 'numeric', 'categorical', or 'image')",

      "description": "string",

      "categories": ["string"], // REQUIRED ONLY IF type is 'categorical'

      "range": [float, float] // REQUIRED ONLY IF type is 'numeric'

    }

  ],

  "clarifications_needed": ["string"]

}"""

@spaces.GPU
def call_llm_for_schema(api_key, problem_statement, system_prompt):
    """

    Tries to query the Google Gemini API using the modern google-genai SDK.

    If no key is supplied, defaults cleanly to sandbox simulation data.

    """
    if not api_key or len(api_key.strip()) < 10:
        p_lower = problem_statement.lower()
        if "sale" in p_lower or "price" in p_lower or "forecast" in p_lower or "demand" in p_lower:
            schema = MOCK_SCHEMAS["retail"]
        else:
            schema = MOCK_SCHEMAS["default"]
        return json.dumps(schema, indent=2), "⚠️ SYSTEM: Running in LOCAL SANDBOX mode (No API Key). Custom data contract simulated."

    try:
        from google import genai
        from google.genai import types
        
        # Initialize Google's GenAI Client
        client = genai.Client(api_key=api_key.strip())
        
        # Build strict JSON Generation config
        config = types.GenerateContentConfig(
            system_instruction=system_prompt,
            response_mime_type="application/json",
            temperature=0.2
        )
        
        # Invoke Gemini 2.5 Flash for high-speed, cost-effective compilation
        response = client.models.generate_content(
            model='gemini-3.5-flash',
            contents=problem_statement,
            config=config
        )
        
        # Clean response string to bypass raw triple backtick blocks if returned
        cleaned_text = response.text.strip()
        if cleaned_text.startswith("```json"):
            cleaned_text = cleaned_text.split("```json", 1)[1].rsplit("```", 1)[0].strip()
        elif cleaned_text.startswith("```"):
            cleaned_text = cleaned_text.split("```", 1)[1].rsplit("```", 1)[0].strip()
            
        return cleaned_text, "✅ Core contract successfully compiled by Gemini 2.5."
    except Exception as e:
        return json.dumps(MOCK_SCHEMAS["default"], indent=2), f"Error querying Gemini API: {str(e)}. Falling back to default mock schema."

@spaces.GPU
def generate_default_python_script(schema_str):
    """

    Generates editable raw Python code matching the JSON contract's inputs/outputs.

    """
    try:
        schema = json.loads(schema_str)
    except Exception:
        return "# Error: Invalid JSON schema generated in Step 1. Please correct it."
    
    script_lines = [
        "import pandas as pd",
        "import random",
        "",
        "def generate_dataset(num_rows=50):",
        "    data = []",
        "    for i in range(num_rows):",
        "        row = {}"
    ]
    
    # Map inputs
    for inp in schema.get("inputs", []):
        name = inp["name"]
        t = inp["type"]
        if t == "categorical":
            cats = inp.get("categories", ["Category A", "Category B"])
            script_lines.append(f"        row['{name}'] = random.choice({cats})")
        elif t == "numeric":
            r = inp.get("range", [0.0, 100.0])
            script_lines.append(f"        row['{name}'] = round(random.uniform({r[0]}, {r[1]}), 2)")
        elif t == "text":
            # FIXED: Double braces {{i+1}} tells Python to treat it as raw text in the output string
            script_lines.append(f"        row['{name}'] = f'Sample text data row {{i+1}}'")
        elif t == "image":
            script_lines.append(f"        row['{name}'] = f'mock_image_path_{{i+1}}.png'")
            
    # Map outputs
    for out in schema.get("outputs", []):
        name = out["name"]
        t = out["type"]
        if t == "categorical":
            cats = out.get("categories", ["Pass", "Fail"])
            script_lines.append(f"        row['{name}'] = random.choice({cats})")
        elif t == "numeric":
            r = out.get("range", [0.0, 1.0])
            script_lines.append(f"        row['{name}'] = round(random.uniform({r[0]}, {r[1]}), 4)")
        elif t == "text":
            # FIXED: Double braces {{i+1}} here too
            script_lines.append(f"        row['{name}'] = f'Target output summary text {{i+1}}'")
        elif t == "image":
            script_lines.append(f"        row['{name}'] = f'mock_processed_image_path_{{i+1}}.png'")

    script_lines.extend([
        "        data.append(row)",
        "    return pd.DataFrame(data)"
    ])
    
    return "\n".join(script_lines)

@spaces.GPU
def execute_custom_script(script_code, num_rows):
    """

    Compiles and executes the user-edited data generation script within a local dictionary.

    Explicitly injects 'random' and 'pandas' to prevent missing module errors during exec().

    """
    try:
        # Pre-populate the execution environment with the required modules
        namespace = {
            "pd": pd,
            "random": random
        }
        
        # Execute the code block inside our prepared environment
        exec(script_code, namespace, namespace)
        
        if "generate_dataset" not in namespace:
            return None, None, "Error: The script must define a function named 'generate_dataset(num_rows)'"
        
        df = namespace["generate_dataset"](int(num_rows))
        csv_filename = "generated_dataset.csv"
        df.to_csv(csv_filename, index=False)
        return df, csv_filename, "✅ Dataset generation executed successfully!"
    except Exception as e:
        return None, None, f"Execution Error: {str(e)}"


# ==========================================
# 3. GRADIO APP INTERFACE LAYOUT
# ==========================================
with gr.Blocks(theme=gr.themes.Soft(), title="Gemini AI Solutions Prototyper") as demo:
    schema_state = gr.State({})
    
    gr.Markdown("# 🚀 Meta-AI Prototyping Sandbox (Powered by Gemini)")
    gr.Markdown("Create a complete AI solution pipeline. Modify, tweak, and approve the structures at every step.")

    with gr.Tabs() as tabs:
        
        # ----------------------------------
        # TAB 1: ARCHITECTURE DESIGN
        # ----------------------------------
        with gr.Tab("Step 1: Architecture Designer", id=0):
            gr.Markdown("### Parse Problem Statement into a JSON Data Contract")
            
            with gr.Row():
                with gr.Column(scale=1):
                    api_key_input = gr.Textbox(
                        label="Google Gemini API Key (Optional)", 
                        placeholder="AIzaSy...", 
                        type="password",
                        info="Leave empty to use local offline simulation mode"
                    )
                    user_problem = gr.Textbox(
                        label="Your AI Problem Statement", 
                        value="Predict housing prices using location metrics, sqft size, and if it is near a transit stop.",
                        lines=4
                    )
                    edit_sys_prompt_btn = gr.Button("⚙️ Show / Edit Architectural System Prompt", size="sm")
                    sys_prompt_box = gr.Textbox(
                        label="Architect System Prompt",
                        value=DEFAULT_SYSTEM_PROMPT,
                        lines=12,
                        visible=False
                    )
                    
                    def toggle_sys_prompt(visible):
                        return gr.update(visible=not visible)
                    edit_sys_prompt_btn.click(toggle_sys_prompt, inputs=[sys_prompt_box], outputs=[sys_prompt_box])

                    generate_btn = gr.Button("Compile Solution Architecture with Gemini", variant="primary")
                    
                with gr.Column(scale=1):
                    engine_status = gr.Markdown("**Status:** Awaiting compilation.")
                    schema_output = gr.Code(
                        label="Generated JSON Data Contract (Editable)", 
                        language="json", 
                        interactive=True,
                        lines=20
                    )
                    
            next_to_step2_btn = gr.Button("Approve Contract & Advance to Data Generation ➡️", variant="secondary")

        # ----------------------------------
        # TAB 2: DATA CREATION
        # ----------------------------------
        with gr.Tab("Step 2: Dummy Dataset Creator", id=1):
            gr.Markdown("### Review and Edit the Custom Generation Script")
            
            with gr.Row():
                with gr.Column(scale=1):
                    row_slider = gr.Slider(minimum=5, maximum=1000, value=50, step=5, label="Number of Rows to Mock")
                    script_editor = gr.Code(
                        label="Data Generation Python Script (Editable)", 
                        language="python", 
                        interactive=True,
                        lines=15
                    )
                    run_script_btn = gr.Button("Execute Script & Build CSV", variant="primary")
                    
                with gr.Column(scale=1):
                    script_status = gr.Markdown("**Status:** Script compiled. Awaiting execution.")
                    data_preview = gr.Dataframe(label="Generated Data Preview", interactive=False)
                    download_link = gr.File(label="Download Generated CSV")

            next_to_step3_btn = gr.Button("Approve Data & Advance to Live Interface ➡️", variant="secondary")

        # ----------------------------------
        # TAB 3: LIVE PROTOTYPE UI
        # ----------------------------------
        with gr.Tab("Step 3 & 4: Live Demo", id=2):
            gr.Markdown("### Interactive Prototype Interface")
            gr.Markdown("The widgets below are generated dynamically using the approved JSON contract in Step 1. Test your model parameters below:")
            
            @gr.render(inputs=schema_state)
            def render_prototype_ui(schema):
                if not schema or "inputs" not in schema:
                    gr.Markdown("### ⚠️ Waiting for Architecture Selection\nPlease generate or copy a valid JSON schema into **Step 1** to activate this view.")
                    return
                
                gr.Markdown(f"### Demo Module: **{schema.get('problem_domain', 'AI Model')}**")
                gr.Markdown(f"**Description:** {schema.get('technical_summary', 'Simulation Module.')}")
                gr.Markdown(f"*Recommended Architecture: `{schema.get('recommended_architecture', 'Vanilla Machine Learning')}`*")
                
                inputs = []
                with gr.Row():
                    # Generate Input Widgets
                    with gr.Column(scale=1, variant="panel"):
                        gr.Markdown("#### Dynamic Parameters (Inputs)")
                        for inp in schema["inputs"]:
                            name = inp["name"]
                            t = inp["type"]
                            desc = inp.get("description", "")
                            
                            if t == "text":
                                inputs.append((name, gr.Textbox(label=name, info=desc)))
                            elif t == "categorical":
                                inputs.append((name, gr.Dropdown(choices=inp.get("categories", ["Option A"]), label=name, info=desc)))
                            elif t == "numeric":
                                r = inp.get("range", [0, 100])
                                inputs.append((name, gr.Slider(minimum=r[0], maximum=r[1], value=(r[0]+r[1])/2, label=name, info=desc)))
                            elif t == "image":
                                inputs.append((name, gr.Image(label=name, type="pil", info=desc)))

                    # Generate Output Display Widgets
                    with gr.Column(scale=1, variant="panel"):
                        gr.Markdown("#### Simulated AI Outputs")
                        outputs = []
                        for out in schema["outputs"]:
                            name = out["name"]
                            t = out["type"]
                            desc = out.get("description", "")
                            
                            if t == "text":
                                outputs.append((name, gr.Textbox(label=name, info=desc, interactive=False)))
                            elif t == "categorical":
                                outputs.append((name, gr.Textbox(label=name, info=desc, interactive=False)))
                            elif t == "numeric":
                                outputs.append((name, gr.Number(label=name, info=desc, interactive=False)))
                            elif t == "image":
                                outputs.append((name, gr.Image(label=name, type="pil", info=desc, interactive=False)))
                
                infer_btn = gr.Button("⚡ Execute Mock Model Inference", variant="primary")
                
                # Dynamic Execution Handler
                def run_inference(*args):
                    input_payload = {inputs[i][0]: args[i] for i in range(len(args))}
                    out_results = []
                    
                    for out in schema["outputs"]:
                        t = out["type"]
                        if t == "categorical":
                            out_results.append(random.choice(out.get("categories", ["N/A"])))
                        elif t == "numeric":
                            r = out.get("range", [0, 100])
                            out_results.append(round(random.uniform(r[0], r[1]), 2))
                        elif t == "text":
                            out_results.append(f"Model inferred successfully based on values: {list(input_payload.values())}")
                        elif t == "image":
                            # Draw a dynamic processing ring in memory
                            img = Image.new("RGB", (300, 300), color=(17, 24, 39)) # Deep charcoal slate
                            draw = ImageDraw.Draw(img)
                            draw.ellipse([100, 100, 200, 200], fill=(16, 185, 129)) # Emerald Green ring
                            out_results.append(img)
                            
                    return out_results

                infer_btn.click(
                    fn=run_inference,
                    inputs=[widget for _, widget in inputs],
                    outputs=[widget for _, widget in outputs]
                )

    # ==========================================
    # INTER-TAB COORDINATION CONTROL FLOW
    # ==========================================
    def step1_action(api_key, problem, prompt):
        raw_json, status_msg = call_llm_for_schema(api_key, problem, prompt)
        try:
            parsed_json = json.loads(raw_json)
        except Exception:
            parsed_json = {}
        script_code = generate_default_python_script(raw_json)
        return raw_json, status_msg, script_code, parsed_json

    generate_btn.click(
        fn=step1_action,
        inputs=[api_key_input, user_problem, sys_prompt_box],
        outputs=[schema_output, engine_status, script_editor, schema_state]
    )

    def advance_to_step2(raw_json):
        try:
            parsed_json = json.loads(raw_json)
        except Exception:
            return gr.update(selected=0), {}, ""
        script_code = generate_default_python_script(raw_json)
        return gr.update(selected=1), parsed_json, script_code

    next_to_step2_btn.click(
        fn=advance_to_step2,
        inputs=[schema_output],
        outputs=[tabs, schema_state, script_editor]
    )

    run_script_btn.click(
        fn=execute_custom_script,
        inputs=[script_editor, row_slider],
        outputs=[data_preview, download_link, script_status]
    )

    def advance_to_step3(raw_json):
        try:
            parsed_json = json.loads(raw_json)
        except Exception:
            parsed_json = {}
        return gr.update(selected=2), parsed_json

    next_to_step3_btn.click(
        fn=advance_to_step3,
        inputs=[schema_output],
        outputs=[tabs, schema_state]
    )

if __name__ == "__main__":
    demo.launch(pwa=True)