与 p2v_tool.run_echomimic_inference 一致的子进程调用，返回生成的 mp4 路径。
(source_image: str, driving_audio: str, save_video_dir: str)
| 66 | |
| 67 | |
| 68 | def _run_echomimic_subprocess(source_image: str, driving_audio: str, save_video_dir: str) -> Path | None: |
| 69 | """与 p2v_tool.run_echomimic_inference 一致的子进程调用，返回生成的 mp4 路径。""" |
| 70 | from ruamel.yaml import YAML |
| 71 | config_path = os.getenv("ECHOMIMIC_CONFIG", DEFAULT_CONFIG) |
| 72 | script_path = os.getenv("ECHOMIMIC_SCRIPT", DEFAULT_SCRIPT) |
| 73 | cwd = os.getenv("ECHOMIMIC_CWD", DEFAULT_ECHOMIMIC_CWD) |
| 74 | python_bin = os.getenv("ECHOMIMIC_PYTHON", DEFAULT_PYTHON) |
| 75 | |
| 76 | env = os.environ.copy() |
| 77 | for key in ["PYTHONHASHSEED", "PYTHONPATH"]: |
| 78 | env.pop(key, None) |
| 79 | env["PYTHONHASHSEED"] = "random" |
| 80 | # 确保子进程用到的 Python 与当前一致，避免从别处启动 API 时未传 ECHOMIMIC_PYTHON |
| 81 | env["ECHOMIMIC_PYTHON"] = python_bin |
| 82 | |
| 83 | audio_basename = os.path.splitext(os.path.basename(driving_audio))[0] |
| 84 | save_path = os.path.join(save_video_dir, audio_basename) |
| 85 | config_bak = config_path.replace(".yaml", "_{}.yaml".format(audio_basename)) |
| 86 | |
| 87 | yaml_rt = YAML() |
| 88 | yaml_rt.preserve_quotes = True |
| 89 | yaml_rt.indent(mapping=2, sequence=4, offset=2) |
| 90 | with open(config_path, "r", encoding="utf-8") as f: |
| 91 | config_data = yaml_rt.load(f) |
| 92 | config_data["test_cases"] = {source_image: [driving_audio]} |
| 93 | with open(config_bak, "w", encoding="utf-8") as f: |
| 94 | yaml_rt.dump(config_data, f) |
| 95 | |
| 96 | cmd = [ |
| 97 | python_bin, "-u", script_path, |
| 98 | "--config", config_bak, |
| 99 | "--save_path", save_path, |
| 100 | ] |
| 101 | import subprocess |
| 102 | try: |
| 103 | subprocess.run(cmd, cwd=cwd, env=env, check=True, timeout=INFER_TIMEOUT) |
| 104 | finally: |
| 105 | if os.path.exists(config_bak): |
| 106 | os.remove(config_bak) |
| 107 | |
| 108 | # 输出一般为 save_path 下 digit_person_withaudio.mp4 或同名目录下 |
| 109 | out_mp4 = Path(save_path) / "digit_person_withaudio.mp4" |
| 110 | if out_mp4.exists(): |
| 111 | return out_mp4 |
| 112 | # 兼容：可能直接写在 save_video_dir 下 |
| 113 | # for p in Path(save_video_dir).rglob("*.mp4"): |
| 114 | # return p |
| 115 | return None |
| 116 | |
| 117 | |
| 118 | @app.post("/infer") |