File size: 4,118 Bytes
c5aa5a6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 | import ast
import json
import os
import numpy as np
import pandas as pd
from autogluon.multimodal import MultiModalPredictor
from constants import (
ALLOWED_INPUT_FORMATS,
ALLOWED_OUTPUT_FORMATS,
BRACKET_FORMATTER,
COMMA_DELIMITER,
IMAGE_COLUMN_NAME,
JSON_FORMAT,
LABELS,
NUM_GPU,
PREDICTED_LABEL,
PROBABILITIES,
PROBABILITY,
SAGEMAKER_INFERENCE_OUTPUT,
)
from utils import infer_type_and_cast_value
INFERENCE_OUTPUT = (
infer_type_and_cast_value(os.getenv(SAGEMAKER_INFERENCE_OUTPUT))
if SAGEMAKER_INFERENCE_OUTPUT in os.environ
else [PREDICTED_LABEL]
)
NUM_GPUS = infer_type_and_cast_value(os.getenv(NUM_GPU))
def generate_single_csv_line_inference_selection(data):
"""Generate a single csv line response.
:param data: list of output generated from the model
:return: csv line for the predictions
"""
contents: str
for single_prediction in data:
contents = (
BRACKET_FORMATTER.format(single_prediction)
if isinstance(single_prediction, list)
else str(single_prediction)
)
return contents
def model_fn(model_dir):
"""Load model from previously saved artifact.
:param model_dir: local path to the model directory
:return: loaded model
"""
predictor = MultiModalPredictor.load(model_dir)
if NUM_GPUS is not None:
predictor._config.env.num_gpus = NUM_GPUS
return predictor
def convert_to_json_compatible_type(value):
"""Convert the input value to a JSON compatible type.
:param value: input value
:return: JSON compatible value
"""
string_value = "{}".format(value)
try:
return ast.literal_eval(string_value)
except Exception:
return string_value
def transform_fn(model, request_body, input_content_type, output_content_type):
"""Transform function for serving inference requests.
If INFERENCE_OUTPUT is provided, then the predictions are generated in the requested format and concatenated in the
same order. Otherwise, prediction_labels are generated by default.
:param model: loaded model
:param request_body: request body
:param input_content_type: content type of the input
:param output_content_type: content type of the response
:return: prediction response
"""
if input_content_type.lower() not in ALLOWED_INPUT_FORMATS:
raise Exception(
f"{input_content_type} input content type not supported. Supported formats are {ALLOWED_INPUT_FORMATS}"
)
if output_content_type.lower() not in ALLOWED_OUTPUT_FORMATS:
raise Exception(
f"{output_content_type} output content type not supported. Supported formats are {ALLOWED_OUTPUT_FORMATS}"
)
data = pd.DataFrame({IMAGE_COLUMN_NAME: [request_body]})
result_dict = dict()
result = []
inference_output_list = (
INFERENCE_OUTPUT if isinstance(INFERENCE_OUTPUT, list) else [INFERENCE_OUTPUT]
)
for output_type in inference_output_list:
if output_type == PREDICTED_LABEL:
prediction = model.predict(data)
result_dict[PREDICTED_LABEL] = convert_to_json_compatible_type(prediction.squeeze())
elif output_type == PROBABILITIES:
predict_probs = model.predict_proba(data)
prediction = predict_probs.to_numpy()
result_dict[PROBABILITIES] = predict_probs.squeeze().tolist()
elif output_type == LABELS:
labels = model.class_labels
prediction = np.array([labels]).astype("str")
result_dict[LABELS] = labels.tolist()
else:
predict_probabilities = model.predict_proba(data).to_numpy()
prediction = np.max(predict_probabilities, axis=1)
result_dict[PROBABILITY] = prediction.squeeze().tolist()
result.append(generate_single_csv_line_inference_selection(prediction.tolist()))
response = COMMA_DELIMITER.join(result)
if output_content_type == JSON_FORMAT:
response = json.dumps(result_dict)
return response, output_content_type
|