anomaly_detection_ecg/servicerequest.py

from typing import Tuple
import json
import keras

import numpy as np
import requests
from src.prepare_data import get_data

#SERVICE_URL = "http://127.0.0.1:3000/predict"

# aws endpoint
SERVICE_URL = "https://libobddgz8.execute-api.us-west-1.amazonaws.com/predict"


def calculate_loss(reconstructions,data,type_loss):
    if type_loss=="mse":
        train_loss  = keras.losses.mse(reconstructions, data)
        return train_loss
    elif type_loss=="mae":
        train_loss  = keras.losses.mae(reconstructions, data) 
        return train_loss 

def sample_random_ecg_data_point() -> Tuple[np.ndarray, np.ndarray]:
    test_data, test_labels = get_data('ecg_data','X_test.pickle'),get_data('ecg_data','y_test.pickle')
    random_index = np.random.randint(0, len(test_data),size=10)
    random_test_data = test_data[random_index]
    print(random_test_data.shape)
    if len(random_index)==1:
       random_test_data = np.expand_dims(random_test_data, 0)
    print(random_test_data.shape)
    return random_test_data, test_labels[random_index]


def make_request_to_bento_service(
    service_url: str, input_array: np.ndarray
) -> str:
    serialized_input_data = json.dumps(input_array.tolist())
    response = requests.post(
        service_url,
        data=serialized_input_data,
        headers={"content-type": "application/json"}
    )
    return response.text


def main():
    input_data, expected_output = sample_random_ecg_data_point()
    prediction = make_request_to_bento_service(SERVICE_URL, input_data)

    print(f"prediction: {prediction}")
    print(f"Expected output: {expected_output.astype(int)}")

if __name__ == "__main__":
    main()