app.py

"""

Learning Objective Taxonomy API

FastAPI backend for Bloom's, Dave's, and CO-PO analysis

"""

import logging
from datetime import datetime
from typing import Any, Dict, List, Optional

import numpy as np
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from sentence_transformers import SentenceTransformer
from transformers import Pipeline, pipeline

# Configure logging
logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# FastAPI app
app = FastAPI(
    title="Learning Objective Taxonomy API",
    description="API for Bloom's Taxonomy, Dave's Psychomotor, and CO-PO Mapping analysis",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc",
)

# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Global model variables
blooms_model: Optional[Pipeline] = None
dave_model: Optional[Pipeline] = None
coppo_model: Optional[SentenceTransformer] = None


# Pydantic models
class TextRequest(BaseModel):
    text: str = Field(
        ...,
        min_length=1,
        max_length=1000,
        description="Learning objective text to analyze",
    )


class PredictionResult(BaseModel):
    level: Optional[str] = Field(None, description="Predicted taxonomy level")
    confidence: Optional[float] = Field(None, description="Confidence score (0-1)")
    all_predictions: List[Dict[str, Any]] = Field(
        default_factory=list, description="All prediction results"
    )


class SingleModelResponse(BaseModel):
    success: bool
    model: str
    text: str
    prediction: PredictionResult
    timestamp: str


class EmbeddingResponse(BaseModel):
    success: bool
    model: str
    text: str
    embeddings: List[float]
    timestamp: str


class CombinedResults(BaseModel):
    blooms: Dict[str, Any]
    dave: Dict[str, Any]
    coppo: Dict[str, Any]


class CombinedResponse(BaseModel):
    success: bool
    text: str
    results: CombinedResults
    timestamp: str


class HealthResponse(BaseModel):
    status: str
    models_loaded: Dict[str, bool]


# Helper functions
def normalize_results(results: Any) -> List[Dict[str, Any]]:
    """Normalize various model output formats to list of dicts."""
    if isinstance(results, list):
        return [
            item if isinstance(item, dict) else {"label": str(item)} for item in results
        ]

    if isinstance(results, dict):
        return [results]

    if isinstance(results, np.ndarray):
        try:
            return [{"value": item.tolist()} for item in results]
        except Exception:
            return [{"value": results.tolist()}]

    try:
        candidate = list(results)
        return [
            item if isinstance(item, dict) else {"label": str(item)}
            for item in candidate
        ]
    except (TypeError, AttributeError):
        return [{"value": results}]


def extract_label_and_score(results: List[Dict[str, Any]]) -> Dict[str, Optional[Any]]:
    """Extract label and score from normalized results with fallback."""
    if not results:
        return {"label": None, "score": None}

    first = results[0]
    label = first.get("label")
    score = first.get("score") or first.get("confidence")

    try:
        score = float(score) if score is not None else None
    except (ValueError, TypeError):
        score = None

    return {"label": label, "score": score}


def get_timestamp() -> str:
    """Get current UTC timestamp in ISO format."""
    return datetime.utcnow().isoformat() + "Z"


# Model loading at startup
@app.on_event("startup")
async def load_models():
    """Load all models at application startup."""
    global blooms_model, dave_model, coppo_model

    logger.info("=" * 60)
    logger.info("Starting model loading...")
    logger.info("=" * 60)

    # Load Bloom's model
    try:
        logger.info("Loading Bloom's Taxonomy model (Jrine/blooms)...")
        from transformers import AutoModelForSequenceClassification, AutoTokenizer

        tokenizer = AutoTokenizer.from_pretrained("Jrine/blooms")
        model = AutoModelForSequenceClassification.from_pretrained("Jrine/blooms")
        blooms_model = pipeline("text-classification", model=model, tokenizer=tokenizer)
        logger.info("✅ Bloom's model loaded successfully")
    except Exception as e:
        logger.error(f"❌ Failed to load Bloom's model: {e}")
        blooms_model = None

    # Load Dave's model
    try:
        logger.info("Loading Dave's Psychomotor model (Jrine/dave)...")
        from transformers import AutoModelForSequenceClassification, AutoTokenizer

        tokenizer = AutoTokenizer.from_pretrained("Jrine/dave")
        model = AutoModelForSequenceClassification.from_pretrained("Jrine/dave")
        dave_model = pipeline("text-classification", model=model, tokenizer=tokenizer)
        logger.info("✅ Dave's model loaded successfully")
    except Exception as e:
        logger.error(f"❌ Failed to load Dave's model: {e}")
        dave_model = None

    # Load CO-PO model - Handle sklearn model
    try:
        logger.info("Loading CO-PO mapping model (Jrine/co-po)...")
        logger.info("CO-PO is a sklearn model - loading with joblib...")

        import joblib
        from huggingface_hub import hf_hub_download

        # Download the sklearn model file
        model_path = hf_hub_download(
            repo_id="Jrine/co-po", filename="sklearn_model.joblib"
        )
        coppo_model = joblib.load(model_path)
        logger.info("✅ CO-PO model loaded successfully as sklearn model")
    except Exception as e:
        logger.error(f"❌ Failed to load CO-PO sklearn model: {e}")
        logger.info("Trying to use sentence-transformers fallback...")
        try:
            # Fallback: use a generic sentence transformer
            coppo_model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
            logger.info("✅ CO-PO using fallback sentence-transformer")
        except Exception as e2:
            logger.error(f"❌ Fallback also failed: {e2}")
            coppo_model = None

    logger.info("=" * 60)
    logger.info(
        f"Model loading complete - "
        f"Bloom's: {blooms_model is not None}, "
        f"Dave's: {dave_model is not None}, "
        f"CO-PO: {coppo_model is not None}"
    )
    logger.info("=" * 60)


# API Endpoints
@app.get("/", tags=["Root"])
async def root():
    """API root endpoint with status information."""
    return {
        "message": "Learning Objective Taxonomy API",
        "version": "1.0.0",
        "documentation": "/docs",
        "endpoints": {
            "blooms": "/api/blooms",
            "dave": "/api/dave",
            "coppo": "/api/coppo",
            "analyze": "/api/analyze",
            "health": "/health",
        },
        "status": {
            "blooms": blooms_model is not None,
            "dave": dave_model is not None,
            "coppo": coppo_model is not None,
        },
    }


@app.post(

    "/api/blooms",

    response_model=SingleModelResponse,

    tags=["Classification"],

    summary="Classify Bloom's Taxonomy Level",

    description="Classifies learning objectives into Bloom's 6 cognitive levels: Remember, Understand, Apply, Analyze, Evaluate, Create",

)
async def predict_blooms(request: TextRequest):
    """Predict Bloom's Taxonomy cognitive level for a learning objective."""
    if blooms_model is None:
        logger.error("Bloom's model not available")
        raise HTTPException(status_code=503, detail="Bloom's model not loaded")

    try:
        logger.info(f"Bloom's prediction for: {request.text[:50]}...")
        raw = blooms_model(request.text)
        results = normalize_results(raw)
        first = extract_label_and_score(results)

        return {
            "success": True,
            "model": "blooms-taxonomy",
            "text": request.text,
            "prediction": {
                "level": first["label"],
                "confidence": first["score"],
                "all_predictions": results,
            },
            "timestamp": get_timestamp(),
        }
    except Exception as e:
        logger.error(f"Bloom's prediction error: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.post(

    "/api/dave",

    response_model=SingleModelResponse,

    tags=["Classification"],

    summary="Classify Dave's Psychomotor Level",

    description="Classifies learning objectives into Dave's 5 psychomotor levels: Imitation, Manipulation, Precision, Articulation, Naturalization",

)
async def predict_dave(request: TextRequest):
    """Predict Dave's Psychomotor motor skill level for a learning objective."""
    if dave_model is None:
        logger.error("Dave's model not available")
        raise HTTPException(status_code=503, detail="Dave's model not loaded")

    try:
        logger.info(f"Dave's prediction for: {request.text[:50]}...")
        raw = dave_model(request.text)
        results = normalize_results(raw)
        first = extract_label_and_score(results)

        return {
            "success": True,
            "model": "dave-psychomotor",
            "text": request.text,
            "prediction": {
                "level": first["label"],
                "confidence": first["score"],
                "all_predictions": results,
            },
            "timestamp": get_timestamp(),
        }
    except Exception as e:
        logger.error(f"Dave's prediction error: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.post(

    "/api/coppo",

    tags=["Embeddings"],

    summary="Generate CO-PO Embeddings",

    description="Generates semantic embeddings (if model supports it)",

)
async def predict_coppo(request: TextRequest):
    """Generate CO-PO semantic embeddings if available."""
    if coppo_model is None:
        logger.error("CO-PO model not available")
        raise HTTPException(status_code=503, detail="CO-PO model not loaded")

    try:
        logger.info(f"CO-PO processing for: {request.text[:50]}...")

        # Check if model supports encoding
        if hasattr(coppo_model, "encode"):
            embeddings = coppo_model.encode(request.text)
            if isinstance(embeddings, np.ndarray):
                emb_list = embeddings.tolist()
            else:
                emb_list = list(embeddings)

            return {
                "success": True,
                "model": "co-po (sentence-transformer fallback)",
                "text": request.text,
                "embeddings": emb_list,
                "timestamp": get_timestamp(),
            }
        else:
            # sklearn model
            return {
                "success": False,
                "model": "co-po (sklearn)",
                "text": request.text,
                "error": "sklearn model requires feature preprocessing - use /api/blooms and /api/dave instead",
                "timestamp": get_timestamp(),
            }
    except Exception as e:
        logger.error(f"CO-PO processing error: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.post(

    "/api/analyze",

    tags=["Combined Analysis"],

    summary="Analyze with Available Models",

    description="Runs all available models on the input text",

)
async def analyze_all(request: TextRequest):
    """Analyze learning objective with all available models."""
    # Check which models are available
    available_models = []
    if blooms_model is not None:
        available_models.append("blooms")
    if dave_model is not None:
        available_models.append("dave")
    if coppo_model is not None:
        available_models.append("coppo")

    if not available_models:
        logger.error("No models available")
        raise HTTPException(status_code=503, detail="No models loaded")

    try:
        logger.info(
            f"Combined analysis for: {request.text[:50]}... (using: {', '.join(available_models)})"
        )

        results = {}

        # Run Bloom's if available
        if blooms_model is not None:
            raw_blooms = blooms_model(request.text)
            blooms_results = normalize_results(raw_blooms)
            first_blooms = extract_label_and_score(blooms_results)
            results["blooms"] = {
                "level": first_blooms["label"],
                "confidence": first_blooms["score"],
                "all_predictions": blooms_results,
                "model": "Jrine/blooms",
            }
        else:
            results["blooms"] = {"error": "Model not loaded", "model": "Jrine/blooms"}

        # Run Dave's if available
        if dave_model is not None:
            raw_dave = dave_model(request.text)
            dave_results = normalize_results(raw_dave)
            first_dave = extract_label_and_score(dave_results)
            results["dave"] = {
                "level": first_dave["label"],
                "confidence": first_dave["score"],
                "all_predictions": dave_results,
                "model": "Jrine/dave",
            }
        else:
            results["dave"] = {"error": "Model not loaded", "model": "Jrine/dave"}

        # Run CO-PO if available
        if coppo_model is not None:
            try:
                # Check if it's sklearn or sentence-transformer
                if hasattr(coppo_model, "encode"):
                    # Sentence transformer
                    coppo_embeddings = coppo_model.encode(request.text)
                    if isinstance(coppo_embeddings, np.ndarray):
                        emb_list = coppo_embeddings.tolist()
                    else:
                        emb_list = list(coppo_embeddings)
                    results["coppo"] = {
                        "embeddings": emb_list,
                        "model": "Jrine/co-po (fallback)",
                    }
                else:
                    # sklearn model - would need feature extraction first
                    results["coppo"] = {
                        "error": "sklearn model requires feature preprocessing",
                        "model": "Jrine/co-po",
                    }
            except Exception as e:
                results["coppo"] = {"error": str(e), "model": "Jrine/co-po"}
        else:
            results["coppo"] = {"error": "Model not loaded", "model": "Jrine/co-po"}

        return {
            "success": True,
            "text": request.text,
            "results": results,
            "available_models": available_models,
            "timestamp": get_timestamp(),
        }
    except Exception as e:
        logger.error(f"Combined analysis error: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.get(

    "/health",

    response_model=HealthResponse,

    tags=["Health"],

    summary="Health Check",

    description="Returns API health status and model availability",

)
async def health():
    """Health check endpoint for monitoring."""
    return {
        "status": "healthy",
        "models_loaded": {
            "blooms": blooms_model is not None,
            "dave": dave_model is not None,
            "coppo": coppo_model is not None,
        },
    }


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=7860, log_level="info")