src/train.py

import os
import torch
import torch.nn as nn
from datetime import datetime

from src.config import (
    CatDogDatasetConfigsInput,
    CatDogClassifierConfigs,
    DataPreprocessorConfigsInput
)
from src.data_ingestion import CatDogDataset
from src.data_preprocessing import DataPreprocessor
from src.model import CatDogClassifier


def train_pipeline():
    # Setup Intialize configurations
    # Data ingestion
    data_path = "datasets/datasets.zip"
    train_data_path = "datasets/train"
    test_data_path = "datasets/test"

    data_path = os.path.join(os.getcwd(), data_path)
    train_data_path = os.path.join(os.getcwd(), train_data_path)
    test_data_path = os.path.join(os.getcwd(), test_data_path)


    # data_ingestion_configs = CatDogDatasetConfigsInput(
    #     data_path=data_path,
    #     train_data_path=train_data_path,
    #     test_data_path=test_data_path,
    #     test_size=0.2,
    #     random_state=42
    # )
    # print(data_ingestion_configs)
    # dataset = CatDogDataset(data_ingestion_configs)
    # dataset.load_data()
    
    # Data preprocessing
    data_preprocessing_configs = DataPreprocessorConfigsInput(
        train_dataset_path=train_data_path,
        test_dataset_path=test_data_path,
        shuffle=True,
        batch_size=32,
        horizontal_flip_prob=0.5,
        image_size=224,
        mean=[0.485, 0.456, 0.406],
        std=[0.229, 0.224, 0.225]
    )
    preprocessor = DataPreprocessor(data_preprocessing_configs)
    train_dataloader, test_dataloader = preprocessor.create_dataloader()
    device = "cuda" if torch.cuda.is_available() else "cpu"
    print(f"Using device: {device}")

    # Model training
    model_configs = CatDogClassifierConfigs(
        device=device,
        input_channels=3,
        num_classes=2,
        learning_rate=0.001,
        kernel_size=3,
        stride=2,
        padding=1,
        num_layers=3,
        use_amp=True
    )

    model = CatDogClassifier(model_configs)
    model.to(device)
    # Set random seeds
    torch.manual_seed(42) 
    torch.cuda.manual_seed(42)
    # Setup loss function and optimizer
    loss_fn = nn.CrossEntropyLoss()
    optimizer = torch.optim.Adam(params=model.parameters(), lr=0.001)

    # # Calculate training time using timeit
    # Start the timer
    from timeit import default_timer as timer 
    start_time = timer()

    model.train_process(
        model=model,
        train_dataloader=train_dataloader,
        test_dataloader=test_dataloader,
        num_epochs=20,
        loss_fn=loss_fn,
        optimizer=optimizer
    )
    end_time = timer()
    print(f"Training completed in {end_time - start_time} seconds.")

    # Save the trained model
    torch.save(model.state_dict(), f"cat_dog_classifier_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pth")
    print("Model saved to cat_dog_classifier.pth")


if __name__ == "__main__":
    train_pipeline()