829 lines
36 KiB
Python
829 lines
36 KiB
Python
import os
|
|
import shutil
|
|
import tempfile
|
|
import subprocess
|
|
from pathlib import Path
|
|
import cv2
|
|
import numpy as np
|
|
|
|
from core.config import SCRIPT_DIR, cv2_imread_unicode, open_videocapture_unicode, cleanup_temp_file
|
|
from core.face_engine import (
|
|
match_face_color_lab, enhance_face_region, apply_soft_face_blend,
|
|
normalize_bbox, swap_effect_ratio, build_face_mask,
|
|
)
|
|
|
|
VIDEO_EXTENSIONS = {".mp4", ".mov", ".avi", ".mkv", ".webm", ".m4v"}
|
|
# Ordnerbestandteile, die als "eigene Ausgabe" gelten und beim erneuten
|
|
# Batch-Lauf NICHT als Eingabe ingestiert werden duerfen.
|
|
OUTPUT_DIR_MARKERS = {"output", "outputs", "ausgabe", "failed", "output_videos", "result", "results"}
|
|
|
|
|
|
def is_output_path(path, out_dir=None):
|
|
"""
|
|
True, wenn `path` im Ausgabe-Baum liegt (Ausgabeordner selbst oder 'failed/').
|
|
|
|
Vorher: `"output" not in p.parts` - das greift fuer den tatsaechlichen
|
|
Config-Default 'Eingabe/output_videos' NICHT, weil 'output_videos' ein
|
|
eigener Part ist. Folge: fertige Videos werden beim naechsten Lauf
|
|
erneut als Quelle eingelesen und immer wieder getauscht.
|
|
"""
|
|
parts = {p.lower() for p in Path(path).parts}
|
|
if parts & OUTPUT_DIR_MARKERS:
|
|
return True
|
|
if out_dir is not None:
|
|
try:
|
|
Path(path).resolve().relative_to(Path(out_dir).resolve())
|
|
return True
|
|
except (ValueError, OSError):
|
|
pass
|
|
return False
|
|
|
|
|
|
def expected_output_path(out_dir, src_path, src_root=None, suffix=None):
|
|
"""
|
|
Erwarteter Zielpfad fuer eine Eingabe - OHNE Kollisionsaufloesung.
|
|
|
|
Wichtig: 'src_path.parts' enthaelt den kompletten Pfad ab der
|
|
Dateisystemwurzel. Deshalb wird ueber src_root relativiert, sonst
|
|
landet die Ausgabe in out_dir/tests/fixtures/ statt out_dir/fixtures/.
|
|
|
|
Existiert der Pfad, ist das eine Wiederaufnahme: der Aufrufer
|
|
ueberspringt, statt die Datei umzubenennen oder zu ueberschreiben.
|
|
"""
|
|
out_dir = Path(out_dir)
|
|
src_path = Path(src_path)
|
|
if src_root is not None:
|
|
try:
|
|
rel = src_path.resolve().relative_to(Path(src_root).resolve())
|
|
parts = [q for q in rel.parts if q not in ("..", ".")]
|
|
except (ValueError, OSError):
|
|
parts = [src_path.name]
|
|
else:
|
|
parts = [src_path.name]
|
|
base = out_dir.joinpath(*parts) if parts else out_dir
|
|
base = base.with_suffix(suffix) if suffix else base
|
|
return base
|
|
|
|
|
|
def is_within(path, directory):
|
|
"""True, wenn path unterhalb von directory liegt."""
|
|
try:
|
|
Path(path).resolve().relative_to(Path(directory).resolve())
|
|
return True
|
|
except (ValueError, OSError):
|
|
return False
|
|
|
|
|
|
def safe_output_path(out_dir, src_path, used, src_root=None):
|
|
"""
|
|
Kollisionsfreier Zielpfad: behaelt die Unterordner-Struktur und macht
|
|
gleichnamige Dateien eindeutig.
|
|
|
|
Vorher: `outdir / img_path.name` - zwei Bilder namens 'name.png' in
|
|
verschiedenen Unterordnern landeten beide auf 'name.png'.
|
|
|
|
src_root: Wurzel des Eingabebaums. Ohne Angabe wird der Elternordner
|
|
der Datei verwendet, sodass die Ausgabe nicht den kompletten absoluten
|
|
Pfad unterhalb von out_dir dupliziert.
|
|
"""
|
|
out_dir = Path(out_dir)
|
|
src_path = Path(src_path)
|
|
|
|
if src_root is not None:
|
|
try:
|
|
rel = src_path.resolve().relative_to(Path(src_root).resolve())
|
|
parts = [p for p in rel.parts if p not in ("..", ".")]
|
|
except (ValueError, OSError):
|
|
parts = [src_path.name]
|
|
else:
|
|
# Absoluten Prefix abschneiden: nur die letzten zwei Ebenen
|
|
parts = list(src_path.parts[-2:]) if len(src_path.parts) >= 2 else [src_path.name]
|
|
parts = [p for p in parts if p not in ("/", "\\", "..", ".")]
|
|
|
|
candidate = out_dir.joinpath(*parts) if parts else out_dir / src_path.name
|
|
if candidate not in used and not candidate.exists():
|
|
used.add(candidate)
|
|
return candidate
|
|
stem, suffix = candidate.stem, candidate.suffix
|
|
for i in range(2, 10000):
|
|
alt = candidate.with_name(f"{stem}__{i}{suffix}")
|
|
if alt not in used and not alt.exists():
|
|
used.add(alt)
|
|
return alt
|
|
raise RuntimeError(f"Kein freier Zielname mehr fuer: {src_path}")
|
|
|
|
|
|
class VideoSwapEngine:
|
|
"""High-quality video face swapping engine with temporal landmark smoothing and tracking."""
|
|
|
|
def __init__(self, swapper, log_fn=print):
|
|
self.swapper = swapper
|
|
self.log = log_fn
|
|
# Tracking & stabilization parameters
|
|
self.use_tracking = True
|
|
self.use_occlusion = True
|
|
# WICHTIG: insightface/model_zoo/inswapper.py positioniert den Swap ueber
|
|
# `target_face.kps` (norm_crop2), NICHT ueber die BBox. Eine BBox-EMA hat
|
|
# daher keinen messbaren Einfluss auf die Swap-Position. Daher wird die
|
|
# Glaettung auf die Landmarken gelegt; ema_alpha_kps ist der wirksame
|
|
# Regler, ema_alpha bleibt fuer die (sekundaere) BBox erhalten.
|
|
self.ema_alpha = 0.65 # BBox-Glaettung (sekundaer)
|
|
# Kalibriert (tests/calibrate_alpha.py): 0.40 ergibt 30% weniger Jitter
|
|
# bei 3 Frames Nachfuehr-Lag. 0.25 waere glueatter, kostet aber 5
|
|
# Frames - dann bleibt das Gesicht bei Bewegung sichtbar stehen.
|
|
self.ema_alpha_kps = 0.40 # Landmarken-Glaettung (primaer, wirkt auf den Swap)
|
|
self.min_det_score = 0.45
|
|
self.start_det_score = 0.60
|
|
self.track_memory = 4
|
|
self.min_iou = 0.05
|
|
self.max_center_jump = 0.20
|
|
self.min_area_ratio = 0.40
|
|
self.max_area_ratio = 2.40
|
|
# Frames, die ein neuer Track bestaetigt sein muss, bevor getauscht wird.
|
|
# Default 1 = sofortiger Start wie zuvor; 2 liefert weniger Fehlstarts,
|
|
# kostet aber den ersten Frame des Abschnitts.
|
|
self.confirm_frames = 1
|
|
self.fade_in_step = 0.45
|
|
self.fade_out_step = 0.50
|
|
# Neu: Occlusion-Gate und Loss-Hysterese
|
|
self.occlusion_tolerance = 3 # Frames ohne Detektion, bevor der Track faellt
|
|
# Ausreisser-Gate in Bildbreiten. Absichtlich GROSSZUEGIG: es soll nur
|
|
# offensichtliche Fehltreffer (Detektor springt ins falsche Bild) fangen,
|
|
# nicht normale Bewegung. Ein zu kleiner Wert (z. B. 0.05) verwarf jede
|
|
# echte Bewegung und machte den Track immun gegen jeden Positionswechsel.
|
|
self.max_kps_jump = 0.45
|
|
self.fade_enabled = False # Fade ein/aus
|
|
|
|
@staticmethod
|
|
def _bbox_iou(a, b):
|
|
ax1, ay1, ax2, ay2 = a
|
|
bx1, by1, bx2, by2 = b
|
|
ix1, iy1 = max(ax1, bx1), max(ay1, by1)
|
|
ix2, iy2 = min(ax2, bx2), min(ay2, by2)
|
|
iw, ih = max(0.0, ix2 - ix1), max(0.0, iy2 - iy1)
|
|
inter = iw * ih
|
|
if inter <= 0:
|
|
return 0.0
|
|
a_area = max(1e-9, (ax2 - ax1) * (ay2 - ay1))
|
|
b_area = max(1e-9, (bx2 - bx1) * (by2 - by1))
|
|
return inter / max(1e-9, a_area + b_area - inter)
|
|
|
|
@staticmethod
|
|
def _embedding_sim(a, b):
|
|
if a is None or b is None:
|
|
return None
|
|
va = np.asarray(a, dtype=np.float32)
|
|
vb = np.asarray(b, dtype=np.float32)
|
|
na, nb = np.linalg.norm(va), np.linalg.norm(vb)
|
|
if na < 1e-6 or nb < 1e-6:
|
|
return None
|
|
return float(np.dot(va, vb) / (na * nb))
|
|
|
|
def _build_face_entry(self, face, w, h):
|
|
x1, y1, x2, y2 = [float(v) for v in face.bbox]
|
|
bw, bh = max(1.0, x2 - x1), max(1.0, y2 - y1)
|
|
cx = (x1 + x2) * 0.5 / max(1.0, float(w))
|
|
cy = (y1 + y2) * 0.5 / max(1.0, float(h))
|
|
norm_bbox = (x1 / w, y1 / h, x2 / w, y2 / h)
|
|
area = (bw * bh) / max(1.0, float(w * h))
|
|
kps = getattr(face, "kps", None)
|
|
norm_kps = None
|
|
if kps is not None:
|
|
nk = np.asarray(kps, dtype=np.float32).copy()
|
|
nk[:, 0] /= max(1.0, float(w))
|
|
nk[:, 1] /= max(1.0, float(h))
|
|
norm_kps = nk
|
|
|
|
return {
|
|
"face": face,
|
|
"bbox": face.bbox,
|
|
"norm_bbox": norm_bbox,
|
|
"cx": cx,
|
|
"cy": cy,
|
|
"area": area,
|
|
"det_score": float(getattr(face, "det_score", 0.0)),
|
|
"embedding": getattr(face, "normed_embedding", None),
|
|
"kps": norm_kps,
|
|
"raw_kps": kps,
|
|
"px_w": bw,
|
|
"px_h": bh,
|
|
}
|
|
|
|
def _smooth_tracked_face(self, current_entry, track_state, frame_shape=None):
|
|
"""
|
|
Glaettet die Landmarken (das ist die Position, die inswapper benutzt)
|
|
und leitet daraus eine konsistente, geklemmte BBox ab.
|
|
|
|
Zwei Probleme des Originals werden hier behoben:
|
|
1) Die BBox-EMA hatte keinen Einfluss auf den Swap, weil inswapper
|
|
ueber `kps` positioniert (insightface inswapper.py:51).
|
|
2) Die BBox wurde ungeclipped uebernommen. Bei Tracking-Ausfall oder
|
|
schneller Bewegung konnte sie negativ/ausserhalb werden.
|
|
"""
|
|
cur_kps = current_entry.get("raw_kps")
|
|
img_w = frame_shape[1] if frame_shape is not None else 1.0
|
|
img_h = frame_shape[0] if frame_shape is not None else 1.0
|
|
|
|
if cur_kps is not None:
|
|
cur_kps = np.asarray(cur_kps, dtype=np.float32)
|
|
prev_kps = track_state.get("smooth_kps")
|
|
|
|
# Ausreisser-Gate: ein Landmarken-Sprung ueber max_kps_jump ist
|
|
# fast sicher ein Fehltreffer (Frame-Mismatch), keine Bewegung.
|
|
if prev_kps is not None and prev_kps.shape == cur_kps.shape:
|
|
jump = float(np.linalg.norm(cur_kps - prev_kps, axis=1).mean())
|
|
if jump > self.max_kps_jump * max(img_w, img_h):
|
|
# Ausreisser verwerfen, letzte stabile Position halten
|
|
cur_kps = prev_kps.copy()
|
|
else:
|
|
a = self.ema_alpha_kps
|
|
cur_kps = a * cur_kps + (1.0 - a) * prev_kps
|
|
|
|
track_state["smooth_kps"] = cur_kps.copy()
|
|
current_entry["face"].kps = cur_kps
|
|
|
|
# BBox aus den geglätteten Landmarken + Detektions-BBox vereinigen
|
|
# (Landmarken definieren das Gesicht, BBox den Kontext)
|
|
kps_x1 = float(cur_kps[:, 0].min())
|
|
kps_y1 = float(cur_kps[:, 1].min())
|
|
kps_x2 = float(cur_kps[:, 0].max())
|
|
kps_y2 = float(cur_kps[:, 1].max())
|
|
det_bbox = np.asarray(current_entry["bbox"], dtype=np.float32)
|
|
kps_w = max(1.0, kps_x2 - kps_x1)
|
|
kps_h = max(1.0, kps_y2 - kps_y1)
|
|
cx = (kps_x1 + kps_x2) * 0.5
|
|
cy = (kps_y1 + kps_y2) * 0.5
|
|
merged = np.array([
|
|
min(det_bbox[0], cx - kps_w * 0.90),
|
|
min(det_bbox[1], cy - kps_h * 1.05),
|
|
max(det_bbox[2], cx + kps_w * 0.90),
|
|
max(det_bbox[3], cy + kps_h * 1.05),
|
|
], dtype=np.float32)
|
|
|
|
prev_bbox = track_state.get("smooth_bbox")
|
|
if prev_bbox is not None and prev_bbox.shape == merged.shape:
|
|
a = self.ema_alpha
|
|
merged = a * merged + (1.0 - a) * prev_bbox
|
|
|
|
# Harte Klemmung auf das Bild - verhindert Out-of-Bounds-Zugriffe
|
|
merged[0] = max(0.0, min(img_w, merged[0]))
|
|
merged[1] = max(0.0, min(img_h, merged[1]))
|
|
merged[2] = max(0.0, min(img_w, merged[2]))
|
|
merged[3] = max(0.0, min(img_h, merged[3]))
|
|
if merged[2] <= merged[0]:
|
|
merged[2] = min(img_w, merged[0] + 1.0)
|
|
if merged[3] <= merged[1]:
|
|
merged[3] = min(img_h, merged[1] + 1.0)
|
|
|
|
track_state["smooth_bbox"] = merged.copy()
|
|
current_entry["face"].bbox = merged
|
|
current_entry["bbox"] = merged
|
|
current_entry["cx"] = float((merged[0] + merged[2]) * 0.5 / max(1.0, float(img_w)))
|
|
current_entry["cy"] = float((merged[1] + merged[3]) * 0.5 / max(1.0, float(img_h)))
|
|
else:
|
|
# Keine Landmarken: BBox wenigstens glaetten und klemmen
|
|
cur_bbox = np.asarray(current_entry["bbox"], dtype=np.float32).copy()
|
|
prev_bbox = track_state.get("smooth_bbox")
|
|
if prev_bbox is not None and prev_bbox.shape == cur_bbox.shape:
|
|
cur_bbox = self.ema_alpha * cur_bbox + (1.0 - self.ema_alpha) * prev_bbox
|
|
cur_bbox[0] = max(0.0, min(img_w, cur_bbox[0]))
|
|
cur_bbox[1] = max(0.0, min(img_h, cur_bbox[1]))
|
|
cur_bbox[2] = max(0.0, min(img_w, cur_bbox[2]))
|
|
cur_bbox[3] = max(0.0, min(img_h, cur_bbox[3]))
|
|
if cur_bbox[2] <= cur_bbox[0]:
|
|
cur_bbox[2] = min(img_w, cur_bbox[0] + 1.0)
|
|
if cur_bbox[3] <= cur_bbox[1]:
|
|
cur_bbox[3] = min(img_h, cur_bbox[1] + 1.0)
|
|
track_state["smooth_bbox"] = cur_bbox.copy()
|
|
current_entry["face"].bbox = cur_bbox
|
|
current_entry["bbox"] = cur_bbox
|
|
|
|
def _apply_frame_fade(self, result, original, swapped_count, track_state):
|
|
"""
|
|
Weiche Ein-/Ausblendung des Swap-Effekts an Track-Grenzen.
|
|
|
|
Vorher existierten fade_in_step/fade_out_step als Parameter, wurden
|
|
aber nie benutzt - der Swap pops bei Track-Aufbau und Track-Verlust
|
|
sichtbar auf. Hier wird der Effekt in den ersten/letzten Frames
|
|
eines durchgehenden Track-Abschnitts weich eingeblendet.
|
|
"""
|
|
if not self.fade_enabled or track_state is None or swapped_count <= 0:
|
|
return result
|
|
if original is None or original.shape != result.shape:
|
|
return result
|
|
|
|
seq = track_state.get("fade_seq")
|
|
if seq is None or seq == 0:
|
|
seq = 1
|
|
track_state["fade_seq"] = seq + 1
|
|
|
|
w = max(1, int(round(seq * self.fade_in_step)))
|
|
# Am Ende wussten wir nicht, dass es endet; Rueckblendung daher
|
|
# beim ersten Frame ohne Swap (siehe swap_video_frame/_reset_track).
|
|
if w >= 1.0:
|
|
a = min(1.0, seq * self.fade_in_step)
|
|
blended = original.astype(np.float32) * (1.0 - a) + result.astype(np.float32) * a
|
|
return np.clip(blended, 0, 255).astype(np.uint8)
|
|
return result
|
|
|
|
def _reset_track(self, track_state):
|
|
"""Track vollstaendig zuruecksetzen."""
|
|
track_state["active"] = None
|
|
track_state["smooth_kps"] = None
|
|
track_state["smooth_bbox"] = None
|
|
track_state["stable_frames"] = 0
|
|
track_state["miss"] = 0
|
|
track_state["fade_seq"] = 0
|
|
|
|
def _register_miss(self, track_state):
|
|
"""
|
|
Einen Detektionsausfall verbuchen. Der Track faellt erst, wenn die
|
|
Ausfalltoleranz ueberschritten ist - kurze Verdeckungen/Blenden
|
|
(Detektionsluecken) sollen den Track nicht zerstoeren.
|
|
"""
|
|
if track_state is None:
|
|
return
|
|
track_state["miss"] = int(track_state.get("miss", 0)) + 1
|
|
tol = self.occlusion_tolerance if self.use_occlusion else self.track_memory
|
|
if track_state["miss"] > tol:
|
|
self._reset_track(track_state)
|
|
|
|
def _occluded_face(self, track_state, work_img):
|
|
"""
|
|
Letztes bekanntes Gesicht fuer einen verdeckten Frame rekonstruieren.
|
|
|
|
Loest das Kernproblem 'bei jedem kurzen Detektionsausfall blitzt das
|
|
Originalgesicht im Video auf': statt das Frame unveraendert
|
|
durchzureichen, wird mit der zuletzt geschaetzten Pose weitergearbeitet.
|
|
"""
|
|
if track_state is None or not self.use_occlusion:
|
|
return None
|
|
kps = track_state.get("smooth_kps")
|
|
bbox = track_state.get("smooth_bbox")
|
|
if kps is None or bbox is None:
|
|
return None
|
|
miss = int(track_state.get("miss", 0))
|
|
if miss <= 0 or miss > (self.occlusion_tolerance if self.use_occlusion else 0):
|
|
return None
|
|
|
|
# Erst ab dem zweiten Ausfallframe rekonstruieren: der erste ist oft
|
|
# nur eine einzelne schwache Detektion.
|
|
if miss < 2:
|
|
return None
|
|
|
|
face = getattr(track_state.get("active"), "face", None)
|
|
import copy
|
|
try:
|
|
ghost = copy.copy(face) if face is not None else None
|
|
except Exception:
|
|
ghost = None
|
|
if ghost is None:
|
|
class _G:
|
|
pass
|
|
ghost = _G()
|
|
ghost.kps = np.asarray(kps, dtype=np.float32).copy()
|
|
nb = normalize_bbox(bbox, work_img.shape)
|
|
if nb is None:
|
|
return None
|
|
ghost.bbox = np.asarray(nb, dtype=np.float32)
|
|
return ghost
|
|
|
|
def _filter_video_faces(self, faces, frame_shape, track_state):
|
|
h, w = frame_shape[:2]
|
|
candidates = []
|
|
for f in faces:
|
|
entry = self._build_face_entry(f, w, h)
|
|
if entry["px_w"] < 40 or entry["px_h"] < 40 or entry["det_score"] < self.min_det_score:
|
|
continue
|
|
candidates.append(entry)
|
|
|
|
if not candidates:
|
|
self._register_miss(track_state)
|
|
return []
|
|
|
|
candidates.sort(key=lambda c: (c["area"], c["det_score"]), reverse=True)
|
|
active = track_state.get("active")
|
|
|
|
if active is None:
|
|
# 'confirm_frames' braucht mehrere belastbare Frames, bevor ein
|
|
# Track als stabil gilt. Vorher war der Parameter definiert, aber
|
|
# nie gelesen - der Track startete also nach einem einzigen
|
|
# schwachen Treffer und konnte sofort wieder fallen.
|
|
seed = next((c for c in candidates if c["det_score"] >= self.start_det_score), None)
|
|
if seed is None:
|
|
# Schwacher Kandidat: bester verbleibender darf den Track halten,
|
|
# sonst zerfaellt die Spur bei jedem kurzzeitigen Score-Einbruch.
|
|
seed = candidates[0] if (candidates and self.min_det_score <= candidates[0]["det_score"]) else None
|
|
if seed is None:
|
|
self._register_miss(track_state)
|
|
return []
|
|
|
|
# Vorgänger zaehlen, die bereits auf denselben Kandidaten passen
|
|
pending = track_state.get("pending_seed")
|
|
if (pending is not None and pending["bbox"] == seed["bbox"]
|
|
and pending["det_score"] >= self.min_det_score):
|
|
track_state["pending_count"] = int(track_state.get("pending_count", 0)) + 1
|
|
else:
|
|
track_state["pending_seed"] = seed
|
|
track_state["pending_count"] = 1
|
|
|
|
needed = max(1, int(self.confirm_frames))
|
|
if int(track_state.get("pending_count", 0)) < needed:
|
|
# Noch nicht bestaetigt: nichts tauschen, aber Track nicht
|
|
# zerstoeren - der Kandidat wartet auf seine Bestaetigung.
|
|
return []
|
|
|
|
track_state["pending_seed"] = None
|
|
track_state["pending_count"] = 0
|
|
track_state["candidate"] = seed
|
|
track_state["miss"] = 0
|
|
track_state["active"] = seed
|
|
track_state["stable_frames"] = 1
|
|
self._smooth_tracked_face(seed, track_state, frame_shape)
|
|
return [seed["face"]]
|
|
|
|
best = None
|
|
best_score = -1.0
|
|
for c in candidates:
|
|
iou = self._bbox_iou(c["bbox"], active["bbox"])
|
|
dx = c["cx"] - active["cx"]
|
|
dy = c["cy"] - active["cy"]
|
|
center_shift = (dx * dx + dy * dy) ** 0.5
|
|
area_ratio = c["area"] / max(1e-9, active["area"])
|
|
|
|
if not (self.min_area_ratio <= area_ratio <= self.max_area_ratio):
|
|
continue
|
|
if iou < self.min_iou and center_shift > self.max_center_jump:
|
|
continue
|
|
|
|
emb_sim = self._embedding_sim(c["embedding"], active["embedding"])
|
|
iou_score = max(0.0, min(1.0, iou / 0.35))
|
|
motion_score = max(0.0, min(1.0, 1.0 - center_shift / self.max_center_jump))
|
|
det_score = max(0.0, min(1.0, c["det_score"]))
|
|
emb_score = 0.5 if emb_sim is None else max(0.0, min(1.0, (emb_sim + 1.0) * 0.5))
|
|
|
|
score = 0.35 * iou_score + 0.30 * motion_score + 0.15 * det_score + 0.20 * emb_score
|
|
if score > best_score:
|
|
best_score = score
|
|
best = c
|
|
|
|
if best is None or best_score < 0.25:
|
|
self._register_miss(track_state)
|
|
return []
|
|
|
|
track_state["candidate"] = best
|
|
track_state["miss"] = 0
|
|
track_state["stable_frames"] = int(track_state.get("stable_frames", 0)) + 1
|
|
track_state["active"] = best
|
|
self._smooth_tracked_face(best, track_state, frame_shape)
|
|
return [best["face"]]
|
|
|
|
def swap_video_frame(self, frame, src_face, track_state=None):
|
|
"""Swap face in a single video frame with temporal stabilization and tracking."""
|
|
if frame is None or getattr(frame, "size", 0) == 0:
|
|
return frame, 0
|
|
if src_face is None:
|
|
return frame, 0
|
|
|
|
faces, work_img, scale = self.swapper.detect_faces(frame)
|
|
if not faces:
|
|
if track_state is not None:
|
|
self._register_miss(track_state)
|
|
# Occlusion-Rekonstruktion: mit letzter Pose weitertauschen,
|
|
# damit das Originalgesicht nicht kurz aufblitzt.
|
|
ghost = self._occluded_face(track_state, frame)
|
|
if ghost is not None:
|
|
result, cnt = self._swap_single(ghost, src_face, work_img, scale, frame)
|
|
return result, cnt
|
|
return frame, 0
|
|
|
|
if track_state is not None and self.use_tracking:
|
|
target_faces = self._filter_video_faces(faces, work_img.shape, track_state)
|
|
if not target_faces:
|
|
self._register_miss(track_state)
|
|
ghost = self._occluded_face(track_state, frame)
|
|
if ghost is not None:
|
|
result, cnt = self._swap_single(ghost, src_face, work_img, scale, frame)
|
|
return result, cnt
|
|
return frame, 0
|
|
else:
|
|
target_faces = [self.swapper.pick_primary_face(faces)]
|
|
|
|
result, cnt = self._swap_single(target_faces, src_face, work_img, scale, frame, track_state)
|
|
return result, cnt
|
|
|
|
def _swap_single(self, target_faces, src_face, work_img, scale, frame, track_state=None):
|
|
"""
|
|
Swap auf einer (bereits validierten) Ziel-Gesichtliste und skaliert
|
|
das Ergebnis auf die native Aufloesung zurueck.
|
|
"""
|
|
if not isinstance(target_faces, (list, tuple)):
|
|
target_faces = [target_faces]
|
|
original = work_img.copy()
|
|
result = work_img.copy()
|
|
swapped_bboxes = []
|
|
|
|
for face in target_faces:
|
|
if face is None:
|
|
continue
|
|
nb = normalize_bbox(getattr(face, "bbox", None), work_img.shape)
|
|
if nb is None:
|
|
continue
|
|
face.bbox = np.asarray(nb, dtype=np.float32)
|
|
|
|
before = result.copy()
|
|
try:
|
|
result = self.swapper.swapper.get(result, face, src_face, paste_back=True)
|
|
except Exception as e:
|
|
self.log(f" Frame-Swap fehlgeschlagen ({type(e).__name__}: {e})")
|
|
result = before
|
|
continue
|
|
|
|
# Normalisierte Effektpruefung (0..1) statt der toten 0.01-Schwelle
|
|
effect = swap_effect_ratio(result, before, nb)
|
|
if effect < getattr(self.swapper, "min_swap_effect", 0.004):
|
|
result = before
|
|
continue
|
|
|
|
swapped_bboxes.append(nb)
|
|
|
|
fmask = None
|
|
if getattr(self.swapper, "face_mask_quality", True):
|
|
try:
|
|
fmask = build_face_mask(work_img.shape, getattr(face, "kps", None), nb)
|
|
except Exception:
|
|
fmask = None
|
|
|
|
if self.swapper.color:
|
|
result = match_face_color_lab(result, original, nb, face_mask=fmask)
|
|
if self.swapper.soft_blend:
|
|
result = apply_soft_face_blend(result, original, nb, face_mask=fmask)
|
|
if self.swapper.enhance:
|
|
result = enhance_face_region(result, nb, sharpen=True, face_mask=fmask)
|
|
|
|
if self.swapper.use_restoration and self.swapper.restorer is not None and swapped_bboxes:
|
|
result = self.swapper.restorer.restore_faces(result, swapped_bboxes)
|
|
|
|
# Fade wirkt auf die Swap-Position und muss daher vor dem
|
|
# Zurueckskalieren passieren (original liegt in work_img-Aufloesung).
|
|
if track_state is not None and swapped_bboxes:
|
|
result = self._apply_frame_fade(result, original, len(swapped_bboxes), track_state)
|
|
|
|
if scale != 1.0:
|
|
result = cv2.resize(result, (frame.shape[1], frame.shape[0]), interpolation=cv2.INTER_AREA)
|
|
self.swapper.last_swapped_bboxes = [
|
|
(b[0] / scale, b[1] / scale, b[2] / scale, b[3] / scale) for b in swapped_bboxes
|
|
]
|
|
else:
|
|
self.swapper.last_swapped_bboxes = list(swapped_bboxes)
|
|
|
|
return result, len(swapped_bboxes)
|
|
|
|
def process_video_file(self, target_video, src_face, out_video, progress_cb=None, cancel_check=None):
|
|
"""
|
|
Process an entire video file, writing swapped frames to a temporary video
|
|
and muxing audio with FFmpeg with high visual quality.
|
|
"""
|
|
cap, cap_tmp_copy = open_videocapture_unicode(target_video, log_fn=self.log)
|
|
if not cap.isOpened():
|
|
raise RuntimeError(f"Video konnte nicht geoeffnet werden: {target_video}")
|
|
|
|
fps = cap.get(cv2.CAP_PROP_FPS)
|
|
if not fps or fps <= 0 or not np.isfinite(fps):
|
|
fps = 25.0
|
|
total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT) or 0)
|
|
|
|
# Erste Frame lesen: aus ihr kommt die WAHRE Groesse. Die
|
|
# CAP_PROP-Werte koennen bei Rotation falsch sein - der Writer
|
|
# wuerde dann still skalieren oder ein leeres Video erzeugen.
|
|
ret, first_frame = cap.read()
|
|
if not ret or first_frame is None:
|
|
cap.release()
|
|
cleanup_temp_file(cap_tmp_copy)
|
|
raise RuntimeError(f"Video enthaelt keine lesbaren Frames: {target_video}")
|
|
|
|
fh, fw = first_frame.shape[:2]
|
|
prop_w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
|
|
prop_h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
|
|
if (prop_w, prop_h) != (fw, fh):
|
|
self.log(f" Hinweis: Video-Metadaten {prop_w}x{prop_h} weichen ab von "
|
|
f"echten Frames {fw}x{fh} - nutze Frame-Groesse.")
|
|
|
|
# Temp-Datei ins echte Temp-Verzeichnis, nicht ins Installationsverzeichnis.
|
|
# SCRIPT_DIR kann read-only sein (Program Files, Netzlaufwerk, volle Platte).
|
|
tmp_video = Path(tempfile.mkstemp(prefix="faceswap_raw_", suffix=".mp4")[1])
|
|
tmp_dir = tmp_video.parent
|
|
try:
|
|
return self._process_frames(cap, first_frame, src_face, tmp_video, out_video,
|
|
fps, (fw, fh), total_frames, progress_cb, cancel_check)
|
|
finally:
|
|
cap.release()
|
|
cleanup_temp_file(cap_tmp_copy)
|
|
cleanup_temp_file(tmp_video)
|
|
try:
|
|
tmp_dir.rmdir()
|
|
except OSError:
|
|
pass
|
|
|
|
def _process_frames(self, cap, first_frame, src_face, tmp_video, out_video, fps,
|
|
frame_size, total_frames, progress_cb, cancel_check):
|
|
"""Frames verarbeiten, muxen und ein ehrliches Ergebnis liefern."""
|
|
width, height = frame_size
|
|
fourcc = cv2.VideoWriter_fourcc(*"mp4v")
|
|
writer = cv2.VideoWriter(str(tmp_video), fourcc, float(fps), (width, height))
|
|
if not writer.isOpened():
|
|
raise RuntimeError(f"Video-Writer konnte nicht initialisiert werden fuer: {out_video}")
|
|
|
|
done = 0
|
|
swapped_frames = 0
|
|
track_state = {} if self.use_tracking else None
|
|
|
|
def _handle(frame):
|
|
nonlocal done, swapped_frames
|
|
result, face_count = self.swap_video_frame(frame, src_face, track_state=track_state)
|
|
writer.write(result)
|
|
if face_count > 0:
|
|
swapped_frames += 1
|
|
done += 1
|
|
if progress_cb and (done % 2 == 0 or done == total_frames):
|
|
progress_cb(done, total_frames, result)
|
|
|
|
try:
|
|
if cancel_check and cancel_check():
|
|
self.log(" Video-Verarbeitung abgebrochen (vor Frame 1).")
|
|
else:
|
|
_handle(first_frame)
|
|
while True:
|
|
if cancel_check and cancel_check():
|
|
self.log(" Video-Verarbeitung abgebrochen.")
|
|
break
|
|
ret, frame = cap.read()
|
|
if not ret or frame is None:
|
|
break
|
|
try:
|
|
_handle(frame)
|
|
except Exception as e:
|
|
# Ein einzelner kaputter Frame darf nicht das ganze
|
|
# Video abbrechen - Originalframe durchreichen.
|
|
self.log(f" Frame {done + 1} fehlgeschlagen ({type(e).__name__}: {e}) - uebersprungen.")
|
|
writer.write(frame)
|
|
done += 1
|
|
finally:
|
|
writer.release()
|
|
|
|
if done == 0:
|
|
raise RuntimeError(f"Kein Frame verarbeitet (Video moeglicherweise defekt): {target_video}")
|
|
|
|
# Mux original audio track using FFmpeg if available
|
|
ffmpeg = shutil.which("ffmpeg")
|
|
out_p = Path(out_video)
|
|
out_p.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
if ffmpeg and tmp_video.exists():
|
|
muxed = False
|
|
try:
|
|
cmd = [
|
|
ffmpeg, "-y", "-i", str(tmp_video), "-i", str(target_video),
|
|
"-c:v", "libx264", "-crf", "18", "-preset", "fast", "-pix_fmt", "yuv420p",
|
|
"-c:a", "aac", "-map", "0:v:0", "-map", "1:a:0?",
|
|
"-shortest", str(out_p)
|
|
]
|
|
r = subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
|
|
# check=True allein genuegt nicht: ffmpeg liefert bei
|
|
# Audio-Mapping-Problemen teils 0 und schreibt trotzdem nichts.
|
|
muxed = (r.returncode == 0 and out_p.exists() and out_p.stat().st_size > 1024)
|
|
if not muxed:
|
|
err = (r.stderr or b"").decode("utf-8", "replace").strip().splitlines()
|
|
self.log(f" FFmpeg-Muxing unvollstaendig (rc={r.returncode}): "
|
|
f"{err[-1] if err else 'kein Detail'}")
|
|
except Exception as e:
|
|
self.log(f" FFmpeg-Muxing fehlgeschlagen ({e}), uebernehme Rohvideo.")
|
|
muxed = False
|
|
if not muxed:
|
|
if tmp_video.exists():
|
|
shutil.move(str(tmp_video), str(out_p))
|
|
elif tmp_video.exists():
|
|
shutil.move(str(tmp_video), str(out_p))
|
|
|
|
if not out_p.exists() or out_p.stat().st_size == 0:
|
|
raise RuntimeError(f"Ausgabevideo konnte nicht erstellt werden: {out_video}")
|
|
|
|
return {
|
|
"frames_processed": done,
|
|
"frames_total": total_frames,
|
|
"frames_swapped": swapped_frames,
|
|
"coverage": (swapped_frames / done) if done else 0.0,
|
|
}
|
|
|
|
def process_batch(self, source_path, input_dir, output_dir, progress_cb=None, cancel_check=None):
|
|
"""Batch swap faces across all videos in input_dir."""
|
|
src_p = Path(source_path)
|
|
if not src_p.is_file():
|
|
raise RuntimeError(f"Quellbild nicht gefunden: {source_path}")
|
|
|
|
src_img = cv2_imread_unicode(src_p)
|
|
if src_img is None:
|
|
raise RuntimeError(f"Quellbild konnte nicht geladen werden: {source_path}")
|
|
|
|
self.swapper.init_models()
|
|
src_face = self.swapper.get_first_face(src_img)
|
|
if src_face is None:
|
|
raise RuntimeError("Kein Gesicht im Quellbild gefunden!")
|
|
|
|
if self.swapper.use_restoration and self.swapper.restorer:
|
|
try:
|
|
self.swapper.restorer.load()
|
|
except Exception as e:
|
|
self.log(f" Hinweis: GFPGAN konnte nicht vorab geladen werden ({e}).")
|
|
|
|
self.log(f"OK Quellgesicht erkannt: {src_p.name}")
|
|
|
|
indir = Path(input_dir)
|
|
outdir = Path(output_dir)
|
|
# Eingaben sammeln: alles im Ausgabe-Baum ausgeschlossen. Der alte
|
|
# Filter (`"output" not in p.parts`) liess den Config-Default
|
|
# 'Eingabe/output_videos' durch, sodass fertige Videos beim
|
|
# naechsten Lauf erneut getauscht wurden.
|
|
videos = []
|
|
skipped_outputs = 0
|
|
for p in indir.rglob("*"):
|
|
if not p.is_file() or p.suffix.lower() not in VIDEO_EXTENSIONS:
|
|
continue
|
|
if is_output_path(p, outdir) or is_within(p, outdir):
|
|
skipped_outputs += 1
|
|
continue
|
|
if is_within(p, outdir / "failed"):
|
|
skipped_outputs += 1
|
|
continue
|
|
videos.append(p)
|
|
if skipped_outputs:
|
|
self.log(f"{skipped_outputs} Ausgabe-Datei(en) aus der Eingabeliste entfernt "
|
|
f"(verhindert erneutes Verarbeiten eigener Ergebnisse).")
|
|
if not videos:
|
|
self.log("Keine Videos im Eingabe-Ordner gefunden.")
|
|
return {"total": 0, "completed": 0, "failed": 0, "skipped_outputs": skipped_outputs}
|
|
|
|
videos.sort()
|
|
outdir.mkdir(parents=True, exist_ok=True)
|
|
|
|
self.log(f"{len(videos)} Video(s) gefunden.")
|
|
failed_dir = outdir / "failed"
|
|
completed = 0
|
|
failed = 0
|
|
used = set()
|
|
|
|
for i, vp in enumerate(videos, 1):
|
|
if cancel_check and cancel_check():
|
|
self.log("Batch-Video-Verarbeitung abgebrochen.")
|
|
break
|
|
|
|
# Erwarteter Zielpfad: Unterordner erhalten, Suffix '_faceswap'.
|
|
out_video = expected_output_path(
|
|
outdir / "swapped", vp, src_root=indir, suffix="_faceswap.mp4"
|
|
)
|
|
if out_video.exists():
|
|
# Wiederaufnahme: vorhandenes Ergebnis unangetastet lassen.
|
|
self.log(f"[{i}/{len(videos)}] {vp.name} -> UEBERSPRUNGEN (bereits vorhanden)")
|
|
continue
|
|
|
|
# Nur Namenskollisionen INNERHALB dieses Laufs aufloesen.
|
|
out_video = safe_output_path(outdir / "swapped", vp, used, src_root=indir)
|
|
out_video = out_video.with_name(f"{out_video.stem}_faceswap.mp4")
|
|
if out_video.exists():
|
|
self.log(f"[{i}/{len(videos)}] {vp.name} -> UEBERSPRUNGEN (Namenskollision aufgeloest)")
|
|
continue
|
|
out_video.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
self.log(f"[{i}/{len(videos)}] Starte {vp.name} ...")
|
|
|
|
def file_progress(done_frames, total_frames, frame, _i=i, _vp=vp):
|
|
if progress_cb:
|
|
progress_cb(_i, len(videos), done_frames, total_frames, frame, _vp.name)
|
|
|
|
try:
|
|
res = self.process_video_file(vp, src_face, out_video, progress_cb=file_progress, cancel_check=cancel_check)
|
|
if res.get("frames_swapped", 0) > 0:
|
|
completed += 1
|
|
self.log(f"[{i}/{len(videos)}] {vp.name} abgeschlossen "
|
|
f"({res['frames_swapped']}/{res['frames_processed']} Frames geswappt, "
|
|
f"Abdeckung {res.get('coverage', 0.0) * 100:.0f}%).")
|
|
else:
|
|
failed += 1
|
|
failed_dir.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
shutil.copy2(vp, failed_dir / vp.name)
|
|
except Exception:
|
|
pass
|
|
self.log(f"[{i}/{len(videos)}] WARNUNG: 0 Frames geswappt in {vp.name} -> failed/{vp.name}")
|
|
except Exception as e:
|
|
failed += 1
|
|
failed_dir.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
shutil.copy2(vp, failed_dir / vp.name)
|
|
except Exception:
|
|
pass
|
|
self.log(f"[{i}/{len(videos)}] FEHLER bei {vp.name}: {e} -> failed/{vp.name}")
|
|
|
|
return {"total": len(videos), "completed": completed, "failed": failed,
|
|
"skipped_outputs": skipped_outputs}
|