Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
bot web server
# 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)
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))'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
# sleep for a small amount to give the model a chance
# to classify the input image
# 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...")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment