File size: 7,186 Bytes
b4f6ffc c3c7748 b4f6ffc db5542d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 |
import sys
from anime_recommender.loggers.logging import logging
from anime_recommender.exception.exception import AnimeRecommendorException
from anime_recommender.components.data_ingestion import DataIngestion
from anime_recommender.components.data_transformation import DataTransformation
from anime_recommender.components.collaborative_recommender import CollaborativeModelTrainer
from anime_recommender.components.content_based_recommender import ContentBasedModelTrainer
from anime_recommender.components.top_anime_recommenders import PopularityBasedRecommendor
from anime_recommender.entity.config_entity import (
TrainingPipelineConfig,
DataIngestionConfig,
DataTransformationConfig,
CollaborativeModelConfig,
ContentBasedModelConfig,
)
from anime_recommender.entity.artifact_entity import (
DataIngestionArtifact,
DataTransformationArtifact,
CollaborativeModelArtifact,
ContentBasedModelArtifact,
)
class TrainingPipeline:
"""
Orchestrates the entire anime recommender training pipeline, including
data ingestion, transformation, model training, and popularity-based recommendations.
"""
def __init__(self):
"""
Initialize the TrainingPipeline with required configurations.
"""
self.training_pipeline_config = TrainingPipelineConfig()
def start_data_ingestion(self) -> DataIngestionArtifact:
"""
Starts the data ingestion process.
Returns:
DataIngestionArtifact: Contains information about ingested data.
"""
try:
logging.info("Initiating Data Ingestion...")
data_ingestion_config = DataIngestionConfig(self.training_pipeline_config)
data_ingestion = DataIngestion(data_ingestion_config=data_ingestion_config)
data_ingestion_artifact = data_ingestion.ingest_data()
logging.info(f"Data Ingestion completed: {data_ingestion_artifact}")
return data_ingestion_artifact
except Exception as e:
raise AnimeRecommendorException(e, sys)
def start_data_transformation(self, data_ingestion_artifact: DataIngestionArtifact) -> DataTransformationArtifact:
"""
Starts the data transformation process.
Returns:
DataTransformationArtifact: Contains transformed data.
"""
try:
logging.info("Initiating Data Transformation...")
data_transformation_config = DataTransformationConfig(self.training_pipeline_config)
data_transformation = DataTransformation(
data_ingestion_artifact=data_ingestion_artifact,
data_transformation_config=data_transformation_config
)
data_transformation_artifact = data_transformation.initiate_data_transformation()
logging.info(f"Data Transformation completed: {data_transformation_artifact}")
return data_transformation_artifact
except Exception as e:
raise AnimeRecommendorException(e, sys)
def start_collaborative_model_training(self, data_transformation_artifact: DataTransformationArtifact) -> CollaborativeModelArtifact:
"""
Starts collaborative filtering model training.
Returns:
CollaborativeModelTrainerArtifact: Trained collaborative model artifact.
"""
try:
logging.info("Initiating Collaborative Model Training...")
collaborative_model_config = CollaborativeModelConfig(self.training_pipeline_config)
collaborative_model_trainer = CollaborativeModelTrainer(
collaborative_model_trainer_config=collaborative_model_config,
data_transformation_artifact=data_transformation_artifact
)
collaborative_model_trainer_artifact = collaborative_model_trainer.initiate_model_trainer()
logging.info(f"Collaborative Model Training completed: {collaborative_model_trainer_artifact}")
return collaborative_model_trainer_artifact
except Exception as e:
raise AnimeRecommendorException(e, sys)
def start_content_based_model_training(self, data_ingestion_artifact: DataIngestionArtifact) -> ContentBasedModelArtifact:
"""
Starts content-based filtering model training.
Returns:
ContentBasedModelTrainerArtifact: Trained content-based model artifact.
"""
try:
logging.info("Initiating Content-Based Model Training...")
content_based_model_config = ContentBasedModelConfig(self.training_pipeline_config)
content_based_model_trainer = ContentBasedModelTrainer(
content_based_model_trainer_config=content_based_model_config,
data_ingestion_artifact=data_ingestion_artifact
)
content_based_model_trainer_artifact = content_based_model_trainer.initiate_model_trainer()
logging.info(f"Content-Based Model Training completed: {content_based_model_trainer_artifact}")
return content_based_model_trainer_artifact
except Exception as e:
raise AnimeRecommendorException(e, sys)
def start_popularity_based_filtering(self, data_ingestion_artifact: DataIngestionArtifact):
"""
Generates popularity-based recommendations.
"""
try:
logging.info("Initiating Popularity-Based Filtering...")
filtering = PopularityBasedRecommendor(data_ingestion_artifact=data_ingestion_artifact)
recommendations = filtering.initiate_model_trainer(filter_type='popular_animes')
logging.info("Popularity-Based Filtering completed.")
return recommendations
except Exception as e:
raise AnimeRecommendorException(e, sys)
def run_pipeline(self):
"""
Executes the entire training pipeline.
"""
try:
# Data Ingestion
data_ingestion_artifact = self.start_data_ingestion()
# Data Transformation
data_transformation_artifact = self.start_data_transformation(data_ingestion_artifact)
# Collaborative Model Training
collaborative_model_trainer_artifact = self.start_collaborative_model_training(data_transformation_artifact)
# Content-Based Model Training
content_based_model_trainer_artifact = self.start_content_based_model_training(data_ingestion_artifact)
# Popularity-Based Filtering
popularity_recommendations = self.start_popularity_based_filtering(data_ingestion_artifact)
logging.info("Training Pipeline executed successfully.")
except Exception as e:
raise AnimeRecommendorException(e, sys)
if __name__ == "__main__":
try:
pipeline = TrainingPipeline()
pipeline.run_pipeline()
except Exception as e:
logging.error(f"Pipeline execution failed: {str(e)}")
raise AnimeRecommendorException(e, sys) |