completed training pipeline

Author: GoJo-Rika

network_security/pipeline/training_pipeline.py

@@ -0,0 +1,128 @@
+import sys
+
+from network_security.components.data_ingestion import DataIngestion
+from network_security.components.data_transformation import DataTransformation
+from network_security.components.data_validation import DataValidation
+from network_security.components.model_trainer import ModelTrainer
+from network_security.entity.artifact_entity import (
+    DataIngestionArtifact,
+    DataTransformationArtifact,
+    DataValidationArtifact,
+    ModelTrainerArtifact,
+)
+from network_security.entity.config_entity import (
+    DataIngestionConfig,
+    DataTransformationConfig,
+    DataValidationConfig,
+    ModelTrainerConfig,
+    TrainingPipelineConfig,
+)
+from network_security.exception.exception import NetworkSecurityException
+from network_security.logging.logger import logging
+
+
+class TrainingPipeline:
+    def __init__(self) -> None:
+        self.training_pipeline_config = TrainingPipelineConfig()
+
+    def start_data_ingestion(self) -> DataIngestionArtifact:
+        try:
+            self.data_ingestion_config = DataIngestionConfig(
+                training_pipeline_config=self.training_pipeline_config,
+            )
+            logging.info("Initiated Data Ingestion")
+            data_ingestion = DataIngestion(
+                data_ingestion_config=self.data_ingestion_config,
+            )
+            data_ingestion_artifact = data_ingestion.initiate_data_ingestion()
+            logging.info(
+                f"Data Ingestion completed and artifact: {data_ingestion_artifact}",
+            )
+            return data_ingestion_artifact
+
+        except Exception as e:
+            raise NetworkSecurityException(e, sys)
+
+    def start_data_validation(
+        self, data_ingestion_artifact: DataIngestionArtifact
+    ) -> DataValidationArtifact:
+        try:
+            self.data_validation_config = DataValidationConfig(
+                training_pipeline_config=self.training_pipeline_config,
+            )
+            logging.info("Initiated Data Validation")
+            data_validation = DataValidation(
+                data_ingestion_artifact=data_ingestion_artifact,
+                data_validation_config=self.data_validation_config,
+            )
+            data_validation_artifact = data_validation.initiate_data_validation()
+            logging.info(
+                f"Data Validation completed and artifact: {data_validation_artifact}",
+            )
+            return data_validation_artifact
+        except Exception as e:
+            raise NetworkSecurityException(e, sys)
+
+    def start_data_transformation(
+        self,
+        data_validation_artifact: DataValidationArtifact,
+    ) -> DataTransformationArtifact:
+        try:
+            self.data_transformation_config = DataTransformationConfig(
+                training_pipeline_config=self.training_pipeline_config,
+            )
+            logging.info("Initiated Data Transformation")
+            data_transformation = DataTransformation(
+                data_validation_artifact=data_validation_artifact,
+                data_transformation_config=self.data_transformation_config,
+            )
+
+            data_transformation_artifact = (
+                data_transformation.initiate_data_transformation()
+            )
+            logging.info(
+                f"Data Transformation completed and artifact: {data_transformation_artifact}",
+            )
+            return data_transformation_artifact
+        except Exception as e:
+            raise NetworkSecurityException(e, sys)
+
+    def start_model_trainer(
+        self,
+        data_transformation_artifact: DataTransformationArtifact,
+    ) -> ModelTrainerArtifact:
+        try:
+            self.model_trainer_config: ModelTrainerConfig = ModelTrainerConfig(
+                training_pipeline_config=self.training_pipeline_config,
+            )
+            logging.info("Initiated Model Training")
+            model_trainer = ModelTrainer(
+                data_transformation_artifact=data_transformation_artifact,
+                model_trainer_config=self.model_trainer_config,
+            )
+
+            model_trainer_artifact = model_trainer.initiate_model_trainer()
+            logging.info(
+                f"Model Training completed and artifact: {model_trainer_artifact}",
+            )
+            return model_trainer_artifact
+
+        except Exception as e:
+            raise NetworkSecurityException(e, sys)
+
+    def run_pipeline(self) -> ModelTrainerArtifact:
+        try:
+            data_ingestion_artifact = self.start_data_ingestion()
+            data_validation_artifact = self.start_data_validation(
+                data_ingestion_artifact=data_ingestion_artifact,
+            )
+            data_transformation_artifact = self.start_data_transformation(
+                data_validation_artifact=data_validation_artifact,
+            )
+            model_trainer_artifact = self.start_model_trainer(
+                data_transformation_artifact=data_transformation_artifact,
+            )
+
+            return model_trainer_artifact
+        except Exception as e:
+            raise NetworkSecurityException(e, sys)