AzureMLEndpointSampleProcessJson.py

AzureMLEndpointSampleProcessJson.py

import urllib.request
import json
import os
import ssl
import base64

def displayPredictions(jsonPrediction):
    global frame_Width, frame_Heigth, labelColors
    jsonObj = json.loads(jsonPrediction)

    scored_label = jsonObj['Results']['WebServiceOutput0'][0]['Scored Labels']
    scored_prob_space_wolf = jsonObj['Results']['WebServiceOutput0'][0]['Scored Probabilities_space_wolf']
    scored_prob_squirrel = jsonObj['Results']['WebServiceOutput0'][0]['Scored Probabilities_squirrel']
    print(f"  > scored label: {scored_label} - squirrel: {scored_prob_squirrel} - space_wolf: {scored_prob_space_wolf}")
    color = (255,255,255)

    # # display labels
    # start_point_label = (10, 20)
    # text = "{}: {:.4f}".format(f"Scored Label: {scored_label}")
    # cv2.putText(frame, text, start_point_label, cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, 2)
    # return frame

def allowSelfSignedHttps(allowed):
    # bypass the server certificate verification on client side
    if allowed and not os.environ.get('PYTHONHTTPSVERIFY', '') and getattr(ssl, '_create_unverified_context', None):
        ssl._create_default_https_context = ssl._create_unverified_context

def processFile(img_filepath):
    # Request data goes here
    data = {
        "Inputs": {
            "WebServiceInput0":
            [
                {
                    'image': "data:image/png;base64,",
                    'id': "0",
                    'category': "space_wolf",
                },
            ],
        },
        "GlobalParameters": {
        }
    }

    # Create base64 encoded string
    with open(img_filepath, "rb") as f:
        image_string = base64.b64encode(f.read()).decode("utf-8")
    image_data = str(f"""data:image/png;base64,{image_string}""")
    data['Inputs']['WebServiceInput0'][0]['image'] = image_data

    body = str.encode(json.dumps(data))

    url = 'http://< API SERVER >:80/api/v1/service/squirrelimageclassification/score'
    api_key = '< API KEY >' # Replace this with the API key for the web service
    headers = {'Content-Type':'application/json', 'Authorization':('Bearer '+ api_key)}

    req = urllib.request.Request(url, body, headers)
    json_result = {}
    try:
        response = urllib.request.urlopen(req)
        result = response.read()
        data = json.loads(result)
        json_result = json.dumps(data)
    except urllib.error.HTTPError as error:
        print("The request failed with status code: " + str(error.code))
        print(error.info())
        print(json.loads(error.read().decode("utf8", 'ignore')))
    return json_result

allowSelfSignedHttps(True) # this line is needed if you use self-signed certificate in your scoring service.

# Save string of image file path below
img_filepath = "./test_images/squirrel_01.jpg"
res = processFile(img_filepath)
print(f"processing {img_filepath}")
displayPredictions(res)

img_filepath = "./test_images/squirrel_02.jpg"
res = processFile(img_filepath)
print(f"processing {img_filepath}")
#print(res)
displayPredictions(res)

img_filepath = "./test_images/space_wolf.jpg"
res = processFile(img_filepath)
print(f"processing {img_filepath}")
#print(res)
displayPredictions(res)