okooooooo/Pixelle-Video
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

新增数字人口播板块与图生视频板块的本地工作流支持，图生视频板块由audio-visual更改为i2v

Browse Source
This commit is contained in:
1152450298 committed 2026-01-20 18:22:36 +08:00
1 parent 1edc0c40e4
commit f6c561db35
8 files changed
+425 -205
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

pyproject.toml
+2 -3
View File
@@ -1,6 +1,6 @@
[project]
name = "pixelle-video"
version = "0.1.11"
version = "0.1.12"
description = "AI-powered video creation platform - Part of Pixelle ecosystem"
authors = [
{name = "Pixelle.AI"}
@@ -53,5 +53,4 @@ ignore = ["E501"]
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
testpaths = ["tests"]
web/components/digital_tts_config.py
+124 -63
View File
@@ -34,71 +34,123 @@ def render_style_config(pixelle_video):
with st.expander(tr("help.feature_description"), expanded=False):
st.markdown(f"**{tr('help.what')}**")
st.markdown(tr("digital_tts.what"))
st.markdown(tr("tts.what"))
st.markdown(f"**{tr('help.how')}**")
st.markdown(tr("tts.how"))
# Get TTS config
comfyui_config = config_manager.get_comfyui_config()
tts_config = comfyui_config["tts"]
tts_mode = "local"
# Inference mode selection
tts_mode = st.radio(
tr("tts.inference_mode"),
["local", "comfyui"],
horizontal=True,
format_func=lambda x: tr(f"tts.mode.{x}"),
index=0 if tts_config.get("inference_mode", "local") == "local" else 1,
key="digital_tts_inference_mode"
)
# Show hint based on mode
if tts_mode == "local":
st.caption(tr("tts.mode.local_hint"))
else:
st.caption(tr("tts.mode.comfyui_hint"))
# ================================================================
# Local Mode UI (simple)
# Local Mode UI
# ================================================================
# Import voice configuration
from pixelle_video.tts_voices import EDGE_TTS_VOICES, get_voice_display_name
# Get saved voice from config
local_config = tts_config.get("local", {})
saved_voice = local_config.get("voice", "zh-CN-YunjianNeural")
saved_speed = local_config.get("speed", 1.2)
# Build voice options with i18n
voice_options = []
voice_ids = []
default_voice_index = 0
for idx, voice_config in enumerate(EDGE_TTS_VOICES):
voice_id = voice_config["id"]
display_name = get_voice_display_name(voice_id, tr, get_language())
voice_options.append(display_name)
voice_ids.append(voice_id)
if tts_mode == "local":
# Import voice configuration
from pixelle_video.tts_voices import EDGE_TTS_VOICES, get_voice_display_name
# Set default index if matches saved voice
if voice_id == saved_voice:
default_voice_index = idx
# Get saved voice from config
local_config = tts_config.get("local", {})
saved_voice = local_config.get("voice", "zh-CN-YunjianNeural")
saved_speed = local_config.get("speed", 1.2)
# Build voice options with i18n
voice_options = []
voice_ids = []
default_voice_index = 0
for idx, voice_config in enumerate(EDGE_TTS_VOICES):
voice_id = voice_config["id"]
display_name = get_voice_display_name(voice_id, tr, get_language())
voice_options.append(display_name)
voice_ids.append(voice_id)
# Set default index if matches saved voice
if voice_id == saved_voice:
default_voice_index = idx
# Two-column layout: Voice | Speed
voice_col, speed_col = st.columns([1, 1])
with voice_col:
# Voice selector
selected_voice_display = st.selectbox(
tr("tts.voice_selector"),
voice_options,
index=default_voice_index,
key="digital_tts_local_voice"
)
# Get actual voice ID
selected_voice_index = voice_options.index(selected_voice_display)
selected_voice = voice_ids[selected_voice_index]
with speed_col:
# Speed slider
tts_speed = st.slider(
tr("tts.speed"),
min_value=0.5,
max_value=2.0,
value=saved_speed,
step=0.1,
format="%.1fx",
key="digital_tts_local_speed"
)
st.caption(tr("tts.speed_label", speed=f"{tts_speed:.1f}"))
# Variables for video generation
tts_workflow_key = None
ref_audio_path = None
# Two-column layout: Voice | Speed
voice_col, speed_col = st.columns([1, 1])
with voice_col:
# Voice selector
selected_voice_display = st.selectbox(
tr("tts.voice_selector"),
voice_options,
index=default_voice_index,
key="digital_tts_local_voice"
# ================================================================
# ComfyUI Mode UI
# ================================================================
else: # comfyui mode
tts_workflow_key = "runninghub/tts_index2.json" # fallback
# Reference audio upload (optional, for voice cloning)
ref_audio_file = st.file_uploader(
tr("tts.ref_audio"),
type=["mp3", "wav", "flac", "m4a", "aac", "ogg"],
help=tr("tts.ref_audio_help"),
key="digital_ref_audio_upload"
)
# Get actual voice ID
selected_voice_index = voice_options.index(selected_voice_display)
selected_voice = voice_ids[selected_voice_index]
with speed_col:
# Speed slider
tts_speed = st.slider(
tr("tts.speed"),
min_value=0.5,
max_value=2.0,
value=saved_speed,
step=0.1,
format="%.1fx",
key="digital_tts_local_speed"
)
st.caption(tr("tts.speed_label", speed=f"{tts_speed:.1f}"))
# Save uploaded ref_audio to temp file if provided
ref_audio_path = None
if ref_audio_file is not None:
# Audio preview player (directly play uploaded file)
st.audio(ref_audio_file)
# Save to temp directory
temp_dir = Path("temp")
temp_dir.mkdir(exist_ok=True)
ref_audio_path = temp_dir / f"ref_audio_{ref_audio_file.name}"
with open(ref_audio_path, "wb") as f:
f.write(ref_audio_file.getbuffer())
# Variables for video generation
selected_voice = None
tts_speed = None
# ================================================================
# TTS Preview (only local mode)
# TTS Preview (works for both modes)
# ================================================================
with st.expander(tr("tts.preview_title"), expanded=False):
# Preview text input
@@ -110,26 +162,34 @@ def render_style_config(pixelle_video):
)
# Preview button
if st.button(tr("tts.preview_button"), key="digital_preview_tts", use_container_width=True):
if st.button(tr("tts.preview_button"), key="gidital_preview_tts", use_container_width=True):
with st.spinner(tr("tts.previewing")):
try:
# Build TTS params for local mode
# Build TTS params based on mode
tts_params = {
"text": preview_text,
"inference_mode": "local",
"voice": selected_voice,
"speed": tts_speed
"inference_mode": tts_mode
}
if tts_mode == "local":
tts_params["voice"] = selected_voice
tts_params["speed"] = tts_speed
else: # comfyui
tts_params["workflow"] = tts_workflow_key
if ref_audio_path:
tts_params["ref_audio"] = str(ref_audio_path)
audio_path = run_async(pixelle_video.tts(**tts_params))
# Play the audio
if audio_path:
st.success(tr("tts.preview_success"))
if audio_path.startswith('http'):
if os.path.exists(audio_path):
st.audio(audio_path, format="audio/mp3")
elif audio_path.startswith('http'):
st.audio(audio_path)
else:
st.audio(audio_path, format="audio/mp3")
st.error("Failed to generate preview audio")
# Show file path
st.caption(f"📁 {audio_path}")
@@ -141,8 +201,9 @@ def render_style_config(pixelle_video):
# Return all style configuration parameters (Simplified version only local TTS)
return {
"tts_inference_mode": "local",
"tts_voice": selected_voice,
"tts_speed": tts_speed
}
"tts_inference_mode": tts_mode,
"tts_voice": selected_voice if tts_mode == "local" else None,
"tts_speed": tts_speed if tts_mode == "local" else None,
"tts_workflow": tts_workflow_key if tts_mode == "comfyui" else None,
"ref_audio": str(ref_audio_path) if ref_audio_path else None,
}
web/i18n/locales/en_US.json
+12 -9
View File
@@ -444,15 +444,18 @@
"digital_human.workflow.ready": "Digital human workflow is ready. You can start video generation",
"digital_human.workflow.missing": "Missing required digital human workflow. Please check the configuration",
"digital_tts.what": "Convert narration text to natural human-like speech",
"audio_visual.input.topic_placeholder": "Enter creative prompts (e.g., audio and rhythm, camera movement rules, overall visual style, color, color grading, etc.)",
"audio_visual.assets.upload": "Upload video first frame image",
"audio_visual.assets.upload_help": "Only single image upload is supported",
"audio_visual.assets.character_sucess": "Video asset uploaded successfully",
"audio_visual.assets.preview": "📷 Asset Preview",
"audio_visual.assets.character_empty_hint": "💡 Please upload a first frame image for your video",
"audio_visual.input_text": "Video prompt",
"audio_visual.assets.image_warning": "Upload the first frame image of your video",
"audio_visual.assets.prompt_warning": "Please enter a video prompt",
"i2v.video_generation": "🛠️ Configuration parameters",
"i2v.assets.image_what": "Reference image of the first frame generated from the uploaded video",
"i2v.assets.how": "Supports JPG/PNG/WebP and other image formats. Clear and relevant images are recommended. Accurate and specific video prompts will enhance the video's quality.",
"i2v.input.topic_placeholder": "Enter creative prompts (e.g., audio and rhythm, camera movement rules, overall visual style, color, color grading, etc.)",
"i2v.assets.upload": "Upload video first frame image",
"i2v.assets.upload_help": "Only single image upload is supported",
"i2v.assets.character_sucess": "Video asset uploaded successfully",
"i2v.assets.preview": "📷 Asset Preview",
"i2v.assets.character_empty_hint": "💡 Please upload a first frame image for your video",
"i2v.input_text": "Video prompt",
"i2v.assets.image_warning": "Upload the first frame image of your video",
"i2v.assets.prompt_warning": "Please enter a video prompt",
"faq.expand_to_view": "FAQ",
"faq.load_error": "Failed to load FAQ content",
web/i18n/locales/zh_CN.json
+13 -10
View File
@@ -444,16 +444,19 @@
"digital_human.workflow.ready": "数字人工作流已就绪，可以开始生成视频",
"digital_human.workflow.missing": "缺少必要的数字人工作流，请检查配置",
"digital_tts.what": "将旁白文本转换为真人般的自然语音",
"audio_visual.input.topic_placeholder" : "输入创作提示词（例如：音频与节奏、镜头与运动规则、整体视觉风格、色彩与调色等）",
"audio_visual.assets.upload": "上传视频首帧图像",
"audio_visual.assets.upload_help": "仅支持单张图片上传",
"audio_visual.assets.character_sucess": "上传视频素材成功",
"audio_visual.assets.preview": "📷 素材预览",
"audio_visual.assets.character_empty_hint": "💡 请上传一个想要生成视频的首帧图像",
"audio_visual.input_text": "视频提示词",
"audio_visual.assets.image_warning": "上传视频首帧图片",
"audio_visual.assets.prompt_warning": "请输入视频提示词",
"i2v.video_generation": "🛠️ 配置参数",
"i2v.assets.image_what": "上传视频生成画面的首帧参考图",
"i2v.assets.how": "支持 JPG/PNG/WebP 等格式图片，建议素材清晰且内容相关\n视频提示词描述准确具体，能使视频效果更佳",
"i2v.input.topic_placeholder" : "输入创作提示词（例如：音频与节奏、镜头与运动规则、整体视觉风格、色彩与调色等）",
"i2v.assets.upload": "上传视频首帧图像",
"i2v.assets.upload_help": "仅支持单张图片上传",
"i2v.assets.character_sucess": "上传视频素材成功",
"i2v.assets.preview": "📷 素材预览",
"i2v.assets.character_empty_hint": "💡 请上传一个想要生成视频的首帧图像",
"i2v.input_text": "视频提示词",
"i2v.assets.image_warning": "上传视频首帧图片",
"i2v.assets.prompt_warning": "请输入视频提示词",
"faq.expand_to_view": "常见问题",
"faq.load_error": "无法加载常见问题内容",
"faq.more_help": "需要更多帮助？"
web/pipelines/__init__.py
+1 -1
View File
@@ -27,7 +27,7 @@ from web.pipelines.base import (
from web.pipelines import standard
from web.pipelines import asset_based
from web.pipelines import digital_human
from web.pipelines import audio_visual
from web.pipelines import i2v
__all__ = [
"PipelineUI",
web/pipelines/digital_human.py
+148 -38
View File
@@ -1,6 +1,5 @@
import os
import time
import uuid
from pathlib import Path
from typing import Any
@@ -9,7 +8,7 @@ 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_bgm_section, render_version_info
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 pixelle_video.config import config_manager
@@ -49,6 +48,7 @@ class DigitalHumanPipelineUI(PipelineUI):
# ====================================================================
with middle_col:
# Style configuration ()
workflow_path = self.workflow_path_config()
mode_params = self.render_digital_human_mode(asset_params["character_assets"])
# # ====================================================================
@@ -59,11 +59,12 @@ class DigitalHumanPipelineUI(PipelineUI):
video_params = {
**mode_params,
**asset_params,
**style_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):
@@ -115,6 +116,68 @@ class DigitalHumanPipelineUI(PipelineUI):
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 available source
if has_runninghub:
default_source_index = 0
elif has_selfhost:
default_source_index = 1
else:
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"
)
# Show hint based on selection
if source == "runninghub":
if not has_runninghub:
st.warning(tr("asset_based.source.runninghub_not_configured"))
else:
st.info(tr("asset_based.source.runninghub_hint"))
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"
}
else:
if not has_selfhost:
st.warning(tr("asset_based.source.selfhost_not_configured"))
else:
st.info(tr("asset_based.source.selfhost_hint"))
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"
}
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')}**")
@@ -281,6 +344,7 @@ class DigitalHumanPipelineUI(PipelineUI):
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
@@ -293,18 +357,32 @@ class DigitalHumanPipelineUI(PipelineUI):
# TTS
audio_path = os.path.join(task_dir, "narration.mp3")
await pixelle_video.tts(
text=generated_text,
output_path=audio_path,
inference_mode="local",
voice=tts_voice,
speed=tts_speed
)
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("workflows/runninghub/digital_combination.json")
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:
@@ -316,7 +394,7 @@ class DigitalHumanPipelineUI(PipelineUI):
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_path)
workflow_input = str(second_workflow_config)
second_result = await kit.execute(workflow_input, second_workflow_params)
# Video Link Extraction
generated_video_url = None
@@ -348,9 +426,9 @@ class DigitalHumanPipelineUI(PipelineUI):
task_dir, task_id = create_task_output_dir()
logger.info(f"[Initialization] Task Directory: {task_dir}")
first_workflow_path = Path("workflows/runninghub/digital_image.json")
third_workflow_path = Path("workflows/runninghub/digital_customize.json")
second_workflow_path = Path("workflows/runninghub/digital_combination.json")
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."
@@ -362,24 +440,40 @@ class DigitalHumanPipelineUI(PipelineUI):
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'))
workflow_input = workflow_config.get("workflow_id", str(workflow_path))
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")
await pixelle_video.tts(
text=goods_text,
output_path=audio_path,
inference_mode="local",
voice=tts_voice,
speed=tts_speed
)
progress_bar.progress(65)
status_text.text(tr("progress.progress.concatenating"))
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"))
second_workflow_path = Path("workflows/runninghub/digital_combination.json")
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:
@@ -391,7 +485,7 @@ class DigitalHumanPipelineUI(PipelineUI):
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_path)
workflow_input = str(second_workflow_config)
second_result = await kit.execute(workflow_input, second_workflow_params)
# Video Link Extraction
generated_video_url = None
@@ -425,7 +519,10 @@ class DigitalHumanPipelineUI(PipelineUI):
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'))
workflow_input = workflow_config.get("workflow_id", str(workflow_path))
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}")
@@ -434,17 +531,30 @@ class DigitalHumanPipelineUI(PipelineUI):
status_text.text(tr("progress.step_audio"))
audio_path = os.path.join(task_dir, "narration.mp3")
await pixelle_video.tts(
text=generated_text,
output_path=audio_path,
inference_mode="local",
voice=tts_voice,
speed=tts_speed
)
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"))
second_workflow_path = Path("workflows/runninghub/digital_combination.json")
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:
@@ -456,7 +566,7 @@ class DigitalHumanPipelineUI(PipelineUI):
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_path)
workflow_input = str(second_workflow_config)
second_result = await kit.execute(workflow_input, second_workflow_params)
# Video Link Extraction
generated_video_url = None
web/pipelines/audio_visual.py → web/pipelines/i2v.py
+125 -81
View File
@@ -1,8 +1,5 @@
from curses import noecho
from multiprocessing import process
import os
import time
import uuid
from pathlib import Path
from typing import Any
@@ -11,12 +8,12 @@ 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_bgm_section, render_version_info
from web.components.content_input import render_version_info
from web.utils.async_helpers import run_async
from pixelle_video.config import config_manager
from pixelle_video.utils.os_util import create_task_output_dir
class AudioVisualPipelineUI(PipelineUI):
class ImageToVideoPipelineUI(PipelineUI):
"""
UI for the Audio-visual synchronization Video Generation Pipeline.
Generates videos from user-provided assets (images&text).
@@ -40,7 +37,7 @@ class AudioVisualPipelineUI(PipelineUI):
# Middle Column: Asset Upload
# ====================================================================
with left_col:
asset_params = self.render_audio_visual_input()
asset_params = self.render_audio_visual_input(pixelle_video)
render_version_info()
with right_col:
@@ -50,75 +47,119 @@ class AudioVisualPipelineUI(PipelineUI):
self._render_output_preview(pixelle_video, video_params)
def render_audio_visual_input(self) -> dict:
def render_audio_visual_input(self, pixelle_video) -> dict:
with st.container(border=True):
st.markdown(f"**{tr('i2v.video_generation')}**")
# File uploader for multiple files
uploaded_files = st.file_uploader(
tr("audio_visual.assets.upload"),
type=["jpg", "jpeg", "png", "webp"],
accept_multiple_files=True,
help=tr("audio_visual.assets.upload_help"),
key="material_files"
)
with st.expander(tr("help.feature_description"), expanded=False):
st.markdown(f"**{tr('help.what')}**")
st.markdown(tr("i2v.assets.image_what"))
st.markdown(f"**{tr('help.how')}**")
st.markdown(tr("i2v.assets.how"))
# Save uploaded files to temp directory with unique session ID
audio_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)
def list_iv2_workflows():
result = []
for source in ("runninghub", "selfhost"):
dir_path = os.path.join("workflows", source)
if not os.path.isdir(dir_path):
continue
for fname in os.listdir(dir_path):
if fname.startswith("iv2_") and fname.endswith(".json"):
display = f"{fname} - {'Runninghub' if source == 'runninghub' else 'Selfhost'}"
result.append({
"key": f"{source}/{fname}",
"display_name": display
})
return result
# File uploader for multiple files
uploaded_files = st.file_uploader(
tr("i2v.assets.upload"),
type=["jpg", "jpeg", "png", "webp"],
accept_multiple_files=True,
help=tr("i2v.assets.upload_help"),
key="material_files"
)
# Save uploaded files to temp directory with unique session ID
audio_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())
audio_asset_paths.append(str(file_path.absolute()))
st.success(tr("i2v.assets.character_sucess"))
# Preview uploaded assets
with st.expander(tr("i2v.assets.preview"), expanded=True):
# Show in a grid (3 columns)
cols = st.columns(3)
for i, (file, path) in enumerate(zip(uploaded_files, audio_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("i2v.assets.character_empty_hint"))
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())
audio_asset_paths.append(str(file_path.absolute()))
prompt_text = st.text_area(
tr("i2v.input_text"),
placeholder=tr("i2v.input.topic_placeholder"),
height=200,
help=tr("input.text_help_audio"),
key="audio_box"
)
st.success(tr("audio_visual.assets.character_sucess"))
iv2_workflows = list_iv2_workflows()
workflow_options = [wf["display_name"] for wf in iv2_workflows]
workflow_keys = [wf["key"] for wf in iv2_workflows]
default_workflow_index = 0
workflow_display = st.selectbox(
tr("iv2.workflow_select"),
workflow_options if workflow_options else ["No workflow found"],
index=default_workflow_index,
label_visibility="collapsed",
key="iv2_workflow_select"
)
if workflow_options:
workflow_selected_index = workflow_options.index(workflow_display)
workflow_key = workflow_keys[workflow_selected_index]
else:
workflow_key = None
# Preview uploaded assets
with st.expander(tr("audio_visual.assets.preview"), expanded=True):
# Show in a grid (3 columns)
cols = st.columns(3)
for i, (file, path) in enumerate(zip(uploaded_files, audio_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("audio_visual.assets.character_empty_hint"))
prompt_text = st.text_area(
tr("audio_visual.input_text"),
placeholder=tr("audio_visual.input.topic_placeholder"),
height=200,
help=tr("input.text_help_audio"),
key="audio_box"
)
return {
"audio_assets": audio_asset_paths,
"prompt_text": prompt_text,
}
return {
"audio_assets": audio_asset_paths,
"prompt_text": prompt_text,
"workflow_key": workflow_key
}
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"))
audio_assets = video_params.get("audio_assets", [])
prompt_text = video_params.get("prompt_text", "")
workflow_key = video_params.get("workflow_key")
logger.info(f" - video_params: {video_params}")
logger.info(f"video_params: {video_params}")
if not audio_assets:
st.info(tr("audio_visual.assets.image_warning"))
st.info(tr("i2v.assets.image_warning"))
st.button(
tr("btn.generate"),
type="primary",
@@ -129,7 +170,7 @@ class AudioVisualPipelineUI(PipelineUI):
return
if not prompt_text:
st.info(tr("audio_visual.assets.prompt_warning"))
st.info(tr("i2v.assets.prompt_warning"))
st.button(
tr("btn.generate"),
type="primary",
@@ -138,47 +179,50 @@ class AudioVisualPipelineUI(PipelineUI):
key="audio_visual_generate"
)
return
# Generate button
if st.button(tr("btn.generate"), type="primary", use_container_width=True, key="digital_human_generate"):
# Validate
if st.button(tr("btn.generate"), type="primary", use_container_width=True, key="iv2_generate"):
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_audio_visual_video():
task_dir, task_id = create_task_output_dir()
logger.info(f"[Initialization] Task Directory: {task_dir}")
kit = await pixelle_video._get_or_create_comfykit()
import json
from pathlib import Path
status_text.text(tr("progress.generation"))
progress_bar.progress(65)
progress_bar.progress(10)
image_path = audio_assets[0]
prompt = prompt_text
workflow_path = Path("workflows/runninghub/audio_visual.json")
workflow_path = Path("workflows") / workflow_key
if not workflow_path.exists():
raise Exception(f"The workflow file does not exist: {workflow_path}")
with open(workflow_path, 'r', encoding='utf-8') as f:
workflow_config = json.load(f)
workflow_params = {
"image": image_path,
"prompt": prompt
}
if workflow_config.get("source") == "runninghub" and "workflow_id" in workflow_config:
workflow_input = workflow_config["workflow_id"]
else:
workflow_input = str(workflow_path)
video_result = await kit.execute(workflow_input, workflow_params)
generated_video_url = None
@@ -191,9 +235,10 @@ class AudioVisualPipelineUI(PipelineUI):
if videos and len(videos) > 0:
generated_video_url = videos[0]
break
if not generated_video_url:
raise Exception("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:
@@ -207,31 +252,30 @@ class AudioVisualPipelineUI(PipelineUI):
# Execute async generation
final_video_path = run_async(generate_audio_visual_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()
@@ -245,12 +289,12 @@ class AudioVisualPipelineUI(PipelineUI):
)
else:
st.error(tr("status.video_not_found", path=final_video_path))
except Exception as e:
logger.exception(e)
status_text.text("")
progress_bar.empty()
st.error(tr("status.error", error=str(e)))
logger.exception(e)
st.stop()
register_pipeline_ui(AudioVisualPipelineUI)
register_pipeline_ui(ImageToVideoPipelineUI)
workflows/runninghub/audio_visual.json → workflows/runninghub/iv2_LTX2.json
View File
File renamed without changes.