recipe_api.py

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, validator
from typing import List, Dict, Optional, Union, Any
from compactrag import CompactAeropressRAG
import uvicorn


# Pydantic models for request validation
class CoffeeAttributes(BaseModel):
    origin: Optional[str] = None
    country: Optional[str] = None
    processing: str

    @validator('processing')
    def processing_must_exist(cls, v):
        if not v:
            raise ValueError("Processing method must be provided")
        return v

    @validator('origin', 'country')
    def validate_origin_or_country(cls, v, values):
        # This validator will be called for both fields
        # For the second field, we can check if either this field or the previous one is set
        if 'origin' in values and values['origin'] or 'country' in values and values['country']:
            return v
        # If we're validating the first field and it's not set, we can't validate yet
        # If we're validating the second field and neither is set, raise an error
        if ('origin' in values and not values['origin'] and not v) or (
                'country' in values and not values['country'] and not v):
            raise ValueError("Either origin or country must be provided")
        return v


class UserPreferences(BaseModel):
    # Define user preferences structure here
    # For example:
    preferred_position: Optional[str] = None
    preferred_dose: Optional[float] = None
    preferred_temperature: Optional[int] = None
    # Add other preferences as needed


class RecipeRequest(BaseModel):
    coffee: CoffeeAttributes
    preferences: Optional[UserPreferences] = None


class ParameterConfidence(BaseModel):
    # Define structure based on your confidence parameters
    position: float
    dose: float
    temperature: float
    grind: float
    brew_time: float
    filter: float
    water_amount: float


class RecipeParameters(BaseModel):
    position: str
    dose: float
    temperature: int
    water_amount: int
    grind: str
    filter: str
    brew_time: str


class Recipe(BaseModel):
    parameters: RecipeParameters
    method: List[str]


class Confidence(BaseModel):
    overall: float
    parameter_confidence: ParameterConfidence


class SimilarRecipe(BaseModel):
    champion: str
    year: int
    placement: str
    similarity: float


class RecipeResponse(BaseModel):
    recipe: Recipe
    confidence: Confidence
    explanation: str
    accuracy: float
    similar_recipes: List[SimilarRecipe]


class SimilarRecipeParameters(BaseModel):
    position: str
    dose: float
    temperature: Optional[int] = None
    brew_time: Optional[str] = None


class DetailedSimilarRecipe(BaseModel):
    champion: str
    year: int
    placement: str
    similarity: float
    recipe_parameters: SimilarRecipeParameters


class SimilarRecipesResponse(BaseModel):
    similar_recipes: List[DetailedSimilarRecipe]


class AeropressRecipeAPI:
    def __init__(self, recipe_database_path="recipes_database.json"):
        self.rag = CompactAeropressRAG(recipe_database_path)

    def generate_recipe(self, coffee_attributes: Dict[str, Any], user_preferences: Optional[Dict[str, Any]] = None) -> \
    Dict[str, Any]:
        """Generate a recipe based on coffee attributes and user preferences"""
        # Generate recipe
        result = self.rag.generate_recipe(coffee_attributes, user_preferences)

        # Format response
        response = {
            "recipe": {
                "parameters": {
                    "position": result['recipe']['position'],
                    "dose": result['recipe']['dose'],
                    "temperature": result['recipe']['water']['temperature'],
                    "water_amount": result['recipe']['water']['amount'],
                    "grind": result['recipe']['grind'],
                    "filter": result['recipe']['filter'],
                    "brew_time": result['recipe']['brew_time']
                },
                "method": result['recipe']['method']
            },
            "confidence": {
                "overall": result['confidence']['overall'],
                "parameter_confidence": result['confidence']['parameters']
            },
            "explanation": result['explanation'],
            "accuracy": result['accuracy'],
            "similar_recipes": [
                {
                    "champion": similar['recipe']['competitor_name'],
                    "year": similar['recipe']['year'],
                    "placement": similar['recipe']['placement'],
                    "similarity": similar['similarity']
                } for similar in result['similar_recipes']
            ]
        }

        return response


# FastAPI implementation
app = FastAPI(
    title="Aeropress Recipe API",
    description="API for generating Aeropress recipes based on coffee attributes",
    version="1.0.0"
)

api = AeropressRecipeAPI()


@app.post("/api/recipe", response_model=RecipeResponse)
async def generate_recipe(request: RecipeRequest):
    """
    Generate an Aeropress recipe based on coffee attributes and optional user preferences
    """
    result = api.generate_recipe(
        coffee_attributes=request.coffee.dict(exclude_none=True),
        user_preferences=request.preferences.dict(exclude_none=True) if request.preferences else None
    )

    return result


@app.post("/api/similar_recipes", response_model=SimilarRecipesResponse)
async def get_similar_recipes(request: RecipeRequest):
    """
    Get similar recipes based on coffee attributes
    """
    coffee_attributes = request.coffee.dict(exclude_none=True)

    rag = CompactAeropressRAG('recipes_database.json')
    similar_recipes = rag.retrieve_similar_recipes(coffee_attributes, k=10)

    return {
        "similar_recipes": [
            {
                "champion": recipe['recipe']['competitor_name'],
                "year": recipe['recipe']['year'],
                "placement": recipe['recipe']['placement'],
                "similarity": recipe['similarity'],
                "recipe_parameters": {
                    "position": recipe['recipe']['recipe']['position'],
                    "dose": recipe['recipe']['recipe']['dose'],
                    "temperature": recipe['recipe']['recipe']['water'].get('temperature'),
                    "brew_time": recipe['recipe']['recipe'].get('brew_time')
                }
            } for recipe in similar_recipes
        ]
    }


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)