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, pipeline_type="detect"):
  19     #from sklearn.pipeline import Pipeline
  20     dependent_modules = [pd, np, 'opencv-python']  # define as dependent libraries
  21
  22     # for now, do nothing specific to transformer...
  23
  24     return transformer, dependent_modules
  25
  26
  27 def main(config={}):
  28     import argparse
  29     parser = argparse.ArgumentParser()
  30     parser.add_argument('-p', '--predict_path', type=str, default='', help="save detections from model (model must be provided via 'dump_model')")
  31     parser.add_argument('-i', '--input', type=str, default='',help='absolute path to input data (image or csv, only during prediction / dump)')
  32     parser.add_argument('-c', '--csv_input', dest='csv_input', action='store_true', default=False, help='input as CSV format not an image')
  33     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')
  34     parser.add_argument('-f', '--function', type=str, default='detect',help='which type of model to generate', choices=['detect', 'pixelate'])
  35     parser.add_argument('-a', '--push_address', help='server address to push the model (e.g. http://localhost:8887/v2/models)', default='')
  36     parser.add_argument('-d', '--dump_model', help='dump model to a pickle directory for local running', default='')
  37     config.update(vars(parser.parse_args()))     #pargs, unparsed = parser.parse_known_args()
  38
  39     if not config['predict_path']:
  40         print("Attempting to create new model for dump or push...")
  41
  42         # refactor the raw samples from upstream image classifier
  43         if config['function'] == "detect":
  44             transform = FaceDetectTransform(include_image=not config['suppress_image'])
  45         elif config['function'] == "pixelate":
  46             transform = RegionTransform()
  47         else:
  48             print("Error: Functional mode '{:}' unknown, aborting create".format(config['function']))
  49         inputDf = transform.generate_in_df()
  50         pipeline, EXTRA_DEPS = model_create_pipeline(transform, "detect")
  51
  52         # formulate the pipeline to be used
  53         model_name = MODEL_NAME+"_"+config['function']
  54         if 'push_address' in config and config['push_address']:
  55             from cognita_client.push import push_sklearn_model # push_skkeras_hybrid_model (keras?)
  56             print("Pushing new model to '{:}'...".format(config['push_address']))
  57             push_sklearn_model(pipeline, inputDf, api=config['push_address'], name=model_name, extra_deps=EXTRA_DEPS)
  58
  59         if 'dump_model' in config and config['dump_model']:
  60             from cognita_client.wrap.dump import dump_sklearn_model # dump_skkeras_hybrid_model (keras?)
  61             print("Dumping new model to '{:}'...".format(config['dump_model']))
  62             dump_sklearn_model(pipeline, inputDf, config['dump_model'], name=model_name, extra_deps=EXTRA_DEPS)
  63
  64     else:
  65         if not config['dump_model'] or not os.path.exists(config['dump_model']):
  66             print("Attempting to predict from a dumped model, but model not found.".format(config['dump_model']))
  67             sys.exit(-1)
  68         if not os.path.exists(config['input']):
  69             print("Predictino requested but target input '{:}' was not found, please check input arguments.".format(config['input']))
  70             sys.exit(-1)
  71
  72         print("Attempting predict/transform on input sample...")
  73         from cognita_client.wrap.load import load_model
  74         model = load_model(config['dump_model'])
  75         if not config['csv_input']:
  76             inputDf = FaceDetectTransform.generate_in_df(config['input'])
  77         else:
  78             inputDf = pd.read_csv(config['input'], converters={FaceDetectTransform.COL_IMAGE_DATA:FaceDetectTransform.read_byte_arrays})
  79         dfPred = model.transform.from_native(inputDf).as_native()
  80
  81         if config['predict_path']:
  82             print("Writing prediction to file '{:}'...".format(config['predict_path']))
  83             if not config['csv_input']:
  84                 dfPred.to_csv(config['predict_path'], sep=",", index=False)
  85             else:
  86                 FaceDetectTransform.generate_out_image(dfPred, config['predict_path'])
  87         if not config['csv_input']:
  88             dfPred = FaceDetectTransform.suppress_image(dfPred)
  89
  90         if dfPred is not None:
  91             print("Predictions:\n{:}".format(dfPred))
  92
  93 if __name__ == '__main__':
  94     main()