bat.apis.deepapi

  1r"""
  2This module implements the DeepAPI client for pretrained VGG16 model on CIFAR-10 dataset.
  3"""
  4
  5import requests
  6from urllib.parse import urlparse
  7import urllib.request, json 
  8from urllib.parse import urljoin
  9
 10import numpy as np
 11np.set_printoptions(suppress=True)
 12
 13from PIL import Image
 14from io import BytesIO
 15import base64
 16
 17import concurrent.futures
 18
 19class DeepAPIBase:
 20    def __init__(self, concurrency=8):
 21        self.concurrency = concurrency
 22        pass
 23
 24    def predict(self, X):
 25        n_targets = 0
 26        if type(X) == list:
 27            n_targets = len(X)
 28        elif type(X) == np.ndarray:
 29            n_targets = X.shape[0]
 30        else:
 31            raise ValueError('Input type not supported...')
 32
 33        return self.predictX(X) if n_targets > 1 else self.predictO(X)
 34
 35    def predictX(self, X):
 36        """
 37        - X: numpy array of shape (N, H, W, C)
 38        """
 39        if isinstance(X, list):
 40            for x in X:
 41                assert len(x.shape) == 3, 'Expecting a 3D tensor'
 42        else:
 43            if len(X.shape) != 4 or X.shape[3] != 3:
 44                raise ValueError(
 45                    "`predict` expects "
 46                    "a batch of images "
 47                    "(i.e. a 4D array of shape (samples, 224, 224, 3)). "
 48                    "Found array with shape: " + str(X.shape)
 49                )
 50
 51        # Single thread
 52        def send_request(url, data):
 53            y_pred_temp = np.zeros(len(self.labels))
 54            for attempt in range(5):
 55                try:
 56                    res = requests.post(url, json=data, timeout=10).json()['predictions']
 57                    for r in res:
 58                        y_pred_temp[self.labels.index(r['label'])] = r['probability']
 59                except Exception as e:
 60                    print('\nError:', e, 'Retrying...', attempt, '\n')
 61                    continue
 62
 63                break
 64
 65            return y_pred_temp
 66
 67        y_preds = []
 68        y_index = []
 69        y_executors = {}
 70
 71        with concurrent.futures.ThreadPoolExecutor(max_workers=self.concurrency) as executor:
 72            for i, x in enumerate(X):
 73                # Load the input image and construct the payload for the request
 74                image = Image.fromarray(np.uint8(x))
 75                buff = BytesIO()
 76                image.save(buff, format="JPEG", subsampling=0, quality=100)
 77
 78                data = {'file': base64.b64encode(buff.getvalue()).decode("utf-8")}
 79                y_executors[executor.submit(send_request, url=self.url, data=data)] = i
 80
 81            for y_executor in concurrent.futures.as_completed(y_executors):
 82                y_index.append(y_executors[y_executor])
 83                y_preds.append(y_executor.result())
 84
 85            y_preds = [y for _, y in sorted(zip(y_index, y_preds))]
 86
 87        return np.array(y_preds)
 88
 89
 90    def predictO(self, X):
 91        """
 92        - X: numpy array of shape (N, H, W, C)
 93        """
 94        if isinstance(X, list):
 95            for x in X:
 96                assert len(x.shape) == 3, 'Expecting a 3D tensor'
 97        else:
 98            if len(X.shape) != 4 or X.shape[3] != 3:
 99                raise ValueError(
100                    "`predict` expects "
101                    "a batch of images "
102                    "(i.e. a 4D array of shape (samples, 224, 224, 3)). "
103                    "Found array with shape: " + str(X.shape)
104                )
105
106        y_preds = []
107        try:
108            for x in X:
109                y_pred_temp = np.zeros(len(self.labels))
110                # Load the input image and construct the payload for the request
111                image = Image.fromarray(np.uint8(x))
112                buff = BytesIO()
113                image.save(buff, format="JPEG")
114
115                data = {'file': base64.b64encode(buff.getvalue()).decode("utf-8")}
116
117                for attempt in range(5):
118                    try:
119                        res = requests.post(self.url, json=data, timeout=10).json()['predictions']
120                        for r in res:
121                            y_pred_temp[self.labels.index(r['label'])] = r['probability']
122                    except Exception as e:
123                        print('\nError:', e, 'Retrying...', attempt, '\n')
124                        continue
125
126                    break
127
128                y_preds.append(y_pred_temp)
129
130        except Exception as e:
131            print(e)
132
133        return np.array(y_preds)
134
135    def print(self, y):
136        """
137        Print the prediction result.
138        """
139        max_len = max([len(x) for x in self.labels])
140
141        print()
142
143        # Unsorted
144        # for i in range(0, len(y)):
145        #     print('{:<{w}s}{:.5f}'.format(self.labels[i], y[i], w=max_len+1))
146
147        # Sorted
148        for p, l in sorted(zip(y, self.labels), reverse=True):
149            print('{:<{w}s}{:.5f}'.format(l, p, w=max_len+1))
150
151    def get_class_name(self, i):
152        """
153        Get the class name from the prediction label 0-10.
154        """
155        return self.labels[i]
156
157
158class DeepAPI_VGG16_Cifar10(DeepAPIBase):
159
160    def __init__(self, url, concurrency=8):
161        """
162        - url: DeepAPI server URL
163        """
164        super().__init__(concurrency)
165
166        url_parse = urlparse(url)
167        self.url = urljoin(url_parse.scheme + '://' + url_parse.netloc, 'vgg16_cifar10')
168
169        # cifar10 labels
170        cifar10_labels = ['frog', 'deer', 'cat', 'bird', 'dog', 'truck', 'ship', 'airplane', 'horse', 'automobile']
171        self.labels = cifar10_labels
172
173
174class DeepAPI_ImageNet(DeepAPIBase):
175    def __init__(self, concurrency=8):
176        super().__init__(concurrency)
177
178        # Load the Keras application labels 
179        with urllib.request.urlopen("https://s3.amazonaws.com/deep-learning-models/image-models/imagenet_class_index.json") as l_url:
180            imagenet_json = json.load(l_url)
181
182            # imagenet_json['134'] = ["n02012849", "crane"],
183            imagenet_json['517'] = ['n02012849', 'crane_bird']
184
185            # imagenet_json['638'] = ['n03710637', 'maillot']
186            imagenet_json['639'] = ['n03710721', 'maillot_tank_suit']
187
188            imagenet_labels = [imagenet_json[str(k)][1] for k in range(len(imagenet_json))]
189
190            self.labels = imagenet_labels
191
192
193class DeepAPI_Inceptionv3_ImageNet(DeepAPI_ImageNet):
194    def __init__(self, url, concurrency=8):
195        """
196        - url: DeepAPI server URL
197        """
198        super().__init__(concurrency)
199
200        url_parse = urlparse(url)
201        self.url = urljoin(url_parse.scheme + '://' + url_parse.netloc, 'inceptionv3')
202
203
204class DeepAPI_Resnet50_ImageNet(DeepAPI_ImageNet):
205    def __init__(self, url, concurrency=8):
206        """
207        - url: DeepAPI server URL
208        """
209        super().__init__(concurrency)
210
211        url_parse = urlparse(url)
212        self.url = urljoin(url_parse.scheme + '://' + url_parse.netloc, 'resnet50')
213
214
215class DeepAPI_VGG16_ImageNet(DeepAPI_ImageNet):
216    def __init__(self, url, concurrency=8):
217        """
218        - url: DeepAPI server URL
219        """
220        super().__init__(concurrency)
221
222        url_parse = urlparse(url)
223        self.url = urljoin(url_parse.scheme + '://' + url_parse.netloc, 'vgg16')
224
225
226bat_deepapi_model_list = {
227    1: ['vgg16_cifar10', DeepAPI_VGG16_Cifar10],
228    2: ['vgg16_imagenet', DeepAPI_VGG16_ImageNet],
229    3: ['resnet50_imagenet', DeepAPI_Resnet50_ImageNet],
230    4: ['inceptionv3_imagenet', DeepAPI_Inceptionv3_ImageNet]
231}