Spaces:
Sleeping
Sleeping
File size: 5,338 Bytes
51f7cb3 | 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 |
from src.entity.config_entity import DataValidationConfig
from src.entity.artifact_entity import DataValidationArtifact,DataIngestionArtifact
from src.utils.main_utils import read_yaml_file
from src.constants import SCHEMA_FILE_PATH
from src.exception import MyException
import sys
import pandas as pd
import logging
import os
import json
from src.utils.main_utils import write_yaml_file
class DataValidation:
def __init__(self):
pass
async def init_config(self,data_validation_config:DataValidationConfig,data_ingestion_artifact:DataIngestionArtifact=None):
try:
logging.info("Initialising init_config in DataValidatin")
self.data_validation_config=data_validation_config
self.data_ingestion_artifact=data_ingestion_artifact
self._schema_config=await read_yaml_file(SCHEMA_FILE_PATH)
except Exception as e:
raise MyException(e,sys)
async def validate_number_of_columns(self,dataframe:pd.DataFrame)->bool:
try:
status=len(dataframe.columns)==len(self._schema_config['columns'])
logging.info(f"Is required colummn present: [{status}]")
return status
except Exception as e:
raise MyException(e,sys)
async def is_column_exists(self,dataframe:pd.DataFrame)->bool:
try:
missing_numerical_columns=[]
missing_categorical_columns=[]
to_check_col=dataframe.columns
for col in self._schema_config['numerical_columns']:
if col not in to_check_col:
missing_numerical_columns.append(col)
for col in self._schema_config['categorical_columns']:
if col not in to_check_col:
missing_categorical_columns.append(col)
if len(missing_categorical_columns)>0:
logging.info(f"Missing categorical columns: {missing_categorical_columns}")
if len(missing_numerical_columns)>0:
logging.info(f"Missing numerical columns: {missing_numerical_columns}")
return False if len(missing_numerical_columns) or len(missing_categorical_columns) else True
except Exception as e:
raise MyException(e,sys)
@staticmethod
async def read_data(file_path:str)->pd.DataFrame:
try:
return pd.read_csv(file_path)
except Exception as e:
raise MyException(e,sys)
async def initiate_data_validation(self,)->DataValidationArtifact:
try:
validation_error_msg=None
logging.info("Starting data validation")
train_df,test_df=(await DataValidation.read_data(file_path=self.data_ingestion_artifact.test_file_path),
await DataValidation.read_data(self.data_ingestion_artifact.trained_file_path))
# Train_df
logging.info("Checking validate_number_of_columns training columns")
status = await self.validate_number_of_columns(dataframe=train_df)
if not status:
validation_error_msg+="Columns are missing in training dataframe.",sys
logging.info(f"All required columns present in train dataframe: {status}")
logging.info("Checking is_column_exists")
status = await self.is_column_exists(dataframe=train_df)
if not status:
validation_error_msg+="Columns are missing in training dataframe.",sys
logging.info(f"All required columns present in train dataframe: {status}")
# Test_df
logging.info("Checking validate_number_of_columns testing columns")
status = await self.validate_number_of_columns(dataframe=test_df)
if not status:
validation_error_msg+="Columns are missing in testing dataframe.",sys
logging.info(f"All required columns present in test dataframe: {status}")
logging.info("Checking is_column_exists testing columns")
status = await self.is_column_exists(dataframe=train_df)
if not status:
validation_error_msg+="Columns are missing in testing dataframe.",sys
logging.info(f"All required columns present in test dataframe: {status}")
data_validation_artifact = DataValidationArtifact(
validation_status=validation_error_msg==None,
message=validation_error_msg,
validation_report_file_path=self.data_validation_config.validation_report_file_path
)
# Ensure the directory for validation_report_file_path exists
report_dir = os.path.dirname(self.data_validation_config.validation_report_file_path)
os.makedirs(report_dir, exist_ok=True)
await write_yaml_file(file_path=self.data_validation_config.validation_report_file_path,content=data_validation_artifact)
logging.info("Data validation artifact created and saved to JSON file.")
logging.info(f"Data validation artifact: {data_validation_artifact}")
return data_validation_artifact
except Exception as e:
raise MyException(e,sys)
|