app.py

"""Gradio application for the smart fridge detector + recipe recommendation pipeline."""

import json
import tempfile
from pathlib import Path
from typing import List, Tuple, Dict, Any

import cv2
import gradio as gr
import numpy as np
from PIL import Image

from frige_detect.detect import (
    detect_and_generate,
    load_roboflow_credentials,
    RoboflowCredentials,
)
from recipe_recommendation.main import (
    load_recipes,
    recommend_recipes,
    save_user_profile,
    get_feedback,
    USER_DATA_DIR,
)

# ---------------------------------------------------------------------------
# Global resources
# ---------------------------------------------------------------------------
CREDENTIALS_PATH = Path("frige_detect/roboflow_credentials.txt")
ROBOFLOW_CREDENTIALS: RoboflowCredentials = load_roboflow_credentials(str(CREDENTIALS_PATH))
RECIPES_DF = load_recipes()

# ---------------------------------------------------------------------------
# Predefined user profiles for examples
# ---------------------------------------------------------------------------
EXAMPLE_PROFILES = {
    "user_1": {
        "vegetarian_type": "flexible",
        "allergies": "",
        "regions": "North America",
        "calorie_min": 250,
        "calorie_max": 2000,
        "protein_min": 50,
        "protein_max": 160,
        "preferred_main": "",
        "disliked_main": "",
        "cooking_time": 45,
    },
    "user_2": {
        "vegetarian_type": "flexible_vegetarian",
        "allergies": "shrimp",
        "regions": "Asia",
        "calorie_min": 400,
        "calorie_max": 1500,
        "protein_min": 40,
        "protein_max": 120,
        "preferred_main": "tofu",
        "disliked_main": "beef",
        "cooking_time": 60,
    },
    "user_3": {
        "vegetarian_type": "non_vegetarian",
        "allergies": "",
        "regions": "Europe",
        "calorie_min": 500,
        "calorie_max": 2000,
        "protein_min": 80,
        "protein_max": 160,
        "preferred_main": "beef, chicken",
        "disliked_main": "",
        "cooking_time": 45,
    },
}

# Predefined example images
EXAMPLE_IMAGES = [
    "frige_detect/demo/t1.jpg",
    "frige_detect/demo/t2.jpg",
    "frige_detect/demo/t3.jpg",
]


# ---------------------------------------------------------------------------
# Helper utilities
# ---------------------------------------------------------------------------
def parse_csv_list(text: str) -> List[str]:
    if not text:
        return []
    parts = [item.strip() for item in text.split(",") if item.strip()]
    return parts


def ensure_numpy_image(image: Any) -> np.ndarray:
    """Convert incoming image (PIL or numpy) to RGB numpy array."""
    if image is None:
        raise ValueError("Please upload a fridge photo before running detection.")
    if isinstance(image, np.ndarray):
        return image
    if isinstance(image, Image.Image):
        return np.array(image.convert("RGB"))
    raise ValueError("Unsupported image format provided.")


def write_temp_image(image: np.ndarray) -> str:
    """Write numpy image to a temporary file and return the path."""
    temp_dir = Path(tempfile.mkdtemp(prefix="fridge_upload_"))
    temp_path = temp_dir / "upload.jpg"
    bgr_image = cv2.cvtColor(image, cv2.COLOR_RGB2BGR)
    cv2.imwrite(str(temp_path), bgr_image)
    return str(temp_path)


def build_user_profile(
    user_id: str,
    vegetarian_type: str,
    allergies: str,
    regions: str,
    calorie_range: Tuple[float, float],
    protein_range: Tuple[float, float],
    preferred_main: str,
    disliked_main: str,
    cooking_time: float,
) -> Dict[str, Any]:
    """
    Build and save user profile. This function ALWAYS creates or overwrites the profile
    with the current input values, enabling users to modify preferences on-the-fly.
    """
    user_id = user_id.strip()
    if not user_id:
        raise ValueError("User ID cannot be empty.")

    profile_dir = USER_DATA_DIR / user_id
    profile_path = profile_dir / "user_profile.json"
    
    # Preserve feedback count if profile exists
    num_feedback = 0
    if profile_path.exists():
        try:
            existing = json.loads(profile_path.read_text(encoding="utf-8"))
            num_feedback = existing.get("num_feedback", 0)
        except Exception:
            pass

    profile = {
        "user_id": user_id,
        "num_feedback": num_feedback,
        "diet": {"vegetarian_type": vegetarian_type},
        "allergies": parse_csv_list(allergies),
        "region_preference": parse_csv_list(regions),
        "nutritional_goals": {
            "calories": {"min": int(calorie_range[0]), "max": int(calorie_range[1])},
            "protein": {"min": int(protein_range[0]), "max": int(protein_range[1])},
        },
        "other_preferences": {
            "preferred_main": parse_csv_list(preferred_main),
            "disliked_main": parse_csv_list(disliked_main),
            "cooking_time_max": int(cooking_time) if cooking_time else None,
        },
    }

    # Always save the profile (create new or overwrite existing)
    save_user_profile(user_id, profile)
    print(f"[app] Profile saved/updated for user '{user_id}'")
    
    return profile


def summarize_ingredients(
    user_parents: List[str],
    high_conf: List[str],
    low_conf: List[str],
) -> str:
    lines = ["### Ingredient Mapping"]
    if user_parents:
        lines.append("- **Mapped parent ingredients:** " + ", ".join(sorted(user_parents)))
    else:
        lines.append("- **Mapped parent ingredients:** none")
    if high_conf:
        lines.append("- **High confidence detections:** " + ", ".join(sorted(high_conf)))
    if low_conf:
        lines.append("- **Low confidence detections:** " + ", ".join(sorted(set(low_conf))))
    return "\n".join(lines)


def _ensure_iterable(value: Any) -> List[str]:
    if value is None:
        return []
    if isinstance(value, set):
        return sorted(value)
    if isinstance(value, list):
        return value
    if isinstance(value, str):
        return [value]
    return list(value)


def render_recommendations(df) -> Tuple[str, List[Dict[str, Any]]]:
    if df is None or df.empty:
        return "No recipes matched the current constraints.", []

    lines = ["### Recommended Recipes"]
    feedback_rows: List[Dict[str, Any]] = []

    for idx, row in df.head(5).iterrows():
        match_score = row.get("match_score") or row.get("ml_score", 0)
        scaled = match_score * 100 if match_score is not None else 0
        name = row.get("name", f"Recipe {idx+1}")
        lines.append(f"{idx + 1}. **{name}** — score {scaled:.1f}%")

        region = row.get("region")
        if region and not (isinstance(region, float) and np.isnan(region)):
            if isinstance(region, (set, list)):
                region_str = ", ".join(sorted(region))
            else:
                region_str = str(region)
            lines.append(f"   - Region: {region_str}")

        cuisine = row.get("cuisine_attr")
        cuisine_items = _ensure_iterable(cuisine)
        if cuisine_items:
            lines.append(f"   - Cuisine: {', '.join(cuisine_items)}")

        calories = row.get("calories")
        protein = row.get("protein")
        if calories is not None:
            lines.append(f"   - Calories: {calories}")
        if protein is not None:
            lines.append(f"   - Protein: {protein}")

        for key in ["main_parent", "staple_parent", "other_parent"]:
            parents = _ensure_iterable(row.get(key))
            if parents:
                pretty_key = key.replace("_", " ").title()
                lines.append(f"   - {pretty_key}: {', '.join(parents)}")

        ingredients = row.get("ingredients")
        if ingredients:
            if isinstance(ingredients, str):
                ingredients_list = parse_csv_list(ingredients)
            else:
                ingredients_list = list(ingredients)
            if ingredients_list:
                lines.append(f"   - Ingredients: {', '.join(ingredients_list[:10])}")
        lines.append("")

        feedback_row = row.to_dict()
        for key in ["main_parent", "staple_parent", "other_parent", "seasoning_parent", "cuisine_attr", "ingredients"]:
            value = feedback_row.get(key)
            if isinstance(value, list):
                feedback_row[key] = set(value)
            elif isinstance(value, str):
                feedback_row[key] = set(parse_csv_list(value))
        feedback_rows.append(feedback_row)

    return "\n".join(lines).strip(), feedback_rows


def load_example_profile(profile_name: str):
    """Load a predefined user profile configuration."""
    if profile_name in EXAMPLE_PROFILES:
        config = EXAMPLE_PROFILES[profile_name]
        return (
            profile_name,
            config["vegetarian_type"],
            config["allergies"],
            config["regions"],
            config["calorie_min"],
            config["calorie_max"],
            config["protein_min"],
            config["protein_max"],
            config["preferred_main"],
            config["disliked_main"],
            config["cooking_time"],
        )
    # Default fallback
    return ("user_custom", "flexible", "", "", 400, 2000, 50, 160, "", "", 45)


def load_example_image(image_path: str):
    """Load an example image."""
    return image_path


def run_pipeline(
    image,
    user_id,
    vegetarian_type,
    allergies,
    regions,
    calorie_min,
    calorie_max,
    protein_min,
    protein_max,
    preferred_main,
    disliked_main,
    cooking_time,
):
    """
    Main pipeline function. 
    This ALWAYS creates/updates the user profile based on current input values,
    then runs detection and recommendation.
    """
    try:
        rgb_image = ensure_numpy_image(image)
        upload_path = write_temp_image(rgb_image)
        temp_dir = Path(tempfile.mkdtemp(prefix="fridge_outputs_"))
        output_json = temp_dir / "recipe_input.json"
        output_image = temp_dir / "annotated_image.jpg"

        detection_result = detect_and_generate(
            image_path=upload_path,
            credentials=ROBOFLOW_CREDENTIALS,
            conf_threshold=0.4,
            overlap_threshold=0.3,
            conf_split=0.7,
            output_json=str(output_json),
            output_image=str(output_image),
        )
        Path(upload_path).unlink(missing_ok=True)

        #2: Always create/update user profile with current UI values
        profile = build_user_profile(
            user_id,
            vegetarian_type,
            allergies,
            regions,
            (calorie_min, calorie_max),
            (protein_min, protein_max),
            preferred_main,
            disliked_main,
            cooking_time,
        )

        
        import time
        time.sleep(0.2) 

        detection_payload = detection_result["recipe_json"]
        ml_top, user_parents, high_conf, low_conf = recommend_recipes(
            detection_payload,
            user_id,
            RECIPES_DF,
            topk=5,
        )

        ingredient_summary = summarize_ingredients(user_parents, high_conf, low_conf)
        recommendation_md, feedback_rows = render_recommendations(ml_top)

        dropdown_choices = [
            f"{idx + 1}. {row.get('name', 'Recipe')}" for idx, row in enumerate(feedback_rows)
        ]

        status = "" if feedback_rows else "No recipes available for feedback yet."
        
        # Add success message about profile creation/update
        profile_status = f"✓ Profile '{user_id}' has been saved/updated with your current preferences."

        return (
            str(output_image),
            detection_payload,
            ingredient_summary,
            recommendation_md,
            gr.Dropdown(choices=dropdown_choices, value=None),
            feedback_rows,
            profile_status,
        )
    except Exception as exc:
        import traceback
        error_detail = traceback.format_exc()
        return (
            None,
            None,
            "",
            f"⚠️ Error: {exc}\n\nDetails:\n{error_detail}",
            gr.Dropdown(choices=[], value=None),
            [],
            f"⚠️ Error: {exc}",
        )


def record_feedback(selected_recipe: str, user_id: str, feedback_rows: List[Dict[str, Any]]):
    if not selected_recipe:
        return "Please select a recipe before submitting feedback."
    if not user_id:
        return "Please provide a valid user ID."
    if not feedback_rows:
        return "No recommendation data available. Run the pipeline first."

    try:
        index = int(selected_recipe.split(".")[0]) - 1
    except (ValueError, IndexError):
        return "Unable to parse the selected recipe."

    if index < 0 or index >= len(feedback_rows):
        return "Selected recipe is out of range."

    recipe_row = feedback_rows[index]
    get_feedback(user_id, recipe_row)

    profile_path = USER_DATA_DIR / user_id / "user_profile.json"
    if profile_path.exists():
        data = json.loads(profile_path.read_text(encoding="utf-8"))
        data["num_feedback"] = data.get("num_feedback", 0) + 1
        save_user_profile(user_id, data)

    return f"✓ Feedback recorded for {recipe_row.get('name', 'selected recipe')}!"


# ---------------------------------------------------------------------------
# Gradio UI definition
# ---------------------------------------------------------------------------
with gr.Blocks(title="Smart Fridge Recipe Assistant", theme=gr.themes.Soft()) as demo:
    gr.Markdown(
        """
        # Smart Fridge Recipe Assistant
        **How to use:**
        1. (Optional) Select an example profile and/or image from dropdowns
        2. Modify any preferences in the form - your profile will be saved automatically when you click Analyze
        3. Upload or select a fridge image
        4. Click "Analyze fridge & recommend recipes"
        """
    )

    with gr.Row():
        with gr.Column(scale=1):
            gr.Markdown("### Quick Start Examples")
            profile_selector = gr.Dropdown(
                label="Choose a predefined user profile",
                choices=list(EXAMPLE_PROFILES.keys()),
                value=None,
            )
            
            image_selector = gr.Dropdown(
                label="Choose an example fridge image",
                choices=[f"Image {i+1}: {img}" for i, img in enumerate(EXAMPLE_IMAGES)],
                value=None,
            )
            
            image_input = gr.Image(
                label="Fridge photo (upload or use example)",
                type="pil",
                height=350,
            )
            
            detection_json = gr.JSON(label="Detection payload")
            annotated_output = gr.Image(label="Annotated detection", height=350)

        with gr.Column(scale=1):
            gr.Markdown("### User Preferences (auto-saved on each run)")
            user_id_box = gr.Textbox(
                label="User ID (will create new profile if doesn't exist)",
                value="user_custom",
                placeholder="e.g. my_new_profile",
            )
            vegetarian_radio = gr.Radio(
                [
                    "flexible",
                    "flexible_vegetarian",
                    "ovo_vegetarian",
                    "lacto_vegetarian",
                    "vegan",
                    "non_vegetarian",
                ],
                label="Vegetarian preference",
                value="flexible",
            )
            allergies_box = gr.Textbox(
                label="Allergies (comma separated)",
                placeholder="peanut, shrimp",
            )
            regions_box = gr.Textbox(
                label="Preferred regions (comma separated)",
                placeholder="Asia, Europe",
            )
            calorie_min = gr.Slider(
                minimum=0,
                maximum=4000,
                value=400,
                label="Minimum Calories",
                step=50,
            )
            calorie_max = gr.Slider(
                minimum=0,
                maximum=4000,
                value=2000,
                label="Maximum Calories",
                step=50,
            )
            protein_min = gr.Slider(
                minimum=0,
                maximum=250,
                value=50,
                label="Minimum Protein (g)",
                step=5,
            )
            protein_max = gr.Slider(
                minimum=0,
                maximum=250,
                value=160,
                label="Maximum Protein (g)",
                step=5,
            )
            preferred_box = gr.Textbox(
                label="Preferred main ingredients",
                placeholder="chicken, tofu",
            )
            disliked_box = gr.Textbox(
                label="Disliked main ingredients",
                placeholder="lamb",
            )
            cooking_slider = gr.Slider(
                minimum=0,
                maximum=180,
                value=45,
                step=5,
                label="Max cooking time (minutes)",
            )
            run_button = gr.Button("Analyze fridge & recommend recipes", variant="primary")
            ingredient_md = gr.Markdown()
            recommendation_md = gr.Markdown()
            feedback_dropdown = gr.Dropdown(label="Select a recipe for positive feedback", choices=[])
            feedback_button = gr.Button("Save feedback")
            feedback_status = gr.Markdown()
            feedback_state = gr.State([])

    # Connect profile selector
    profile_selector.change(
        fn=load_example_profile,
        inputs=[profile_selector],
        outputs=[
            user_id_box,
            vegetarian_radio,
            allergies_box,
            regions_box,
            calorie_min,
            calorie_max,
            protein_min,
            protein_max,
            preferred_box,
            disliked_box,
            cooking_slider,
        ],
    )

    # Connect image selector
    def select_image(choice):
        if choice:
            idx = int(choice.split(":")[0].replace("Image ", "")) - 1
            return EXAMPLE_IMAGES[idx]
        return None
    
    image_selector.change(
        fn=select_image,
        inputs=[image_selector],
        outputs=[image_input],
    )

    run_button.click(
        fn=run_pipeline,
        inputs=[
            image_input,
            user_id_box,
            vegetarian_radio,
            allergies_box,
            regions_box,
            calorie_min,
            calorie_max,
            protein_min,
            protein_max,
            preferred_box,
            disliked_box,
            cooking_slider,
        ],
        outputs=[
            annotated_output,
            detection_json,
            ingredient_md,
            recommendation_md,
            feedback_dropdown,
            feedback_state,
            feedback_status,
        ],
    )

    feedback_button.click(
        fn=record_feedback,
        inputs=[feedback_dropdown, user_id_box, feedback_state],
        outputs=feedback_status,
    )

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