|
| 1 | +from singa import singa_wrap as singa |
| 2 | +from singa import device |
| 3 | +from singa import tensor |
| 4 | +from singa import opt |
| 5 | +import numpy as np |
| 6 | +import time |
| 7 | +import argparse |
| 8 | +import sys |
| 9 | +sys.path.append("../../..") |
| 10 | + |
| 11 | +from PIL import Image |
| 12 | + |
| 13 | +from healthcare.data import diaret |
| 14 | +from healthcare.models import diabetic_retinopthy_net |
| 15 | + |
| 16 | +np_dtype = {"float16": np.float16, "float32": np.float32} |
| 17 | + |
| 18 | +singa_dtype = {"float16": tensor.float16, "float32": tensor.float32} |
| 19 | + |
| 20 | + |
| 21 | +# Data augmentation |
| 22 | +def augmentation(x, batch_size): |
| 23 | + xpad = np.pad(x, [[0, 0], [0, 0], [4, 4], [4, 4]], 'symmetric') |
| 24 | + for data_num in range(0, batch_size): |
| 25 | + offset = np.random.randint(8, size=2) |
| 26 | + x[data_num, :, :, :] = xpad[data_num, :, |
| 27 | + offset[0]:offset[0] + x.shape[2], |
| 28 | + offset[1]:offset[1] + x.shape[2]] |
| 29 | + if_flip = np.random.randint(2) |
| 30 | + if (if_flip): |
| 31 | + x[data_num, :, :, :] = x[data_num, :, :, ::-1] |
| 32 | + return x |
| 33 | + |
| 34 | + |
| 35 | +# Calculate accuracy |
| 36 | +def accuracy(pred, target): |
| 37 | + # y is network output to be compared with ground truth (int) |
| 38 | + y = np.argmax(pred, axis=1) |
| 39 | + a = y == target |
| 40 | + correct = np.array(a, "int").sum() |
| 41 | + return correct |
| 42 | + |
| 43 | + |
| 44 | +# Data partition according to the rank |
| 45 | +def partition(global_rank, world_size, train_x, train_y, val_x, val_y): |
| 46 | + # Partition training data |
| 47 | + data_per_rank = train_x.shape[0] // world_size |
| 48 | + idx_start = global_rank * data_per_rank |
| 49 | + idx_end = (global_rank + 1) * data_per_rank |
| 50 | + train_x = train_x[idx_start:idx_end] |
| 51 | + train_y = train_y[idx_start:idx_end] |
| 52 | + |
| 53 | + # Partition evaluation data |
| 54 | + data_per_rank = val_x.shape[0] // world_size |
| 55 | + idx_start = global_rank * data_per_rank |
| 56 | + idx_end = (global_rank + 1) * data_per_rank |
| 57 | + val_x = val_x[idx_start:idx_end] |
| 58 | + val_y = val_y[idx_start:idx_end] |
| 59 | + return train_x, train_y, val_x, val_y |
| 60 | + |
| 61 | + |
| 62 | +# Function to all reduce NUMPY accuracy and loss from multiple devices |
| 63 | +def reduce_variable(variable, dist_opt, reducer): |
| 64 | + reducer.copy_from_numpy(variable) |
| 65 | + dist_opt.all_reduce(reducer.data) |
| 66 | + dist_opt.wait() |
| 67 | + output = tensor.to_numpy(reducer) |
| 68 | + return output |
| 69 | + |
| 70 | + |
| 71 | +def resize_dataset(x, image_size): |
| 72 | + num_data = x.shape[0] |
| 73 | + dim = x.shape[1] |
| 74 | + X = np.zeros(shape=(num_data, dim, image_size, image_size), |
| 75 | + dtype=np.float32) |
| 76 | + for n in range(0, num_data): |
| 77 | + for d in range(0, dim): |
| 78 | + X[n, d, :, :] = np.array(Image.fromarray(x[n, d, :, :]).resize( |
| 79 | + (image_size, image_size), Image.BILINEAR), |
| 80 | + dtype=np.float32) |
| 81 | + return X |
| 82 | + |
| 83 | + |
| 84 | +def run(global_rank, |
| 85 | + world_size, |
| 86 | + dir_path, |
| 87 | + max_epoch, |
| 88 | + batch_size, |
| 89 | + model, |
| 90 | + data, |
| 91 | + sgd, |
| 92 | + graph, |
| 93 | + verbosity, |
| 94 | + dist_option='plain', |
| 95 | + spars=None, |
| 96 | + precision='float32'): |
| 97 | + # now CPU version only, could change to GPU device for GPU-support machines |
| 98 | + dev = device.get_default_device() |
| 99 | + dev.SetRandSeed(0) |
| 100 | + np.random.seed(0) |
| 101 | + if data == 'diaret': |
| 102 | + train_x, train_y, val_x, val_y = diaret.load(dir_path=dir_path) |
| 103 | + else: |
| 104 | + print( |
| 105 | + 'Wrong dataset!' |
| 106 | + ) |
| 107 | + sys.exit(0) |
| 108 | + |
| 109 | + num_channels = train_x.shape[1] |
| 110 | + image_size = train_x.shape[2] |
| 111 | + data_size = np.prod(train_x.shape[1:train_x.ndim]).item() |
| 112 | + num_classes = (np.max(train_y) + 1).item() |
| 113 | + |
| 114 | + if model == 'cnn': |
| 115 | + model = diabetic_retinopthy_net.create_model(num_channels=num_channels, |
| 116 | + num_classes=num_classes) |
| 117 | + else: |
| 118 | + print( |
| 119 | + 'Wrong model!' |
| 120 | + ) |
| 121 | + sys.exit(0) |
| 122 | + |
| 123 | + # For distributed training, sequential has better performance |
| 124 | + if hasattr(sgd, "communicator"): |
| 125 | + DIST = True |
| 126 | + sequential = True |
| 127 | + else: |
| 128 | + DIST = False |
| 129 | + sequential = False |
| 130 | + |
| 131 | + if DIST: |
| 132 | + train_x, train_y, val_x, val_y = partition(global_rank, world_size, |
| 133 | + train_x, train_y, val_x, |
| 134 | + val_y) |
| 135 | + |
| 136 | + if model.dimension == 4: |
| 137 | + tx = tensor.Tensor( |
| 138 | + (batch_size, num_channels, model.input_size, model.input_size), dev, |
| 139 | + singa_dtype[precision]) |
| 140 | + elif model.dimension == 2: |
| 141 | + tx = tensor.Tensor((batch_size, data_size), |
| 142 | + dev, singa_dtype[precision]) |
| 143 | + np.reshape(train_x, (train_x.shape[0], -1)) |
| 144 | + np.reshape(val_x, (val_x.shape[0], -1)) |
| 145 | + |
| 146 | + ty = tensor.Tensor((batch_size,), dev, tensor.int32) |
| 147 | + num_train_batch = train_x.shape[0] // batch_size |
| 148 | + num_val_batch = val_x.shape[0] // batch_size |
| 149 | + idx = np.arange(train_x.shape[0], dtype=np.int32) |
| 150 | + |
| 151 | + # Attach model to graph |
| 152 | + model.set_optimizer(sgd) |
| 153 | + model.compile([tx], is_train=True, use_graph=graph, sequential=sequential) |
| 154 | + dev.SetVerbosity(verbosity) |
| 155 | + |
| 156 | + # Training and evaluation loop |
| 157 | + for epoch in range(max_epoch): |
| 158 | + start_time = time.time() |
| 159 | + np.random.shuffle(idx) |
| 160 | + |
| 161 | + if global_rank == 0: |
| 162 | + print('Starting Epoch %d:' % (epoch)) |
| 163 | + |
| 164 | + # Training phase |
| 165 | + train_correct = np.zeros(shape=[1], dtype=np.float32) |
| 166 | + test_correct = np.zeros(shape=[1], dtype=np.float32) |
| 167 | + train_loss = np.zeros(shape=[1], dtype=np.float32) |
| 168 | + |
| 169 | + model.train() |
| 170 | + for b in range(num_train_batch): |
| 171 | + # if b % 100 == 0: |
| 172 | + # print ("b: \n", b) |
| 173 | + # Generate the patch data in this iteration |
| 174 | + x = train_x[idx[b * batch_size:(b + 1) * batch_size]] |
| 175 | + if model.dimension == 4: |
| 176 | + x = augmentation(x, batch_size) |
| 177 | + if (image_size != model.input_size): |
| 178 | + x = resize_dataset(x, model.input_size) |
| 179 | + x = x.astype(np_dtype[precision]) |
| 180 | + y = train_y[idx[b * batch_size:(b + 1) * batch_size]] |
| 181 | + |
| 182 | + # Copy the patch data into input tensors |
| 183 | + tx.copy_from_numpy(x) |
| 184 | + ty.copy_from_numpy(y) |
| 185 | + |
| 186 | + # Train the model |
| 187 | + out, loss = model(tx, ty, dist_option, spars) |
| 188 | + train_correct += accuracy(tensor.to_numpy(out), y) |
| 189 | + train_loss += tensor.to_numpy(loss)[0] |
| 190 | + |
| 191 | + if DIST: |
| 192 | + # Reduce the evaluation accuracy and loss from multiple devices |
| 193 | + reducer = tensor.Tensor((1,), dev, tensor.float32) |
| 194 | + train_correct = reduce_variable(train_correct, sgd, reducer) |
| 195 | + train_loss = reduce_variable(train_loss, sgd, reducer) |
| 196 | + |
| 197 | + if global_rank == 0: |
| 198 | + print('Training loss = %f, training accuracy = %f' % |
| 199 | + (train_loss, train_correct / |
| 200 | + (num_train_batch * batch_size * world_size)), |
| 201 | + flush=True) |
| 202 | + |
| 203 | + # Evaluation phase |
| 204 | + model.eval() |
| 205 | + for b in range(num_val_batch): |
| 206 | + x = val_x[b * batch_size:(b + 1) * batch_size] |
| 207 | + if model.dimension == 4: |
| 208 | + if (image_size != model.input_size): |
| 209 | + x = resize_dataset(x, model.input_size) |
| 210 | + x = x.astype(np_dtype[precision]) |
| 211 | + y = val_y[b * batch_size:(b + 1) * batch_size] |
| 212 | + tx.copy_from_numpy(x) |
| 213 | + ty.copy_from_numpy(y) |
| 214 | + out_test = model(tx) |
| 215 | + test_correct += accuracy(tensor.to_numpy(out_test), y) |
| 216 | + |
| 217 | + if DIST: |
| 218 | + # Reduce the evaulation accuracy from multiple devices |
| 219 | + test_correct = reduce_variable(test_correct, sgd, reducer) |
| 220 | + |
| 221 | + # Output the evaluation accuracy |
| 222 | + if global_rank == 0: |
| 223 | + print('Evaluation accuracy = %f, Elapsed Time = %fs' % |
| 224 | + (test_correct / (num_val_batch * batch_size * world_size), |
| 225 | + time.time() - start_time), |
| 226 | + flush=True) |
| 227 | + |
| 228 | + dev.PrintTimeProfiling() |
| 229 | + |
| 230 | + |
| 231 | +if __name__ == '__main__': |
| 232 | + # Use argparse to get command config: max_epoch, model, data, etc., for single gpu training |
| 233 | + parser = argparse.ArgumentParser( |
| 234 | + description='Training using the autograd and graph.') |
| 235 | + parser.add_argument( |
| 236 | + 'model', |
| 237 | + choices=['cnn'], |
| 238 | + default='cnn') |
| 239 | + parser.add_argument('data', |
| 240 | + choices=['diaret'], |
| 241 | + default='diaret') |
| 242 | + parser.add_argument('-p', |
| 243 | + choices=['float32', 'float16'], |
| 244 | + default='float32', |
| 245 | + dest='precision') |
| 246 | + parser.add_argument('-dir', |
| 247 | + '--dir-path', |
| 248 | + default="/tmp/diaret", |
| 249 | + type=str, |
| 250 | + help='the directory to store the Diabetic Retinopathy dataset', |
| 251 | + dest='dir_path') |
| 252 | + parser.add_argument('-m', |
| 253 | + '--max-epoch', |
| 254 | + default=300, |
| 255 | + type=int, |
| 256 | + help='maximum epochs', |
| 257 | + dest='max_epoch') |
| 258 | + parser.add_argument('-b', |
| 259 | + '--batch-size', |
| 260 | + default=64, |
| 261 | + type=int, |
| 262 | + help='batch size', |
| 263 | + dest='batch_size') |
| 264 | + parser.add_argument('-l', |
| 265 | + '--learning-rate', |
| 266 | + default=0.005, |
| 267 | + type=float, |
| 268 | + help='initial learning rate', |
| 269 | + dest='lr') |
| 270 | + parser.add_argument('-g', |
| 271 | + '--disable-graph', |
| 272 | + default='True', |
| 273 | + action='store_false', |
| 274 | + help='disable graph', |
| 275 | + dest='graph') |
| 276 | + parser.add_argument('-v', |
| 277 | + '--log-verbosity', |
| 278 | + default=0, |
| 279 | + type=int, |
| 280 | + help='logging verbosity', |
| 281 | + dest='verbosity') |
| 282 | + |
| 283 | + args = parser.parse_args() |
| 284 | + |
| 285 | + sgd = opt.SGD(lr=args.lr, momentum=0.9, weight_decay=1e-5, |
| 286 | + dtype=singa_dtype[args.precision]) |
| 287 | + run(0, |
| 288 | + 1, |
| 289 | + args.dir_path, |
| 290 | + args.max_epoch, |
| 291 | + args.batch_size, |
| 292 | + args.model, |
| 293 | + args.data, |
| 294 | + sgd, |
| 295 | + args.graph, |
| 296 | + args.verbosity, |
| 297 | + precision=args.precision) |
0 commit comments