Abeshith's picture
Add pipeline stages implementation
a7d80f2
from pathlib import Path
from mlpipeline.constants import CONFIG_FILE_PATH, AUTOML_CONFIG_FILE_PATH
from mlpipeline.utils.common import read_yaml, create_directories
from mlpipeline.entity.config_entity import (
DataIngestionConfig,
DataValidationConfig,
DataTransformationConfig,
FeatureEngineeringConfig,
ModelTrainerConfig,
ModelEvaluationConfig,
ModelPusherConfig,
)
class ConfigurationManager:
def __init__(self, config_filepath=CONFIG_FILE_PATH):
self.config = read_yaml(config_filepath)
create_directories([self.config.artifacts_root])
def get_data_ingestion_config(self) -> DataIngestionConfig:
config = self.config.data_ingestion
create_directories([config.root_dir])
return DataIngestionConfig(
root_dir=Path(config.root_dir),
source_url=config.source_url,
local_data_file=Path(config.local_data_file),
unzip_dir=Path(config.unzip_dir),
)
def get_data_validation_config(self) -> DataValidationConfig:
config = self.config.data_validation
create_directories([config.root_dir])
return DataValidationConfig(
root_dir=Path(config.root_dir),
data_dir=Path(config.data_dir),
status_file=Path(config.status_file),
schema_file=Path(config.schema_file),
)
def get_data_transformation_config(self) -> DataTransformationConfig:
config = self.config.data_transformation
create_directories([config.root_dir])
return DataTransformationConfig(
root_dir=Path(config.root_dir),
data_path=Path(config.data_path),
train_path=Path(config.train_path),
test_path=Path(config.test_path),
test_size=config.test_size,
random_state=config.random_state,
)
def get_feature_engineering_config(self) -> FeatureEngineeringConfig:
config = self.config.feature_engineering
create_directories([config.root_dir])
return FeatureEngineeringConfig(
root_dir=Path(config.root_dir),
train_path=Path(config.train_path),
test_path=Path(config.test_path),
output_train_path=Path(config.output_train_path),
output_test_path=Path(config.output_test_path),
)
def get_model_trainer_config(self) -> ModelTrainerConfig:
config = self.config.model_trainer
automl_config = read_yaml(Path(AUTOML_CONFIG_FILE_PATH))
create_directories([config.root_dir])
return ModelTrainerConfig(
root_dir=Path(config.root_dir),
train_data_path=Path(config.train_data_path),
test_data_path=Path(config.test_data_path),
model_path=Path(config.model_path),
target_column=config.target_column,
automl_library=automl_config.automl_library,
)
def get_model_evaluation_config(self) -> ModelEvaluationConfig:
config = self.config.model_evaluation
automl_config = read_yaml(Path(AUTOML_CONFIG_FILE_PATH))
create_directories([config.root_dir])
return ModelEvaluationConfig(
root_dir=Path(config.root_dir),
model_path=Path(config.model_path),
test_data_path=Path(config.test_data_path),
metrics_file=Path(config.metrics_file),
target_column=config.target_column,
automl_library=automl_config.automl_library,
)
def get_model_pusher_config(self) -> ModelPusherConfig:
config = self.config.model_pusher
create_directories([config.root_dir])
return ModelPusherConfig(
root_dir=Path(config.root_dir),
model_path=Path(config.model_path),
model_registry_path=Path(config.model_registry_path),
)