app.py

import tensorflow as tf
from pydantic import BaseModel
from typing import List
from fastapi import FastAPI
import numpy as np
import requests
from fastapi.middleware.cors import CORSMiddleware
import pandas as pd
from sklearn.preprocessing import MinMaxScaler
from sklearn.cluster import KMeans
from keras.models import load_model


model = load_model('models/big5_gru.h5')

class PersonalityRequest(BaseModel):
    responses: List[float] 

class PersonalityResponse(BaseModel):
    personality_type: int 
    personality_name: str

personality_mapping = {
    0: 'Extroverted',
    1: 'Neurotic',
    2: 'Agreeable',
    3: 'Conscientious',
    4: 'Open'
}

origins = ["*"]

app = FastAPI()

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

def five2one(item):
    return (item - 0) / (5 - 0)

model_List = ["gpt-4", "gpt-4-0613", "gpt-3.5-turbo-16k-0613", "gpt-3.5-long"]

def get_response(model, prompt):
    url = "https://gptapi.lumaticai.com/v1/chat/completions"
    body = {
        "model": model,
        "stream": False,
        "messages": [
            {"role": "assistant", "content": prompt}
        ]
    }

    try:
        json_response = requests.post(url, json=body).json().get('choices', [])
        print(model)
        for choice in json_response:
            # print(choice.get('message', {}).get('content', ''))
            res = choice.get('message', {}).get('content', '')
            return res
    except Exception as e:
        # print('Error : ', e)
        return e

def get_response_with_fallback(prompt):
  for model in model_List:
    try:
      response = get_response(model, prompt)
      if not isinstance(response, Exception):  # Check if it's an error
        return {"response": response, "model": model}
    except Exception as e:
      return f"Error with model {model}: {e}"
  # If none succeed, return an error message
  return "Failed to generate diagnosis with any model."


def preprocess_data(new_data_path):
    # Load new data
    new_data = pd.DataFrame(new_data_path)
    
    # Scale the data using the same scaler used during training
    scaler = MinMaxScaler(feature_range=(0, 1))
    scaled_data = scaler.fit_transform(new_data)
    
    # Reshape the data to fit the model input
    reshaped_data = scaled_data.reshape((scaled_data.shape[1], scaled_data.shape[0], 1))
    
    return reshaped_data

def predict_clusters(model, preprocessed_data):
    # Predict the cluster for each instance
    predictions = model.predict(preprocessed_data)
    
    # Get the cluster with the highest probability
    predicted_clusters = np.argmax(predictions, axis=1)
    
    return predicted_clusters

@app.post("/predict", response_model=PersonalityResponse)
def predict_personality(request: PersonalityRequest):
    preprocessed_data = preprocess_data(request.responses)

    predicted_clusters = predict_clusters(model, preprocessed_data)
    personality_type = predicted_clusters[0]
    
    personality_name = personality_mapping[predicted_clusters[0]]
    print(personality_name)
    
    return PersonalityResponse(personality_type=personality_type, personality_name=personality_name)

@app.post('/personality/description')
def personality_description(personality_name : str):
    prompt = f"\nYou are an psychology expert. I will give you a personality name based on The Big Five OCEAN Personality types. you will provide me with the description of the personality, tell everything about the personality type and the person who have this type of personality. \nPersonality type: {personality_name}. Answer in this format: \nPersonality description: "

    try:
        resultt = get_response_with_fallback(prompt)

        modell = resultt.get('model')

        resultsj = resultt.get('response')

        results = str(resultsj)

        # rst = resultt.response

        if isinstance(results, Exception):
            return {"message": "Error generating diagnosis: " + results, "model": modell}
        
        for line in results.split('\n'):
            if line.startswith('"**Personality description:'):
                ps = line.split(':**')[1].strip()
            elif line.startswith('**Personality description:'):
                ps = line.split(':**')[1].strip()
            elif line.startswith('Personality description:'):
                ps = line.split(':')[1].strip()
        
        return ps

    except Exception as e:
        return {"message": "couldn't get description for " + personality_name}



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