My data preprocess pipeline is: construct training set and test set of CIFAR10; add different perturbations to training set and test set; apply image transformations, thus I can’t do image transformations by specifying transform argument in torchvision.datasets.cifar.CIFAR10. So I custom the CIFAR10 dataset.
cifar.py
import pickle
import os
import numpy as np
from PIL import Image
from torch.utils.data.dataset import Dataset
class CIFAR10(Dataset):
def __init__(self, data_dir, transform=None, evaluation=False):
self.classes = 10
self.transform = transform
if not evaluation:
data_list = [
"data_batch_1", "data_batch_2", "data_batch_3", "data_batch_4",
"data_batch_5"
]
else:
data_list = ["test_batch"]
data = []
targets = []
for file_name in data_list:
file_path = os.path.join(data_dir, file_name)
with open(file_path, "rb") as f:
entry = pickle.load(f, encoding="latin1")
data.append(entry["data"])
targets.extend(entry["labels"])
data = np.vstack(data).reshape(-1, 3, 32, 32)
# convert to NHWC (np.ndarray) works better with transformations.
self.data = data.transpose((0, 2, 3, 1))
self.targets = np.asarray(targets)
def __getitem__(self, index):
img, target = self.data[index], self.targets[index]
img = Image.fromarray(img)
if self.transform is not None:
img = self.transform(img)
return img, target
def __len__(self):
return len(self.data)
And I write a function to apply image transformations to a pre-constructed dataset.
def dst_transforms(dataset, transform):
data_loader = DataLoader(dataset, batch_size=len(dataset), num_workers=3)
data, targets = list(data_loader)[0] # data_loader only has one batch
transformed_data = torch.zeros(data.shape).permute(0, 3, 1, 2)
for i, d in enumerate(data):
transformed_data[i] = transform(d.numpy()) # type(d) is torch.Tensor
transformed_dst = TensorDataset(transformed_data, targets)
return transformed_dst
However, when I preprocess data with the same transformation, I got lower accuracy with dst_transforms than specifying transform argument. What’s wrong in my dst_transformation? Any help is appreciated!
Following with an example code.
main.py
import numpy as np
import setGPU
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
import torchvision
import torchvision.transforms as transforms
from torch.utils.data import DataLoader, TensorDataset
import cifar
def dst_transforms(dataset, transform):
data_loader = DataLoader(dataset, batch_size=len(dataset), num_workers=3)
data, targets = list(data_loader)[0] # data_loader only has one batch
transformed_data = torch.zeros(data.shape).permute(0, 3, 1, 2)
for i, d in enumerate(data):
transformed_data[i] = transform(d.numpy()) # type(d) is torch.Tensor
transformed_dst = TensorDataset(transformed_data, targets)
return transformed_dst
def train(model, train_loader, epoch):
device = torch.device("cuda")
model = model.to(device)
model.train()
criterion = nn.CrossEntropyLoss()
optim = torch.optim.Adam(model.parameters(), lr=0.001)
train_loss = 0
correct = 0
for batch_idx, (data, target) in enumerate(train_loader):
data, target = data.to(device), target.to(device)
optim.zero_grad()
output = model(data)
loss = F.cross_entropy(output, target)
loss.backward()
optim.step()
train_loss += loss.item()
pred = output.argmax(dim=1, keepdim=True)
correct += pred.eq(target.view_as(pred)).sum().item()
train_loss /= len(train_loader.dataset)
acc = 100. * correct / len(train_loader.dataset)
return train_loss, acc
def test(model, test_loader, epoch):
device = torch.device("cuda")
model.eval()
test_loss = 0
correct = 0
with torch.no_grad():
for batch_idx, (data, target) in enumerate(test_loader):
data, target = data.to(device), target.to(device)
output = model(data)
loss = F.cross_entropy(output, target)
test_loss += loss.item()
pred = output.argmax(dim=1, keepdim=True)
correct += pred.eq(target.view_as(pred)).sum().item()
test_loss /= len(test_loader.dataset)
acc = 100. * correct / len(test_loader.dataset)
return test_loss, acc
def main():
print('==> Preparing data..')
train_transform = transforms.Compose([
# cifar dataset consists of np.ndarray (NHWC).
#transforms.ToPILImage(),
#transforms.RandomCrop(32, padding=4),
transforms.RandomHorizontalFlip(),
transforms.ToTensor(), # NHWC => NCHW
transforms.Normalize((0.4914, 0.4822, 0.4465),
(0.2023, 0.1994, 0.2010))
])
test_transform = transforms.Compose([
transforms.ToTensor(), # NHWC => NCHW
transforms.Normalize((0.4914, 0.4822, 0.4465),
(0.2023, 0.1994, 0.2010))
])
train_data = cifar.CIFAR10("./cifar10/cifar-10-batches-py",
transform=train_transform)
test_data = cifar.CIFAR10("./cifar10/cifar-10-batches-py",
transform=train_transform,
evaluation=True)
#train_data = dst_transforms(train_data, train_transform)
train_loader = DataLoader(train_data,
batch_size=128,
num_workers=3,
shuffle=True)
#test_data = dst_transforms(test_data, test_transform)
test_loader = DataLoader(test_data,
batch_size=100,
num_workers=3,
shuffle=False)
model = models.resnet18()
for epoch in range(200):
train_loss, train_acc = train(model, train_loader, epoch)
test_loss, test_acc = test(model, test_loader, epoch)
print(
"train loss: {}, train accuracy: {}, test loss: {}, test accuracy: {}\n"
.format(train_loss, train_acc, test_loss, test_acc))
if __name__ == "__main__":
main()