|
| 1 | +# |
| 2 | +# Licensed to the Apache Software Foundation (ASF) under one |
| 3 | +# or more contributor license agreements. See the NOTICE file |
| 4 | +# distributed with this work for additional information |
| 5 | +# regarding copyright ownership. The ASF licenses this file |
| 6 | +# to you under the Apache License, Version 2.0 (the |
| 7 | +# "License"); you may not use this file except in compliance |
| 8 | +# with the License. You may obtain a copy of the License at |
| 9 | +# |
| 10 | +# http://www.apache.org/licenses/LICENSE-2.0 |
| 11 | +# |
| 12 | +# Unless required by applicable law or agreed to in writing, |
| 13 | +# software distributed under the License is distributed on an |
| 14 | +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| 15 | +# KIND, either express or implied. See the License for the |
| 16 | +# specific language governing permissions and limitations |
| 17 | +# under the License. |
| 18 | +# |
| 19 | + |
| 20 | +try: |
| 21 | + import pickle |
| 22 | +except ImportError: |
| 23 | + import cPickle as pickle |
| 24 | + |
| 25 | +from singa import singa_wrap as singa |
| 26 | +from singa import autograd |
| 27 | +from singa import tensor |
| 28 | +from singa import device |
| 29 | +from singa import opt |
| 30 | +from PIL import Image |
| 31 | +import numpy as np |
| 32 | +import os |
| 33 | +import sys |
| 34 | +import time |
| 35 | + |
| 36 | + |
| 37 | +def load_dataset(filepath): |
| 38 | + with open(filepath, 'rb') as fd: |
| 39 | + try: |
| 40 | + cifar10 = pickle.load(fd, encoding='latin1') |
| 41 | + except TypeError: |
| 42 | + cifar10 = pickle.load(fd) |
| 43 | + image = cifar10['data'].astype(dtype=np.uint8) |
| 44 | + image = image.reshape((-1, 3, 32, 32)) |
| 45 | + label = np.asarray(cifar10['labels'], dtype=np.uint8) |
| 46 | + label = label.reshape(label.size, 1) |
| 47 | + return image, label |
| 48 | + |
| 49 | + |
| 50 | +def load_train_data(dir_path='cifar-10-batches-py', num_batches=5): |
| 51 | + labels = [] |
| 52 | + batchsize = 10000 |
| 53 | + images = np.empty((num_batches * batchsize, 3, 32, 32), dtype=np.uint8) |
| 54 | + for did in range(1, num_batches + 1): |
| 55 | + fname_train_data = dir_path + "/data_batch_{}".format(did) |
| 56 | + image, label = load_dataset(check_dataset_exist(fname_train_data)) |
| 57 | + images[(did - 1) * batchsize:did * batchsize] = image |
| 58 | + labels.extend(label) |
| 59 | + images = np.array(images, dtype=np.float32) |
| 60 | + labels = np.array(labels, dtype=np.int32) |
| 61 | + return images, labels |
| 62 | + |
| 63 | + |
| 64 | +def load_test_data(dir_path='cifar-10-batches-py'): |
| 65 | + images, labels = load_dataset(check_dataset_exist(dir_path + "/test_batch")) |
| 66 | + return np.array(images, dtype=np.float32), np.array(labels, dtype=np.int32) |
| 67 | + |
| 68 | + |
| 69 | +def check_dataset_exist(dirpath): |
| 70 | + if not os.path.exists(dirpath): |
| 71 | + print( |
| 72 | + 'Please download the cifar10 dataset using download_data.py (e.g. python ~/singa/examples/cifar10/download_data.py py)' |
| 73 | + ) |
| 74 | + sys.exit(0) |
| 75 | + return dirpath |
| 76 | + |
| 77 | + |
| 78 | +def normalize_for_resnet(train_x, test_x): |
| 79 | + mean = [0.4914, 0.4822, 0.4465] |
| 80 | + std = [0.2023, 0.1994, 0.2010] |
| 81 | + train_x /= 255 |
| 82 | + test_x /= 255 |
| 83 | + for ch in range(0, 2): |
| 84 | + train_x[:, ch, :, :] -= mean[ch] |
| 85 | + train_x[:, ch, :, :] /= std[ch] |
| 86 | + test_x[:, ch, :, :] -= mean[ch] |
| 87 | + test_x[:, ch, :, :] /= std[ch] |
| 88 | + return train_x, test_x |
| 89 | + |
| 90 | + |
| 91 | +def resize_dataset(x, IMG_SIZE): |
| 92 | + num_data = x.shape[0] |
| 93 | + dim = x.shape[1] |
| 94 | + X = np.zeros(shape=(num_data, dim, IMG_SIZE, IMG_SIZE), dtype=np.float32) |
| 95 | + for n in range(0, num_data): |
| 96 | + for d in range(0, dim): |
| 97 | + X[n, d, :, :] = np.array(Image.fromarray(x[n, d, :, :]).resize( |
| 98 | + (IMG_SIZE, IMG_SIZE), Image.BILINEAR), |
| 99 | + dtype=np.float32) |
| 100 | + return X |
| 101 | + |
| 102 | + |
| 103 | +def augmentation(x, batch_size): |
| 104 | + xpad = np.pad(x, [[0, 0], [0, 0], [4, 4], [4, 4]], 'symmetric') |
| 105 | + for data_num in range(0, batch_size): |
| 106 | + offset = np.random.randint(8, size=2) |
| 107 | + x[data_num, :, :, :] = xpad[data_num, :, offset[0]:offset[0] + 32, |
| 108 | + offset[1]:offset[1] + 32] |
| 109 | + if_flip = np.random.randint(2) |
| 110 | + if (if_flip): |
| 111 | + x[data_num, :, :, :] = x[data_num, :, :, ::-1] |
| 112 | + return x |
| 113 | + |
| 114 | + |
| 115 | +def accuracy(pred, target): |
| 116 | + y = np.argmax(pred, axis=1) |
| 117 | + t = np.argmax(target, axis=1) |
| 118 | + a = y == t |
| 119 | + return np.array(a, "int").sum() |
| 120 | + |
| 121 | + |
| 122 | +def to_categorical(y, num_classes): |
| 123 | + y = np.array(y, dtype="int") |
| 124 | + n = y.shape[0] |
| 125 | + categorical = np.zeros((n, num_classes)) |
| 126 | + for i in range(0, n): |
| 127 | + categorical[i, y[i]] = 1 |
| 128 | + categorical = categorical.astype(np.float32) |
| 129 | + return categorical |
| 130 | + |
| 131 | + |
| 132 | +# Function to all reduce NUMPY accuracy and loss from multiple devices |
| 133 | +def reduce_variable(variable, dist_opt, reducer): |
| 134 | + reducer.copy_from_numpy(variable) |
| 135 | + dist_opt.all_reduce(reducer.data) |
| 136 | + dist_opt.wait() |
| 137 | + output = tensor.to_numpy(reducer) |
| 138 | + return output |
| 139 | + |
| 140 | + |
| 141 | +# Function to synchronize SINGA TENSOR initial model parameters |
| 142 | +def synchronize(tensor, dist_opt): |
| 143 | + dist_opt.all_reduce(tensor.data) |
| 144 | + dist_opt.wait() |
| 145 | + tensor /= dist_opt.world_size |
| 146 | + |
| 147 | + |
| 148 | +# Data partition |
| 149 | +def data_partition(dataset_x, dataset_y, global_rank, world_size): |
| 150 | + data_per_rank = dataset_x.shape[0] // world_size |
| 151 | + idx_start = global_rank * data_per_rank |
| 152 | + idx_end = (global_rank + 1) * data_per_rank |
| 153 | + return dataset_x[idx_start:idx_end], dataset_y[idx_start:idx_end] |
0 commit comments