app.py

import joblib
import pandas as pd
from flask import Flask, request, jsonify
import os
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

#initialise flask app
sales_forecast_api = Flask('Sales forecasting')

# load the model and training columns
try:
    # Log current working directory and files
    current_dir = os.getcwd()
    logger.info(f"Current working directory: {current_dir}")
    files_in_dir = os.listdir(current_dir)
    logger.info(f"Files in current directory: {files_in_dir}")

    # Assuming model and columns are saved as a dictionary using joblib
    model_and_columns_path = 'sales_forecast_model_and_columns.joblib' # Update path if needed
    loaded_object = joblib.load(model_and_columns_path)

    model = loaded_object['model']
    training_columns = loaded_object['columns']


    logger.info("Model and training columns loaded successfully.")
except FileNotFoundError:
    logger.error(f"Model and training columns file not found at {model_and_columns_path}!")
    model = None
    training_columns = None # Ensure training_columns is also None if file not found
except Exception as e:
    logger.error(f"Error loading model or training columns: {e}")
    model = None
    training_columns = None


#define home page
@sales_forecast_api.get('/')
def home():
  return 'Welcome to the sales forecase api'

#define an endpoint for prediction
@sales_forecast_api.post('/v1/sales')
def sales_predict():
  if model is None or training_columns is None:
      return jsonify({"error": "Model or training columns not loaded"}), 500

  #get data from json request
  sales_data = request.get_json()

  #get relevant details
  sample = {
      'Product_Weight': sales_data['Product_Weight'],
      'Product_Sugar_Content': sales_data['Product_Sugar_Content'],
      'Product_Allocated_Area': sales_data['Product_Allocated_Area'],
      'Product_Type': sales_data['Product_Type'],
      'Product_MRP': sales_data['Product_MRP'],
      'Store_Establishment_Year': sales_data['Store_Establishment_Year'],
      'Store_Size': sales_data['Store_Size'],
      'Store_Location_City_Type': sales_data['Store_Location_City_Type'],
      'Store_Type': sales_data['Store_Type']
      }

  input_data = pd.DataFrame([sample])

#convert the categorical to dummies
  categorical_columns_for_dummies = ['Product_Sugar_Content','Product_Type','Store_Size','Store_Location_City_Type','Store_Type']
  input_df_dummies = pd.get_dummies(input_data, columns=categorical_columns_for_dummies, drop_first=True)

  # Reindex input_df_dummies to match the columns of X_train used during training
  input_df_aligned = input_df_dummies.reindex(columns=training_columns, fill_value=0)

  #make model to predict
  prediction = model.predict(input_df_aligned)

  return jsonify({'Prediction':prediction[0]})


#defining endpoint for batch
@sales_forecast_api.post('/v1/salesbatch')

def sales_batch_predict():
  if model is None or training_columns is None:
      return jsonify({"error": "Model or training columns not loaded"}), 500

  #get the file from the request
  file = request.files['file']
  #read the file to df
  input_data = pd.read_csv(file)

#convert the categorical to dummies
  categorical_columns_for_dummies = ['Product_Sugar_Content','Product_Type','Store_Size','Store_Location_City_Type','Store_Type']
  input_df_dummies = pd.get_dummies(input_data, columns=categorical_columns_for_dummies, drop_first=True)

  # Reindex input_df_dummies to match the columns of X_train used during training
  input_df_aligned =input_df_dummies.reindex(columns=training_columns, fill_value=0)

#predict
  predictions = model.predict(input_df_aligned).tolist() # Predict and convert to list
  product_id_list = input_data.Product_Id.tolist() # Convert to list
  store_id_list = input_data.Store_Id.tolist() # Convert to list

  # Create a list of dictionaries for the output
  output_list = []
  for i in range(len(product_id_list)):
    output_list.append({
        'Product_Id': product_id_list[i],
        'Store_Id': store_id_list[i],
        'Prediction': predictions[i]
    })

  return jsonify(output_list)

#run the flask app in debug mode
if __name__ == '__main__':
  sales_forecast_api.run(debug=True)