def run_classification_inference()

in src/edge/run.py [0:0]

• 26 lines of code
• 10 McCabe index (conditional complexity)

def run_classification_inference(agent, filename):
    """Runs inference on the given image file. Returns prediction and model latency."""
    # Check if the model for image classification is available
    # The application always uses the latest version of the model in the list of loaded models
    model_name_img_clf = config['mappings']['image-classification-app']
    model_is_loaded = any([m['name']==model_name_img_clf for m in models_loaded])
    if not model_is_loaded:
        logging.info('Model for image classification not loaded, waiting for deployment...')
        return None, None

    # Get the identifier of the currently loaded model
    model_dict_img_clf = next((x for x in models_loaded if x['name'] == model_name_img_clf), None)
    if not model_dict_img_clf:
        logging.info('Model for image classification not loaded, waiting for deployment...')
        return None, None
    model_id_img_clf = model_dict_img_clf['identifier']

    logging.info('\nClassification inference with %s' % filename)
    image = PIL.Image.open(filename)
    image = image.convert(mode='RGB')

    # Preprocessing
    x_batchified = preprocess_image(image, IMG_WIDTH, IMG_HEIGHT)

    # Run inference with agent and time taken
    t_start = timer()
    y = agent.predict(model_id_img_clf, x_batchified)
    t_stop = timer()
    t_ms = np.round((t_stop - t_start) * 1000, decimals=0)

    agent.capture_data(model_id_img_clf, x_batchified, y)
    y = y.ravel()
    logging.info(y)

    img_clf_class_labels = ['normal', 'anomalous']

    for indx, l in enumerate(img_clf_class_labels):
        logging.info('Class probability label "%s": %f' % (img_clf_class_labels[indx], y[indx]))
    return y, t_ms