Predicting and Monitoring without Python SDK

import requests
import json


def generate_authorization_header(host: str, deployment_id: str, api_token: str):
    """This method calls our prediction service to create a jwt and build authorization header.
    Our jwt are usually available for 1 hour, you might need to recall this method at some point to maintain connection.

    :param host: of our prediction service
    :param deployment_id: deployment identifier that you want to use
    :param api_token: your picsellia api token
    :return: a dict representing headers with a JWT of connection to our service.
    """
    auth_url = host + "/api/login"
    auth_headers = {
        "Authorization": "Token " + api_token,
    }
    # json.dumps() serialize a python dictionary to a json string
    auth_payload = json.dumps({"deployment_id": deployment_id, "api_token": api_token})

    # requests.post() makes a HTTP 'POST', on given url, with given headers, and given json payload
    # calling .json() return the response as a dict, that have the wanted key "jwt"
    response = requests.post(
        url=auth_url, headers=auth_headers, data=auth_payload
    ).json()
    return {
        "Authorization": "Bearer " + response["jwt"],
    }


if __name__ == "__main__":
    # Host of our picsellia prediction service.
    host = "https://serving.picsellia.com"

    # Fill these variables with your data
    deployment_id = "<YOUR_DEPLOYMENT_ID>"
    api_token = "<YOUR_PICSELLIA_API_TOKEN>"

    # Generate authorization headers by calling our prediction service
    headers = generate_authorization_header(host, deployment_id, api_token)

    # Url of our service to make a prediction
    url = host + "/api/deployment/" + deployment_id + "/predict"

    # Some optional metadata of your prediction
    data = {
        # "source": "source-of-your-prediction",
        # "tags": ["some", "tag"]
    }

    image_path = "path/to/image.png"
    with open(image_path, "rb") as file:
        # this requests.post() needs a file object, that we open with open() primitive
        # note requests module does not need us to serialize data and files here
        r = requests.post(url=url, files={"media": file}, headers=headers, data=data)
        prediction = r.json()
        print(prediction)

import base64
import mimetypes

import requests
import json


def generate_authorization_header(host: str, deployment_id: str, api_token: str):
    """This method calls our monitoring service to create a jwt and build authorization header.
    Our jwt are usually available for 1 hour, you might need to recall this method at some point to maintain connection.

    :param host: of our monitoring service
    :param deployment_id: deployment identifier that you want to use
    :param api_token: your picsellia api token
    :return: a dict representing headers with a JWT of connection to our service.
    """
    # Take care, this auth url is different from prediction service one
    auth_url = host + "/api/auth/login"
    auth_headers = {
        "Authorization": "Token " + api_token,
    }
    # json.dumps() serialize a python dictionary to a json string
    auth_payload = json.dumps({"deployment_id": deployment_id, "api_token": api_token})

    # requests.post() makes an HTTP 'POST', on given url, with given headers, and given json payload
    # calling .json() return the response as a dict, that have the wanted key "jwt"
    response = requests.post(
        url=auth_url, headers=auth_headers, data=auth_payload
    ).json()
    return {
        "Authorization": "Bearer " + response["jwt"],
    }


if __name__ == "__main__":
    # Host of our picsellia prediction service.
    host = "https://serving.picsellia.com"

    # Fill these variables with your data
    deployment_id = "<YOUR_DEPLOYMENT_ID>"
    api_token = "<YOUR_PICSELLIA_API_TOKEN>"

    # Generate authorization headers by calling our prediction service
    headers = generate_authorization_header(host, deployment_id, api_token)

    # Url of our service to make a prediction
    url = host + "/api/deployment/" + deployment_id + "/predictions"

    # This block is needed in our snippet, because we want to find content type, filename
    # also we have to encode image to a base64 data : monitoring api only allows this type of encoding at the moment
    # image_width and image_height are not computed in this snippet but need to be sent to monitoring service
    # You might already have opened this image, you might want to do it differently
    image_path = "/path/to/your/image"
    image_height = 600
    image_width = 800
    with open(image_path, "rb") as img_file:
        content_type = mimetypes.guess_type(image_path, strict=False)[0]
        filename = image_path.split("/")[-1]
        encoded_image = base64.b64encode(img_file.read()).decode("utf-8")

    # latency given by your model to predict on this image
    latency = 0.259

    # prediction is a python dictionary representing shapes predicted by your model,
    # it needs different keys depending on your model. For example for an ObjectDetection model:
    raw_predictions = {
        "detection_scores": [
            0.7279638051986694,  # first shape's score and should be between 0 and 1
            0.7203983664512634,
            0.7142382264137268,
        ],
        "detection_boxes": [
            [57, 213, 118, 256],  # first shape's box as [y1, x1, y2, x2] coordinates
            [186, 165, 246, 248],
            [80, 73, 169, 199],
        ],
        "detection_classes": [
            3,  # first shape's class, please make sure your labelmap is well-defined
            3,
            4,
        ],
    }

    data = json.dumps(
        {
            "filename": filename,
            "content_type": content_type,
            "height": image_height,
            "width": image_width,
            "image": encoded_image,
            "raw_predictions": raw_predictions,
            "latency": latency,
        }
    )
    # requests.post() will send your prediction to our monitoring service
    response = requests.post(url=url, headers=headers, data=data)

    monitoring_response = response.json()
    print(monitoring_response)