bot web server

bot_web_server.py

# import the necessary packages
import ast
import logging
from PIL import Image
import numpy as np
import settings
import helpers
from classes import disease_classes
from quart import Quart, request, jsonify
import redis
import uuid
import time
import json
import io
from asyncio import Future

LOGGER = logging.getLogger(__name__)

# initialize our Quart application and Redis server
app = Quart(__name__)

db = redis.StrictRedis(host=settings.REDIS_HOST,
	port=settings.REDIS_PORT, db=settings.REDIS_DB)

@app.route("/")
async def homepage():
	return "Welcome to the Bot REST API!"

@app.route("/dataset", methods=["GET"])
async def dataset():
	return jsonify(disease_classes)

@app.route("/predict", methods=["POST"])
async 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 request.method == "POST":
		data = await request.json
		image_path = data.get("image_path")
		if image_path:
			# # generate an ID for the classification then add the
			# # classification ID + image to the queue
			k = str(uuid.uuid4())
			d = {"id": k, "image_path": image_path}
			db.rpush(settings.IMAGE_QUEUE, json.dumps(d))
			LOGGER.info(f'Image ID: {k} stored in Redis')
			# 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')
					# import pdb; pdb.set_trace()
					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 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()