path: root/megapixels/app/server/tasks/demo.py

import app.settings.app_cfg as cfg
from app.server.tasks import celery

from celery.utils.log import get_task_logger
log = get_task_logger(__name__)

opt_size = (256, 256,)

@celery.task(bind=True)
def demo_task(self, uuid_name, fn):

  import sys
  import os
  from os.path import join
  from pathlib import Path
  import time

  import numpy as np
  import cv2 as cv
  import dlib
  from PIL import Image
  import matplotlib.pyplot as plt

  from app.utils import logger_utils, file_utils, im_utils, display_utils, draw_utils
  from app.utils import plot_utils
  from app.processors import face_detector, face_landmarks, face_age_gender, face_beauty
  # , face_emotion
  from app.models.data_store import DataStore

  # TODO add selective testing
  opt_gpu = -1
  opt_run_pose = True
  opt_run_2d_68 = True
  opt_run_3d_68 = True
  opt_run_3d_68 = True

  opt_gif_size = (256, 256,)
  opt_gif_frames = 15

  meta = {
    'step': 0,
    'total': 10,
    'message': 'Starting',
    'uuid': uuid_name,
    'data': { 'statistics': {} },
  }
  paths = []

  def step(msg, step=1):
    meta['message'] = msg
    meta['step'] += step
    log.debug('> {}'.format(msg))
    self.update_state(state='PROCESSING', meta=meta)

  def save_image(key, title, data):
    fn =  '{}_{}.jpg'.format(uuid_name, key)
    fpath = os.path.join(cfg.DIR_SITE_USER_CONTENT, fn)
    paths.append(fpath)
    cv.imwrite(fpath, data)

    meta['data'][key] = {
      'title': title,
      'url': os.path.join('/user_content/', fn),
    }

  step('Loading image')
  self.update_state(state='PROCESSING', meta=meta)

  # -------------------------------------------------
  # init here

  # load image
  im = cv.imread(fn)
  im_resized = im_utils.resize(im, width=opt_size[0], height=opt_size[1])
  os.remove(fn)

  # ----------------------------------------------------------------------------
  # detect face

  face_detector_instance = face_detector.DetectorDLIBCNN(gpu=opt_gpu)  # -1 for CPU
  step('Detecting face')
  st = time.time()
  bboxes = face_detector_instance.detect(im_resized, largest=True, pyramids=2)
  bbox = bboxes[0]
  dim = im_resized.shape[:2][::-1]
  bbox_dim = bbox.to_dim(dim)
  if not bbox:
    log.error('No face detected')
    meta['error'] = 'No face detected'
    self.update_state(state='FAILURE', meta=meta)
    return meta
  else:
    log.info(f'Detected face in {(time.time() - st):.2f}s')


  # ----------------------------------------------------------------------------
  # detect 3D landmarks
  
  step('Generating 3D Landmarks')
  log.info('loading 3D landmark generator files...')
  landmark_detector_3d_68 = face_landmarks.FaceAlignment3D_68(gpu=opt_gpu)  # -1 for CPU
  log.info('generating 3D landmarks...')
  st = time.time()
  points_3d_68 = landmark_detector_3d_68.landmarks(im_resized, bbox_dim.to_xyxy())
  log.info(f'generated 3D landmarks in {(time.time() - st):.2f}s')
  log.info('')

  # draw 3d landmarks
  im_landmarks_3d_68 = im_resized.copy()
  draw_utils.draw_landmarks3D(im_landmarks_3d_68, points_3d_68)
  draw_utils.draw_bbox(im_landmarks_3d_68, bbox_dim)

  save_image('landmarks_3d_68', '3D Landmarks', im_landmarks_3d_68)

  # ----------------------------------------------------------------------------
  # generate 3D GIF animation
  
  step('Generating GIF Animation')
  log.info('generating 3D animation...')

  fn =  '{}_{}.gif'.format(uuid_name, '3d')
  fp_out = os.path.join(cfg.DIR_SITE_USER_CONTENT, fn)
  paths.append(fp_out)

  st = time.time()
  plot_utils.generate_3d_landmark_anim(np.array(points_3d_68), fp_out, 
    size=opt_gif_size, num_frames=opt_gif_frames)
  log.info(f'Generated animation in {(time.time() - st):.2f}s')
  log.info(f'Saved to: {fp_out}')
  log.info('')

  meta['data']['points_3d_68'] = points_3d_68
  meta['data']['points_3d_68'] = {
    'title': '3D Animated GIF',
    'url': os.path.join('/user_content/', fn),
  }

  # ----------------------------------------------------------------------------
  # generate 68 point landmarks using dlib
  
  step('Generating 2D 68PT landmarks')
  log.info('initializing face landmarks 68 dlib...')
  from app.processors import face_landmarks
  landmark_detector_2d_68 = face_landmarks.Dlib2D_68()
  log.info('generating 2D 68PT landmarks...')
  st = time.time()
  points_2d_68 = landmark_detector_2d_68.landmarks(im_resized, bbox_dim)
  log.info(f'generated 2D 68PT face landmarks in {(time.time() - st):.2f}s')
  log.info('')

  # draw 2d landmarks
  im_landmarks_2d_68 = im_resized.copy()
  draw_utils.draw_landmarks2D(im_landmarks_2d_68, points_2d_68)
  draw_utils.draw_bbox(im_landmarks_2d_68, bbox_dim)
  save_image('landmarks_2d_68', '2D Landmarks', im_landmarks_2d_68)

  # ----------------------------------------------------------------------------
  # generate pose from 68 point 2D landmarks
  
  if opt_run_pose:
    step('Generating pose')
    log.info('initialize pose...')
    from app.processors import face_pose
    pose_detector = face_pose.FacePoseDLIB()
    log.info('generating pose...')
    st = time.time()
    pose_data = pose_detector.pose(points_2d_68, dim)
    log.info(f'generated pose {(time.time() - st):.2f}s')
    log.info('')

    im_pose = im_resized.copy()
    draw_utils.draw_pose(im_pose, pose_data['point_nose'], pose_data['points'])
    draw_utils.draw_degrees(im_pose, pose_data)
    save_image('pose', 'Pose', im_pose)

  # ----------------------------------------------------------------------------
  # age

  # real
  step('Running age predictor')
  age_real_predictor = face_age_gender.FaceAgeReal()
  st = time.time()
  age_real = age_real_predictor.predict(im_resized, bbox_dim)
  log.info(f'age real took: {(time.time()-st)/1000:.5f}s')
  meta['data']['statistics']['age_real'] = f'{(age_real):.2f}'
  
  # apparent
  age_apparent_predictor = face_age_gender.FaceAgeApparent()
  st = time.time()
  age_apparent = age_apparent_predictor.predict(im_resized, bbox_dim)
  log.info(f'age apparent took: {(time.time()-st)/1000:.5f}s')
  meta['data']['statistics']['age_apparent'] = f'{(age_apparent):.2f}'

  # gender
  step('Running gender predictor')
  gender_predictor = face_age_gender.FaceGender()
  st = time.time()
  gender = gender_predictor.predict(im_resized, bbox_dim)
  log.info(f'gender took: {(time.time()-st)/1000:.5f}s')
  meta['data']['statistics']['gender'] = f"M: {gender['m']:.2f}, F: {gender['f']:.2f}"

  # # ----------------------------------------------------------------------------
  # # emotion

  # emotion_predictor = face_emotion.FaceEmotion(gpu=opt_gpu)
  # emotion_score = emotion_predictor.emotion(im_resized, bbox_dim)
  # log.info(f'emotion score: {(100*emotion_score):.2f}')

  # im_emotion = im_resized.copy()
  # draw_utils.draw_bbox(im_emotion, bbox_dim)
  # txt = f'emotion score: {(100*emotion_score):.2f}'
  # draw_utils.draw_text(im_emotion, bbox_dim.pt_tl, txt)
  # save_image('emotion', 'Emotion', im_emotion)


  # ----------------------------------------------------------------------------
  # beauty

  # TODO fix Keras CPU/GPU device selection issue
  # NB: GPU visibility issues with dlib/keras
  # Wrap this with cuda toggle and run before init dlib GPU

  step('Running beauty predictor')
  device_cur = os.getenv('CUDA_VISIBLE_DEVICES', '')
  os.environ['CUDA_VISIBLE_DEVICES'] = ''
  beauty_predictor = face_beauty.FaceBeauty()
  os.environ['CUDA_VISIBLE_DEVICES'] = device_cur

  beauty_score = beauty_predictor.beauty(im_resized, bbox_dim)
  log.info(f'beauty score: {(100*beauty_score):.2f}')

  # # draw 2d landmarks
  # im_beauty = im_resized.copy()
  # draw_utils.draw_bbox(im_beauty, bbox_dim)
  # txt = f'Beauty score: {(100*beauty_score):.2f}'
  # draw_utils.draw_text(im_beauty, bbox_dim.pt_tl, txt)
  # save_image('beauty', 'Beauty', im_beauty)
  meta['data']['statistics']['beauty'] = f'{(100*beauty_score):.2f}'

  step('Done')
  
  # # 3DDFA
  # self.log.debug('Add depth')
  # self.log.debug('Add pncc')

  # # TODO
  # self.log.debug('Add 3D face model')
  # self.log.debug('Add face texture flat')
  # self.log.debug('Add ethnicity')

  log.debug('done!!')

  time.sleep(3)
  for path in paths:
    if os.path.exists(path):
      os.remove(path)
  
  meta['step'] = meta['total']
  meta['state'] = 'SUCCESS'
  return meta