app.py

from flask import Flask, request, jsonify
from flask_restx import Api, Resource, fields
from flask_cors import CORS
import pandas as pd
import numpy as np
import joblib
import os

app = Flask(__name__)
CORS(app)  # Enable CORS for frontend access

# Swagger API setup
api = Api(
    app,
    version='1.0',
    title='Effort Prediction API',
    description='ML-powered API to predict software development effort in person-hours based on project metrics',
    doc='/docs'  # Swagger UI available at /docs
)

ns = api.namespace('predict', description='Effort prediction operations')

# ----------------- Load Model & Artifacts -----------------
def load_artifacts():
    model = joblib.load("model/best_model.pkl")
    scaler = joblib.load("model/scaler.pkl")
    feature_cols = joblib.load("model/features.pkl")
    median_afp = joblib.load("model/median_afp.pkl")
    return model, scaler, feature_cols, median_afp

# Load once at startup
MODEL, SCALER, FEATURE_COLS, MEDIAN_AFP = load_artifacts()

# ----------------- Feature Engineering -----------------
def engineer_features(project, median_afp):
    """Engineer features for prediction"""
    df = pd.DataFrame([project])
    
    # IO Ratio
    df['IO_Ratio'] = df['Input'] / (df['Output'] + 1)
    
    # Total Code Churn
    df['Total_Churn'] = df['Added'] + df['Changed'] + df['Deleted']
    
    # Dominant Complexity Type
    complexity_cols = ['Input', 'Output', 'Enquiry', 'File', 'Interface']
    df['Dominant_Type'] = df[complexity_cols].idxmax(axis=1)
    
    # Size Category
    df['Size_Category'] = (df['AFP'] > median_afp).astype(int)
    
    # Complexity Score
    df['Complexity_Score'] = (
        df['Input'] * 1.0 +
        df['Output'] * 1.5 +
        df['Enquiry'] * 0.8 +
        df['File'] * 1.2 +
        df['Interface'] * 1.3
    )
    
    # Log transformations
    df['Log_AFP'] = np.log1p(df['AFP'])
    df['Log_Duration'] = np.log1p(df['Duration'])
    
    return df

# ----------------- API Models for Swagger -----------------
project_input = api.model('ProjectInput', {
    'AFP': fields.Integer(required=True, description='Adjusted Function Points', example=1587),
    'Input': fields.Integer(required=True, description='Number of input transactions', example=774),
    'Output': fields.Integer(required=True, description='Number of output transactions', example=260),
    'Enquiry': fields.Integer(required=True, description='Number of enquiry transactions', example=340),
    'File': fields.Integer(required=True, description='Number of internal logical files', example=128),
    'Interface': fields.Integer(required=True, description='Number of external interface files', example=0),
    'Duration': fields.Integer(required=True, description='Project duration in months', example=4),
    'Resource': fields.Integer(required=True, description='Number of team resources', example=4),
    'Added': fields.Integer(required=True, description='Lines of code added', example=1502),
    'Changed': fields.Integer(required=True, description='Lines of code changed', example=0),
    'Deleted': fields.Integer(required=True, description='Lines of code deleted', example=0),
    'PDR_AFP': fields.Float(required=True, description='Productivity Delivery Rate per AFP', example=4.7)
})

prediction_output = api.model('PredictionOutput', {
    'predicted_effort': fields.Float(description='Predicted effort in person-hours'),
    'predicted_effort_days': fields.Float(description='Predicted effort in person-days (8h/day)'),
    'status': fields.String(description='Request status')
})

# ----------------- API Endpoints -----------------
@ns.route('/')
class EffortPrediction(Resource):
    @ns.doc('predict_effort')
    @ns.expect(project_input)
    @ns.marshal_with(prediction_output)
    def post(self):
        """Predict software development effort based on project metrics"""
        try:
            project = request.json
            
            # Feature engineering
            df_features = engineer_features(project, MEDIAN_AFP)
            
            # Prepare and scale features
            X = df_features[FEATURE_COLS].fillna(0)
            X_scaled = SCALER.transform(X)
            
            # Predict
            predicted_effort = MODEL.predict(X_scaled)[0]
            
            return {
                'predicted_effort': round(float(predicted_effort), 2),
                'predicted_effort_days': round(float(predicted_effort) / 8, 2),
                'status': 'success'
            }
        except Exception as e:
            api.abort(400, f"Prediction failed: {str(e)}")

@ns.route('/health')
class HealthCheck(Resource):
    def get(self):
        """Check API health status"""
        return {'status': 'healthy', 'model_loaded': MODEL is not None}

@ns.route('/features')
class FeatureInfo(Resource):
    def get(self):
        """Get list of required input features"""
        return {
            'required_features': [
                'AFP', 'Input', 'Output', 'Enquiry', 'File', 
                'Interface', 'Duration', 'Resource', 'Added', 
                'Changed', 'Deleted', 'PDR_AFP'
            ],
            'description': 'All features are required for prediction'
        }

if __name__ == '__main__':
    port = int(os.environ.get('PORT', 5000))
    app.run(host='0.0.0.0', port=port, debug=False)