def get_anomaly_prediction()

in source/lambda/model-invocation/index.py [0:0]

• 8 lines of code
• 1 McCabe index (conditional complexity)

def get_anomaly_prediction(data):
    sagemaker_endpoint_name = "{}-rcf".format(SOLUTION_PREFIX)
    sagemaker_runtime = boto3.client('sagemaker-runtime')
    response = sagemaker_runtime.invoke_endpoint(
        EndpointName=sagemaker_endpoint_name, ContentType='text/csv', Body=data)
    # Extract anomaly score from the endpoint response
    anomaly_score = json.loads(response['Body'].read().decode())["scores"][0]["score"]
    logger.info("anomaly score: {}".format(anomaly_score))

    return {"score": anomaly_score}