fix: upload machine learning models code sources

Author: tph-kds

src/ano_detection/config/__init__.py

@@ -4,7 +4,8 @@
     LoggerArgumentsConfig,
     DataIngestionConfig,
     DataProcessingConfig,
-    DataTrainingConfig
+    DataTrainingConfig,
+    ModelArgumentsConfig
 )
 
 __all__ = [
@@ -13,5 +14,6 @@
     "LoggerArgumentsConfig",
     "DataIngestionConfig",
     "DataProcessingConfig",
-    "DataTrainingConfig"
+    "DataTrainingConfig",
+    "ModelArgumentsConfig",
 ]

src/ano_detection/config/arguments_config.py

@@ -222,6 +222,44 @@ class DataTrainingConfig:
         }
     )
 
+@dataclass(frozen=True)
+class ModelArgumentsConfig:
+    root_dir: str = field(
+        default="src/artifacts/models",
+        metadata={
+            "help": "Root directory of the dataset.",
+            "type": "string",
+        }
+    )
+    model_name: str = field(
+        default="name_model",
+        metadata={
+            "help": "Name of the model.",
+            "type": "string",
+        }
+    )
+    model_path: str = field(
+        default="model.pkl",
+        metadata={
+            "help": "Path of the model.",
+            "type": "string",
+        }
+    )
+    model_params: dict = field(
+        default={"n_estimators": 100, "max_depth": 5, "random_state": 42},
+        metadata={
+            "help": "Parameters of the model.",
+            "type": "dict",
+        }
+    )
+    model_description: str = field(
+        default="Model Description",
+        metadata={
+            "help": "Description of the model.",
+            "type": "string",
+        }
+    )
+
 
 
 

src/ano_detection/config/manager_config.py

@@ -10,6 +10,7 @@
     DataIngestionConfig,
     DataProcessingConfig,
     DataTrainingConfig,
+    ModelArgumentsConfig
 
 )
 
@@ -104,4 +105,20 @@ def get_data_training_arguments_config(self) -> DataTrainingConfig:
         )
 
         return data_training_config
+    
+    ### MODEL TRAINING CONFIG PARAMS PHASE ###
+    def get_model_training_arguments_config(self) -> ModelArgumentsConfig:
+        config = self.config.model.stages.models
+
+        create_directories([config.root_dir])
+
+        model_training_config = ModelArgumentsConfig(
+            root_dir=config.root_dir,
+            model_name=config.model_name,
+            model_path=config.model_path,
+            model_params=config.model_params,
+            model_description=config.model_description,
+        )
+
+        return model_training_config
 

src/ano_detection/models/base_model.py

@@ -4,7 +4,20 @@
 from abc import ABC, abstractmethod # ( Abstract Base Classes Library )
 from src.ano_detection.logger import logger
 from src.ano_detection.exception import MyException
+from src.ano_detection.config import ModelArgumentsConfig
+from src.ano_detection.visualization import VILModel
+from src.config_params import ROOT_PROJECT
 
+from sklearn.metrics import (
+    accuracy_score, 
+    recall_score, 
+    precision_score, 
+    f1_score,
+    roc_curve, 
+    auc, 
+    precision_recall_curve
+)
+from sklearn.preprocessing import label_binarize
 
 class BaseModel(ABC):
     """
@@ -19,9 +32,18 @@ class BaseModel(ABC):
     """
     @abstractmethod
     def __init__(self, 
-                 config: object,
+                 config: ModelArgumentsConfig,
                  **kwargs):
-        pass
+        super(BaseModel, self).__init__(**kwargs)   
+        self.config = config
+        self.root_dir = self.config.root_dir
+        self.model_name = self.config.model_name
+        self.model_path = ROOT_PROJECT / self.root_dir / self.config.model_path
+        self.model_params = self.config.model_params
+        self.model_description = self.config.model_description
+
+        self.visualization = VILModel()
+
 
     @abstractmethod
     def __repr__(self):
@@ -47,4 +69,41 @@ def save_model(self):
     def load_model(self):
         pass
 
+    @abstractmethod
+    def get_score(self, y_true, y_pred, y_scores, n_classes):
+        accuracy = accuracy_score(y_true, y_pred)
+        recall = recall_score(y_true, y_pred, average='weighted')
+        precision = precision_score(y_true, y_pred, average='weighted')
+        f1 = f1_score(y_true, y_pred, average='weighted')
+
+        y_test_bin = label_binarize(y_true, classes=[0, 1, 2])  # Convert y_test to binary format
+
+        # Compute ROC curve and AUC for each class
+        fpr = {}
+        tpr = {}
+        roc_auc = {}
+        for i in range(n_classes):
+            fpr[i], tpr[i], _ = roc_curve(y_test_bin[:, i], y_scores[:, i])
+            roc_auc[i] = auc(fpr[i], tpr[i])
+
+        # Compute Precision-Recall curve and AUC for each class
+        precision_curve_pr = {}
+        recall_curve_pr = {}
+        pr_auc = {}
+        for i in range(n_classes):
+            precision_curve_pr[i], recall_curve_pr[i], _ = precision_recall_curve(y_test_bin[:, i], y_scores[:, i])
+            pr_auc[i] = auc(recall_curve_pr[i], precision_curve_pr[i])
+
+
+        print(f"Accuracy: {accuracy:.2f}")
+        print(f"Recall: {recall:.2f}")
+        print(f"Precision: {precision:.2f}")
+        print(f"F1-score: {f1:.2f}")
+
+        self.visualization.auc_roc_curve(fpr, tpr, roc_auc, n_classes)
+        self.visualization.auc_pr_curve(precision_curve_pr, recall_curve_pr, pr_auc, n_classes)
+        print("Complete!")
+
+        return accuracy, recall, precision, f1
+
 

src/ano_detection/models/gradient_boosting.py

@@ -0,0 +1,110 @@
+import os
+import sys
+
+from .base_model import BaseModel
+# from src.ano_detection.config import ModelArgumentsConfig
+from src.ano_detection.logger import logger
+from src.ano_detection.exception import MyException
+
+from sklearn.ensemble import GradientBoostingClassifier
+from sklearn.metrics import classification_report
+
+from src.ano_detection.utils import save_pickle, load_pickle
+
+
+
+class GradientBoostingModel(BaseModel):
+    """
+     Configuration for Gradient Boosting Model 
+
+    Params:
+        config: object
+
+    Returns:
+        model for training and prediction tasks
+    
+    """
+    def __init__(self, **kwargs):
+        super(GradientBoostingModel, self).__init__(**kwargs)   
+
+        # self.config = kwargs.get("config")
+        
+        self.gradient_boosting_model = GradientBoostingClassifier(random_state=2024,
+                                                n_estimators=100,
+                                                learning_rate=0.1,
+                                                max_depth=10,
+                                                min_samples_leaf=1,
+                                                min_weight_fraction_leaf=0.0,
+                                                max_features=1.0)
+
+    def __repr__(self):
+        print(f"GradientBoostingModel config: {self.config}")
+        logger.log_message("info", f"GradientBoostingModel config: {self.config}")
+        return f"{self.__class__.__name__}"
+
+    def fit(self, X_train, y_train):
+        try:
+            logger.log_message("info", "Fitting Gradient Boosting Model....")
+            self.gradient_boosting_model.fit(X_train, y_train)
+
+            logger.log_message("info", "Fitted Gradient Boosting Model successfully....")
+
+        except Exception as e:
+            logger.log_message("error", f"Error in fitting Gradient Boosting Model: {e}")
+            my_exception = MyException(
+                error_message="Error in fitting Gradient Boosting Model", 
+                error_details=sys
+            )
+            print(my_exception)
+
+    def predict(self, X_test, y_test):
+        try:
+            logger.log_message("info", "Predicting Gradient Boosting Model....")
+            y_pred_gbc = self.gradient_boosting_model.predict(X_test)
+            print(classification_report(y_test, y_pred_gbc))
+
+            logger.log_message("info", "Predicted Gradient Boosting Model successfully....")
+
+            return y_pred_gbc
+        
+        except Exception as e:
+            logger.log_message("error", f"Error in predicting Gradient Boosting Model: {e}")
+            my_exception = MyException(
+                error_message="Error in predicting Gradient Boosting Model", 
+                error_details=sys
+            )
+            print(my_exception)
+
+
+    def save_model(self):
+        try:
+            logger.log_message("info", "Saving Gradient Boosting Model....")
+            save_pickle(self.gradient_boosting_model, self.model_path)
+
+            logger.log_message("info", "Saved Gradient Boosting Model successfully....")
+
+        except Exception as e:
+            logger.log_message("error", f"Error in saving Gradient Boosting Model: {e}")
+            my_exception = MyException(
+                error_message="Error in saving Gradient Boosting Model", 
+                error_details=sys
+            )
+            print(my_exception)
+
+    def load_model(self):
+        try:
+            logger.log_message("info", "Loading Gradient Boosting Model....")
+            self.gradient_boosting_model = load_pickle(self.model_path)
+
+            logger.log_message("info", "Loaded Gradient Boosting Model successfully....")
+
+        except Exception as e:
+            logger.log_message("error", f"Error in loading Gradient Boosting Model: {e}")
+            my_exception = MyException(
+                error_message="Error in loading Gradient Boosting Model",
+                error_details=sys
+
+            )
+            print(my_exception)
+
+

src/ano_detection/models/logistic_regression.py

@@ -0,0 +1,107 @@
+import os
+import sys
+
+from .base_model import BaseModel
+# from src.ano_detection.config import ModelArgumentsConfig
+from src.ano_detection.logger import logger
+from src.ano_detection.exception import MyException
+
+from sklearn.linear_model import LogisticRegression
+from sklearn.metrics import classification_report
+
+from src.ano_detection.utils import save_pickle, load_pickle
+
+
+
+class LogisticRegressionModel(BaseModel):
+    """
+     Configuration for Logistic Regression Model 
+
+    Params:
+        config: object
+
+    Returns:
+        model for training and prediction tasks
+    
+    """
+    def __init__(self, **kwargs):
+        super(LogisticRegressionModel, self).__init__(**kwargs)   
+
+        # self.config = kwargs.get("config")
+        
+        self.lr_model = LogisticRegression(random_state=2024,
+                                    max_iter=1000,
+                                    solver='lbfgs',
+                                    multi_class='multinomial')
+
+    def __repr__(self):
+        print(f"LogisticRegressionModel config: {self.config}")
+        logger.log_message("info", f"LLogisticRegressionModel config: {self.config}")
+        return f"{self.__class__.__name__}"
+
+    def fit(self, X_train, y_train):
+        try:
+            logger.log_message("info", "Fitting Logistic Regression Model....")
+            self.lr_model.fit(X_train, y_train)
+
+            logger.log_message("info", "Fitted Logistic Regression Model successfully....")
+
+        except Exception as e:
+            logger.log_message("error", f"Error in fitting Logistic Regression Model: {e}")
+            my_exception = MyException(
+                error_message="Error in fitting Logistic Regression Model", 
+                error_details=sys
+            )
+            print(my_exception)
+
+    def predict(self, X_test, y_test):
+        try:
+            logger.log_message("info", "Predicting Logistic Regression Model....")
+            y_pred_gbc = self.lr_model.predict(X_test)
+            print(classification_report(y_test, y_pred_gbc))
+
+            logger.log_message("info", "Predicted Logistic Regression Model successfully....")
+
+            return y_pred_gbc
+        
+        except Exception as e:
+            logger.log_message("error", f"Error in predicting Logistic Regression Model: {e}")
+            my_exception = MyException(
+                error_message="Error in predicting Logistic Regression Model", 
+                error_details=sys
+            )
+            print(my_exception)
+
+
+    def save_model(self):
+        try:
+            logger.log_message("info", "Saving Logistic Regression Model....")
+            save_pickle(self.lr_model, self.model_path)
+
+            logger.log_message("info", "Saved Logistic Regression Model successfully....")
+
+        except Exception as e:
+            logger.log_message("error", f"Error in saving Logistic Regression Model: {e}")
+            my_exception = MyException(
+                error_message="Error in saving Logistic Regression Model", 
+                error_details=sys
+            )
+            print(my_exception)
+
+    def load_model(self):
+        try:
+            logger.log_message("info", "Loading Logistic Regression Model....")
+            self.lr_model = load_pickle(self.model_path)
+
+            logger.log_message("info", "Loaded Logistic Regression Model successfully....")
+
+        except Exception as e:
+            logger.log_message("error", f"Error in loading Logistic Regression Model: {e}")
+            my_exception = MyException(
+                error_message="Error in loading Logistic Regression Model",
+                error_details=sys
+
+            )
+            print(my_exception)
+
+