anomaly_detection_ecg/service.py

import numpy as np
import pickle
import yaml
import bentoml
from bentoml.io import NumpyNdarray


def calculate_loss(reconstructions,data,type_loss):
    if type_loss=="mse":
        train_loss  = np.mean((reconstructions - data)**2, axis=1)
        return train_loss
    elif type_loss=="mae":
        train_loss  = np.mean(np.abs(reconstructions - data), axis=1)
        return train_loss 

BENTO_MODEL_TAG = "keras_model:latest"

ad_runner = bentoml.keras.get(BENTO_MODEL_TAG).to_runner()

service = bentoml.Service("keras_model_ad", runners=[ad_runner])

f = open('src/threshold.pickle', 'rb')
threshold = pickle.load(f)
f.close()

f = open('src/parameters.yaml','rb')
params = yaml.load(f, Loader=yaml.FullLoader)
f.close()

@service.api(input=NumpyNdarray(), output=NumpyNdarray())
def predict(input_data: np.ndarray) -> np.ndarray:
    rec = ad_runner.predict.run(input_data)
    print(type(rec))
    loss = calculate_loss(np.array(rec),input_data,params['type_loss'])
    return np.array(loss>threshold).astype(int)