Modules

This project is a Data Science application built with FastAPI, designed to facilitate model training, prediction, and data processing. The application uses Poetry for dependency management and follows modular design principles to ensure scalability and maintainability.

Modules:

This application is designed to help automate and streamline the workflow of training machine learning models and making predictions via an API interface, with easy-to-understand routes and clear separation of concerns between different modules.

This module centralizes all configuration settings for the project, ensuring consistent and maintainable configuration management. It provides access to environment variables, default settings, and other configurations required across different parts of the application.

Modules:

from config.env_config import ENVIRONMENT, DEBUG_MODE
from config.db_config import DATABASE_URL

This package defines global constants used throughout the project. Constants help in maintaining consistency and avoiding magic numbers or strings in the codebase.

from constants import APP_NAME, ENVIRONMENT
from constants import STATUS_OK, STATUS_BAD_REQUEST

This package contains modular and reusable components used across the project. Each component is designed to encapsulate specific functionality, promoting code reuse, scalability, and maintainability.

Modules:

from components.api_component import APIManager

This module provides centralized logging utilities for the data science pipeline. It standardizes logging practices, ensures consistency across components, and facilitates easy debugging and monitoring of the pipeline's execution, including data preprocessing, model training, evaluation, and predictions.

Functions:

from sample_ds_project.logging import logger

logger.info("Starting the model training process...")
logger.error("An error occurred during data validation.")

The utils module provides various utility functions for file I/O, data encoding/decoding, and directory management.

Functions:

41
42
43
44
45
46
47
48
49
50
51

@ensure_annotations
def create_directories(path_to_directories: list, verbose: bool = True) -> None:
    """create list of directories

    Args:
        path_to_directories (list): list of path of directories
    """
    for path in path_to_directories:
        os.makedirs(path, exist_ok=True)
        if verbose:
            logger.info(f"created directory at: {path}")

126
127
128
129
130
131
132
133
134
135
136
137
138

@ensure_annotations
def decode_image(imgstring: str, file_name: str) -> None:
    """
    Decodes a base64 string into an image and saves it at the given path

    Args:
        imgstring (str): base64 string of the image
        file_name (str): path at which to save the image
    """
    imgdata = base64.b64decode(imgstring)
    with open(file_name, "wb") as f:
        f.write(imgdata)
        f.close()

141
142
143
144
145
146
147
148
149
150
151
152
153

@ensure_annotations
def encode_image_into_base64(cropped_image_path: str) -> Any:
    """
    Encodes an image file into a base64 string.

    Args:
        cropped_image_path (str): Path to the image file to be encoded.

    Returns:
        bytes: Base64 encoded string of the image.
    """
    with open(cropped_image_path, "rb") as f:
        return base64.b64encode(f.read())

112
113
114
115
116
117
118
119
120
121
122
123

@ensure_annotations
def get_size(path: Path) -> str:
    """get size in KB

    Args:
        path (Path): path of the file

    Returns:
        str: size in KB
    """
    size_in_kb = round(os.path.getsize(path) / 1024)
    return f"~ {size_in_kb} KB"

 97
 98
 99
100
101
102
103
104
105
106
107
108
109

@ensure_annotations
def load_bin(path: Path) -> Any:
    """load binary data

    Args:
        path (Path): path to binary file

    Returns:
        Any: object stored in the file
    """
    data = joblib.load(path)
    logger.info(f"binary file loaded from: {path}")
    return data

68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

@ensure_annotations
def load_json(path: Path) -> ConfigBox:
    """load json files data

    Args:
        path (Path): path to json file

    Returns:
        ConfigBox: data as class attributes instead of dict
    """
    with open(path) as f:
        content = json.load(f)

    logger.info(f"json file loaded succesfully from: {path}")
    return ConfigBox(content)

18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

@ensure_annotations
def read_yaml(path_to_yaml: Path) -> ConfigBox:
    """reads yaml file and returns

    Args:
        path_to_yaml (str): path like input

    Raises:
        ValueError: if yaml file is empty
        e: empty file

    Returns:
        ConfigBox: ConfigBox type
    """
    try:
        with open(path_to_yaml) as yaml_file:
            content = yaml.safe_load(yaml_file)
            logger.info(f"yaml file: {path_to_yaml} loaded successfully")
            return ConfigBox(content)
    except BoxValueError as box_err:
        raise ValueError(YAML_EMPTY_ERROR) from box_err

85
86
87
88
89
90
91
92
93
94

@ensure_annotations
def save_bin(data: Any, path: Path) -> None:
    """save binary file

    Args:
        data (Any): data to be saved as binary
        path (Path): path to binary file
    """
    joblib.dump(value=data, filename=path)
    logger.info(f"binary file saved at: {path}")

54
55
56
57
58
59
60
61
62
63
64
65

@ensure_annotations
def save_json(path: Path, data: dict) -> None:
    """save json data

    Args:
        path (Path): path to json file
        data (dict): data to be saved in json file
    """
    with open(path, "w") as f:
        json.dump(data, f, indent=4)

    logger.info(f"json file saved at: {path}")

This module defines custom exception classes and error-handling utilities tailored to the needs of a data science pipeline. It helps standardize error reporting, improve debugging, and provide meaningful feedback during model training, data preprocessing, and prediction processes.

Classes:

from exception import DataValidationError, ModelTrainingError

try:
    validate_data(input_data)
except DataValidationError as e:
    logger.error(f"Data validation failed: {e}")
    raise

Examples:

• Data Validation: Raise a DataValidationError if the input data schema is incorrect or missing required fields.
• Model Training: Raise a ModelTrainingError if the model fails to converge due to invalid hyperparameters.
• Prediction: Raise a PredictionError when incompatible input data is passed to the model.

51
52
53
54
55
56
57
58
59

class CustomException(HTTPException):
    def __init__(self, status_code: int, detail: str):
        """
        Custom exception for handling API errors.

        :param status_code: The HTTP status code to return.
        :param detail: A string describing the error in detail.
        """
        super().__init__(status_code=status_code, detail=detail)

52
53
54
55
56
57
58
59

def __init__(self, status_code: int, detail: str):
    """
    Custom exception for handling API errors.

    :param status_code: The HTTP status code to return.
    :param detail: A string describing the error in detail.
    """
    super().__init__(status_code=status_code, detail=detail)

This module defines the core entities and data structures used throughout the data science pipeline. Entities are designed to represent the inputs, outputs, and intermediate states of the model training and prediction processes, ensuring consistency and validation across the project.

Modules:

from entity.data_schema import InputSchema
from entity.model_params import ModelConfig
from entity.prediction_result import PredictionOutput

Examples:

• Data Schema: Define the expected input structure for data preprocessing.
• Model Parameters: Store configurations like learning rate, batch size, and optimizer type.
• Prediction Results: Represent the model's outputs in a structured format, including predicted classes, probabilities, and confidence scores.

The pipeline module orchestrates the end-to-end flow of the data science process, from raw data ingestion to final predictions. It consists of multiple submodules, each responsible for a specific stage in the pipeline. This modular structure ensures scalability, reusability, and ease of maintenance. The pipeline is designed to handle data preprocessing, model training, evaluation, and predictions in a systematic and automated manner.

Modules:

from pipeline.stage_01_data_ingestion import DataIngestionTrainingPipeline
from pipeline.stage_04_model_trainer import ModelTrainingPipeline

# Perform data ingestion
data_ingestion = DataIngestionTrainingPipeline(config)
raw_data = data_ingestion.run()

# Train the model
model_trainer = ModelTrainingPipeline(config, raw_data)
trained_model = model_trainer.run()

The routes module defines the API routes that enable interaction with the machine learning model. This module contains endpoints for initiating model training and making predictions on new data.