-
Notifications
You must be signed in to change notification settings - Fork 0
/
utils.py
139 lines (101 loc) · 4.89 KB
/
utils.py
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
import os
import scipy
import numpy as np
import tensorflow as tf
from config import cfg
import data.smallNORB as norb
from keras.datasets import cifar10, cifar100
from keras import backend as K
import logging
import daiquiri
import warnings
warnings.filterwarnings("ignore")
daiquiri.setup(level=logging.DEBUG)
logger = daiquiri.getLogger(__name__)
def create_inputs_norb(is_train: bool, epochs: int):
import re
if is_train:
CHUNK_RE = re.compile(r"train\d+\.tfrecords")
else:
CHUNK_RE = re.compile(r"test\d+\.tfrecords")
processed_dir = './data'
chunk_files = [os.path.join(processed_dir, fname)
for fname in os.listdir(processed_dir)
if CHUNK_RE.match(fname)]
image, label = norb.read_norb_tfrecord(chunk_files, epochs)
if is_train:
# TODO: is it the right order: add noise, resize, then corp?
image = tf.image.random_brightness(image, max_delta=32. / 255.)
image = tf.image.random_contrast(image, lower=0.5, upper=1.5)
image = tf.image.resize_images(image, [48, 48])
image = tf.random_crop(image, [32, 32, 1])
else:
image = tf.image.resize_images(image, [48, 48])
image = tf.slice(image, [8, 8, 0], [32, 32, 1])
x, y = tf.train.shuffle_batch([image, label], num_threads=cfg.num_threads, batch_size=cfg.batch_size, capacity=cfg.batch_size * 64,
min_after_dequeue=cfg.batch_size * 32, allow_smaller_final_batch=False)
return x, y
def create_inputs_mnist(is_train):
tr_x, tr_y = load_mnist(cfg.dataset, is_train)
data_queue = tf.train.slice_input_producer([tr_x, tr_y], capacity=64 * 8)
x, y = tf.train.shuffle_batch(data_queue, num_threads=8, batch_size=cfg.batch_size, capacity=cfg.batch_size * 64,
min_after_dequeue=cfg.batch_size * 32, allow_smaller_final_batch=False)
return (x, y)
def create_inputs_fashion_mnist(is_train):
tr_x, tr_y = load_mnist(cfg.dataset_fashion_mnist, is_train)
data_queue = tf.train.slice_input_producer([tr_x, tr_y], capacity=64 * 8)
x, y = tf.train.shuffle_batch(data_queue, num_threads=8, batch_size=cfg.batch_size, capacity=cfg.batch_size * 64,
min_after_dequeue=cfg.batch_size * 32, allow_smaller_final_batch=False)
return (x, y)
def load_mnist(path, is_training):
fd = open(os.path.join(cfg.dataset, 'train-images-idx3-ubyte'))
loaded = np.fromfile(file=fd, dtype=np.uint8)
trX = loaded[16:].reshape((60000, 28, 28, 1)).astype(np.float32)
fd = open(os.path.join(cfg.dataset, 'train-labels-idx1-ubyte'))
loaded = np.fromfile(file=fd, dtype=np.uint8)
trY = loaded[8:].reshape((60000)).astype(np.int32)
fd = open(os.path.join(cfg.dataset, 't10k-images-idx3-ubyte'))
loaded = np.fromfile(file=fd, dtype=np.uint8)
teX = loaded[16:].reshape((10000, 28, 28, 1)).astype(np.float32)
fd = open(os.path.join(cfg.dataset, 't10k-labels-idx1-ubyte'))
loaded = np.fromfile(file=fd, dtype=np.uint8)
teY = loaded[8:].reshape((10000)).astype(np.int32)
# normalization and convert to a tensor [60000, 28, 28, 1]
# trX = tf.convert_to_tensor(trX, tf.float32)
# teX = tf.convert_to_tensor(teX, tf.float32)
# => [num_samples, 10]
# trY = tf.one_hot(trY, depth=10, axis=1, dtype=tf.float32)
# teY = tf.one_hot(teY, depth=10, axis=1, dtype=tf.float32)
if is_training:
return trX, trY
else:
return teX, teY
def create_inputs_cifar10(is_train):
tr_x, tr_y = load_cifar10(is_train)
data_queue = tf.train.slice_input_producer([tr_x, tr_y], capacity=64 * 8)
x, y = tf.train.shuffle_batch(data_queue, num_threads=8, batch_size=cfg.batch_size, capacity=cfg.batch_size * 64,
min_after_dequeue=cfg.batch_size * 32, allow_smaller_final_batch=False)
return (x, y)
def load_cifar10(is_training):
# https://keras.io/datasets/
assert(K.image_data_format() == 'channels_last')
if is_training:
return cifar10.load_data()[0]
else:
return cifar10.load_data()[1]
def create_inputs_cifar100(is_train):
tr_x, tr_y = load_cifar100(is_train)
data_queue = tf.train.slice_input_producer([tr_x, tr_y], capacity=64 * 8)
x, y = tf.train.shuffle_batch(data_queue, num_threads=8, batch_size=cfg.batch_size, capacity=cfg.batch_size * 64,
min_after_dequeue=cfg.batch_size * 32, allow_smaller_final_batch=False)
return (x, y)
def load_cifar100(is_training):
# https://keras.io/datasets/
# https://www.cs.toronto.edu/~kriz/cifar.html:
# "Each image comes with a 'fine' label (the class to which it belongs)
# and a 'coarse' label (the superclass to which it belongs)."
assert(K.image_data_format() == 'channels_last')
if is_training:
return cifar100.load_data(label_mode='fine')[0]
else:
return cifar100.load_data(label_mode='fine')[1]