Create pipeline/training_pipeline.py

src/pipeline/training_pipeline.py

@@ -0,0 +1,122 @@
+import os 
+import sys 
+
+from src.exception.exception import DeliveryTimeException
+from src.logging.logger import logging
+
+from src.components.data_ingestion import DataIngestion
+from src.components.data_validation import DataValidation
+from src.components.data_transformation import DataTransformation
+from src.components.model_trainer import ModelTrainer
+
+from src.entity.config_entity import (
+    TrainingPipelineConfig,
+    DataIngestionConfig,
+    DataValidationConfig,
+    DataTransformationConfig,
+    ModelTrainerConfig
+)
+
+from src.entity.artifact_entity import (
+    DataIngestionArtifact,
+    DataValidationArtifact,
+    DataTransformationArtifact,
+    ModelTrainerArtifact
+)
+
+from src.constants.training_pipeline import TRAINING_BUCKET_NAME
+from src.cloud.s3_syncer import S3Sync
+from src.constants.training_pipeline import SAVED_MODEL_DIR
+import sys
+
+class TrainingPipeline:
+    def __init__(self):
+        self.training_pipeline_config=TrainingPipelineConfig()
+        self.s3_sync = S3Sync()
+
+    def start_data_ingestion(self):
+        try:
+            self.data_ingestion_config=DataIngestionConfig(training_pipeline_config=self.training_pipeline_config)
+            logging.info("Start data Ingestion")
+            data_ingestion=DataIngestion(data_ingestion_config=self.data_ingestion_config)
+            data_ingestion_artifact=data_ingestion.initiate_date_ingestion()
+            logging.info(f"Data Ingestion completed with artifact: {data_ingestion_artifact}")
+            return data_ingestion_artifact
+        except Exception as e:
+            raise DeliveryTimeException(e, sys)
+    
+    def start_data_validation(self, data_ingestion_artifact:DataIngestionArtifact):
+        try:
+            data_validation_config=DataValidationConfig(training_pipeline_config=self.training_pipeline_config)
+            data_validation=DataValidation(data_ingestion_artifact=data_ingestion_artifact, data_validation_config=data_validation_config)
+            logging.info("Initiate data validation")
+            data_validation_artifact=data_validation.initiate_data_validation()
+            return data_validation_artifact
+        except Exception as e:
+            raise DeliveryTimeException(e, sys)
+    
+    def start_data_transformation(self, data_validation_artifact:DataValidationArtifact):
+        try:
+            data_transformation_config=DataTransformationConfig(training_pipeline_config=self.training_pipeline_config)
+            data_transformation=DataTransformation(data_validation_artifact=data_validation_artifact,
+                                                   data_transformation_config=data_transformation_config)
+            data_transformation_artifact=data_transformation.initiate_data_transformation()
+            return data_transformation_artifact
+        except Exception as e:
+            raise DeliveryTimeException(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
+            )
+
+            model_trainer=ModelTrainer(
+                data_transformation_artifact=data_transformation_artifact,
+                model_trainer_config=self.model_trainer_config
+            )
+
+            model_trainer_artifact=model_trainer.initiate_model_trainer()
+
+            return model_trainer_artifact
+        except Exception as e:
+            raise DeliveryTimeException(e, sys)
+        
+
+    def sync_artifact_dir_to_s3(self):
+        try:
+            aws_bucket_url = f"s3://{TRAINING_BUCKET_NAME}/artifact/{self.training_pipeline_config.timestamp}"
+            self.s3_sync.sync_folder_to_s3(folder=self.training_pipeline_config.artifact_dir, aws_bucket_url=aws_bucket_url)
+
+        except Exception as e:
+            raise DeliveryTimeException(e, sys)
+        
+    def sync_saved_model_dir_to_s3(self):
+        try:
+            aws_bucket_url = f"s3://{TRAINING_BUCKET_NAME}/final_model/{self.training_pipeline_config.timestamp}"
+            self.s3_sync.sync_folder_to_s3(folder=self.training_pipeline_config.model_dir, aws_bucket_url=aws_bucket_url)
+        except Exception as e:
+            raise DeliveryTimeException(e, sys)
+        
+
+    def run_pipeline(self):
+        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)
+
+            self.sync_artifact_dir_to_s3()
+            self.sync_saved_model_dir_to_s3()
+
+            return model_trainer_artifact
+            
+        except Exception as e:
+            raise DeliveryTimeException(e, sys)
+        
+if __name__ == "__main__":
+    try:
+        training_pipeline = TrainingPipeline()
+        training_pipeline.run_pipeline()
+    except Exception as e:
+        raise DeliveryTimeException(e, sys)