연예인 얼굴 인식 서비스 데이터 수집

1. 개요

 페이스북의 글을 읽던 중 Google Cloud Platform Korea User Group의 조대협님께서 작성하신 글을 보고 흥미가 있어 따라하게 되었고, 거기에 대한 과정을 작성한 내용입니다. 최종 목표는 CNN(Convolutional Neural Network)와 뎉서 플로우를 이용한 얼굴인식 서비스 구현까지입니다. 이번 글은 기나긴 여정의 첫번째이며, 원본 글의 첫번째 내용인 학습할 이미지 데이터를 받아오는 부분입니다.

원본 글을 조대협님의 Blog(http://bcho.tistory.com/1166)에 가시면 확인하실 수 있습니다.

2. 공개 데이터 다운

 머신러닝을 하기위해서는 일정량 이상의 데이터가 필요한데 특정 인물의 사진을 적정량을 일일이 수집하기엔 많은 시간이 들어갑니다. 그렇기 때문에 공개되어 있는 데이터를 이용하도록 하겠습니다.

위 싸이트(http://www.cs.columbia.edu/CAVE/databases/pubfig/)에서 Download로 가시면 중단 쯤에 dev_urls.txt라는 이미지 URL이 담긴 텍스트 파일이 있습니다. 해당 파일은 사람이름, 번호, URL, 사진 크기, 체크섬을 필드로 가지고 있습니다. 

dev_urls.txt 첨부 파일은 제가 사용했던 파일입니다.

3. 이미지 다운

 앞서 다운받은 txt 파일에는 16,336개의 이미지에 대한 url이 있습니다. 손으로 하나하나 다운받기엔 무리가 있고, 조대협님의 블로그를 토대로 따라하고 있기 때문에 조대협님의 동료분께서 작성하신 프로그램을 이용하겠습니다. 소스코드는  https://github.com/wwoo/tf_face/blob/master/tf/face_extract/pubfig_get.py에서 다운받으실 수 있습니다.

import sys import threading import os import socket import urllib2 from Queue import Queue from PIL import Image NUM_THREADS = 4 URL_TIMEOUT = 4 IMAGE_CROP = False ESCAPE_SPACES = False class LabelWriterThread(threading.Thread): def __init__(self, queue, dest_dir): super(LabelWriterThread, self).__init__() self.queue = queue self.daemon = True self.dest_dir = dest_dir def run(self): file_path = os.path.join(dest_dir, "manifest.txt") f = open(file_path, 'w') while True: f.write(self.queue.get() + "\n") self.queue.task_done() f.close() class DownloadThread(threading.Thread): def __init__(self, url_queue, print_queue, classes, image_crop, dest_dir): super(DownloadThread, self).__init__() socket.setdefaulttimeout(URL_TIMEOUT) self.url_queue = url_queue self.classes = classes self.dest_dir = dest_dir self.daemon = True self.image_crop = image_crop self.print_queue = print_queue def run(self): while True: dict = self.url_queue.get() try: name = dict["url"].split('/')[-1] person_dir = os.path.join(self.dest_dir, dict["rel_dir"]) if not os.path.exists(person_dir): os.makedirs(person_dir) dest_file = os.path.join(person_dir, name) self.download_url(dest_file, dict["url"]) if os.path.isfile(dest_file): if self.image_crop: crop_dir = os.path.join(person_dir, "crop") if not os.path.exists(crop_dir): os.makedirs(crop_dir) out_filename = os.path.join(crop_dir, 'crop_' + name) self.crop_image(dest_file, out_filename, dict["crop_dims"]) if ESCAPE_SPACES: out_filename = out_filename.replace(' ', '\ ') self.print_queue.put(out_filename + '|0|0|0') else: if ESCAPE_SPACES: dest_file = dest_file.replace(' ', '\ ') self.print_queue.put(dest_file) except Exception, e: print("[%s] Error: %s" % (self.ident, e)) self.url_queue.task_done() def download_url(self, dest_file, url): try: print("[%s] Downloading %s -> %s" % (self.ident, url, dest_file)) u = urllib2.urlopen(url) with open(dest_file, "wb") as f: f.write(u.read()) f.close() except urllib2.HTTPError, e: print("[%s] HTTP Error: %s %s" % (self.ident, e.code, url)) except urllib2.URLError, e: print("[%s] URL Error: %s %s" % (self.ident, e.reason, url)) def crop_image(self, dest_file, crop_file, crop_dims): print("[%s] Cropping %s -> %s" % (self.ident, dest_file, crop_file)) c = crop_dims.split(',') img = Image.open(dest_file) img2 = img.crop((float(c[0]), float(c[1]), float(c[2]), float(c[3]))) img2.save(crop_file) def read_url_file(file_path): f = open(file_path) queue = Queue() classes = {} for line in f: if not line.startswith('#'): tokens = line.split('\t') queue.put({ "rel_dir": tokens[0], "url": tokens[2], "crop_dims": tokens[3]}) if not tokens[0] in classes: classes[tokens[0]] = len(classes) f.close() return queue, classes def write_class_file(classes, file): f = open(file, 'w') for key in classes: f.write(key + "\n") f.close() if __name__ == "__main__": if len(sys.argv) <> 3: print("Usage: pub_fig_get.py <url_file> <dest_folder>") exit(0) url_file = sys.argv[1] dest_dir = sys.argv[2] class_file = os.path.join(dest_dir, "classes.txt") url_queue, classes = read_url_file(url_file) write_class_file(classes, class_file) print_queue = Queue() for i in range(NUM_THREADS): t = DownloadThread(url_queue, print_queue, classes, IMAGE_CROP, dest_dir) t.start() t = LabelWriterThread(print_queue, dest_dir) t.start() url_queue.join() print_queue.join()

사용법은 https://github.com/wwoo/tf_face에 잘 설명되어 있으며 main함수의 파라메터로 URL이 담긴 txt 파일과 이미지가 저장될 폴더 위치입니다.

$> python tf/face_extract/pubfig_get.py eval_urls.txt ./data

4. 이미지 선별

 구글 얼굴 추출 API를 사용하기 위해서는 하나의 사진에 2명의 얼굴이 있으면 안됩니다. 또한 깨진 사진이나 URL이 없어져 다운에 실패한 사진들도 걸러주셔야합니다.

5. Google Vision

 이미지 선별이 끝나셨으면 Google Console로 갑니다. 프로젝트를 생성해 주신 다음에 API 관리자 페이지(https://console.cloud.google.com/apis)로 갑니다. 그리고 라이브러리 항목으로 이동해 주신 다음 Vision API를 활성화 시킵니다.

이후 사용자 인증 정보 항목을 눌러주시고 아래와 같이 서비스 계정 키 항목을 눌러 Json 파일을 다운받습니다.

6. Google Vision Example Code

 우선 예제 코드는 https://github.com/bwcho75/facerecognition/blob/master/com/terry/face/extract/crop_face.py에서 확인하실 수 있습니다.

from googleapiclient import discovery from oauth2client.client import GoogleCredentials import sys import io import base64 from PIL import Image from PIL import ImageDraw from genericpath import isfile import os from oauth2client.service_account import ServiceAccountCredentials NUM_THREADS = 10 MAX_RESULTS = 1 IMAGE_SIZE = 96,96 class FaceDetector(): def __init__(self): # initialize library #credentials = GoogleCredentials.get_application_default() scopes = ['https://www.googleapis.com/auth/cloud-platform'] credentials = ServiceAccountCredentials.from_json_keyfile_name( './terrycho-ml-80abc460730c.json', scopes=scopes) self.service = discovery.build('vision', 'v1', credentials=credentials) #print ("Getting vision API client : %s" ,self.service) #def extract_face(selfself,image_file,output_file): def detect_face(self,image_file): try: with io.open(image_file,'rb') as fd: image = fd.read() batch_request = [{ 'image':{ 'content':base64.b64encode(image).decode('utf-8') }, 'features':[{ 'type':'FACE_DETECTION', 'maxResults':MAX_RESULTS, }] }] fd.close() request = self.service.images().annotate(body={ 'requests':batch_request, }) response = request.execute() if 'faceAnnotations' not in response['responses'][0]: print('[Error] %s: Cannot find face ' % image_file) return None face = response['responses'][0]['faceAnnotations'] box = face[0]['fdBoundingPoly']['vertices'] left = box[0]['x'] top = box[1]['y'] right = box[2]['x'] bottom = box[2]['y'] rect = [left,top,right,bottom] print("[Info] %s: Find face from in position %s" % (image_file,rect)) return rect except Exception as e: print('[Error] %s: cannot process file : %s' %(image_file,str(e)) ) def rect_face(self,image_file,rect,outputfile): try: fd = io.open(image_file,'rb') image = Image.open(fd) draw = ImageDraw.Draw(image) draw.rectangle(rect,fill=None,outline="green") image.save(outputfile) fd.close() print('[Info] %s: Mark face with Rect %s and write it to file : %s' %(image_file,rect,outputfile) ) except Exception as e: print('[Error] %s: Rect image writing error : %s' %(image_file,str(e)) ) def crop_face(self,image_file,rect,outputfile): try: fd = io.open(image_file,'rb') image = Image.open(fd) crop = image.crop(rect) im = crop.resize(IMAGE_SIZE,Image.ANTIALIAS) im.save(outputfile,"JPEG") fd.close() print('[Info] %s: Crop face %s and write it to file : %s' %(image_file,rect,outputfile) ) except Exception as e: print('[Error] %s: Crop image writing error : %s' %(image_file,str(e)) ) def getfiles(self,src_dir): files = [] for f in os.listdir(src_dir): if isfile(os.path.join(src_dir,f)): if not f.startswith('.'): files.append(os.path.join(src_dir,f)) return files def rect_faces_dir(self,src_dir,des_dir): if not os.path.exists(des_dir): os.makedirs(des_dir) files = self.getfiles(src_dir) for f in files: des_file = os.path.join(des_dir,os.path.basename(f)) rect = self.detect_face(f) if rect != None: self.rect_face(f, rect, des_file) def crop_faces_dir(self,src_dir,des_dir): # training data will be written in $des_dir/training # validation data will be written in $des_dir/validate des_dir_training = os.path.join(des_dir,'training') des_dir_validate = os.path.join(des_dir,'validate') if not os.path.exists(des_dir): os.makedirs(des_dir) if not os.path.exists(des_dir_training): os.makedirs(des_dir_training) if not os.path.exists(des_dir_validate): os.makedirs(des_dir_validate) path,folder_name = os.path.split(src_dir) label = folder_name # create label file. it will contains file location # and label for each file training_file = open('training_file.txt','a') validate_file = open('validate_file.txt','a') files = self.getfiles(src_dir) cnt = 0 for f in files: rect = self.detect_face(f) # replace ',' in file name to '.' # because ',' is used for deliminator of image file name and its label des_file_name = os.path.basename(f) des_file_name = des_file_name.replace(',','_') if rect != None: # 70% of file will be stored in training data directory if(cnt < 8): des_file = os.path.join(des_dir_training,des_file_name) self.crop_face(f, rect, des_file ) training_file.write("%s,%s\n"%(des_file,label) ) # 30% of files will be stored in validation data directory else: # for validation data des_file = os.path.join(des_dir_validate,des_file_name) self.crop_face(f, rect, des_file) validate_file.write("%s,%s\n"%(des_file,label) ) if(cnt>9): cnt = 0 cnt = cnt + 1 training_file.close() validate_file.close() def getdirs(self,dir): dirs = [] for f in os.listdir(dir): f=os.path.join(dir,f) if os.path.isdir(f): if not f.startswith('.'): dirs.append(f) return dirs def crop_faces_rootdir(self,src_dir,des_dir): # crop file from sub-directoris in src_dir dirs = self.getdirs(src_dir) #list sub directory for d in dirs: print('[INFO] : ### Starting cropping in directory %s ###'%d) self.crop_faces_dir(d, des_dir) #loop and run face crop def main(argv): srcdir= argv[1] desdir = argv[2] detector = FaceDetector() detector.crop_faces_rootdir(srcdir, desdir) #detector.crop_faces_dir(inputfile,outputfile) #rect = detector.detect_face(inputfile) #detector.rect_image(inputfile, rect, outputfile) #detector.crop_face(inputfile, rect, outputfile) if __name__ == "__main__": main(sys.argv)

소스 코드를 잠시 보도록 하겠습니다.

1. Main

Main 함수입니다. 사진 파일이 있는 폴더와 얼굴부분만 잘라낸 사진을 저장할 폴더의 위치가 파라메터로 들어갑니다.

$> python com/terry/face/extract/crop_face.py “원본 파일이있는 디렉토리" “아웃풋 디렉토리"

2. FaceDetector Init

Google Vision API 초기화 부분입니다. 중간에 json 파일 경로를 넣는 부분이 있는데 자신의 json 파일을 집어 넣으시면 됩니다.

3. crop_faces_rootdir

원본 사진 폴더 위치와 저장할 폴더 위지가 파라메터로 들어가며 원본 사진 폴더에서 특정 인물의 사진이 모여 있는 폴더를 하나 하나 읽어 crop_faces_dir 함수를 실행시킵니다.

4. crop_faces_dir

해당 프로그램의 메인 함수입니다. 폴더의 각각의 사진 파일을 읽어 얼굴부분을 자르는 함수인 detect_face()를 실행시킵니다. 또한 작업한 사진 파일의 로그를 남길 뿐만 나이라 결과물을 일정 비율에 따라 다른 위치에 저장합니다.

5. detect_face

구글 Vision API를 사용하는 함수가 있는 함수입니다. 이미지 파일을 읽어 batch_request라는 문자열을 만들고 base64 인코딩 방식으로 인코딩을 하여 패킷을 전송합니다. 보낼 때 VISION API 중 얼굴 인식을 사용할것이기 때문에 'FACE_DETECTION'으로 정의합니다.

이후 response가 올텐데 response는 배열로 오며 다음과 같은 항목에 얼굴 데이터가 들어있습니다. Array['responses'][0]['faceAnnotations'] 여기에서 ['fdBoundingPoly']['vertices']는 얼굴 영역의 각 모서리 위치 정보가 들어있고 저희는 이 정보를 저장 후 리턴합니다.

6. crop_face

앞 함수에서 찾은 얼굴 위치 정보를 가지고 원본 사진내에서 얼굴 영역만 잘라내 새로운 이미지 파일을 만듭니다.

7. 결과

 소스코드 수정 없이 실행만 잘 시키시면 아래와 같은 결과를 얻을 수 있습니다.