Files

676 lines
36 KiB
Python

import os
import time
from pathlib import Path
from typing import Any
import streamlit as st
from loguru import logger
import httpx
from web.i18n import tr, get_language
from web.pipelines.base import PipelineUI, register_pipeline_ui
from web.components.content_input import render_version_info
from web.components.digital_tts_config import render_style_config
from web.utils.async_helpers import run_async
from web.utils.streamlit_helpers import check_and_warn_selfhost_workflow
from pixelle_video.config import config_manager
from pixelle_video.utils.os_util import create_task_output_dir
class DigitalHumanPipelineUI(PipelineUI):
"""
UI for the Digital_Human Video Generation Pipeline.
Generates videos from user-provided assets (images&videos&audio).
"""
name = "digital_human"
icon = "🤖"
@property
def display_name(self):
return tr("pipeline.digital_human.name")
@property
def description(self):
return tr("pipeline.digital_human.description")
def render(self, pixelle_video: Any):
# Three-column layout
left_col, middle_col, right_col = st.columns([1, 1, 1])
# ====================================================================
# Left Column: Asset Upload
# ====================================================================
with left_col:
asset_params = self.render_digital_human_input()
style_params = render_style_config(pixelle_video)
# bgm_params = render_bgm_section(key_prefix="asset_")
render_version_info()
# ====================================================================
# Middle Column: Video Configuration
# ====================================================================
with middle_col:
# Style configuration ()
workflow_path = self.workflow_path_config()
mode_params = self.render_digital_human_mode(asset_params["character_assets"])
# ====================================================================
# Right Column: Output Preview
# ====================================================================
with right_col:
# Combine all parameters
video_params = {
**mode_params,
**asset_params,
**style_params,
"workflow_path": workflow_path
}
self._render_output_preview(pixelle_video, video_params)
def render_digital_human_input(self) -> dict:
"""Render digital human character image upload section"""
with st.container(border=True):
st.markdown(f"**{tr('digital_human.section.character_assets')}**")
with st.expander(tr("help.feature_description"), expanded=False):
st.markdown(f"**{tr('help.what')}**")
st.markdown(tr("digital_human.assets.character_what"))
st.markdown(f"**{tr('help.how')}**")
st.markdown(tr("digital_human.assets.how"))
# File uploader for multiple files
uploaded_files = st.file_uploader(
tr("digital_human.assets.upload"),
type=["jpg", "jpeg", "png", "webp"],
accept_multiple_files=True,
help=tr("digital_human.assets.upload_help"),
key="character_files"
)
# Save uploaded files to temp directory with unique session ID
character_asset_paths = []
if uploaded_files:
import uuid
session_id = str(uuid.uuid4()).replace('-', '')[:12]
temp_dir = Path(f"temp/assets_{session_id}")
temp_dir.mkdir(parents=True, exist_ok=True)
for uploaded_file in uploaded_files:
file_path = temp_dir / uploaded_file.name
with open(file_path, "wb") as f:
f.write(uploaded_file.getbuffer())
character_asset_paths.append(str(file_path.absolute()))
st.success(tr("digital_human.assets.character_sucess"))
# Preview uploaded assets
with st.expander(tr("digital_human.assets.preview"), expanded=True):
# Show in a grid (3 columns)
cols = st.columns(3)
for i, (file, path) in enumerate(zip(uploaded_files, character_asset_paths)):
with cols[i % 3]:
# Check if image
ext = Path(path).suffix.lower()
if ext in [".jpg", ".jpeg", ".png", ".webp"]:
st.image(file, caption=file.name, use_container_width=True)
else:
st.info(tr("digital_human.assets.character_empty_hint"))
return {"character_assets": character_asset_paths}
def workflow_path_config(self) -> dict:
# Workflow source selection
with st.container(border=True):
st.markdown(f"**{tr('asset_based.section.source')}**")
with st.expander(tr("help.feature_description"), expanded=False):
st.markdown(f"**{tr('help.what')}**")
st.markdown(tr("asset_based.source.what"))
st.markdown(f"**{tr('help.how')}**")
st.markdown(tr("asset_based.source.how"))
source_options = {
"runninghub": tr("asset_based.source.runninghub"),
"selfhost": tr("asset_based.source.selfhost")
}
# Check if RunningHub API key is configured
comfyui_config = config_manager.get_comfyui_config()
has_runninghub = bool(comfyui_config.get("runninghub_api_key"))
has_selfhost = bool(comfyui_config.get("comfyui_url"))
# Default to runninghub always
default_source_index = 0
source = st.radio(
tr("asset_based.source.select"),
options=list(source_options.keys()),
format_func=lambda x: source_options[x],
index=default_source_index,
horizontal=True,
key="digital_human_workflow_source",
label_visibility="collapsed"
)
# Initialize workflow_config with default value based on source selection
# This ensures the variable is always defined even if the backend is not configured
if source == "runninghub":
workflow_config = {
"first_workflow_path": "workflows/runninghub/digital_image.json",
"second_workflow_path": "workflows/runninghub/digital_combination.json",
"third_workflow_path": "workflows/runninghub/digital_customize.json"
}
if not has_runninghub:
st.warning(tr("asset_based.source.runninghub_not_configured"))
else:
st.info(tr("asset_based.source.runninghub_hint"))
else:
workflow_config = {
"first_workflow_path": "workflows/selfhost/digital_image.json",
"second_workflow_path": "workflows/selfhost/digital_combination.json",
"third_workflow_path": "workflows/selfhost/digital_customize.json"
}
if not has_selfhost:
st.warning(tr("asset_based.source.selfhost_not_configured"))
else:
st.info(tr("asset_based.source.selfhost_hint"))
# Check and warn for selfhost workflows (auto popup if not confirmed)
# Warn for the first workflow as representative
# TODO: need to check if the workflow is valid
# check_and_warn_selfhost_workflow("selfhost/digital_image.json")
return workflow_config
def render_digital_human_mode(self, character_asset_paths: list) -> dict:
with st.container(border=True):
st.markdown(f"**{tr('digital_human.section.select_mode')}**")
with st.expander(tr("help.feature_description"), expanded=False):
st.markdown(f"**{tr('help.what')}**")
st.markdown(tr("digital_human.assets.mode_what"))
st.markdown(f"**{tr('help.how')}**")
st.markdown(tr("digital_human.assets.select_how"))
mode = st.radio(
"Processing Mode",
["digital", "customize"],
horizontal=True,
format_func=lambda x: tr(f"mode.{x}"),
label_visibility="collapsed",
key="mode_selection"
)
# Text input (unified for both modes)
text_placeholder = tr("digital_human.input.topic_placeholder") if mode == "digital" else tr("digital_human.input.content_placeholder")
text_height = 120 if mode == "digital" else 200
text_help = tr("input.text_help_digital") if mode == "digital" else tr("input.text_help_fixed")
if mode == "digital":
# File uploader for multiple files
uploaded_files = st.file_uploader(
tr("digital_human.assets.upload"),
type=["jpg", "jpeg", "png", "webp"],
accept_multiple_files=True,
help=tr("digital_human.assets.upload_help"),
key="digital_files"
)
# Save uploaded files to temp directory with unique session ID
goods_asset_paths = []
if uploaded_files:
import uuid
session_id = str(uuid.uuid4()).replace('-', '')[:12]
temp_dir = Path(f"temp/assets_{session_id}")
temp_dir.mkdir(parents=True, exist_ok=True)
for uploaded_file in uploaded_files:
file_path = temp_dir / uploaded_file.name
with open(file_path, "wb") as f:
f.write(uploaded_file.getbuffer())
goods_asset_paths.append(str(file_path.absolute()))
st.success(tr("digital_human.assets.goods_sucess"))
# Preview uploaded assets
with st.expander(tr("digital_human.assets.preview"), expanded=True):
# Show in a grid (3 columns)
cols = st.columns(3)
for i, (file, path) in enumerate(zip(uploaded_files, goods_asset_paths)):
with cols[i % 3]:
# Check if image
ext = Path(path).suffix.lower()
if ext in [".jpg", ".jpeg", ".png", ".webp"]:
st.image(file, caption=file.name, use_container_width=True)
else:
st.info(tr("digital_human.assets.goods_empty_hint"))
# Text input
goods_text = st.text_area(
tr("digital_human.input_text"),
placeholder=text_placeholder,
height=text_height,
help=text_help,
key="digital_box"
)
goods_title = st.text_input(
tr("digital_human.goods_title"),
placeholder=tr("digital_human.goods_title_placeholder"),
help=tr("digital_human.goods_title_help"),
key="goods_title"
)
return {
"character_assets": character_asset_paths,
"goods_title": goods_title,
"goods_assets": goods_asset_paths,
"goods_text": goods_text,
"mode": mode
}
else:
goods_text = st.text_area(
tr("digital_human.customize_text"),
placeholder=text_placeholder,
height=text_height,
help=text_help,
key="customize_box"
)
return {
"character_assets": character_asset_paths,
"goods_text": goods_text,
"mode": mode
}
def _render_output_preview(self, pixelle_video: Any, video_params: dict):
"""Render output preview section"""
with st.container(border=True):
st.markdown(f"**{tr('section.video_generation')}**")
# Check configuration
if not config_manager.validate():
st.warning(tr("settings.not_configured"))
# Get input data
character_assets = video_params.get("character_assets", [])
goods_assets = video_params.get("goods_assets", [])
goods_title = video_params.get("goods_title", "")
goods_text = video_params.get("goods_text", "")
mode = video_params.get("mode")
tts_voice = video_params.get("tts_voice", "zh-CN-YunjianNeural")
tts_speed = video_params.get("tts_speed", 1.2)
logger.info(f"🔧 The obtained TTS parameters:")
logger.info(f" - tts_voice: {tts_voice}")
logger.info(f" - tts_speed: {tts_speed}")
logger.info(f" - video_params中的tts_voice: {video_params.get('tts_voice', 'NOT_FOUND')}")
logger.info(f" - video_params: {video_params}")
# Validation
if not character_assets:
st.info(tr("digital_human.assets.character_warning"))
st.button(
tr("btn.generate"),
type="primary",
use_container_width=True,
disabled=True,
key="digital_human_generate_disabled"
)
return
if mode == "digital" and not goods_assets:
st.info(tr("digital_human.assets.goods_warning"))
st.button(
tr("btn.generate"),
type="primary",
use_container_width=True,
disabled=True,
key="digital_human_goods_vaiidation"
)
return
if mode == "digital" and not (goods_text or goods_title):
st.info(tr("digital_human.assets.digital_mode"))
st.button(
tr("btn.generate"),
type="primary",
use_container_width=True,
disabled=True,
key="digital_human_digital_disable"
)
return
if mode == "digital" and (goods_text or goods_title):
st.warning(tr("digital_human.assets.digital_mode_warning"))
if mode == "customize" and not goods_text:
st.info(tr("digital_human.assets.customize_mode"))
st.button(
tr("btn.generate"),
type="primary",
use_container_width=True,
disabled=True,
key="digital_human_customize_disable"
)
return
# Generate button
if st.button(tr("btn.generate"), type="primary", use_container_width=True, key="digital_human_generate"):
# Validate
if not config_manager.validate():
st.error(tr("settings.not_configured"))
st.stop()
# Show progress
progress_bar = st.progress(0)
status_text = st.empty()
start_time = time.time()
try:
# Define async generation function
async def generate_digital_human_video():
task_dir, task_id = create_task_output_dir()
kit = await pixelle_video._get_or_create_comfykit()
workflow_path = video_params["workflow_path"]
import json
from pathlib import Path
if mode == "customize":
status_text.text(tr("progress.step_audio"))
progress_bar.progress(25)
generated_image_path = character_assets[0]
generated_text = goods_text
# TTS
audio_path = os.path.join(task_dir, "narration.mp3")
tts_inference_mode = video_params.get("tts_inference_mode", "local")
tts_voice = video_params.get("tts_voice")
tts_speed = video_params.get("tts_speed")
tts_workflow = video_params.get("tts_workflow")
ref_audio = video_params.get("ref_audio")
tts_kwargs = {
"text": generated_text,
"output_path": audio_path,
"inference_mode": tts_inference_mode
}
if tts_inference_mode == "local":
tts_kwargs["voice"] = tts_voice
tts_kwargs["speed"] = tts_speed
elif tts_inference_mode == "comfyui":
if tts_workflow:
tts_kwargs["workflow"] = tts_workflow
if ref_audio:
tts_kwargs["ref_audio"] = ref_audio
await pixelle_video.tts(**tts_kwargs)
progress_bar.progress(65)
status_text.text(tr("progress.concatenating"))
# Directly call the second workflow
second_workflow_path = Path(workflow_path.get("second_workflow_path"))
if not second_workflow_path.exists():
raise Exception(f"The second step workflow file does not exist:{second_workflow_path}")
with open(second_workflow_path, 'r', encoding='utf-8') as f:
second_workflow_config = json.load(f)
second_workflow_params = {
"videoimage": generated_image_path,
"audio": audio_path
}
if second_workflow_config.get("source") == "runninghub" and "workflow_id" in second_workflow_config:
workflow_input = second_workflow_config["workflow_id"]
else:
workflow_input = str(second_workflow_config)
second_result = await kit.execute(workflow_input, second_workflow_params)
# Video Link Extraction
generated_video_url = None
if hasattr(second_result, 'videos') and second_result.videos:
generated_video_url = second_result.videos[0]
elif hasattr(second_result, 'outputs') and second_result.outputs:
for node_id, node_output in second_result.outputs.items():
if isinstance(node_output, dict) and 'videos' in node_output:
videos = node_output['videos']
if videos and len(videos) > 0:
generated_video_url = videos[0]
break
if not generated_video_url:
raise Exception("The second step of the workflow did not return a video. Please check the workflow configuration.")
final_video_path = os.path.join(task_dir, "final.mp4")
timeout = httpx.Timeout(300.0)
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.get(generated_video_url)
response.raise_for_status()
with open(final_video_path, 'wb') as f:
f.write(response.content)
progress_bar.progress(100)
status_text.text(tr("status.success"))
return final_video_path
else:
#Initialization and parameter preparation
task_dir, task_id = create_task_output_dir()
logger.info(f"[Initialization] Task Directory: {task_dir}")
first_workflow_path = Path(workflow_path.get("first_workflow_path"))
third_workflow_path = Path(workflow_path.get("third_workflow_path"))
second_workflow_path = Path(workflow_path.get("second_workflow_path"))
assert first_workflow_path.exists(), "The first_workflow file does not exist."
assert third_workflow_path.exists(), "The third_workflow file does not exist."
assert second_workflow_path.exists(), "The second_workflow file does not exist."
if goods_text and goods_text.strip():
workflow_path = third_workflow_path
workflow_params = {"firstimage": character_assets[0], "secondimage": goods_assets[0]}
generated_text = goods_text
status_text.text(tr("progress.step_image"))
kit = await pixelle_video._get_or_create_comfykit()
workflow_config = json.load(open(workflow_path, 'r', encoding='utf8'))
if workflow_config.get("source") == "runninghub" and "workflow_id" in workflow_config:
workflow_input = workflow_config["workflow_id"]
else:
workflow_input = str(workflow_config)
combine_image = await kit.execute(workflow_input, workflow_params)
if combine_image.status != "completed":
raise Exception(f"workflow execution failed: {combine_image.msg}")
generated_image_url = getattr(combine_image, "images", [None])[0]
status_text.text(tr("progress.step_audio"))
audio_path = os.path.join(task_dir, "narration.mp3")
tts_inference_mode = video_params.get("tts_inference_mode", "local")
tts_voice = video_params.get("tts_voice")
tts_speed = video_params.get("tts_speed")
tts_workflow = video_params.get("tts_workflow")
ref_audio = video_params.get("ref_audio")
tts_kwargs = {
"text": generated_text,
"output_path": audio_path,
"inference_mode": tts_inference_mode
}
if tts_inference_mode == "local":
tts_kwargs["voice"] = tts_voice
tts_kwargs["speed"] = tts_speed
elif tts_inference_mode == "comfyui":
if tts_workflow:
tts_kwargs["workflow"] = tts_workflow
if ref_audio:
tts_kwargs["ref_audio"] = ref_audio
await pixelle_video.tts(**tts_kwargs)
progress_bar.progress(65)
status_text.text(tr("progress.concatenating"))
if not second_workflow_path.exists():
raise Exception(f"The second step workflow file does not exist:{second_workflow_path}")
with open(second_workflow_path, 'r', encoding='utf-8') as f:
second_workflow_config = json.load(f)
second_workflow_params = {
"videoimage": generated_image_url,
"audio": audio_path
}
if second_workflow_config.get("source") == "runninghub" and "workflow_id" in second_workflow_config:
workflow_input = second_workflow_config["workflow_id"]
else:
workflow_input = str(second_workflow_config)
second_result = await kit.execute(workflow_input, second_workflow_params)
# Video Link Extraction
generated_video_url = None
if hasattr(second_result, 'videos') and second_result.videos:
generated_video_url = second_result.videos[0]
elif hasattr(second_result, 'outputs') and second_result.outputs:
for node_id, node_output in second_result.outputs.items():
if isinstance(node_output, dict) and 'videos' in node_output:
videos = node_output['videos']
if videos and len(videos) > 0:
generated_video_url = videos[0]
break
if not generated_video_url:
raise Exception("The second step of the workflow did not return a video. Please check the workflow configuration.")
final_video_path = os.path.join(task_dir, "final.mp4")
timeout = httpx.Timeout(300.0)
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.get(generated_video_url)
response.raise_for_status()
with open(final_video_path, 'wb') as f:
f.write(response.content)
progress_bar.progress(100)
status_text.text(tr("status.success"))
return final_video_path
else:
workflow_path = first_workflow_path
workflow_params = {"firstimage": character_assets[0], "secondimage": goods_assets[0], "goodstype": goods_title}
status_text.text(tr("progress.step_image"))
kit = await pixelle_video._get_or_create_comfykit()
workflow_config = json.load(open(workflow_path, 'r', encoding='utf8'))
if workflow_config.get("source") == "runninghub" and "workflow_id" in workflow_config:
workflow_input = workflow_config["workflow_id"]
else:
workflow_input = str(workflow_config)
synthesis_result = await kit.execute(workflow_input, workflow_params)
if synthesis_result.status != "completed":
raise Exception(f"workflow execution failed: {synthesis_result.msg}")
generated_image_url = getattr(synthesis_result, "images", [None])[0]
generated_text = getattr(synthesis_result, "texts", [None])[0]
status_text.text(tr("progress.step_audio"))
audio_path = os.path.join(task_dir, "narration.mp3")
tts_inference_mode = video_params.get("tts_inference_mode", "local")
tts_voice = video_params.get("tts_voice")
tts_speed = video_params.get("tts_speed")
tts_workflow = video_params.get("tts_workflow")
ref_audio = video_params.get("ref_audio")
tts_kwargs = {
"text": generated_text,
"output_path": audio_path,
"inference_mode": tts_inference_mode
}
if tts_inference_mode == "local":
tts_kwargs["voice"] = tts_voice
tts_kwargs["speed"] = tts_speed
elif tts_inference_mode == "comfyui":
if tts_workflow:
tts_kwargs["workflow"] = tts_workflow
if ref_audio:
tts_kwargs["ref_audio"] = ref_audio
await pixelle_video.tts(**tts_kwargs)
progress_bar.progress(65)
status_text.text(tr("progress.concatenating"))
if not second_workflow_path.exists():
raise Exception(f"The second step workflow file does not exist:{second_workflow_path}")
with open(second_workflow_path, 'r', encoding='utf-8') as f:
second_workflow_config = json.load(f)
second_workflow_params = {
"videoimage": generated_image_url,
"audio": audio_path
}
if second_workflow_config.get("source") == "runninghub" and "workflow_id" in second_workflow_config:
workflow_input = second_workflow_config["workflow_id"]
else:
workflow_input = str(second_workflow_config)
second_result = await kit.execute(workflow_input, second_workflow_params)
# Video Link Extraction
generated_video_url = None
if hasattr(second_result, 'videos') and second_result.videos:
generated_video_url = second_result.videos[0]
elif hasattr(second_result, 'outputs') and second_result.outputs:
for node_id, node_output in second_result.outputs.items():
if isinstance(node_output, dict) and 'videos' in node_output:
videos = node_output['videos']
if videos and len(videos) > 0:
generated_video_url = videos[0]
break
if not generated_video_url:
raise Exception("The second step of the workflow did not return a video. Please check the workflow configuration.")
final_video_path = os.path.join(task_dir, "final.mp4")
timeout = httpx.Timeout(300.0)
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.get(generated_video_url)
response.raise_for_status()
with open(final_video_path, 'wb') as f:
f.write(response.content)
progress_bar.progress(100)
status_text.text(tr("status.success"))
return final_video_path
# Execute async generation
final_video_path = run_async(generate_digital_human_video())
total_time = time.time() - start_time
progress_bar.progress(100)
status_text.text(tr("status.success"))
# Display result
st.success(tr("status.video_generated", path=final_video_path))
st.markdown("---")
# Video info
if os.path.exists(final_video_path):
file_size_mb = os.path.getsize(final_video_path) / (1024 * 1024)
info_text = (
f"⏱️ {tr('info.generation_time')} {total_time:.1f}s "
f"📦 {file_size_mb:.2f}MB"
)
st.caption(info_text)
st.markdown("---")
# Video preview
st.video(final_video_path)
# Download button
with open(final_video_path, "rb") as video_file:
video_bytes = video_file.read()
video_filename = os.path.basename(final_video_path)
st.download_button(
label="⬇️ 下载视频" if get_language() == "zh_CN" else "⬇️ Download Video",
data=video_bytes,
file_name=video_filename,
mime="video/mp4",
use_container_width=True
)
else:
st.error(tr("status.video_not_found", path=final_video_path))
except Exception as e:
status_text.text("")
progress_bar.empty()
st.error(tr("status.error", error=str(e)))
logger.exception(e)
st.stop()
# Register self
register_pipeline_ui(DigitalHumanPipelineUI)