File size: 12,786 Bytes
c930369
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
from argparse import ArgumentParser
from functools import lru_cache
from types import ModuleType
from typing import List

import cv2
import numpy

import facefusion.choices
import facefusion.jobs.job_manager
import facefusion.jobs.job_store
from facefusion import config, content_analyser, face_classifier, face_detector, face_landmarker, face_masker, face_recognizer, inference_manager, logger, state_manager, translator, video_manager
from facefusion.common_helper import create_int_metavar, get_middle
from facefusion.download import conditional_download_hashes, conditional_download_sources, resolve_download_url
from facefusion.face_creator import scale_face
from facefusion.face_helper import merge_matrix, paste_back, scale_face_landmark_5, warp_face_by_face_landmark_5
from facefusion.face_masker import create_box_mask, create_occlusion_mask
from facefusion.face_selector import select_faces
from facefusion.filesystem import in_directory, is_image, is_video, resolve_relative_path, same_file_extension
from facefusion.processors.modules.age_modifier import choices as age_modifier_choices
from facefusion.processors.modules.age_modifier.types import AgeModifierDirection, AgeModifierInputs
from facefusion.processors.types import ProcessorOutputs
from facefusion.program_helper import find_argument_group
from facefusion.thread_helper import thread_semaphore
from facefusion.types import ApplyStateItem, Args, DownloadScope, Face, InferencePool, ModelOptions, ModelSet, ProcessMode, VisionFrame
from facefusion.vision import match_frame_color, read_static_image, read_static_video_frame


@lru_cache()
def create_static_model_set(download_scope : DownloadScope) -> ModelSet:
	return\
	{
		'fran':
		{
			'__metadata__':
			{
				'vendor': 'ry-lu',
				'license': 'mit',
				'year': 2024
			},
			'hashes':
			{
				'age_modifier':
				{
					'url': resolve_download_url('models-3.6.0', 'fran.hash'),
					'path': resolve_relative_path('../.assets/models/fran.hash')
				}
			},
			'sources':
			{
				'age_modifier':
				{
					'url': resolve_download_url('models-3.6.0', 'fran.onnx'),
					'path': resolve_relative_path('../.assets/models/fran.onnx')
				}
			},
			'templates':
			{
				'target': 'ffhq_512',
			},
			'sizes':
			{
				'target': (1024, 1024),
			},
			'mean': [ 0.0, 0.0, 0.0 ],
			'standard_deviation': [ 1.0, 1.0, 1.0 ]
		},
		'styleganex_age':
		{
			'__metadata__':
			{
				'vendor': 'williamyang1991',
				'license': 'S-Lab-1.0',
				'year': 2023
			},
			'hashes':
			{
				'age_modifier':
				{
					'url': resolve_download_url('models-3.1.0', 'styleganex_age.hash'),
					'path': resolve_relative_path('../.assets/models/styleganex_age.hash')
				}
			},
			'sources':
			{
				'age_modifier':
				{
					'url': resolve_download_url('models-3.1.0', 'styleganex_age.onnx'),
					'path': resolve_relative_path('../.assets/models/styleganex_age.onnx')
				}
			},
			'templates':
			{
				'target': 'ffhq_512',
				'target_with_background': 'styleganex_384'
			},
			'sizes':
			{
				'target': (256, 256),
				'target_with_background': (384, 384)
			},
			'mean': [ 0.5, 0.5, 0.5 ],
			'standard_deviation': [ 0.5, 0.5, 0.5 ]
		}
	}


def get_inference_pool() -> InferencePool:
	model_names = [ state_manager.get_item('age_modifier_model') ]
	model_source_set = get_model_options().get('sources')

	return inference_manager.get_inference_pool(__name__, model_names, model_source_set)


def clear_inference_pool() -> None:
	model_names = [ state_manager.get_item('age_modifier_model') ]
	inference_manager.clear_inference_pool(__name__, model_names)


def get_model_options() -> ModelOptions:
	model_name = state_manager.get_item('age_modifier_model')
	return create_static_model_set('full').get(model_name)


def register_args(program : ArgumentParser) -> None:
	group_processors = find_argument_group(program, 'processors')
	if group_processors:
		group_processors.add_argument('--age-modifier-model', help = translator.get('help.model', __package__), default = config.get_str_value('processors', 'age_modifier_model', 'fran'), choices = age_modifier_choices.age_modifier_models)
		group_processors.add_argument('--age-modifier-direction', help = translator.get('help.direction', __package__), type = int, default = config.get_int_value('processors', 'age_modifier_direction', '0'), choices = age_modifier_choices.age_modifier_direction_range, metavar = create_int_metavar(age_modifier_choices.age_modifier_direction_range))
		facefusion.jobs.job_store.register_step_keys([ 'age_modifier_model', 'age_modifier_direction' ])


def apply_args(args : Args, apply_state_item : ApplyStateItem) -> None:
	apply_state_item('age_modifier_model', args.get('age_modifier_model'))
	apply_state_item('age_modifier_direction', args.get('age_modifier_direction'))


def get_common_modules() -> List[ModuleType]:
	return [ content_analyser, face_classifier, face_detector, face_landmarker, face_masker, face_recognizer ]


def pre_check() -> bool:
	model_hash_set = get_model_options().get('hashes')
	model_source_set = get_model_options().get('sources')

	for common_module in get_common_modules():
		if not common_module.pre_check():
			return False

	return conditional_download_hashes(model_hash_set) and conditional_download_sources(model_source_set)


def pre_process(mode : ProcessMode) -> bool:
	if mode in [ 'output', 'preview' ] and not is_image(state_manager.get_item('target_path')) and not is_video(state_manager.get_item('target_path')):
		logger.error(translator.get('choose_image_or_video_target') + translator.get('exclamation_mark'), __name__)
		return False
	if mode == 'output' and not in_directory(state_manager.get_item('output_path')):
		logger.error(translator.get('specify_image_or_video_output') + translator.get('exclamation_mark'), __name__)
		return False
	if mode == 'output' and not same_file_extension(state_manager.get_item('target_path'), state_manager.get_item('output_path')):
		logger.error(translator.get('match_target_and_output_extension') + translator.get('exclamation_mark'), __name__)
		return False
	return True


def post_process() -> None:
	read_static_image.cache_clear()
	read_static_video_frame.cache_clear()
	video_manager.clear_video_pool()

	if state_manager.get_item('video_memory_strategy') in [ 'strict', 'moderate' ]:
		clear_inference_pool()

	if state_manager.get_item('video_memory_strategy') == 'strict':
		for common_module in get_common_modules():
			common_module.clear_inference_pool()


def modify_age(target_face : Face, temp_vision_frame : VisionFrame) -> VisionFrame:
	model_templates = get_model_options().get('templates')
	model_sizes = get_model_options().get('sizes')
	face_landmark_5 = target_face.landmark_set.get('5/68').copy()
	crop_vision_frame, affine_matrix = warp_face_by_face_landmark_5(temp_vision_frame, face_landmark_5, model_templates.get('target'), model_sizes.get('target'))

	if state_manager.get_item('age_modifier_model') == 'fran':
		box_mask = create_box_mask(crop_vision_frame, state_manager.get_item('face_mask_blur'), (0, 0, 0, 0))
		crop_masks =\
		[
			box_mask
		]

		if 'occlusion' in state_manager.get_item('face_mask_types'):
			occlusion_mask = create_occlusion_mask(crop_vision_frame)
			crop_masks.append(occlusion_mask)

		crop_vision_frame = prepare_vision_frame(crop_vision_frame)
		target_age = numpy.mean(target_face.age)
		age_modifier_direction = numpy.array([ target_age, target_age + state_manager.get_item('age_modifier_direction') ], dtype = numpy.float32) / 100
		age_modifier_direction = age_modifier_direction.clip(0, 1)
		crop_vision_frame = forward(crop_vision_frame, crop_vision_frame, age_modifier_direction)
		crop_vision_frame = normalize_vision_frame(crop_vision_frame)
		crop_mask = numpy.minimum.reduce(crop_masks).clip(0, 1)
		paste_vision_frame = paste_back(temp_vision_frame, crop_vision_frame, crop_mask, affine_matrix)
		return paste_vision_frame

	if state_manager.get_item('age_modifier_model') == 'styleganex_age':
		extend_face_landmark_5 = scale_face_landmark_5(face_landmark_5, 0.875)
		extend_vision_frame, extend_affine_matrix = warp_face_by_face_landmark_5(temp_vision_frame, extend_face_landmark_5, model_templates.get('target_with_background'), model_sizes.get('target_with_background'))
		extend_vision_frame_raw = extend_vision_frame.copy()
		box_mask = create_box_mask(extend_vision_frame, state_manager.get_item('face_mask_blur'), (0, 0, 0, 0))
		crop_masks =\
		[
			box_mask
		]

		if 'occlusion' in state_manager.get_item('face_mask_types'):
			occlusion_mask = create_occlusion_mask(crop_vision_frame)
			temp_matrix = merge_matrix([ extend_affine_matrix, cv2.invertAffineTransform(affine_matrix) ])
			occlusion_mask = cv2.warpAffine(occlusion_mask, temp_matrix, model_sizes.get('target_with_background'))
			crop_masks.append(occlusion_mask)

		crop_vision_frame = prepare_vision_frame(crop_vision_frame)
		extend_vision_frame = prepare_vision_frame(extend_vision_frame)
		age_modifier_direction = numpy.array(numpy.interp(state_manager.get_item('age_modifier_direction'), [ -100, 100 ], [ 2.5, -2.5 ])).astype(numpy.float32)
		extend_vision_frame = forward(crop_vision_frame, extend_vision_frame, age_modifier_direction)
		extend_vision_frame = normalize_extend_frame(extend_vision_frame)
		extend_vision_frame = match_frame_color(extend_vision_frame_raw, extend_vision_frame)
		extend_affine_matrix *= (model_sizes.get('target')[0] * 4) / model_sizes.get('target_with_background')[0]
		crop_mask = numpy.minimum.reduce(crop_masks).clip(0, 1)
		crop_mask = cv2.resize(crop_mask, (model_sizes.get('target')[0] * 4, model_sizes.get('target')[1] * 4))
		paste_vision_frame = paste_back(temp_vision_frame, extend_vision_frame, crop_mask, extend_affine_matrix)
		return paste_vision_frame

	return temp_vision_frame


def forward(crop_vision_frame : VisionFrame, extend_vision_frame : VisionFrame, age_modifier_direction : AgeModifierDirection) -> VisionFrame:
	age_modifier = get_inference_pool().get('age_modifier')
	age_modifier_inputs = {}

	for age_modifier_input in age_modifier.get_inputs():
		if age_modifier_input.name == 'target':
			age_modifier_inputs[age_modifier_input.name] = crop_vision_frame
		if age_modifier_input.name == 'target_with_background':
			age_modifier_inputs[age_modifier_input.name] = extend_vision_frame
		if age_modifier_input.name == 'direction':
			age_modifier_inputs[age_modifier_input.name] = age_modifier_direction

	with thread_semaphore():
		crop_vision_frame = age_modifier.run(None, age_modifier_inputs)[0][0]

	return crop_vision_frame


def prepare_vision_frame(vision_frame : VisionFrame) -> VisionFrame:
	model_mean = get_model_options().get('mean')
	model_standard_deviation = get_model_options().get('standard_deviation')
	vision_frame = vision_frame[:, :, ::-1] / 255.0
	vision_frame = (vision_frame - model_mean) / model_standard_deviation
	vision_frame = numpy.expand_dims(vision_frame.transpose(2, 0, 1), axis = 0).astype(numpy.float32)
	return vision_frame


def normalize_vision_frame(vision_frame : VisionFrame) -> VisionFrame:
	model_mean = get_model_options().get('mean')
	model_standard_deviation = get_model_options().get('standard_deviation')
	vision_frame = vision_frame.transpose(1, 2, 0)
	vision_frame = vision_frame * model_standard_deviation + model_mean
	vision_frame = vision_frame.clip(0, 1)
	vision_frame = vision_frame[:, :, ::-1] * 255
	return vision_frame


def normalize_extend_frame(extend_vision_frame : VisionFrame) -> VisionFrame:
	model_sizes = get_model_options().get('sizes')
	extend_vision_frame = numpy.clip(extend_vision_frame, -1, 1)
	extend_vision_frame = (extend_vision_frame + 1) / 2
	extend_vision_frame = extend_vision_frame.transpose(1, 2, 0).clip(0, 255)
	extend_vision_frame = (extend_vision_frame * 255.0)
	extend_vision_frame = extend_vision_frame.astype(numpy.uint8)[:, :, ::-1]
	extend_vision_frame = cv2.resize(extend_vision_frame, (model_sizes.get('target')[0] * 4, model_sizes.get('target')[1] * 4), interpolation = cv2.INTER_AREA)
	return extend_vision_frame


def process_frame(inputs : AgeModifierInputs) -> ProcessorOutputs:
	reference_vision_frame = inputs.get('reference_vision_frame')
	source_vision_frames = inputs.get('source_vision_frames')
	target_vision_frames = inputs.get('target_vision_frames')
	temp_vision_frame = inputs.get('temp_vision_frame')
	temp_vision_mask = inputs.get('temp_vision_mask')

	target_vision_frame = get_middle(target_vision_frames)
	target_faces = select_faces(reference_vision_frame, source_vision_frames, target_vision_frames)

	if target_faces:
		for target_face in target_faces:
			target_face = scale_face(target_face, target_vision_frame, temp_vision_frame)
			temp_vision_frame = modify_age(target_face, temp_vision_frame)

	return temp_vision_frame, temp_vision_mask