Code

import logging

import os

from flask import Flask, request, jsonify

from dotenv import load_dotenv

# Assuming processmodel is in the same directory or accessible via PYTHONPATH

import processmodel

# Load environment variables from .env file

load_dotenv()

app = Flask(__name__)

# Configure logging

LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO").upper()

logging.basicConfig(

    level=LOG_LEVEL,

    format="%(asctime)s - %(levelname)s - %(module)s - %(message)s",

    datefmt="%Y-%m-%d %H:%M:%S",

)

# --- Health Check Endpoint ---

@app.route("/epd", methods=["GET"])

def get_health():

    """

    Health check endpoint to verify the API is running.

    """

    return jsonify({"status": "OK", "message": "Everything is working fine"}), 200

# --- Model Execution Endpoint ---

@app.route("/epd", methods=["POST"])

def run_model():

    """

    Endpoint to receive model data and execute the EPD model.

    Expects a JSON payload in the request body.

    Returns the model output as a JSON response.

    """

    try:

        model_data = request.get_json()

        logging.info(f"Received input request: {model_data}")

        if model_data is None:

            logging.error("Received empty or invalid JSON payload.")

            return jsonify({"error": "Invalid JSON payload"}), 400

        response = processmodel.runEPDModel(model_data)

        # Convert numpy array to list if necessary

        if hasattr(response, 'tolist'):

            list_data = response.tolist()

            logging.info(f"Model output: {list_data}")

            return jsonify(list_data), 200

        else:

            logging.info(f"Model output: {response}")

            return jsonify(response), 200

    except Exception as e:

        logging.error(f"Error processing request: {str(e)}", exc_info=True)

        return jsonify({"error": "Internal server error"}), 500

if __name__ == "__main__":

    # Determine the port to run the application on

    port = int(os.environ.get("PORT", 5000))

    # Run the Flask development server in debug mode only if explicitly enabled

    debug_mode = os.environ.get("DEBUG", "False").lower() == "true"

    app.run(debug=debug_mode, host="0.0.0.0", port=port)

import xgboost as xgb

import numpy as np

import pandas as pd

import logging

import os

# Configure logging

logging.basicConfig(

    level=logging.INFO,  # Set your desired logging level

    format='%(asctime)s - %(levelname)s - %(module)s - %(message)s',

    datefmt='%Y-%m-%d %H:%M:%S'

)

def runEPDModel(single_input_data):

    """

    Runs the EPD model to predict based on the provided input data.

    Args:

        single_input_data (dict): A dictionary containing the feature values for a single prediction.

                                     The keys should match the expected feature names of the model.

    Returns:

        numpy.ndarray: A NumPy array containing the prediction probability for the positive class.

                       Returns None if an error occurs.

    """

    try:

        logging.info(f"Received input data: {single_input_data}")

        # Load the XGBoost model

        try:

            poc_xgb3_V2 = xgb.XGBClassifier(early_stopping_rounds=15)

            base_dir = os.path.dirname(__file__)

            file_name = "xgb6_v2_latest.json"

            file_path = os.path.join(base_dir, file_name)

            poc_xgb3_V2.load_model(file_path)

            logging.info(f"Model loaded successfully from: {file_path}")

        except FileNotFoundError:

            logging.error(f"Model file not found at: {file_path}")

            return None

        except xgb.core.XGBoostError as e:

            logging.error(f"Error loading XGBoost model: {e}")

            return None

        # Get feature names from the loaded model

        try:

            features_poc_xgb3_V2 = poc_xgb3_V2.feature_names_in_

            logging.info(f"Expected features: {features_poc_xgb3_V2}")

        except AttributeError:

            logging.error("Could not retrieve feature names from the loaded model.")

            return None

        # Convert the dictionary to a Pandas DataFrame (important for correct order)

        try:

            single_input_df = pd.DataFrame([single_input_data])

            # Ensure the input data has the expected columns in the correct order

            if not all(feature in single_input_df.columns for feature in features_poc_xgb3_V2):

                missing_features = [f for f in features_poc_xgb3_V2 if f not in single_input_df.columns]

                logging.error(f"Input data is missing the following expected features: {missing_features}")

                return None

            single_input_df = single_input_df[features_poc_xgb3_V2] # Reorder columns

        except Exception as e:

            logging.error(f"Error creating DataFrame from input data: {e}")

            return None

        # Make the prediction

        try:

            prediction_single = np.round(poc_xgb3_V2.predict_proba(single_input_df)[:, 1] * 100, 2) # Multiply by 100 and round to 2 decimals for percentage

            logging.info(f"Prediction probability: {prediction_single}")

            return prediction_single

        except Exception as e:

            logging.error(f"Error during prediction: {e}")

            return None

    except Exception as overall_e:

        logging.critical(f"An unexpected error occurred in runEPDModel: {overall_e}", exc_info=True)

        return None