206 lines
8.3 KiB
Python
206 lines
8.3 KiB
Python
import keras
|
|
from keras.src.legacy.preprocessing.image import ImageDataGenerator
|
|
from flask import Flask, render_template, send_file, request, jsonify
|
|
import traceback
|
|
import numpy as np
|
|
import tensorflow
|
|
import sys
|
|
import os
|
|
import io
|
|
import PIL.Image
|
|
import uuid
|
|
|
|
#init flaskapp
|
|
app = Flask(__name__)
|
|
|
|
@app.errorhandler(Exception)
|
|
def handle_exception(e):
|
|
traceback.print_exc()
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
# use this on image passed in to the API
|
|
def save_image(imgData):
|
|
filename=str(uuid.uuid4())+'.jpg'
|
|
print('saving {filename}'.format(filename=filename))
|
|
with open(filename,'wb') as output:
|
|
output.write(imgData)
|
|
|
|
# Use this after processing with PIL
|
|
def save_pil_image(image):
|
|
filename=str(uuid.uuid4())+'.jpg'
|
|
print('saving {filename}'.format(filename=filename))
|
|
imageByteArray = io.BytesIO()
|
|
image.save(imageByteArray, format='JPEG')
|
|
imageByteArray = imageByteArray.getvalue()
|
|
with open(filename,'wb') as output:
|
|
output.write(imageByteArray)
|
|
|
|
|
|
def render_index():
|
|
return """<!doctype html>
|
|
<html>
|
|
<head>
|
|
<title>Convolutional Neural Network Hosting Platform</title>
|
|
</head>
|
|
<body>
|
|
<p>You have reached the main page of the Convolutional Neural Network Hosting Platform. This platform hosts the following models:<strong>resnet50</strong>,<strong>vgg16</strong>, and <strong>inception</strong></p>
|
|
<p>The models can be invoked by submitting a POST request with 128,128 grayscale image in jpg format</p>
|
|
<p>The following are some examples</p>
|
|
<p>http://127.0.0.1/predict_vgg16</p>
|
|
<p>http://127.0.0.1/predict_resnet50</p>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
@app.route('/')
|
|
def index():
|
|
return render_index()
|
|
|
|
@app.route('/ping', methods=['GET','POST'])
|
|
def ping():
|
|
return "Alive"
|
|
|
|
@app.route('/predict_vgg16', methods=['GET','POST'])
|
|
def predict_vgg16():
|
|
test_image=request.get_data()
|
|
test_image = PIL.Image.open(io.BytesIO(test_image))
|
|
# save_pil_image(test_image)
|
|
# test_image = test_image.convert('L')
|
|
test_array=keras.preprocessing.image.img_to_array(test_image)
|
|
batch_test_array=np.array([test_array])
|
|
predictions=vgg16_model.predict(batch_test_array)
|
|
if type(predictions) == list:
|
|
average_prediction = sum(predictions)/len(predictions)
|
|
threshold_output = np.where(average_prediction > 0.5, 1, 0)
|
|
else :
|
|
threshold_output = np.where(predictions > 0.5, 1, 0)
|
|
response=str(predictions)+'-->'+str(threshold_output)
|
|
print('/predict_vgg16:'+response)
|
|
return response
|
|
|
|
@app.route('/predict_resnet50', methods=['GET','POST'])
|
|
def predict_resnet50():
|
|
test_image=request.get_data()
|
|
test_image = PIL.Image.open(io.BytesIO(test_image))
|
|
test_image = test_image.convert('L')
|
|
test_array=keras.preprocessing.image.img_to_array(test_image)
|
|
batch_test_array=np.array([test_array])
|
|
predictions=resnet50_model.predict(batch_test_array)
|
|
if type(predictions) == list:
|
|
average_prediction = sum(predictions)/len(predictions)
|
|
threshold_output = np.where(average_prediction > 0.5, 1, 0)
|
|
else :
|
|
threshold_output = np.where(predictions > 0.5, 1, 0)
|
|
response=str(predictions)+'-->'+str(threshold_output)
|
|
print('/predict_resnet50:'+response)
|
|
return response
|
|
|
|
# New model updated in October 2024
|
|
# To Test : curl -X POST --data-binary @ADT_11_0_0_Training_BollingerBand_20191129_270d.jpg http://127.0.0.1:5000/predict_resnet50_20241024_270
|
|
@app.route('/predict_resnet50_20241024_270', methods=['GET','POST'])
|
|
def predict_resnet50_20241024_270():
|
|
test_image=request.get_data()
|
|
test_image = PIL.Image.open(io.BytesIO(test_image))
|
|
test_image = test_image.convert('L')
|
|
test_array=keras.preprocessing.image.img_to_array(test_image)
|
|
batch_test_array=np.array([test_array])
|
|
predictions=resnet50_20241024_270_model.predict(batch_test_array)
|
|
if type(predictions) == list:
|
|
average_prediction = sum(predictions)/len(predictions)
|
|
threshold_output = np.where(average_prediction > 0.5, 1, 0)
|
|
else :
|
|
threshold_output = np.where(predictions > 0.5, 1, 0)
|
|
response=str(predictions)+'-->'+str(threshold_output)
|
|
print('/predict_resnet50_20241024_270:'+response)
|
|
return response
|
|
|
|
@tensorflow.function(reduce_retracing=True)
|
|
def infer_convnext(x):
|
|
return convnext_model(x, training=False)
|
|
|
|
# New model updated in February 2026 ConvNexT)
|
|
@app.route('/predict_convnext', methods=['GET','POST'])
|
|
def predict_convnext():
|
|
test_image=request.get_data()
|
|
test_image = PIL.Image.open(io.BytesIO(test_image))
|
|
# Convert grayscale → RGB
|
|
test_image = test_image.convert('RGB')
|
|
# Resize to match training size
|
|
if test_image.size != (224, 224):
|
|
test_image = test_image.resize((224, 224))
|
|
test_array=keras.preprocessing.image.img_to_array(test_image)
|
|
batch_test_array = np.expand_dims(test_array, axis=0)
|
|
# predictions=convnext_model.predict(batch_test_array)
|
|
# predictions = convnext_model(batch_test_array, training=False)
|
|
predictions = infer_convnext(batch_test_array)
|
|
threshold_output = np.where(predictions > 0.5, 1, 0)
|
|
response=str(predictions)+'-->'+str(threshold_output)
|
|
print('/predict_convnext:'+response)
|
|
return response
|
|
|
|
|
|
@app.route('/predict_lenet5', methods=['GET','POST'])
|
|
def predict_lenet5():
|
|
test_image=request.get_data()
|
|
test_image = PIL.Image.open(io.BytesIO(test_image))
|
|
test_image = test_image.convert('L')
|
|
test_array=keras.preprocessing.image.img_to_array(test_image)
|
|
batch_test_array=np.array([test_array])
|
|
predictions=lenet_model.predict(batch_test_array)
|
|
if type(predictions) == list:
|
|
average_prediction = sum(predictions)/len(predictions)
|
|
threshold_output = np.where(average_prediction > 0.5, 1, 0)
|
|
else :
|
|
threshold_output = np.where(predictions > 0.5, 1, 0)
|
|
response=str(predictions)+'-->'+str(threshold_output)
|
|
print('/predict_lenet5:'+response)
|
|
return response
|
|
|
|
# This method is used to process an image through PIL and send it back to the client. The client can then used this processed image as part of the training data
|
|
# so that the model can adapt to images that are processed through PIL
|
|
@app.route('/process_image', methods=['GET','POST'])
|
|
def process_image():
|
|
print('/process_image')
|
|
image=request.get_data()
|
|
image = PIL.Image.open(io.BytesIO(image))
|
|
imageByteArray = io.BytesIO()
|
|
image.save(imageByteArray, format='JPEG')
|
|
imageByteArray = imageByteArray.getvalue()
|
|
print('processed {length} bytes.'.format(length=len(imageByteArray)))
|
|
return send_file(io.BytesIO(imageByteArray), mimetype='image/jpeg', as_attachment=True, download_name='%s.jpg' % str(uuid.uuid4()))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
path = os.getcwd()
|
|
if(not path.endswith('Models')):
|
|
os.chdir(path + '/Weights')
|
|
path=os.getcwd()
|
|
print('Current directory is ..{current_directory}'.format(current_directory=path))
|
|
resnet50_model_name='../Weights/resnet50.h5.keras'
|
|
resnet50_model_name = os.path.realpath(path + '/' +resnet50_model_name)
|
|
print('Loading {model_name}'.format(model_name=resnet50_model_name))
|
|
resnet50_model = keras.models.load_model(resnet50_model_name)
|
|
|
|
resnet50_20241024_270_model_name='../Weights/resnet50_20241024_270.h5.keras'
|
|
resnet50_20241024_270_model_name = os.path.realpath(path + '/' + resnet50_20241024_270_model_name)
|
|
print('Loading {model_name}'.format(model_name=resnet50_20241024_270_model_name))
|
|
resnet50_20241024_270_model = keras.models.load_model(resnet50_20241024_270_model_name)
|
|
|
|
convnext_model_name='../Weights/convnext_20260228_90.h5.keras'
|
|
convnext_model_name=os.path.realpath(path + '/' + convnext_model_name)
|
|
print('Loading {model_name}'.format(model_name=convnext_model_name))
|
|
convnext_model=keras.models.load_model(convnext_model_name);
|
|
|
|
vgg16_model_name='../Weights/vggnet16.h5.keras'
|
|
vgg16_model_name = os.path.realpath(path + '/' + vgg16_model_name)
|
|
print('Loading {model_name}'.format(model_name=vgg16_model_name))
|
|
vgg16_model=keras.models.load_model(vgg16_model_name)
|
|
|
|
lenet_model_name='../Weights/lenet5.h5.keras'
|
|
lenet_model_name = os.path.realpath(path + '/' +lenet_model_name)
|
|
print('Loading {model_name}'.format(model_name=lenet_model_name))
|
|
lenet_model=keras.models.load_model(lenet_model_name)
|
|
|
|
port = int(os.environ.get('PORT',5000))
|
|
app.run(host='0.0.0.0',port=port)
|