doc_moi_tieng_Viet / preprocessing /crop_mouth_from_video.py
aiface's picture
Upload 12 files
327b68f
raw
history blame
15 kB
#! /usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright 2020 Imperial College London (Pingchuan Ma)
# Apache 2.0 (http://www.apache.org/licenses/LICENSE-2.0)
""" Crop Mouth ROIs from videos for lipreading"""
# from msilib.schema import File
from ast import Pass
import os
import cv2 # OpenCV 라이브러리
import glob # 리눅스식 경로 표기법을 사용하여 원하는 폴더/파일 리스트 얻음
import argparse # 명령행 인자를 파싱해주는 모듈
import numpy as np
from collections import deque # collections 모듈에 있는 데크 불러오기 # 데크: 스택과 큐를 합친 자료구조
from utils import * # utils.py 모듈에 있는 모든 함수 불러오기
from transform import * # transform.py 모듈에 있는 모든 함수 불러오기
import dlib # face landmark 찾는 라이브러리
import face_alignment # face landmark 찾는 라이브러리
from PIL import Image
# 인자값을 받아서 처리하는 함수
def load_args(default_config=None):
# 인자값을 받아서 처리하는 함수
parser = argparse.ArgumentParser(description='Lipreading Pre-processing')
# 입력받을 인자값 등록
# -- utils
parser.add_argument('--video-direc', default=None, help='raw video directory')
parser.add_argument('--video-format', default='.mp4', help='raw video format')
parser.add_argument('--landmark-direc', default=None, help='landmark directory')
parser.add_argument('--filename-path', default='./vietnamese_detected_face_30.csv', help='list of detected video and its subject ID')
parser.add_argument('--save-direc', default=None, help='the directory of saving mouth ROIs')
# -- mean face utils
parser.add_argument('--mean-face', default='./20words_mean_face.npy', help='mean face pathname')
# -- mouthROIs utils
parser.add_argument('--crop-width', default=96, type=int, help='the width of mouth ROIs')
parser.add_argument('--crop-height', default=96, type=int, help='the height of mouth ROIs')
parser.add_argument('--start-idx', default=48, type=int, help='the start of landmark index')
parser.add_argument('--stop-idx', default=68, type=int, help='the end of landmark index')
parser.add_argument('--window-margin', default=12, type=int, help='window margin for smoothed_landmarks')
# -- convert to gray scale
parser.add_argument('--convert-gray', default=False, action='store_true', help='convert2grayscale')
# -- test set only
parser.add_argument('--testset-only', default=False, action='store_true', help='process testing set only')
# 입력받은 인자값을 args에 저장 (type: namespace)
args = parser.parse_args()
return args
args = load_args() # args 파싱 및 로드
# -- mean face utils
STD_SIZE = (256, 256)
mean_face_landmarks = np.load(args.mean_face) # 20words_mean_face.npy
stablePntsIDs = [33, 36, 39, 42, 45]
# 영상에서 랜드마크 받아서 입술 잘라내기
def crop_patch( video_pathname, landmarks):
"""Crop mouth patch
:param str video_pathname: pathname for the video_dieo # 영상 위치
:param list landmarks: interpolated landmarks # 보간된 랜드마크
"""
frame_idx = 0 # 프레임 인덱스 번호 0 으로 초기화
frame_gen = read_video(video_pathname) # 비디오 불러오기
# 무한 반복
while True:
try:
frame = frame_gen.__next__() ## -- BGR # 이미지 프레임 하나씩 불러오기
except StopIteration: # 더 이상 next 요소가 없으면 StopIterraion Exception 발생
break # while 빠져나가기
if frame_idx == 0: # 프레임 인덱스 번호가 0일 경우
q_frame, q_landmarks = deque(), deque() # 데크 생성
sequence = []
q_landmarks.append(landmarks[frame_idx]) # 프레임 인덱스 번호에 맞는 랜드마크 정보 추가
q_frame.append(frame) # 프레임 정보 추가
if len(q_frame) == args.window_margin:
smoothed_landmarks = np.mean(q_landmarks, axis=0) # 각 그룹의 같은 원소끼리 평균
cur_landmarks = q_landmarks.popleft() # 데크 제일 왼쪽 값 꺼내기
cur_frame = q_frame.popleft() # 데크 제일 왼쪽 값 꺼내기
# -- affine transformation # 아핀 변환
trans_frame, trans = warp_img( smoothed_landmarks[stablePntsIDs, :],
mean_face_landmarks[stablePntsIDs, :],
cur_frame,
STD_SIZE)
trans_landmarks = trans(cur_landmarks)
# -- crop mouth patch # 입술 잘라내기
sequence.append( cut_patch( trans_frame,
trans_landmarks[args.start_idx:args.stop_idx],
args.crop_height//2,
args.crop_width//2,))
if frame_idx == len(landmarks)-1:
while q_frame:
cur_frame = q_frame.popleft() # 데크 제일 왼쪽 값 꺼내기
# -- transform frame # 프레임 변환
trans_frame = apply_transform( trans, cur_frame, STD_SIZE)
# -- transform landmarks # 랜드마크 변환
trans_landmarks = trans(q_landmarks.popleft())
# -- crop mouth patch # 입술 잘라내기
sequence.append( cut_patch( trans_frame,
trans_landmarks[args.start_idx:args.stop_idx],
args.crop_height//2,
args.crop_width//2,))
return np.array(sequence) # 입술 numpy 반환
frame_idx += 1 # 프레임 인덱스 번호 증가
return None
# 랜드마크 보간
def landmarks_interpolate(landmarks):
"""Interpolate landmarks
param list landmarks: landmarks detected in raw videos # 원본 영상 데이터에서 검출한 랜드마크
"""
valid_frames_idx = [idx for idx, _ in enumerate(landmarks) if _ is not None] # 랜드마크 번호 list 생성
# 랜드마크 번호 list 가 비어있다면
if not valid_frames_idx:
return None
# 1부터 (랜드마크 번호 list 개수-1)만큼 for 문 반복
for idx in range(1, len(valid_frames_idx)):
if valid_frames_idx[idx] - valid_frames_idx[idx-1] == 1: # 현재 랜드마크 번호 - 이전 랜드마크 번호 == 1 일 경우
continue # 코드 실행 건너뛰기
else: # 아니라면
landmarks = linear_interpolate(landmarks, valid_frames_idx[idx-1], valid_frames_idx[idx]) # 랜드마크 업데이트(보간)
valid_frames_idx = [idx for idx, _ in enumerate(landmarks) if _ is not None] # 랜드마크 번호 list 생성
# -- Corner case: keep frames at the beginning or at the end failed to be detected. # 시작 또는 끝 프레임을 보관하지 못함
if valid_frames_idx:
landmarks[:valid_frames_idx[0]] = [landmarks[valid_frames_idx[0]]] * valid_frames_idx[0] # 랜드마크 첫번째 프레임 정보 저장
landmarks[valid_frames_idx[-1]:] = [landmarks[valid_frames_idx[-1]]] * (len(landmarks) - valid_frames_idx[-1]) # 랜드마크 마지막 프레임 정보 저장
valid_frames_idx = [idx for idx, _ in enumerate(landmarks) if _ is not None] # 랜드마크 번호 list 생성
# 랜드마크 번호 list 개수 == 보간한 랜드마크 개수 확인, 아니면 AssertionError 메시지를 띄움
assert len(valid_frames_idx) == len(landmarks), "not every frame has landmark" # 원하는 조건의 변수값을 보증하기 위해 사용
return landmarks # 랜드마크 반환
def get_yield(output_video):
for frame in output_video:
yield frame
lines = open(args.filename_path).read().splitlines() # 문자열을 '\n' 기준으로 쪼갠 후 list 생성
lines = list(filter(lambda x: 'test' == x.split('/')[-2], lines)) if args.testset_only else lines # args.testset_only 값이 있다면 test 폴더 속 파일명만 불러와서 list 생성, 아니라면 원래 lines 그대로 값 유지
# lines 개수만큼 반복문 실행
for filename_idx, line in enumerate(lines):
# 파일명, 사람id
filename, person_id = line.split(',')
print('idx: {} \tProcessing.\t{}'.format(filename_idx, filename)) # 파일 인덱스번호, 파일명 출력
video_pathname = os.path.join(args.video_direc, filename+args.video_format) # 영상디렉토리 + 파일명.비디오포맷/
landmarks_pathname = os.path.join(args.landmark_direc, filename+'.npz') # 저장디렉토리 + 랜드마크 파일명.npz
dst_pathname = os.path.join( args.save_direc, filename+'.npz') # 저장디렉토리 + 결과영상 파일명.npz
# 파일이 있는지 확인, 없으면 AssertionError 메시지를 띄움
assert os.path.isfile(video_pathname), "File does not exist. Path input: {}".format(video_pathname) # 원하는 조건의 변수값을 보증하기 위해 사용
# video 에 대한 face landmark npz 파일이 없고 영상 확장자 avi 인 경우 dlib 으로 직접 npz 파일 생성
if not os.path.exists(landmarks_pathname) and video_pathname.split('.')[-1] == 'mp4':
# dlib 사용해서 face landmark 찾기
def get_face_landmark(img):
detector_hog = dlib.get_frontal_face_detector()
dlib_rects = detector_hog(img, 1)
model_path = os.path.dirname(os.path.abspath(__file__)) + '/shape_predictor_68_face_landmarks.dat'
landmark_predictor = dlib.shape_predictor(model_path)
# dlib 으로 face landmark 찾기
list_landmarks = []
for dlib_rect in dlib_rects:
points = landmark_predictor(img, dlib_rect)
list_points = list(map(lambda p: (p.x, p.y), points.parts()))
list_landmarks.append(list_points)
input_width, input_height = img.shape
output_width, output_height = (256, 256)
width_rate = input_width / output_width
height_rate = input_height / output_height
img_rate = [(width_rate, height_rate)]*68
face_rate = np.array(img_rate)
eye_rate = np.array(img_rate[36:48])
# face landmark list 가 비어있지 않은 경우
if list_landmarks:
for dlib_rect, landmark in zip(dlib_rects, list_landmarks):
face_landmark = np.array(landmark) # face landmark
eye_landmark = np.array(landmark[36:48]) # eye landmark
return face_landmark, eye_landmark
# face landmark list 가 비어있는 경우
else:
landmark = [(0.0, 0.0)] * 68
face_landmark = np.array(landmark) # face landmark
eye_landmark = np.array(landmark[36:48]) # eye landmark
return face_landmark, eye_landmark
target_frames = 29 # 원하는 프레임 개수
video = videoToArray(video_pathname, is_gray=args.convert_gray) # 영상 정보 앞에 영상 프레임 개수를 추가한 numpy
output_video = frameAdjust(video, target_frames) # frame sampling (프레임 개수 맞추기)
multi_sub_landmarks = []
person_landmarks = []
frame_landmarks = []
for frame_idx, frame in enumerate(get_yield(output_video)):
print(f'\n ------------frame {frame_idx}------------ ')
facial_landmarks, eye_landmarks = get_face_landmark(frame) # dlib 사용해서 face landmark 찾기
person_landmarks = {
'id': 0,
'most_recent_fitting_scores': np.array([2.0,2.0,2.0]),
'facial_landmarks': facial_landmarks,
'roll': 7,
'yaw': 3.5,
'eye_landmarks': eye_landmarks,
'fitting_scores_updated': True,
'pitch': -0.05
}
frame_landmarks.append(person_landmarks)
multi_sub_landmarks.append(np.array(frame_landmarks.copy(), dtype=object))
multi_sub_landmarks = np.array(multi_sub_landmarks) # list to numpy
save2npz(landmarks_pathname, data=multi_sub_landmarks) # face landmark npz 저장
print('\n ------------ save npz ------------ \n')
# video 에 대한 face landmark npz 파일이 있는 경우
else:
# 파일이 있는지 확인, 없으면 AssertionError 메시지를 띄움
assert os.path.isfile(landmarks_pathname), "File does not exist. Path input: {}".format(landmarks_pathname) # 원하는 조건의 변수값을 보증하기 위해 사용
# 파일이 존재할 경우
if os.path.exists(dst_pathname):
continue # 코드 실행 건너뛰기
multi_sub_landmarks = np.load( landmarks_pathname, allow_pickle=True)['data'] # numpy 파일 열기
landmarks = [None] * len( multi_sub_landmarks) # 랜드마크 변수 초기화
for frame_idx in range(len(landmarks)):
try:
landmarks[frame_idx] = multi_sub_landmarks[frame_idx][int(person_id)]['facial_landmarks'].astype(np.float64) # 프레임 인덱스 번호에서 사람id의 얼굴 랜드마크 정보 가져오기
except IndexError: # 해당 인덱스 번호에 깂이 없으면 IndexError 발생
continue # 코드 실행 건너뛰기
# face landmark 가 [(0,0)]*68 이 아니면 랜드마크 보간 후 npz 파일 생성
landmarks_empty_list = []
landmarks_empty = [(0, 0)]*68
landmarks_empty = np.array(landmarks_empty, dtype=object)
for i in range(len(landmarks_empty)):
landmarks_empty_list.append(landmarks_empty.copy())
condition = landmarks != landmarks_empty_list
if condition:
# -- pre-process landmarks: interpolate frames not being detected.
preprocessed_landmarks = landmarks_interpolate(landmarks) # 랜드마크 보간
# 변수가 비어있지 않다면
if not preprocessed_landmarks:
continue # 코드 실행 건너뛰기
# -- crop
sequence = crop_patch(video_pathname, preprocessed_landmarks) # 영상에서 랜드마크 받아서 입술 잘라내기
# sequence가 비어있는지 확인, 비어있으면 AssertionError 메시지를 띄움
assert sequence is not None, "cannot crop from {}.".format(filename) # 원하는 조건의 변수값을 보증하기 위해 사용
# -- save
data = convert_bgr2gray(sequence) if args.convert_gray else sequence[...,::-1] # gray 변환
save2npz(dst_pathname, data=data) # 데이터를 npz 형식으로 저장
print('Done.')