face_privacy_filter/filter_image.py

   1 #! python
   2 # -*- coding: utf-8 -*-
   3 """
   4 Wrapper for image emotion classification task
   5 """
   6
   7 import os.path
   8 import sys
   9
  10 import numpy as np
  11 import pandas as pd
  12
  13 from face_privacy_filter.transform_detect import FaceDetectTransform
  14 from face_privacy_filter.transform_region import RegionTransform
  15 from face_privacy_filter._version import MODEL_NAME
  16
  17
  18 def model_create_pipeline(transformer):
  19     from acumos.modeling import Model
  20     from acumos.session import Requirements
  21     import sklearn
  22     import cv2
  23     from os import path
  24
  25     type_in = transformer._acumos_type_in
  26     type_out = transformer._acumos_type_out
  27
  28     def predict_class(df: type_in) -> type_out:
  29         '''Returns an array of float predictions'''
  30         return transformer.predict(df)
  31
  32     # compute path of this package to add it as a dependency
  33     package_path = path.dirname(path.realpath(__file__))
  34     return Model(classify=predict_class), Requirements(packages=[package_path], reqs=[pd, np, sklearn],
  35                                                        req_map={cv2: 'opencv-python'})
  36
  37
  38 def main(config={}):
  39     import argparse
  40     parser = argparse.ArgumentParser()
  41     parser.add_argument('-p', '--predict_path', type=str, default='', help="save detections from model (model must be provided via 'dump_model')")
  42     parser.add_argument('-i', '--input', type=str, default='', help='absolute path to input data (image or csv, only during prediction / dump)')
  43     parser.add_argument('-c', '--csv_input', dest='csv_input', action='store_true', default=False, help='input as CSV format not an image')
  44     parser.add_argument('-s', '--suppress_image', dest='suppress_image', action='store_true', default=False, help='do not create an extra row for a returned image')
  45     parser.add_argument('-f', '--function', type=str, default='detect', help='which type of model to generate', choices=['detect', 'pixelate'])
  46     parser.add_argument('-a', '--push_address', help='server address to push the model (e.g. http://localhost:8887/v2/models)', default='')
  47     parser.add_argument('-d', '--dump_model', help='dump model to a pickle directory for local running', default='')
  48     config.update(vars(parser.parse_args()))     # pargs, unparsed = parser.parse_known_args()
  49
  50     if not config['predict_path']:
  51         print("Attempting to create new model for dump or push...")
  52
  53         # refactor the raw samples from upstream image classifier
  54         if config['function'] == "detect":
  55             transform = FaceDetectTransform(include_image=not config['suppress_image'])
  56         elif config['function'] == "pixelate":
  57             transform = RegionTransform()
  58         else:
  59             print("Error: Functional mode '{:}' unknown, aborting create".format(config['function']))
  60         inputDf = transform.generate_in_df()
  61         pipeline, reqs = model_create_pipeline(transform)
  62
  63         # formulate the pipeline to be used
  64         model_name = MODEL_NAME + "_" + config['function']
  65         if 'push_address' in config and config['push_address']:
  66             from acumos.session import AcumosSession
  67             print("Pushing new model to '{:}'...".format(config['push_address']))
  68             session = AcumosSession(push_api=config['push_address'], auth_api=config['auth_address'])
  69             session.push(pipeline, model_name, reqs)  # creates ./my-iris.zip
  70
  71         if 'dump_model' in config and config['dump_model']:
  72             from acumos.session import AcumosSession
  73             from os import makedirs
  74             if not os.path.exists(config['dump_model']):
  75                 makedirs(config['dump_model'])
  76             print("Dumping new model to '{:}'...".format(config['dump_model']))
  77             session = AcumosSession()
  78             session.dump(pipeline, model_name, config['dump_model'], reqs)  # creates ./my-iris.zip
  79
  80     else:
  81         if not config['dump_model'] or not os.path.exists(config['dump_model']):
  82             print("Attempting to predict from a dumped model, but model not found.".format(config['dump_model']))
  83             sys.exit(-1)
  84         if not os.path.exists(config['input']):
  85             print("Predictino requested but target input '{:}' was not found, please check input arguments.".format(config['input']))
  86             sys.exit(-1)
  87
  88         print("Attempting predict/transform on input sample...")
  89         from acumos.wrapped import load_model
  90         model = load_model(config['dump_model'])
  91         if not config['csv_input']:
  92             inputDf = FaceDetectTransform.generate_in_df(config['input'])
  93         else:
  94             inputDf = pd.read_csv(config['input'], converters={FaceDetectTransform.COL_IMAGE_DATA: FaceDetectTransform.read_byte_arrays})
  95         dfPred = model.transform.from_native(inputDf).as_wrapped()
  96         dfPred = dfPred[0]
  97
  98         if config['predict_path']:
  99             print("Writing prediction to file '{:}'...".format(config['predict_path']))
 100             if not config['csv_input']:
 101                 dfPred.to_csv(config['predict_path'], sep=",", index=False)
 102             else:
 103                 FaceDetectTransform.generate_out_image(dfPred, config['predict_path'])
 104         if not config['csv_input']:
 105             dfPred = FaceDetectTransform.suppress_image(dfPred)
 106
 107         if dfPred is not None:
 108             print("Predictions:\n{:}".format(dfPred))
 109
 110
 111 if __name__ == '__main__':
 112     main()