From 987f6b392b1740623b3fa8a5cb46fdd0b7e185b9 Mon Sep 17 00:00:00 2001 From: cuyua9 <2114364329@qq.com> Date: Fri, 14 Aug 2026 06:44:22 +0800 Subject: [PATCH] fix: extract frames for map faces fallback (#1824) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Verified this fix. Confirmed the bug by reverting just the `modules/core.py` hunk and re-running the new regression test — with the old code, `process_video`/`create_video` run against a temp directory that was never populated when `map_faces=True`, since `create_temp`/`extract_frames` were skipped for that case. That means map-faces video runs were silently broken (empty or failed output). The fix removes the `map_faces` guard so extraction always runs before the disk-based fallback, which is correct for both cases that reach this branch (map_faces=True, and non-map-faces pipe failures). `create_temp` is idempotent (mkdir exist_ok=True), so the double-call for the non-map-faces path is harmless. --- modules/core.py | 7 +- tests/test_core_map_faces_fallback.py | 137 ++++++++++++++++++++++++++ 2 files changed, 140 insertions(+), 4 deletions(-) create mode 100644 tests/test_core_map_faces_fallback.py diff --git a/modules/core.py b/modules/core.py index ed68389..520e410 100644 --- a/modules/core.py +++ b/modules/core.py @@ -276,10 +276,9 @@ def start() -> None: update_status('Falling back to disk-based processing...') extraction_start = time.time() - if not modules.globals.map_faces: - create_temp(modules.globals.target_path) - update_status('Extracting frames...') - extract_frames(modules.globals.target_path) + create_temp(modules.globals.target_path) + update_status('Extracting frames...') + extract_frames(modules.globals.target_path) extraction_time = time.time() - extraction_start temp_frame_paths = get_temp_frame_paths(modules.globals.target_path) diff --git a/tests/test_core_map_faces_fallback.py b/tests/test_core_map_faces_fallback.py new file mode 100644 index 0000000..2a7e7c1 --- /dev/null +++ b/tests/test_core_map_faces_fallback.py @@ -0,0 +1,137 @@ +import importlib +import sys +import types +import unittest +from contextlib import contextmanager +from unittest.mock import patch + + +@contextmanager +def _patched_core_import_stubs(calls, pipe_result=False): + class Processor: + NAME = "test_processor" + + def pre_start(self): + return True + + def pre_check(self): + return True + + def process_image(self, *_args, **_kwargs): + raise AssertionError("image path should not be used") + + def process_video(self, source_path, frame_paths): + calls.append(("process_video", source_path, tuple(frame_paths))) + + stubs = { + "cv2": types.SimpleNamespace( + IMREAD_COLOR=1, + imdecode=lambda *_args, **_kwargs: None, + imencode=lambda *_args, **_kwargs: ( + True, + types.SimpleNamespace(tofile=lambda *_a, **_k: None), + ), + ), + "numpy": types.SimpleNamespace(uint8=object, fromfile=lambda *_args, **_kwargs: b""), + "torch": types.SimpleNamespace( + cuda=types.SimpleNamespace(empty_cache=lambda: None) + ), + "onnxruntime": types.SimpleNamespace( + get_available_providers=lambda: ["CPUExecutionProvider"] + ), + "tensorflow": types.SimpleNamespace(), + "modules.metadata": types.SimpleNamespace(name="Deep-Live-Cam", version="test"), + "modules.ui": types.SimpleNamespace( + check_and_ignore_nsfw=lambda *_args, **_kwargs: False, + update_status=lambda *_args, **_kwargs: None, + init=lambda *_args, **_kwargs: types.SimpleNamespace(mainloop=lambda: None), + ), + "modules.processors.frame.core": types.SimpleNamespace( + get_frame_processors_modules=lambda _names: [Processor()], + process_video_in_memory=lambda *_args, **_kwargs: calls.append(("pipe",)) + or pipe_result, + ), + "modules.utilities": types.SimpleNamespace( + has_image_extension=lambda _path: False, + is_image=lambda _path: False, + is_video=lambda _path: True, + detect_fps=lambda _path: 24.0, + create_video=lambda target_path, fps: calls.append( + ("create_video", target_path, fps) + ) + or True, + extract_frames=lambda target_path: calls.append( + ("extract_frames", target_path) + ), + get_temp_frame_paths=lambda target_path: [f"{target_path}/0001.png"], + restore_audio=lambda *_args, **_kwargs: calls.append(("restore_audio",)), + create_temp=lambda target_path: calls.append(("create_temp", target_path)), + move_temp=lambda target_path, output_path: calls.append( + ("move_temp", target_path, output_path) + ), + clean_temp=lambda target_path: calls.append(("clean_temp", target_path)), + normalize_output_path=lambda _source, _target, output: output, + ), + } + with patch.dict(sys.modules, stubs, clear=False): + sys.modules.pop("modules.core", None) + yield importlib.import_module("modules.core") + sys.modules.pop("modules.core", None) + + +def _configure_video_run(core, *, map_faces): + core.modules.globals.source_path = "source.jpg" + core.modules.globals.target_path = "target.mp4" + core.modules.globals.output_path = "output.mp4" + core.modules.globals.frame_processors = ["face_swapper"] + core.modules.globals.headless = True + core.modules.globals.keep_fps = False + core.modules.globals.keep_audio = False + core.modules.globals.keep_frames = False + core.modules.globals.map_faces = map_faces + core.modules.globals.nsfw_filter = False + core.modules.globals.execution_threads = 1 + core.modules.globals.execution_providers = ["CPUExecutionProvider"] + core.modules.globals.max_memory = None + + +class MapFacesFallbackTests(unittest.TestCase): + def test_map_faces_disk_fallback_extracts_frames_before_processing(self): + calls = [] + with _patched_core_import_stubs(calls, pipe_result=False) as core: + _configure_video_run(core, map_faces=True) + + with patch.object(core.os.path, "isfile", return_value=True): + core.start() + + self.assertNotIn(("pipe",), calls) + self.assertIn(("create_temp", "target.mp4"), calls) + self.assertIn(("extract_frames", "target.mp4"), calls) + self.assertIn(("process_video", "source.jpg", ("target.mp4/0001.png",)), calls) + self.assertIn(("create_video", "target.mp4", 30.0), calls) + self.assertIn(("move_temp", "target.mp4", "output.mp4"), calls) + + step_indices = {} + for index, call in enumerate(calls): + step_indices.setdefault(call[0], index) + + self.assertLess(step_indices["create_temp"], step_indices["extract_frames"]) + self.assertLess(step_indices["extract_frames"], step_indices["process_video"]) + self.assertLess(step_indices["process_video"], step_indices["create_video"]) + self.assertLess(step_indices["create_video"], step_indices["move_temp"]) + + def test_non_map_faces_pipe_success_does_not_extract_frames(self): + calls = [] + with _patched_core_import_stubs(calls, pipe_result=True) as core: + _configure_video_run(core, map_faces=False) + + with patch.object(core.os.path, "isfile", return_value=True): + core.start() + + self.assertIn(("pipe",), calls) + self.assertNotIn(("extract_frames", "target.mp4"), calls) + self.assertNotIn(("process_video", "source.jpg", ("target.mp4/0001.png",)), calls) + + +if __name__ == "__main__": + unittest.main()