Ledmush421 commited on
Commit
691e51a
·
1 Parent(s): 20dbbf0

Add video quality selector

Browse files
Files changed (1) hide show
  1. app.py +127 -962
app.py CHANGED
@@ -1,977 +1,142 @@
1
  import gradio as gr
2
- import os
3
  import cv2
4
  import numpy as np
5
- import shutil
6
- import subprocess
7
- import time
8
- from SinglePhoto import FaceSwapper
9
- import argparse
10
-
11
- wellcomingMessage = """
12
- <h1>Face Swapping Suite</h1>
13
- <p>All-in-one face swapping: single photo, video, multi-source, and multi-destination!</p>
14
- """
15
-
16
- swapper = FaceSwapper()
17
-
18
- #Photo Swapping Functions
19
- def swap_single_photo(src_img, src_idx, dst_img, dst_idx, progress=gr.Progress(track_tqdm=True)):
20
- log = ""
21
- start_time = time.time()
22
- try:
23
- progress(0, desc="Preparing files")
24
- src_path = "SinglePhoto/data_src.jpg"
25
- dst_path = "SinglePhoto/data_dst.jpg"
26
- output_path = "SinglePhoto/output_swapped.jpg"
27
- os.makedirs(os.path.dirname(src_path), exist_ok=True)
28
- os.makedirs(os.path.dirname(dst_path), exist_ok=True)
29
- os.makedirs(os.path.dirname(output_path), exist_ok=True)
30
- src_img_bgr = cv2.cvtColor(src_img, cv2.COLOR_RGB2BGR)
31
- dst_img_bgr = cv2.cvtColor(dst_img, cv2.COLOR_RGB2BGR)
32
- cv2.imwrite(src_path, src_img_bgr)
33
- cv2.imwrite(dst_path, dst_img_bgr)
34
- log += f"Saved source to {src_path}, destination to {dst_path}\n"
35
- progress(0.5, desc="Swapping faces")
36
- result = swapper.swap_faces(src_path, int(src_idx), dst_path, int(dst_idx))
37
- cv2.imwrite(output_path, result)
38
- log += f"Swapped and saved result to {output_path}\n"
39
- progress(0.8, desc="Cleaning up")
40
- try:
41
- if os.path.exists(src_path):
42
- os.remove(src_path)
43
- if os.path.exists(dst_path):
44
- os.remove(dst_path)
45
- log += "Cleaned up temp files.\n"
46
- except Exception as cleanup_error:
47
- log += f"Cleanup error: {cleanup_error}\n"
48
- progress(1, desc="Done")
49
- elapsed = time.time() - start_time
50
- log += f"Elapsed time: {elapsed:.2f} seconds\n"
51
- return output_path, log
52
- except Exception as e:
53
- log += f"Error: {e}\n"
54
- progress(1, desc="Error")
55
- elapsed = time.time() - start_time
56
- log += f"Elapsed time: {elapsed:.2f} seconds\n"
57
- return None, log
58
-
59
- def swap_single_src_multi_dst(src_img, dst_imgs, dst_indices, progress=gr.Progress(track_tqdm=True)):
60
- log = ""
61
- results = []
62
- src_dir = "SingleSrcMultiDst/src"
63
- dst_dir = "SingleSrcMultiDst/dst"
64
- output_dir = "SingleSrcMultiDst/output"
65
- os.makedirs(src_dir, exist_ok=True)
66
- os.makedirs(dst_dir, exist_ok=True)
67
- os.makedirs(output_dir, exist_ok=True)
68
-
69
- if isinstance(dst_img, tuple):
70
- dst_img = dst_img[0]
71
- dst_img_bgr = cv2.cvtColor(dst_img, cv2.COLOR_RGB2BGR)
72
- dst_path = os.path.join(dst_dir, "data_dst.jpg")
73
- cv2.imwrite(dst_path, dst_img_bgr)
74
- log += f"Saved destination image to {dst_path}\n"
75
- progress(0.05, desc="Saved destination image")
76
-
77
- for i, src_img in enumerate(src_imgs):
78
- if isinstance(src_img, tuple):
79
- src_img = src_img[0]
80
- src_path = os.path.join(src_dir, f"data_src_{i}.jpg")
81
- output_path = os.path.join(output_dir, f"output_swapped_{i}.jpg")
82
- src_img_bgr = cv2.cvtColor(src_img, cv2.COLOR_RGB2BGR)
83
- cv2.imwrite(src_path, src_img_bgr)
84
- log += f"Saved source image {i} to {src_path}\n"
85
- try:
86
- result = swapper.swap_faces(src_path, 1, dst_path, int(dst_idx))
87
- cv2.imwrite(output_path, result)
88
- results.append(output_path)
89
- log += f"Swapped and saved result to {output_path}\n"
90
- except Exception as e:
91
- results.append(f"Error: {e}")
92
- log += f"Error swapping source {i}: {e}\n"
93
- progress((i + 1) / len(src_imgs), desc=f"Swapping source {i+1}/{len(src_imgs)}")
94
- progress(1, desc="Done")
95
- return results, log
96
-
97
- def swap_multi_src_single_dst(src_imgs, dst_img, dst_idx, progress=gr.Progress(track_tqdm=True)):
98
- log = ""
99
- results = []
100
- src_dir = "MultiSrcSingleDst/src"
101
- dst_dir = "MultiSrcSingleDst/dst"
102
- output_dir = "MultiSrcSingleDst/output"
103
- os.makedirs(src_dir, exist_ok=True)
104
- os.makedirs(dst_dir, exist_ok=True)
105
- os.makedirs(output_dir, exist_ok=True)
106
-
107
- if isinstance(dst_img, tuple):
108
- dst_img = dst_img[0]
109
- dst_img_bgr = cv2.cvtColor(dst_img, cv2.COLOR_RGB2BGR)
110
- dst_path = os.path.join(dst_dir, "data_dst.jpg")
111
- cv2.imwrite(dst_path, dst_img_bgr)
112
- log += f"Saved destination image to {dst_path}\n"
113
- progress(0.05, desc="Saved destination image")
114
-
115
- for i, src_img in enumerate(src_imgs):
116
- if isinstance(src_img, tuple):
117
- src_img = src_img[0]
118
- src_path = os.path.join(src_dir, f"data_src_{i}.jpg")
119
- output_path = os.path.join(output_dir, f"output_swapped_{i}.jpg")
120
- src_img_bgr = cv2.cvtColor(src_img, cv2.COLOR_RGB2BGR)
121
- cv2.imwrite(src_path, src_img_bgr)
122
- log += f"Saved source image {i} to {src_path}\n"
123
- try:
124
- result = swapper.swap_faces(src_path, 1, dst_path, int(dst_idx))
125
- cv2.imwrite(output_path, result)
126
- results.append(output_path)
127
- log += f"Swapped and saved result to {output_path}\n"
128
- except Exception as e:
129
- results.append(f"Error: {e}")
130
- log += f"Error swapping source {i}: {e}\n"
131
- progress((i + 1) / len(src_imgs), desc=f"Swapping source {i+1}/{len(src_imgs)}")
132
- progress(1, desc="Done")
133
- return results, log
134
-
135
- def swap_multi_src_multi_dst(src_imgs, dst_imgs, dst_indices, progress=gr.Progress(track_tqdm=True)):
136
- log = ""
137
- results = []
138
- src_dir = "MultiSrcMultiDst/src"
139
- dst_dir = "MultiSrcMultiDst/dst"
140
- output_dir = "MultiSrcMultiDst/output"
141
- os.makedirs(src_dir, exist_ok=True)
142
- os.makedirs(dst_dir, exist_ok=True)
143
- os.makedirs(output_dir, exist_ok=True)
144
-
145
- if isinstance(dst_indices, str):
146
- dst_indices_list = [int(idx.strip()) for idx in dst_indices.split(",") if idx.strip().isdigit()]
147
- else:
148
- dst_indices_list = [int(idx) for idx in dst_indices]
149
-
150
- total = max(1, len(src_imgs) * len(dst_imgs))
151
- count = 0
152
- for i, src_img in enumerate(src_imgs):
153
- if isinstance(src_img, tuple):
154
- src_img = src_img[0]
155
- if src_img is None:
156
- results.append(f"Error: Source image at index {i} is None")
157
- log += f"Source image at index {i} is None\n"
158
- continue
159
- src_path = os.path.join(src_dir, f"data_src_{i}.jpg")
160
- if isinstance(src_img, np.ndarray):
161
- src_img_bgr = cv2.cvtColor(src_img, cv2.COLOR_RGB2BGR)
162
- cv2.imwrite(src_path, src_img_bgr)
163
- log += f"Saved source image {i} to {src_path}\n"
164
- elif isinstance(src_img, str) and os.path.exists(src_img):
165
- shutil.copy(src_img, src_path)
166
- log += f"Copied source image {i} from {src_img} to {src_path}\n"
167
- else:
168
- results.append(f"Error: Invalid source image at index {i}")
169
- log += f"Invalid source image at index {i}\n"
170
- continue
171
- for j, dst_img in enumerate(dst_imgs):
172
- if isinstance(dst_img, tuple):
173
- dst_img = dst_img[0]
174
- if dst_img is None:
175
- results.append(f"Error: Destination image at index {j} is None")
176
- log += f"Destination image at index {j} is None\n"
177
- continue
178
- dst_path = os.path.join(dst_dir, f"data_dst_{j}.jpg")
179
- output_path = os.path.join(output_dir, f"output_swapped_{i}_{j}.jpg")
180
- if isinstance(dst_img, np.ndarray):
181
- dst_img_bgr = cv2.cvtColor(dst_img, cv2.COLOR_RGB2BGR)
182
- cv2.imwrite(dst_path, dst_img_bgr)
183
- log += f"Saved destination image {j} to {dst_path}\n"
184
- elif isinstance(dst_img, str) and os.path.exists(dst_img):
185
- shutil.copy(dst_img, dst_path)
186
- log += f"Copied destination image {j} from {dst_img} to {dst_path}\n"
187
- else:
188
- results.append(f"Error: Invalid destination image at index {j}")
189
- log += f"Invalid destination image at index {j}\n"
190
- continue
191
- try:
192
- dst_idx = dst_indices_list[j] if j < len(dst_indices_list) else 1
193
- result = swapper.swap_faces(src_path, 1, dst_path, int(dst_idx))
194
- cv2.imwrite(output_path, result)
195
- results.append(output_path)
196
- log += f"Swapped src {i} with dst {j} and saved to {output_path}\n"
197
- except Exception as e:
198
- results.append(f"Error: {e}")
199
- log += f"Error swapping src {i} with dst {j}: {e}\n"
200
- count += 1
201
- progress(count / total, desc=f"Swapping ({count}/{total})")
202
- progress(1, desc="Done")
203
- return results, log
204
-
205
- def swap_single_src_multi_dst(src_img, dst_imgs, dst_indices, progress=gr.Progress(track_tqdm=True)):
206
- log = ""
207
- results = []
208
- src_dir = "SingleSrcMultiDst/src"
209
- dst_dir = "SingleSrcMultiDst/dst"
210
- output_dir = "SingleSrcMultiDst/output"
211
- os.makedirs(src_dir, exist_ok=True)
212
- os.makedirs(dst_dir, exist_ok=True)
213
- os.makedirs(output_dir, exist_ok=True)
214
-
215
- if isinstance(src_img, tuple):
216
- src_img = src_img[0]
217
- src_path = os.path.join(src_dir, "data_src.jpg")
218
- src_img_bgr = cv2.cvtColor(src_img, cv2.COLOR_RGB2BGR)
219
- cv2.imwrite(src_path, src_img_bgr)
220
- log += f"Saved source image to {src_path}\n"
221
- progress(0.05, desc="Saved source image")
222
-
223
- if isinstance(dst_indices, str):
224
- dst_indices_list = [int(idx.strip()) for idx in dst_indices.split(",") if idx.strip().isdigit()]
225
- else:
226
- dst_indices_list = [int(idx) for idx in dst_indices]
227
-
228
- for j, dst_img in enumerate(dst_imgs):
229
- if isinstance(dst_img, tuple):
230
- dst_img = dst_img[0]
231
- dst_path = os.path.join(dst_dir, f"data_dst_{j}.jpg")
232
- output_path = os.path.join(output_dir, f"output_swapped_{j}.jpg")
233
- dst_img_bgr = cv2.cvtColor(dst_img, cv2.COLOR_RGB2BGR)
234
- cv2.imwrite(dst_path, dst_img_bgr)
235
- log += f"Saved destination image {j} to {dst_path}\n"
236
- try:
237
- dst_idx = dst_indices_list[j] if j < len(dst_indices_list) else 1
238
- result = swapper.swap_faces(src_path, 1, dst_path, int(dst_idx))
239
- cv2.imwrite(output_path, result)
240
- results.append(output_path)
241
- log += f"Swapped and saved result to {output_path}\n"
242
- except Exception as e:
243
- results.append(f"Error: {e}")
244
- log += f"Error swapping with destination {j}: {e}\n"
245
- progress((j + 1) / len(dst_imgs), desc=f"Swapping destination {j+1}/{len(dst_imgs)}")
246
- progress(1, desc="Done")
247
- return results, log
248
-
249
- def swap_faces_custom(src_imgs, dst_img, mapping_str, progress=gr.Progress(track_tqdm=True)):
250
- """
251
- src_imgs: list of source images (numpy arrays)
252
- dst_img: destination image (numpy array)
253
- mapping_str: comma-separated string, e.g. "2,1,3"
254
- """
255
- log = ""
256
- start_time = time.time()
257
- dst_path = "CustomSwap/data_dst.jpg"
258
- output_path = "CustomSwap/output_swapped.jpg"
259
- src_dir = "CustomSwap/src"
260
- temp_dir = "CustomSwap/temp"
261
- os.makedirs(src_dir, exist_ok=True)
262
- os.makedirs(temp_dir, exist_ok=True)
263
- os.makedirs(os.path.dirname(dst_path), exist_ok=True)
264
- os.makedirs(os.path.dirname(output_path), exist_ok=True)
265
-
266
- # Save destination image
267
- dst_img_bgr = cv2.cvtColor(dst_img, cv2.COLOR_RGB2BGR)
268
- cv2.imwrite(dst_path, dst_img_bgr)
269
- log += f"Saved destination image to {dst_path}\n"
270
-
271
- # Save all source images
272
- src_paths = []
273
- for i, src_img in enumerate(src_imgs):
274
- src_path = os.path.join(src_dir, f"data_src_{i+1}.jpg")
275
- if isinstance(src_img, tuple):
276
- src_img = src_img[0]
277
- if src_img is None:
278
- log += f"Source image {i+1} is None, skipping.\n"
279
- continue
280
- if isinstance(src_img, np.ndarray):
281
- src_img_bgr = cv2.cvtColor(src_img, cv2.COLOR_RGB2BGR)
282
- cv2.imwrite(src_path, src_img_bgr)
283
- src_paths.append(src_path)
284
- log += f"Saved source image {i+1} to {src_path}\n"
285
- elif isinstance(src_img, str) and os.path.exists(src_img):
286
- shutil.copy(src_img, src_path)
287
- src_paths.append(src_path)
288
- log += f"Copied source image {i+1} from {src_img} to {src_path}\n"
289
- else:
290
- log += f"Source image {i+1} is not a valid image, skipping.\n"
291
-
292
- # Parse mapping
293
- try:
294
- mapping = [int(x.strip()) for x in mapping_str.split(",") if x.strip().isdigit()]
295
- except Exception as e:
296
- log += f"Error parsing mapping: {e}\n"
297
- elapsed = time.time() - start_time
298
- log += f"Elapsed time: {elapsed:.2f} seconds\n"
299
- return None, log
300
-
301
- # Use a temp file for iterative swapping
302
- temp_dst_path = os.path.join(temp_dir, "temp_dst.jpg")
303
- shutil.copy(dst_path, temp_dst_path)
304
-
305
- for face_idx, src_idx in enumerate(mapping, start=1):
306
- if src_idx < 1 or src_idx > len(src_paths):
307
- log += f"Invalid source index {src_idx} for face {face_idx}, skipping.\n"
308
- continue
309
- try:
310
- swapped_img = swapper.swap_faces(src_paths[src_idx-1], 1, temp_dst_path, face_idx)
311
- cv2.imwrite(temp_dst_path, swapped_img)
312
- log += f"Swapped face {face_idx} in destination with source {src_idx}\n"
313
- except Exception as e:
314
- log += f"Failed to swap face {face_idx} with source {src_idx}: {e}\n"
315
-
316
- shutil.copy(temp_dst_path, output_path)
317
- log += f"Saved swapped image to {output_path}\n"
318
- if os.path.exists(temp_dst_path):
319
- os.remove(temp_dst_path)
320
- elapsed = time.time() - start_time
321
- log += f"Elapsed time: {elapsed:.2f} seconds\n"
322
- return output_path, log
323
-
324
- #Video Swapping Functions
325
- def add_audio_to_video(original_video_path, video_no_audio_path, output_path):
326
- cmd = [
327
- "ffmpeg",
328
- "-y",
329
- "-i", video_no_audio_path,
330
- "-i", original_video_path,
331
- "-c:v", "copy",
332
- "-c:a", "aac",
333
- "-map", "0:v:0",
334
- "-map", "1:a:0?",
335
- "-shortest",
336
- output_path
337
- ]
338
- try:
339
- subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
340
- return True, ""
341
- except subprocess.CalledProcessError as e:
342
- return False, e.stderr.decode()
343
-
344
- def swap_video(src_img, src_idx, video, dst_idx, delete_frames_dir=True, add_audio=True, copy_to_drive=False, progress=gr.Progress()):
345
- log = ""
346
- start_time = time.time()
347
- src_path = "VideoSwapping/data_src.jpg"
348
- dst_video_path = "VideoSwapping/data_dst.mp4"
349
- frames_dir = "VideoSwapping/video_frames"
350
- swapped_dir = "VideoSwapping/swapped_frames"
351
- output_video_path = "VideoSwapping/output_tmp_output_video.mp4"
352
- final_output_path = "VideoSwapping/output_with_audio.mp4"
353
-
354
- os.makedirs(os.path.dirname(src_path), exist_ok=True)
355
- os.makedirs(os.path.dirname(dst_video_path), exist_ok=True)
356
- os.makedirs(frames_dir, exist_ok=True)
357
- os.makedirs(swapped_dir, exist_ok=True)
358
-
359
- src_img_bgr = cv2.cvtColor(src_img, cv2.COLOR_RGB2BGR)
360
- cv2.imwrite(src_path, src_img_bgr)
361
- log += f"Saved source image to {src_path}\n"
362
- progress(0.05, desc="Saved source image")
363
-
364
- if isinstance(video, str) and os.path.exists(video):
365
- shutil.copy(video, dst_video_path)
366
- log += f"Copied video to {dst_video_path}\n"
367
- else:
368
- dst_video_path = video
369
-
370
- from VideoSwapping import extract_frames, frames_to_video
371
-
372
- frame_paths = extract_frames(dst_video_path, frames_dir)
373
- log += f"Extracted {len(frame_paths)} frames to {frames_dir}\n"
374
- progress(0.15, desc="Extracted frames")
375
-
376
- swapped_files = set(os.listdir(swapped_dir))
377
- total_frames = len(frame_paths)
378
- start_loop_time = time.time()
379
- for idx, frame_path in enumerate(frame_paths):
380
- swapped_name = f"swapped_{idx:05d}.jpg"
381
- out_path = os.path.join(swapped_dir, swapped_name)
382
- if swapped_name in swapped_files and os.path.exists(out_path):
383
- log += f"Frame {idx}: already swapped, skipping.\n"
384
- elapsed = time.time() - start_loop_time
385
- avg_time = elapsed / (idx + 1) if idx + 1 > 0 else 0
386
- remaining = avg_time * (total_frames - (idx + 1))
387
- mins, secs = divmod(int(remaining), 60)
388
- progress(0.15 + 0.6 * (idx + 1) / total_frames, desc=f"Swapping {idx+1}/{total_frames} | {mins:02d}:{secs:02d} left")
389
- continue
390
- try:
391
- try:
392
- swapped = swapper.swap_faces(src_path, int(src_idx), frame_path, int(dst_idx))
393
- except ValueError as ve:
394
- if int(dst_idx) != 1 and "Target image contains" in str(ve):
395
- swapped = swapper.swap_faces(src_path, int(src_idx), frame_path, 1)
396
- log += f"Frame {idx}: dst_idx {dst_idx} not found, used 1 instead.\n"
397
- else:
398
- raise ve
399
- cv2.imwrite(out_path, swapped)
400
- log += f"Swapped frame {idx} and saved to {out_path}\n"
401
- except Exception as e:
402
- cv2.imwrite(out_path, cv2.imread(frame_path))
403
- log += f"Failed to swap frame {idx}: {e}\n"
404
- elapsed = time.time() - start_loop_time
405
- avg_time = elapsed / (idx + 1) if idx + 1 > 0 else 0
406
- remaining = avg_time * (total_frames - (idx + 1))
407
- mins, secs = divmod(int(remaining), 60)
408
- progress(0.15 + 0.6 * (idx + 1) / total_frames, desc=f"Swapping {idx+1}/{total_frames} | {mins:02d}:{secs:02d} left")
409
- cap = cv2.VideoCapture(dst_video_path)
410
- fps = cap.get(cv2.CAP_PROP_FPS)
411
- cap.release()
412
- frames_to_video(swapped_dir, output_video_path, fps)
413
- log += f"Combined swapped frames into video {output_video_path}\n"
414
- progress(0.8, desc="Muxing audio")
415
-
416
- # Copy to Google Drive if requested
417
- if copy_to_drive:
418
- drive_path = "/content/drive/MyDrive/" + os.path.basename(output_video_path)
419
- try:
420
- shutil.copy(output_video_path, drive_path)
421
- log += f"Copied swapped video without audio to Google Drive: {drive_path}\n"
422
- except Exception as e:
423
- log += f"Failed to copy to Google Drive: {e}\n"
424
-
425
- if add_audio:
426
- ok, audio_log = add_audio_to_video(dst_video_path, output_video_path, final_output_path)
427
- if ok:
428
- log += f"Added audio to {final_output_path}\n"
429
- else:
430
- log += f"Audio muxing failed: {audio_log}\n"
431
- final_output_path = output_video_path
432
- else:
433
- final_output_path = output_video_path
434
- log += "Audio was not added as per user request.\n"
435
-
436
- try:
437
- if os.path.exists(src_path):
438
- os.remove(src_path)
439
- if os.path.exists(dst_video_path):
440
- os.remove(dst_video_path)
441
- if delete_frames_dir and os.path.exists(frames_dir):
442
- shutil.rmtree(frames_dir)
443
- log += "Deleted video_frames directory.\n"
444
- elif not delete_frames_dir:
445
- log += "Kept video_frames directory as requested.\n"
446
- if os.path.exists(swapped_dir):
447
- shutil.rmtree(swapped_dir)
448
- log += "Cleaned up temp files and folders.\n"
449
- except Exception as cleanup_error:
450
- log += f"Cleanup error: {cleanup_error}\n"
451
- progress(1, desc="Done")
452
- elapsed = time.time() - start_time
453
- log += f"Elapsed time: {elapsed:.2f} seconds\n"
454
- return final_output_path, log
455
-
456
- def swap_video_all_faces(src_img, video, num_faces_to_swap, delete_frames_dir=True, add_audio=True, copy_to_drive=False, progress=gr.Progress()):
457
- log = ""
458
- start_time = time.time()
459
- src_path = "VideoSwappingAllFaces/data_src.jpg"
460
- dst_video_path = "VideoSwappingAllFaces/data_dst.mp4"
461
- frames_dir = "VideoSwappingAllFaces/video_frames"
462
- swapped_dir = "VideoSwappingAllFaces/swapped_frames"
463
- output_video_path = "VideoSwappingAllFaces/output_tmp_output_video.mp4"
464
- final_output_path = "VideoSwappingAllFaces/output_with_audio.mp4"
465
-
466
- os.makedirs(os.path.dirname(src_path), exist_ok=True)
467
- os.makedirs(os.path.dirname(dst_video_path), exist_ok=True)
468
- os.makedirs(frames_dir, exist_ok=True)
469
- os.makedirs(swapped_dir, exist_ok=True)
470
-
471
- src_img_bgr = cv2.cvtColor(src_img, cv2.COLOR_RGB2BGR)
472
- cv2.imwrite(src_path, src_img_bgr)
473
- log += f"Saved source image to {src_path}\n"
474
- progress(0.05, desc="Saved source image")
475
-
476
- if isinstance(video, str) and os.path.exists(video):
477
- shutil.copy(video, dst_video_path)
478
- log += f"Copied video to {dst_video_path}\n"
479
- else:
480
- dst_video_path = video
481
-
482
- from VideoSwapping import extract_frames, frames_to_video
483
-
484
- frame_paths = extract_frames(dst_video_path, frames_dir)
485
- log += f"Extracted {len(frame_paths)} frames to {frames_dir}\n"
486
- progress(0.15, desc="Extracted frames")
487
-
488
- swapped_files = set(os.listdir(swapped_dir))
489
- temp_dir = os.path.join(swapped_dir, "temp_swap")
490
- os.makedirs(temp_dir, exist_ok=True)
491
- total_frames = len(frame_paths)
492
- start_loop_time = time.time()
493
-
494
- for idx, frame_path in enumerate(frame_paths):
495
- swapped_name = f"swapped_{idx:05d}.jpg"
496
- out_path = os.path.join(swapped_dir, swapped_name)
497
- temp_frame_path = os.path.join(temp_dir, "temp.jpg")
498
- if swapped_name in swapped_files and os.path.exists(out_path):
499
- log += f"Frame {idx}: already swapped, skipping.\n"
500
- elapsed = time.time() - start_loop_time
501
- avg_time = elapsed / (idx + 1) if idx + 1 > 0 else 0
502
- remaining = avg_time * (total_frames - (idx + 1))
503
- mins, secs = divmod(int(remaining), 60)
504
- progress(0.15 + 0.6 * (idx + 1) / total_frames, desc=f"Swapping {idx+1}/{total_frames} | {mins:02d}:{secs:02d} left")
505
- continue
506
- try:
507
- shutil.copy(frame_path, temp_frame_path)
508
- for face_idx in range(1, int(num_faces_to_swap) + 1):
509
- try:
510
- swapped_img = swapper.swap_faces(src_path, 1, temp_frame_path, face_idx)
511
- cv2.imwrite(temp_frame_path, swapped_img)
512
- except Exception as e:
513
- log += f"Failed to swap face {face_idx} in frame {idx}: {e}\n"
514
- shutil.copy(temp_frame_path, out_path)
515
- log += f"Swapped all faces in frame {idx} and saved to {out_path}\n"
516
- if os.path.exists(temp_frame_path):
517
- os.remove(temp_frame_path)
518
- except Exception as e:
519
- cv2.imwrite(out_path, cv2.imread(frame_path))
520
- log += f"Failed to swap frame {idx}: {e}\n"
521
- elapsed = time.time() - start_loop_time
522
- avg_time = elapsed / (idx + 1) if idx + 1 > 0 else 0
523
- remaining = avg_time * (total_frames - (idx + 1))
524
- mins, secs = divmod(int(remaining), 60)
525
- progress(0.15 + 0.6 * (idx + 1) / total_frames, desc=f"Swapping {idx+1}/{total_frames} | {mins:02d}:{secs:02d} left")
526
- if os.path.exists(temp_dir):
527
- shutil.rmtree(temp_dir)
528
-
529
- cap = cv2.VideoCapture(dst_video_path)
530
- fps = cap.get(cv2.CAP_PROP_FPS)
531
- cap.release()
532
- frames_to_video(swapped_dir, output_video_path, fps)
533
- log += f"Combined swapped frames into video {output_video_path}\n"
534
- progress(0.8, desc="Muxing audio")
535
-
536
- # Copy to Google Drive if requested
537
- if copy_to_drive:
538
- drive_path = "/content/drive/MyDrive/" + os.path.basename(output_video_path)
539
- try:
540
- shutil.copy(output_video_path, drive_path)
541
- log += f"Copied swapped video without audio to Google Drive: {drive_path}\n"
542
- except Exception as e:
543
- log += f"Failed to copy to Google Drive: {e}\n"
544
-
545
- if add_audio:
546
- ok, audio_log = add_audio_to_video(dst_video_path, output_video_path, final_output_path)
547
- if ok:
548
- log += f"Added audio to {final_output_path}\n"
549
- else:
550
- log += f"Audio muxing failed: {audio_log}\n"
551
- final_output_path = output_video_path
552
- else:
553
- final_output_path = output_video_path
554
- log += "Audio was not added as per user request.\n"
555
-
556
- try:
557
- if os.path.exists(src_path):
558
- os.remove(src_path)
559
- if os.path.exists(dst_video_path):
560
- os.remove(dst_video_path)
561
- if delete_frames_dir and os.path.exists(frames_dir):
562
- shutil.rmtree(frames_dir)
563
- log += "Deleted video_frames directory.\n"
564
- elif not delete_frames_dir:
565
- log += "Kept video_frames directory as requested.\n"
566
- if os.path.exists(swapped_dir):
567
- shutil.rmtree(swapped_dir)
568
- log += "Cleaned up temp files and folders.\n"
569
- except Exception as cleanup_error:
570
- log += f"Cleanup error: {cleanup_error}\n"
571
- progress(1, desc="Done")
572
- elapsed = time.time() - start_time
573
- log += f"Elapsed time: {elapsed:.2f} seconds\n"
574
- return final_output_path, log
575
-
576
- def swap_video_custom_mapping(src_imgs, video, mapping_str, delete_frames_dir=True, add_audio=True, copy_to_drive=False, progress=gr.Progress()):
577
- log = ""
578
- start_time = time.time()
579
- src_dir = "CustomVideoSwap/src"
580
- temp_dir = "CustomVideoSwap/temp"
581
- frames_dir = "CustomVideoSwap/frames"
582
- swapped_dir = "CustomVideoSwap/swapped_frames"
583
- output_video_path = "CustomVideoSwap/output_tmp_output_video.mp4"
584
- final_output_path = "CustomVideoSwap/output_with_audio.mp4"
585
- dst_video_path = "CustomVideoSwap/data_dst.mp4"
586
-
587
- os.makedirs(src_dir, exist_ok=True)
588
- os.makedirs(temp_dir, exist_ok=True)
589
- os.makedirs(frames_dir, exist_ok=True)
590
- os.makedirs(swapped_dir, exist_ok=True)
591
-
592
- # Save all source images
593
- src_paths = []
594
- for i, src_img in enumerate(src_imgs):
595
- src_path = os.path.join(src_dir, f"data_src_{i+1}.jpg")
596
- if isinstance(src_img, tuple):
597
- src_img = src_img[0]
598
- if src_img is None:
599
- log += f"Source image {i+1} is None, skipping.\n"
600
- continue
601
- if isinstance(src_img, np.ndarray):
602
- src_img_bgr = cv2.cvtColor(src_img, cv2.COLOR_RGB2BGR)
603
- cv2.imwrite(src_path, src_img_bgr)
604
- src_paths.append(src_path)
605
- log += f"Saved source image {i+1} to {src_path}\n"
606
- elif isinstance(src_img, str) and os.path.exists(src_img):
607
- shutil.copy(src_img, src_path)
608
- src_paths.append(src_path)
609
- log += f"Copied source image {i+1} from {src_img} to {src_path}\n"
610
- else:
611
- log += f"Source image {i+1} is not a valid image, skipping.\n"
612
-
613
- # Parse mapping
614
- try:
615
- mapping = [int(x.strip()) for x in mapping_str.split(",") if x.strip().isdigit()]
616
- except Exception as e:
617
- log += f"Error parsing mapping: {e}\n"
618
- elapsed = time.time() - start_time
619
- log += f"Elapsed time: {elapsed:.2f} seconds\n"
620
- return None, log
621
-
622
- # Prepare video
623
- if isinstance(video, str) and os.path.exists(video):
624
- shutil.copy(video, dst_video_path)
625
- log += f"Copied video to {dst_video_path}\n"
626
- else:
627
- dst_video_path = video
628
-
629
- from VideoSwapping import extract_frames, frames_to_video
630
-
631
- frame_paths = extract_frames(dst_video_path, frames_dir)
632
- log += f"Extracted {len(frame_paths)} frames to {frames_dir}\n"
633
- progress(0.1, desc="Extracted frames")
634
-
635
- swapped_files = set(os.listdir(swapped_dir))
636
- temp_frame_path = os.path.join(temp_dir, "temp.jpg")
637
- total_frames = len(frame_paths)
638
- start_loop_time = time.time()
639
 
640
- for idx, frame_path in enumerate(frame_paths):
641
- swapped_name = f"swapped_{idx:05d}.jpg"
642
- out_path = os.path.join(swapped_dir, swapped_name)
643
- if swapped_name in swapped_files and os.path.exists(out_path):
644
- log += f"Frame {idx}: already swapped, skipping.\n"
645
- elapsed = time.time() - start_loop_time
646
- avg_time = elapsed / (idx + 1) if idx + 1 > 0 else 0
647
- remaining = avg_time * (total_frames - (idx + 1))
648
- mins, secs = divmod(int(remaining), 60)
649
- progress(0.1 + 0.7 * (idx + 1) / total_frames, desc=f"Swapping {idx+1}/{total_frames} | {mins:02d}:{secs:02d} left")
650
- continue
651
- try:
652
- shutil.copy(frame_path, temp_frame_path)
653
- for face_idx, src_idx in enumerate(mapping, start=1):
654
- if src_idx < 1 or src_idx > len(src_paths):
655
- log += f"Invalid source index {src_idx} for face {face_idx} in frame {idx}, skipping.\n"
656
- continue
657
- try:
658
- swapped_img = swapper.swap_faces(src_paths[src_idx-1], 1, temp_frame_path, face_idx)
659
- cv2.imwrite(temp_frame_path, swapped_img)
660
- log += f"Frame {idx}: Swapped face {face_idx} with source {src_idx}\n"
661
- except Exception as e:
662
- log += f"Frame {idx}: Failed to swap face {face_idx} with source {src_idx}: {e}\n"
663
- shutil.copy(temp_frame_path, out_path)
664
- log += f"Swapped all faces in frame {idx} and saved to {out_path}\n"
665
- if os.path.exists(temp_frame_path):
666
- os.remove(temp_frame_path)
667
- except Exception as e:
668
- cv2.imwrite(out_path, cv2.imread(frame_path))
669
- log += f"Failed to swap frame {idx}: {e}\n"
670
- elapsed = time.time() - start_loop_time
671
- avg_time = elapsed / (idx + 1) if idx + 1 > 0 else 0
672
- remaining = avg_time * (total_frames - (idx + 1))
673
- mins, secs = divmod(int(remaining), 60)
674
- progress(0.1 + 0.7 * (idx + 1) / total_frames, desc=f"Swapping {idx+1}/{total_frames} | {mins:02d}:{secs:02d} left")
675
- if os.path.exists(temp_dir):
676
- shutil.rmtree(temp_dir)
677
 
678
- cap = cv2.VideoCapture(dst_video_path)
679
- fps = cap.get(cv2.CAP_PROP_FPS)
680
- cap.release()
681
- frames_to_video(swapped_dir, output_video_path, fps)
682
- log += f"Combined swapped frames into video {output_video_path}\n"
683
- progress(0.9, desc="Muxing audio")
684
 
685
- if add_audio:
686
- ok, audio_log = add_audio_to_video(dst_video_path, output_video_path, final_output_path)
687
- if ok:
688
- log += f"Added audio to {final_output_path}\n"
689
- else:
690
- log += f"Audio muxing failed: {audio_log}\n"
691
- final_output_path = output_video_path
692
- else:
693
- final_output_path = output_video_path
694
- log += "Audio was not added as per user request.\n"
695
 
 
696
  try:
697
- if os.path.exists(dst_video_path):
698
- os.remove(dst_video_path)
699
- if delete_frames_dir and os.path.exists(frames_dir):
700
- shutil.rmtree(frames_dir)
701
- log += "Deleted video_frames directory.\n"
702
- elif not delete_frames_dir:
703
- log += "Kept video_frames directory as requested.\n"
704
- if os.path.exists(swapped_dir):
705
- shutil.rmtree(swapped_dir)
706
- log += "Cleaned up temp files and folders.\n"
707
- except Exception as cleanup_error:
708
- log += f"Cleanup error: {cleanup_error}\n"
709
- progress(1, desc="Done")
710
- elapsed = time.time() - start_time
711
- log += f"Elapsed time: {elapsed:.2f} seconds\n"
712
- return final_output_path, log
713
-
714
- def swap_single_src_multi_video(src_img, dst_videos, dst_indices, delete_frames_dir=True, add_audio=True, copy_to_drive=False, progress=gr.Progress(track_tqdm=True)):
715
- """
716
- Swaps a single source image onto multiple videos, with a mapping for destination face indices.
717
- Each video is processed one by one.
718
- """
719
- log = ""
720
- results = []
721
- start_time = time.time()
722
- base_dir = "SingleSrcMultiVideo"
723
- dst_dir = os.path.join(base_dir, "dst")
724
- frames_dir = os.path.join(base_dir, "video_frames")
725
- swapped_dir = os.path.join(base_dir, "swap_frames")
726
- output_dir = os.path.join(base_dir, "output")
727
- os.makedirs(dst_dir, exist_ok=True)
728
- os.makedirs(frames_dir, exist_ok=True)
729
- os.makedirs(swapped_dir, exist_ok=True)
730
- os.makedirs(output_dir, exist_ok=True)
731
-
732
- # Parse indices
733
- if isinstance(dst_indices, str):
734
- dst_indices_list = [int(idx.strip()) for idx in dst_indices.split(",") if idx.strip().isdigit()]
735
- else:
736
- dst_indices_list = [int(idx) for idx in dst_indices]
737
-
738
- # Save source image
739
- src_path = os.path.join(base_dir, "data_src.jpg")
740
- src_img_bgr = cv2.cvtColor(src_img, cv2.COLOR_RGB2BGR)
741
- cv2.imwrite(src_path, src_img_bgr)
742
- log += f"Saved source image to {src_path}\n"
743
-
744
- from VideoSwapping import extract_frames, frames_to_video
745
-
746
- for i, video in enumerate(dst_videos):
747
- dst_idx = dst_indices_list[i] if i < len(dst_indices_list) else 1
748
- video_name = f"video_{i}.mp4"
749
- dst_video_path = os.path.join(dst_dir, video_name)
750
- output_video_path = os.path.join(output_dir, f"output_{i}.mp4")
751
- output_video_with_audio = os.path.join(output_dir, f"output_with_audio_{i}.mp4")
752
-
753
- # Copy video to dst_dir
754
- if isinstance(video, str) and os.path.exists(video):
755
- shutil.copy(video, dst_video_path)
756
- log += f"Copied video {video} to {dst_video_path}\n"
757
  else:
758
- dst_video_path = video
759
-
760
- # Extract frames
761
- frame_paths = extract_frames(dst_video_path, frames_dir)
762
- log += f"Extracted {len(frame_paths)} frames from {dst_video_path} to {frames_dir}\n"
763
- progress(i / len(dst_videos), desc=f"Processing video {i+1}/{len(dst_videos)}")
764
-
765
- swapped_files = set(os.listdir(swapped_dir))
766
- total_frames = len(frame_paths)
767
- temp_frame_path = os.path.join(swapped_dir, "temp.jpg")
768
- start_loop_time = time.time()
769
-
770
- for idx, frame_path in enumerate(frame_paths):
771
- swapped_name = f"swapped_{idx:05d}.jpg"
772
- out_path = os.path.join(swapped_dir, swapped_name)
773
- if swapped_name in swapped_files and os.path.exists(out_path):
774
- continue
775
- try:
776
- shutil.copy(frame_path, temp_frame_path)
777
- try:
778
- swapped_img = swapper.swap_faces(src_path, 1, temp_frame_path, int(dst_idx))
779
- cv2.imwrite(temp_frame_path, swapped_img)
780
- except Exception as e:
781
- log += f"Failed to swap face in frame {idx} of video {i}: {e}\n"
782
- shutil.copy(temp_frame_path, out_path)
783
- if os.path.exists(temp_frame_path):
784
- os.remove(temp_frame_path)
785
- except Exception as e:
786
- cv2.imwrite(out_path, cv2.imread(frame_path))
787
- log += f"Failed to swap frame {idx} of video {i}: {e}\n"
788
-
789
- # Combine frames to video
790
- cap = cv2.VideoCapture(dst_video_path)
791
- fps = cap.get(cv2.CAP_PROP_FPS)
 
 
 
 
 
 
 
 
 
 
 
792
  cap.release()
793
- frames_to_video(swapped_dir, output_video_path, fps)
794
- log += f"Combined swapped frames into video {output_video_path}\n"
795
-
796
- # Copy to Google Drive if requested
797
- if copy_to_drive:
798
- drive_path = "/content/drive/MyDrive/" + os.path.basename(output_video_path)
799
- try:
800
- shutil.copy(output_video_path, drive_path)
801
- log += f"Copied swapped video without audio to Google Drive: {drive_path}\n"
802
- except Exception as e:
803
- log += f"Failed to copy to Google Drive: {e}\n"
804
-
805
- # Mux audio if requested
806
- if add_audio:
807
- ok, audio_log = add_audio_to_video(dst_video_path, output_video_path, output_video_with_audio)
808
- if ok:
809
- log += f"Added audio to {output_video_with_audio}\n"
810
- results.append(output_video_with_audio)
811
- else:
812
- log += f"Audio muxing failed for video {i}: {audio_log}\n"
813
- results.append(output_video_path)
814
- else:
815
- results.append(output_video_path)
816
- log += f"Audio was not added for video {i} as per user request.\n"
817
-
818
- # Cleanup frames for next video
819
- if delete_frames_dir and os.path.exists(frames_dir):
820
- shutil.rmtree(frames_dir)
821
- os.makedirs(frames_dir, exist_ok=True)
822
- log += f"Deleted video_frames directory after video {i}.\n"
823
- elif not delete_frames_dir:
824
- log += f"Kept video_frames directory after video {i} as requested.\n"
825
- if os.path.exists(swapped_dir):
826
- shutil.rmtree(swapped_dir)
827
- os.makedirs(swapped_dir, exist_ok=True)
828
- log += f"Cleared swap_frames directory after video {i}.\n"
829
-
830
- elapsed = time.time() - start_time
831
- log += f"Elapsed time: {elapsed:.2f} seconds\n"
832
- return results, log
833
-
834
-
835
 
836
- # Gradio UI
837
- with gr.Blocks() as demo:
838
- gr.Markdown(wellcomingMessage)
839
- with gr.Tab("Single Photo Swapping"):
840
- gr.Interface(
841
- fn=swap_single_photo,
842
- inputs=[
843
- gr.Image(label="Source Image"),
844
- gr.Number(value=1, label="Source Face Index"),
845
- gr.Image(label="Destination Image"),
846
- gr.Number(value=1, label="Destination Face Index"),
847
- ],
848
- outputs=[
849
- gr.Image(label="Swapped Image"),
850
- gr.Textbox(label="Log Output", lines=8, interactive=False)
851
- ],
852
- )
 
 
 
 
 
 
853
 
854
- with gr.Tab("SingleSrc MultiDst"):
855
- gr.Interface(
856
- fn=swap_single_src_multi_dst,
857
- inputs=[
858
- gr.Image(label="Source Image"),
859
- gr.Gallery(label="Destination Images", type="numpy", columns=3),
860
- gr.Textbox(label="Destination Face Indices (comma-separated, e.g. 1,1,2)"),
861
- ],
862
- outputs=[
863
- gr.Gallery(label="Swapped Images"),
864
- gr.Textbox(label="Log Output", lines=8, interactive=False)
865
- ],
866
- )
867
- with gr.Tab("MultiSrc SingleDst"):
868
- gr.Interface(
869
- fn=swap_multi_src_single_dst,
870
- inputs=[
871
- gr.Gallery(label="Source Images", type="numpy", columns=3),
872
- gr.Image(label="Destination Image"),
873
- gr.Number(value=1, label="Destination Face Index"),
874
- ],
875
- outputs=[
876
- gr.Gallery(label="Swapped Images"),
877
- gr.Textbox(label="Log Output", lines=8, interactive=False)
878
- ],
879
- )
880
- with gr.Tab("MultiSrc MultiDst"):
881
- gr.Interface(
882
- fn=swap_multi_src_multi_dst,
883
- inputs=[
884
- gr.Gallery(label="Source Images", type="numpy", columns=3),
885
- gr.Gallery(label="Destination Images", type="numpy", columns=3),
886
- gr.Textbox(label="Destination Face Indices (comma-separated, e.g. 1,1,2)"),
887
- ],
888
- outputs=[
889
- gr.Gallery(label="Swapped Images"),
890
- gr.Textbox(label="Log Output", lines=8, interactive=False)
891
- ],
892
- )
893
- with gr.Tab("Custom Face Mapping"):
894
- gr.Interface(
895
- fn=swap_faces_custom,
896
- inputs=[
897
- gr.Gallery(label="Source Images", type="numpy", columns=3),
898
- gr.Image(label="Destination Image"),
899
- gr.Textbox(label="Mapping (comma-separated, e.g. 2,1,3)"),
900
- ],
901
- outputs=[
902
- gr.Image(label="Swapped Image"),
903
- gr.Textbox(label="Log Output", lines=8, interactive=False)
904
- ],
905
- )
906
- with gr.Tab("Video Swapping"):
907
- gr.Interface(
908
- fn=swap_video,
909
- inputs=[
910
- gr.Image(label="Source Image"),
911
- gr.Number(value=1, label="Source Face Index"),
912
- gr.Video(label="Target Video"),
913
- gr.Number(value=1, label="Destination Face Index"),
914
- gr.Checkbox(label="Delete video_frames directory after processing", value=True),
915
- gr.Checkbox(label="Add audio from original video", value=True),
916
- gr.Checkbox(label="Copy swapped video (no audio) to Google Drive root", value=False)
917
- ],
918
- outputs=[
919
- gr.Video(label="Swapped Video"),
920
- gr.Textbox(label="Log Output", lines=8, interactive=False)
921
- ],
922
- )
923
- with gr.Tab("Video All Faces"):
924
- gr.Interface(
925
- fn=swap_video_all_faces,
926
- inputs=[
927
- gr.Image(label="Source Image"),
928
- gr.Video(label="Target Video"),
929
- gr.Number(value=1, label="Number of Faces to Swap"),
930
- gr.Checkbox(label="Delete video_frames directory after processing", value=True),
931
- gr.Checkbox(label="Add audio from original video", value=True),
932
- gr.Checkbox(label="Copy swapped video (no audio) to Google Drive root", value=False)
933
- ],
934
- outputs=[
935
- gr.Video(label="Swapped Video"),
936
- gr.Textbox(label="Log Output", lines=8, interactive=False)
937
- ],
938
- )
939
- with gr.Tab("Custom Video Face Mapping"):
940
- gr.Interface(
941
- fn=swap_video_custom_mapping,
942
- inputs=[
943
- gr.Gallery(label="Source Images", type="numpy", columns=3),
944
- gr.Video(label="Target Video"),
945
- gr.Textbox(label="Mapping (comma-separated, e.g. 2,1,3)"),
946
- gr.Checkbox(label="Delete video_frames directory after processing", value=True),
947
- gr.Checkbox(label="Add audio from original video", value=True),
948
- gr.Checkbox(label="Copy swapped video (no audio) to Google Drive root", value=False)
949
- ],
950
- outputs=[
951
- gr.Video(label="Swapped Video"),
952
- gr.Textbox(label="Log Output", lines=8, interactive=False)
953
- ],
954
- )
955
- with gr.Tab("SingleSrcMultiVideo"):
956
- gr.Interface(
957
- fn=swap_single_src_multi_video,
958
- inputs=[
959
- gr.Image(label="Source Image"),
960
- gr.File(label="Target Videos (select multiple)", file_count="multiple", type="filepath"),
961
- gr.Textbox(label="Destination Face Indices (comma-separated, e.g. 1,2,1)"),
962
- gr.Checkbox(label="Delete video_frames directory after processing", value=True),
963
- gr.Checkbox(label="Add audio from original video", value=True),
964
- gr.Checkbox(label="Copy swapped video (no audio) to Google Drive root", value=False)
965
- ],
966
- outputs=[
967
- gr.Gallery(label="Swapped Videos", type="filepath"),
968
- gr.Textbox(label="Log Output", lines=8, interactive=False)
969
- ],
970
- )
971
 
 
 
 
 
 
 
 
972
 
973
  if __name__ == "__main__":
974
- parser = argparse.ArgumentParser()
975
- parser.add_argument("--share", action="store_true", help="Launch Gradio with share=True")
976
- args = parser.parse_args()
977
- demo.launch(share=args.share)
 
1
  import gradio as gr
 
2
  import cv2
3
  import numpy as np
4
+ import insightface
5
+ from insightface.app import FaceAnalysis
6
+ from insightface.model_zoo import get_model
7
+ import urllib.request
8
+ import os
9
+ import tempfile
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
10
 
11
+ # Download model if not exists
12
+ MODEL_URL = "https://huggingface.co/ezioruan/inswapper_128.onnx/resolve/main/inswapper_128.onnx"
13
+ MODEL_PATH = "inswapper_128.onnx"
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
14
 
15
+ if not os.path.exists(MODEL_PATH):
16
+ print("📥 Downloading face swap model...")
17
+ urllib.request.urlretrieve(MODEL_URL, MODEL_PATH)
 
 
 
18
 
19
+ # Initialize face analysis
20
+ print("🔄 Loading face analysis models...")
21
+ face_app = FaceAnalysis(name="buffalo_l", providers=["CPUExecutionProvider"])
22
+ face_app.prepare(ctx_id=0, det_size=(640, 640))
23
+ face_swapper = get_model(MODEL_PATH, providers=["CPUExecutionProvider"])
24
+ print(" Models ready!")
 
 
 
 
25
 
26
+ def process_video(source_img, target_video, quality_choice):
27
  try:
28
+ # Quality settings map
29
+ quality_map = {
30
+ "320p (Fastest)": {"width": 320, "fps_reduction": 3, "bitrate": "300k"},
31
+ "480p (Balanced)": {"width": 480, "fps_reduction": 2, "bitrate": "500k"},
32
+ "720p (Good)": {"width": 720, "fps_reduction": 1, "bitrate": "1000k"},
33
+ "1080p (Original)": {"width": None, "fps_reduction": 1, "bitrate": "2000k"}
34
+ }
35
+
36
+ settings = quality_map[quality_choice]
37
+
38
+ # Open video
39
+ cap = cv2.VideoCapture(target_video)
40
+ original_fps = cap.get(cv2.CAP_PROP_FPS)
41
+ original_width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
42
+ original_height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
43
+
44
+ # Calculate new dimensions
45
+ if settings["width"]:
46
+ new_width = settings["width"]
47
+ aspect = original_height / original_width
48
+ new_height = int(new_width * aspect)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
49
  else:
50
+ new_width = original_width
51
+ new_height = original_height
52
+
53
+ # Calculate new fps
54
+ new_fps = original_fps // settings["fps_reduction"]
55
+
56
+ # Setup output
57
+ temp_output = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False).name
58
+ fourcc = cv2.VideoWriter_fourcc(*"mp4v")
59
+ out = cv2.VideoWriter(temp_output, fourcc, new_fps, (new_width, new_height))
60
+
61
+ # Get source face
62
+ source_rgb = cv2.cvtColor(source_img, cv2.COLOR_BGR2RGB)
63
+ source_faces = face_app.get(source_rgb)
64
+
65
+ if len(source_faces) == 0:
66
+ return None, "❌ No face detected in source image"
67
+
68
+ source_face = source_faces[0]
69
+
70
+ # Process video
71
+ frame_count = 0
72
+ while True:
73
+ ret, frame = cap.read()
74
+ if not ret:
75
+ break
76
+
77
+ # Resize frame
78
+ if settings["width"]:
79
+ frame = cv2.resize(frame, (new_width, new_height))
80
+
81
+ # Convert and swap face
82
+ frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
83
+ target_faces = face_app.get(frame_rgb)
84
+
85
+ if len(target_faces) > 0:
86
+ result = face_swapper.get(frame_rgb, target_faces[0], source_face, paste_back=True)
87
+ frame_rgb = result
88
+
89
+ # Convert back and write
90
+ frame_bgr = cv2.cvtColor(frame_rgb, cv2.COLOR_RGB2BGR)
91
+ out.write(frame_bgr)
92
+
93
+ frame_count += 1
94
+
95
  cap.release()
96
+ out.release()
97
+
98
+ return temp_output, f"✅ Video processed at {quality_choice} - {frame_count} frames"
99
+
100
+ except Exception as e:
101
+ return None, f" Error: {str(e)}"
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
102
 
103
+ # Create UI
104
+ with gr.Blocks(theme=gr.themes.Soft(), title="FaceSwapAll") as demo:
105
+ gr.Markdown("# FaceSwapAll with Quality Control")
106
+
107
+ with gr.Row():
108
+ with gr.Column():
109
+ source = gr.Image(label="Source Face", type="numpy", height=300)
110
+ with gr.Column():
111
+ target = gr.Video(label="Target Video", height=300)
112
+
113
+ # Quality dropdown
114
+ quality = gr.Dropdown(
115
+ label="Video Output Quality",
116
+ choices=["320p (Fastest)", "480p (Balanced)", "720p (Good)", "1080p (Original)"],
117
+ value="480p (Balanced)",
118
+ info="Lower quality = faster processing on free CPU"
119
+ )
120
+
121
+ swap_btn = gr.Button("Swap Face in Video", variant="primary", size="lg")
122
+
123
+ with gr.Row():
124
+ result = gr.Video(label="Result Video")
125
+ status = gr.Textbox(label="Status", lines=3)
126
 
127
+ swap_btn.click(
128
+ fn=process_video,
129
+ inputs=[source, target, quality],
130
+ outputs=[result, status]
131
+ )
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
132
 
133
+ gr.Markdown("""
134
+ ### ⚡ Speed Tips:
135
+ - **320p**: Fastest (8-10x speedup) - Best for testing
136
+ - **480p**: Balanced (4-5x speedup) - Good quality/speed tradeoff
137
+ - **720p**: Good quality (2x speedup)
138
+ - **1080p**: Original quality - Slowest on free CPU
139
+ """)
140
 
141
  if __name__ == "__main__":
142
+ demo.launch()