keras-complete-rest-api
├── helpers.py
├── jemma.png
├── keras_rest_api_app.wsgi
├── run_model_server.py
├── run_web_server.py
├── settings.py
├── simple_request.py
└── stress_test.py

# initialize Redis connection settings
REDIS_HOST = "localhost"
REDIS_PORT = 6379
REDIS_DB = 0
# initialize constants used to control image spatial dimensions and
# data type
IMAGE_WIDTH = 224
IMAGE_HEIGHT = 224
IMAGE_CHANS = 3
IMAGE_DTYPE = "float32"
# initialize constants used for server queuing
IMAGE_QUEUE = "image_queue"
BATCH_SIZE = 32
SERVER_SLEEP = 0.25
CLIENT_SLEEP = 0.25

# import the necessary packages
import numpy as np
import base64
import sys
def base64_encode_image(a):
  # base64 encode the input NumPy array
  return base64.b64encode(a).decode("utf-8")
def base64_decode_image(a, dtype, shape):
  # if this is Python 3, we need the extra step of encoding the
  # serialized NumPy string as a byte object
  if sys.version_info.major == 3:
    a = bytes(a, encoding="utf-8")
  # convert the string to a NumPy array using the supplied data
  # type and target shape
  a = np.frombuffer(base64.decodestring(a), dtype=dtype)
  a = a.reshape(shape)
  # return the decoded image
  return a

# import the necessary packages
from tensorflow.keras.preprocessing.image import img_to_array
from tensorflow.keras.applications.resnet50 import preprocess_input
from PIL import Image
import numpy as np
import settings
import helpers
import flask
import redis
import uuid
import time
import json
import io
# initialize our Flask application and Redis server
app = flask.Flask(__name__)
db = redis.StrictRedis(host=settings.REDIS_HOST,
  port=settings.REDIS_PORT, db=settings.REDIS_DB)
def prepare_image(image, target):
  # if the image mode is not RGB, convert it
  if image.mode != "RGB":
    image = image.convert("RGB")
  # resize the input image and preprocess it
  image = image.resize(target)
  image = img_to_array(image)
  image = np.expand_dims(image, axis=0)
  image = preprocess_input(image)
  # return the processed image
  return image
@app.route("/")
def homepage():
  return "Welcome to the PyImageSearch Keras REST API!"
@app.route("/predict", methods=["POST"])
def predict():
  # initialize the data dictionary that will be returned from the
  # view
  data = {"success": False}
  # ensure an image was properly uploaded to our endpoint
  if flask.request.method == "POST":
    if flask.request.files.get("image"):
      # read the image in PIL format and prepare it for
      # classification
      image = flask.request.files["image"].read()
      image = Image.open(io.BytesIO(image))
      image = prepare_image(image,
        (settings.IMAGE_WIDTH, settings.IMAGE_HEIGHT))
      # ensure our NumPy array is C-contiguous as well,
      # otherwise we won't be able to serialize it
      image = image.copy(order="C")
      # generate an ID for the classification then add the
      # classification ID + image to the queue
      k = str(uuid.uuid4())
      image = helpers.base64_encode_image(image)
      d = {"id": k, "image": image}
      db.rpush(settings.IMAGE_QUEUE, json.dumps(d))
      # keep looping until our model server returns the output
      # predictions
      while True:
        # attempt to grab the output predictions
        output = db.get(k)
        # check to see if our model has classified the input
        # image
        if output is not None:
          # add the output predictions to our data
          # dictionary so we can return it to the client
          output = output.decode("utf-8")
          data["predictions"] = json.loads(output)
          # delete the result from the database and break
          # from the polling loop
          db.delete(k)
          break
        # sleep for a small amount to give the model a chance
        # to classify the input image
        time.sleep(settings.CLIENT_SLEEP)
      # indicate that the request was a success
      data["success"] = True
  # return the data dictionary as a JSON response
  return flask.jsonify(data)
# for debugging purposes, it's helpful to start the Flask testing
# server (don't use this for production
if __name__ == "__main__":
  print("* Starting web service...")
  app.run()

# import the necessary packages
from tensorflow.keras.applications import ResNet50
from tensorflow.keras.applications.resnet50 import decode_predictions
import numpy as np
import settings
import helpers
import redis
import time
import json
# connect to Redis server
db = redis.StrictRedis(host=settings.REDIS_HOST,
  port=settings.REDIS_PORT, db=settings.REDIS_DB)
def classify_process():
  # load the pre-trained Keras model (here we are using a model
  # pre-trained on ImageNet and provided by Keras, but you can
  # substitute in your own networks just as easily)
  print("* Loading model...")
  model = ResNet50(weights="imagenet")
  print("* Model loaded")
  # continually pool for new images to classify
  while True:
    # attempt to grab a batch of images from the database, then
    # initialize the image IDs and batch of images themselves
    queue = db.lrange(settings.IMAGE_QUEUE, 0,
      settings.BATCH_SIZE - 1)
    imageIDs = []
    batch = None
    # loop over the queue
    for q in queue:
      # deserialize the object and obtain the input image
      q = json.loads(q.decode("utf-8"))
      image = helpers.base64_decode_image(q["image"],
        settings.IMAGE_DTYPE,
        (1, settings.IMAGE_HEIGHT, settings.IMAGE_WIDTH,
          settings.IMAGE_CHANS))
      # check to see if the batch list is None
      if batch is None:
        batch = image
      # otherwise, stack the data
      else:
        batch = np.vstack([batch, image])
      # update the list of image IDs
      imageIDs.append(q["id"])
    # check to see if we need to process the batch
    if len(imageIDs) > 0:
      # classify the batch
      print("* Batch size: {}".format(batch.shape))
      preds = model.predict(batch)
      results = decode_predictions(preds)
      # loop over the image IDs and their corresponding set of
      # results from our model
      for (imageID, resultSet) in zip(imageIDs, results):
        # initialize the list of output predictions
        output = []
        # loop over the results and add them to the list of
        # output predictions
        for (imagenetID, label, prob) in resultSet:
          r = {"label": label, "probability": float(prob)}
          output.append(r)
        # store the output predictions in the database, using
        # the image ID as the key so we can fetch the results
        db.set(imageID, json.dumps(output))
      # remove the set of images from our queue
      db.ltrim(settings.IMAGE_QUEUE, len(imageIDs), -1)
    # sleep for a small amount
    time.sleep(settings.SERVER_SLEEP)
# if this is the main thread of execution start the model server
# process
if __name__ == "__main__":
  classify_process()

# import the necessary packages
from threading import Thread
import requests
import time
# initialize the Keras REST API endpoint URL along with the input
# image path
KERAS_REST_API_URL = "http://localhost/predict"
IMAGE_PATH = "jemma.png"
# initialize the number of requests for the stress test along with
# the sleep amount between requests
NUM_REQUESTS = 500
SLEEP_COUNT = 0.05
def call_predict_endpoint(n):
  # load the input image and construct the payload for the request
  image = open(IMAGE_PATH, "rb").read()
  payload = {"image": image}
  # submit the request
  r = requests.post(KERAS_REST_API_URL, files=payload).json()
  # ensure the request was sucessful
  if r["success"]:
    print("[INFO] thread {} OK".format(n))
  # otherwise, the request failed
  else:
    print("[INFO] thread {} FAILED".format(n))
# loop over the number of threads
for i in range(0, NUM_REQUESTS):
  # start a new thread to call the API
  t = Thread(target=call_predict_endpoint, args=(i,))
  t.daemon = True
  t.start()
  time.sleep(SLEEP_COUNT)
# insert a long sleep so we can wait until the server is finished
# processing the images
time.sleep(300)

$ wget http://download.redis.io/redis-stable.tar.gz
$ tar xvzf redis-stable.tar.gz
$ cd redis-stable
$ make
$ sudo make install

$ workon dl4cv
$ pip install flask
$ pip install gevent
$ pip install requests
$ pip install redis

<VirtualHost *:80>
  ...
  WSGIDaemonProcess keras_rest_api_app threads=10
  WSGIScriptAlias / /var/www/html/keras-complete-rest-api/keras_rest_api_app.wsgi
  <Directory /var/www/html/keras-complete-rest-api>
    WSGIProcessGroup keras_rest_api_app
    WSGIApplicationGroup %{GLOBAL}
    Order deny,allow
    Allow from all
  </Directory>
  ...
</VirtualHost>

$ curl -X POST -F image=@jemma.png 'http://localhost/predict'
{
  "predictions": [
    {
      "label": "beagle", 
      "probability": 0.9461532831192017
    }, 
    {
      "label": "bluetick", 
      "probability": 0.031958963721990585
    }, 
    {
      "label": "redbone", 
      "probability": 0.0066171870566904545
    }, 
    {
      "label": "Walker_hound", 
      "probability": 0.003387963864952326
    }, 
    {
      "label": "Greater_Swiss_Mountain_dog", 
      "probability": 0.0025766845792531967
    }
  ], 
  "success": true
}

$ python stress_test.py 
[INFO] thread 3 OK
[INFO] thread 0 OK
[INFO] thread 1 OK
...
[INFO] thread 497 OK
[INFO] thread 499 OK
[INFO] thread 498 OK

* Batch size: (4, 224, 224, 3)
* Batch size: (9, 224, 224, 3)
* Batch size: (9, 224, 224, 3)
* Batch size: (8, 224, 224, 3)
...
* Batch size: (2, 224, 224, 3)
* Batch size: (10, 224, 224, 3)
* Batch size: (7, 224, 224, 3)