face_privacy_filter/transform_region.py

   1 #! python
   2 # -*- coding: utf-8 -*-
   3 """
   4 Wrapper for region processing task; wrapped in classifier for pipieline terminus
   5 """
   6 import cv2
   7 import os
   8 import pandas as pd
   9 import numpy as np
  10 from sklearn.base import BaseEstimator, ClassifierMixin
  11
  12 # NOTE: If this class were built in another model (e.g. another vendor, class, etc), we would need to
  13 #       *exactly match* the i/o for the upstream (detection) and downstream (this processing)
  14 from face_privacy_filter.transform_detect import FaceDetectTransform
  15
  16 class RegionTransform(BaseEstimator, ClassifierMixin):
  17     '''
  18     A sklearn classifier mixin that manpulates image content based on input
  19     '''
  20
  21     def __init__(self, transform_mode="pixelate"):
  22         self.transform_mode = transform_mode    # specific image processing mode to utilize
  23
  24     def get_params(self, deep=False):
  25         return {'transform_mode': self.transform_mode}
  26
  27     @staticmethod
  28     def generate_out_df(media_type="", bin_stream=b""):
  29         # munge stream and mimetype into input sample
  30         return pd.DataFrame([[media_type, bin_stream]], columns=[FaceDetectTransform.COL_IMAGE_MIME, FaceDetectTransform.COL_IMAGE_DATA])
  31
  32     @staticmethod
  33     def generate_in_df(idx=FaceDetectTransform.VAL_REGION_IMAGE_ID, x=0, y=0, w=0, h=0, image=0, bin_stream=b"", media=""):
  34         return pd.DataFrame([[idx,x,y,w,h,image,media,bin_stream]],
  35                             columns=[FaceDetectTransform.COL_REGION_IDX, FaceDetectTransform.COL_FACE_X, FaceDetectTransform.COL_FACE_Y,
  36                                      FaceDetectTransform.COL_FACE_W, FaceDetectTransform.COL_FACE_H,
  37                                      FaceDetectTransform.COL_IMAGE_IDX, FaceDetectTransform.COL_IMAGE_MIME,
  38                                      FaceDetectTransform.COL_IMAGE_DATA])
  39
  40     @property
  41     def output_names_(self):
  42         return [FaceDetectTransform.COL_IMAGE_MIME, FaceDetectTransform.COL_IMAGE_DATA]
  43
  44     @property
  45     def output_types_(self):
  46         list_name = self.output_names_
  47         list_type = self.classes_
  48         return [{list_name[i]:list_type[i]} for i in range(len(list_name))]
  49
  50     @property
  51     def n_outputs_(self):
  52         return 8
  53
  54     @property
  55     def classes_(self):
  56         return [str, str]
  57
  58     def score(self, X, y=None):
  59         return 0
  60
  61     def fit(self, X, y=None):
  62         return self
  63
  64     def predict(self, X, y=None):
  65         """
  66         Assumes a numpy array of [[mime_type, binary_string] ... ]
  67            where mime_type is an image-specifying mime type and binary_string is the raw image bytes
  68         """
  69
  70         # group by image index first
  71         #   decode image at region -1
  72         #   collect all remaining regions, operate with each on input image
  73         #   generate output image, send to output
  74
  75         dfReturn = None
  76         image_region_list = RegionTransform.transform_raw_sample(X)
  77         for image_data in image_region_list:
  78             #print(image_data)
  79             img = image_data['data']
  80             for r in image_data['regions']:  # loop through regions
  81                 x_max = min(r[0]+r[2], img.shape[1])
  82                 y_max = min(r[1]+r[3], img.shape[0])
  83                 if self.transform_mode=="pixelate":
  84                     img[r[1]:y_max, r[0]:x_max] = \
  85                         RegionTransform.pixelate_image(img[r[1]:y_max, r[0]:x_max])
  86
  87             # for now, we hard code to jpg output; TODO: add more encoding output (or try to match source?)
  88             img_binary = cv2.imencode(".jpg", img)[1].tostring()
  89             img_mime = 'image/jpeg'  # image_data['mime']
  90
  91             df = RegionTransform.generate_out_df(media_type=img_mime, bin_stream=img_binary)
  92             if dfReturn is None:  # create an NP container for all images
  93                 dfReturn = df.reindex_axis(self.output_names_, axis=1)
  94             else:
  95                 dfReturn = dfReturn.append(df, ignore_index=True)
  96             print("IMAGE {:} found {:} total rows".format(image_data['image'], len(df)))
  97         return dfReturn
  98
  99     @staticmethod
 100     def transform_raw_sample(raw_sample):
 101         """Method to transform raw samples into dict of image and regions"""
 102         raw_sample.sort_values([FaceDetectTransform.COL_IMAGE_IDX], ascending=True, inplace=True)
 103         groupImage = raw_sample.groupby(FaceDetectTransform.COL_IMAGE_IDX)
 104         return_set = []
 105
 106         for nameG, rowsG in groupImage:
 107             local_image = {'image': -1, 'data': b"", 'regions': [], 'mime': ''}
 108             image_row = rowsG[rowsG[FaceDetectTransform.COL_REGION_IDX]==FaceDetectTransform.VAL_REGION_IMAGE_ID]
 109             if len(image_row) < 1:  # must have at least one image set
 110                 print("Error: RegionTransform could not find a valid image reference for image set {:}".format(nameG))
 111                 continue
 112             if not len(image_row[FaceDetectTransform.COL_IMAGE_DATA]):  # must have valid image data
 113                 print("Error: RegionTransform expected image data, but found empty binary string {:}".format(nameG))
 114                 continue
 115             file_bytes = np.asarray(image_row[FaceDetectTransform.COL_IMAGE_DATA][0], dtype=np.uint8)
 116             local_image['data'] = cv2.imdecode(file_bytes, cv2.IMREAD_COLOR)
 117             local_image['image'] = nameG
 118             local_image['mime'] = image_row[FaceDetectTransform.COL_IMAGE_MIME]
 119
 120             # now proceed to loop around regions detected
 121             for index, row in rowsG.iterrows():
 122                 if row[FaceDetectTransform.COL_REGION_IDX]!=FaceDetectTransform.VAL_REGION_IMAGE_ID:  # skip bad regions
 123                     local_image['regions'].append([row[FaceDetectTransform.COL_FACE_X], row[FaceDetectTransform.COL_FACE_Y],
 124                                                    row[FaceDetectTransform.COL_FACE_W], row[FaceDetectTransform.COL_FACE_H]])
 125             return_set.append(local_image)
 126         return return_set
 127
 128     ################################################################
 129     # image processing routines (using opencv)
 130
 131     # http://www.jeffreythompson.org/blog/2012/02/18/pixelate-and-posterize-in-processing/
 132     @staticmethod
 133     def pixelate_image(img, blockSize=None):
 134         if not img.shape[0] or not img.shape[1]:
 135             return img
 136         if blockSize is None:
 137             blockSize = round(max(img.shape[0], img.shape[2]) / 8)
 138         ratio = (img.shape[1] / img.shape[0]) if img.shape[0] < img.shape[1] else (img.shape[0] / img.shape[1])
 139         blockHeight = round(blockSize * ratio)  # so that we cover all image
 140         for x in range(0, img.shape[0], blockSize):
 141             for y in range(0, img.shape[1], blockHeight):
 142                 max_x = min(x+blockSize, img.shape[0])
 143                 max_y = min(y+blockSize, img.shape[1])
 144                 fill_color = img[x,y] # img[x:max_x, y:max_y].mean()
 145                 img[x:max_x, y:max_y] = fill_color
 146         return img
 147