File size: 4,165 Bytes
a18e884
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
import os
import numpy as np
from pathlib import Path
import torch
from torch.utils.data import DataLoader, random_split
from torchvision import datasets, transforms
import tensorflow as tf



IMAGE_SIZE = (150, 150)   
BATCH_SIZE = 32
VAL_SPLIT = 0.2        
SEED = 42

CLASS_NAMES = ["buildings", "forest", "glacier", "mountain", "sea", "street"]



#  PyTorch  Data Pipeline
def get_pytorch_loaders(train_dir: str, test_dir: str):
    mean = [0.485, 0.456, 0.406]
    std = [0.229, 0.224, 0.225]

    
    train_transform = transforms.Compose([
        transforms.Resize(IMAGE_SIZE),
        transforms.RandomHorizontalFlip(p=0.5),          
        transforms.RandomRotation(degrees=15),          
        transforms.RandomResizedCrop(IMAGE_SIZE,scale=(0.8, 1.0)),
        transforms.ColorJitter(brightness=0.2,contrast=0.2),
        transforms.ToTensor(),                          
        transforms.Normalize(mean, std),   
    ])

    eval_transform = transforms.Compose([
        transforms.Resize(IMAGE_SIZE),
        transforms.ToTensor(),
        transforms.Normalize(mean, std),
    ])

    full_train = datasets.ImageFolder(root=train_dir,transform=train_transform)

    # Split into train / validation
    n_val = int(len(full_train) * VAL_SPLIT)
    n_train = len(full_train) - n_val
    train_ds, val_ds = random_split(
        full_train, [n_train, n_val],
        generator=torch.Generator().manual_seed(SEED)
    )

    
    val_ds.dataset = datasets.ImageFolder(root=train_dir,transform=eval_transform)
    test_ds = datasets.ImageFolder(root=test_dir, transform=eval_transform)

    
    train_loader = DataLoader(train_ds, batch_size=BATCH_SIZE, shuffle=True, num_workers=2, pin_memory=True)
    val_loader = DataLoader(val_ds, batch_size=BATCH_SIZE, shuffle=False, num_workers=2, pin_memory=True)
    test_loader  = DataLoader(test_ds, batch_size=BATCH_SIZE, shuffle=False, num_workers=2, pin_memory=True)

    print(f"[PyTorch] Train: {n_train} | Val: {n_val} | Test: {len(test_ds)}")
    return train_loader, val_loader, test_loader, full_train.classes





#  TensorFlow  Data Pipeline
def get_tensorflow_datasets(train_dir: str, test_dir: str):
    img_h, img_w = IMAGE_SIZE

    raw_train = tf.keras.utils.image_dataset_from_directory(
        train_dir,
        validation_split=VAL_SPLIT,
        subset="training",
        seed=SEED,
        image_size=IMAGE_SIZE,
        batch_size=BATCH_SIZE,
        label_mode="categorical", 
    )

    raw_val = tf.keras.utils.image_dataset_from_directory(
        train_dir,
        validation_split=VAL_SPLIT,
        subset="validation",
        seed=SEED,
        image_size=IMAGE_SIZE,
        batch_size=BATCH_SIZE,
        label_mode="categorical",
    )

    raw_test = tf.keras.utils.image_dataset_from_directory(
        test_dir,
        image_size=IMAGE_SIZE,
        batch_size=BATCH_SIZE,
        label_mode="categorical",
        shuffle=False,
    )
    

    class_names = raw_train.class_names

    
    normalization = tf.keras.layers.Rescaling(1.0 / 255)
    augmentation = tf.keras.Sequential([
        tf.keras.layers.RandomFlip("horizontal"),
        tf.keras.layers.RandomRotation(0.1),       
        tf.keras.layers.RandomZoom(0.2),           
        tf.keras.layers.RandomContrast(0.1),
    ])

    def preprocess_train(images, labels):
        images = normalization(images)
        images = augmentation(images, training=True)
        return images, labels

    def preprocess_eval(images, labels):
        images = normalization(images)
        return images, labels

    AUTOTUNE = tf.data.AUTOTUNE

    train_ds = (raw_train
                .map(preprocess_train, num_parallel_calls=AUTOTUNE)
                .cache()
                .shuffle(1000)
                .prefetch(AUTOTUNE))

    val_ds = (raw_val
              .map(preprocess_eval, num_parallel_calls=AUTOTUNE)
              .cache()
              .prefetch(AUTOTUNE))

    test_ds = (raw_test
               .map(preprocess_eval, num_parallel_calls=AUTOTUNE)
               .prefetch(AUTOTUNE))

    print(f"[TensorFlow] Classes: {class_names}")
    return train_ds, val_ds, test_ds, class_names