def transformation()

in source/containers/face-detection/detector/predictor.py [0:0]

• 49 lines of code
• 6 McCabe index (conditional complexity)

def transformation():
    """
    Do an inference on a single batch of data. In this sample server, we take image data as base64 formation,
    decode it for internal use and then convert the predictions to json format

    :return:
    """
    t_start = time.time()

    if flask.request.content_type == 'application/json':
        request_body = flask.request.data.decode('utf-8')
        request_body = json.loads(request_body)
        image_bytes = request_body['image_bytes']
        short_size = request_body['short_size']
    else:
        return flask.Response(
            response='Object detector only supports application/json data',
            status=415,
            mimetype='text/plain')

    # pre-process
    img = mx.img.imdecode(base64.b64decode(image_bytes))
    height, width, channels = img.shape[0], img.shape[1], img.shape[2]

    mean, std = (0.485, 0.456, 0.406), (0.229, 0.224, 0.225)
    resized_img = mx.image.resize_short(img, size=short_size)
    resized_rescaled_img = mx.nd.image.to_tensor(resized_img)
    resized_rescaled_normalized_img = mx.nd.image.normalize(resized_rescaled_img, mean=mean, std=std)
    resized_rescaled_normalized_img = resized_rescaled_normalized_img.expand_dims(0)
    if mx.context.num_gpus() != 0:
        resized_rescaled_normalized_img = resized_rescaled_normalized_img.copyto(mx.gpu())

    # inference
    class_ids, mx_scores, mx_bounding_boxes = ObjectDetectionService.predict(resized_rescaled_normalized_img)

    # post-process
    class_ids = class_ids.asnumpy()
    mx_scores = mx_scores.asnumpy()
    mx_bounding_boxes = mx_bounding_boxes.asnumpy()

    # resize detection results back to original image size
    scale_ratio = short_size / height if height < width else short_size / width
    bbox_coords, bbox_scores = list(), list()
    for index, bbox in enumerate(mx_bounding_boxes[0]):
        prob = float(mx_scores[0][index][0])
        if prob < 0.0:
            continue

        [x_min, y_min, x_max, y_max] = bbox
        x_min = int(x_min / scale_ratio)
        y_min = int(y_min / scale_ratio)
        x_max = int(x_max / scale_ratio)
        y_max = int(y_max / scale_ratio)
        bbox_coords.append([x_min, y_min, x_max, y_max])
        bbox_scores.append([prob])

    body = {
        'width': width,
        'height': height,
        'channels': channels,
        'bbox_scores': bbox_scores,  # shape = (N, 1)
        'bbox_coords': bbox_coords,  # shape = (N, 4)
    }
    t_end = time.time()
    print('Time consumption = {} second'.format(t_end - t_start))
    print('Response = {}'.format(body))

    return flask.Response(response=json.dumps(body), status=200, mimetype='application/json')