This commit is contained in:
wjluoxiao
2026-11-10 03:00:46 +08:00
parent 62318ceee2
commit 666ea32a5c
226 changed files with 11903 additions and 8550 deletions
+60
View File
@@ -211,6 +211,66 @@ Nodes that are fully self-contained say so in their own docstring.
---
### 🗃️ Repository layout
One feature per folder; the root only keeps the assembly entry points:
```text
XB_ToolBox/
├─ __init__.py node registration + route wiring
├─ prestartup_script.py startup cleanup (ComfyUI convention: must stay at root)
├─ nodes/ node implementations
│ ├─ core/ parameter bus / relay / label / string / switch / accel
│ ├─ image/ prompt presets / encoders / scale / segmentation / comic
│ ├─ video/ Wan / LTX / Hailuo / digital human / video I-O
│ ├─ audio/ audio slicer / YuE2 / play sound
│ ├─ llm/ XB-llama / GGUF loaders
│ ├─ hardware/ VRAM & memory monitoring
│ └─ cosyvoice3/ CosyVoice3 node package
├─ libs/ node-free logic (xb_paths / xb_presets / xb_gallery / xb_skills / yue2_* …)
├─ jzl_workflows/ "JZL Workflows" (data + routes + icon/style)
├─ web/ frontend assets (formerly js/)
├─ locales/ docs/ tests/ i18n / docs / tests
├─ tools/ one-off dev scripts (not used at runtime)
├─ support_llama/ xb_skill/ video_formats/ data folders
└─ LibreHardwareMonitorLib.dll
```
---
### 🗂️ JZL Workflows (sidebar entry)
Once XB_ToolBox is installed, ComfyUI's sidebar gains a **"机智罗 / JZL Workflows"** entry,
sitting right above the built-in **Templates** tab. Clicking it opens a **centered dialog** just
like the official Templates browser — category list on the left, search on top, thumbnail cards,
and a card click opens the workflow in a new workflow tab (`Esc` / backdrop click / ✕ closes it).
No ComfyUI frontend files are patched, so frontend upgrades won't break it.
Everything for this feature lives in **`jzl_workflows/`**:
```text
XB_ToolBox/
jzl_workflows/
__init__.py registration entry (root __init__.py calls it)
routes.py routes: index / workflow / preview / icon+css assets
data/
index.json English index
index.zh.json Chinese index (same shape as comfyui-workflow-templates)
anima_t2i_basic.json the workflow itself
anima_t2i_basic-1.webp preview image; a placeholder card is shown if missing
web/
xb_workflow_gallery.css
xb_icon_dark.png / xb_icon_light.png
web/jzl_workflows/
xb_workflow_gallery.js panel script (must stay under web/ to be loaded)
```
To add a workflow: drop the JSON into `jzl_workflows/data/` (ASCII file name), add an entry to
both index files (`name` must match the file name), and put the preview next to it as
`<name>-1.webp`.
---
### 🙏 Credits
XB_ToolBox stands on the shoulders of the ComfyUI ecosystem. Several nodes are ports, wrappers or adaptations — thanks to their authors:
+59
View File
@@ -210,6 +210,65 @@ pip install -r requirements.txt
---
### 🗃️ 目录结构
按「一个功能一个目录」组织,根目录只留装配入口:
```text
XB_ToolBox/
├─ __init__.py 节点注册 + 路由装配
├─ prestartup_script.py ComfyUI 启动清理(约定:必须留在根)
├─ nodes/ 节点实现
│ ├─ core/ 参数总线 / 接力 / 标签 / 字符串 / 开关 / 加速
│ ├─ image/ 提示词预设 / 编码 / 缩放 / 分割 / 漫画 / 神笔
│ ├─ video/ Wan / LTX / Hailuo / 数字人 / 视频读写
│ ├─ audio/ 音频切片 / YuE2 / 播放声音
│ ├─ llm/ XB-llama / GGUF 加载
│ ├─ hardware/ 显存与内存监控
│ └─ cosyvoice3/ CosyVoice3 节点包
├─ libs/ 无节点的纯逻辑(xb_paths / xb_presets / xb_gallery / xb_skills / yue2_* …)
├─ jzl_workflows/ 「机智罗的工作流」(数据 + 路由 + 图标样式)
├─ web/ 前端资源(原 js/)
├─ locales/ docs/ tests/ 国际化 / 文档 / 测试
├─ tools/ 开发用一次性脚本(不参与运行)
├─ support_llama/ xb_skill/ video_formats/ 数据目录
└─ LibreHardwareMonitorLib.dll
```
---
### 🗂️ 机智罗的工作流(侧栏入口)
装上 XB_ToolBox 后,ComfyUI 左侧栏会多出一个 **「机智罗」** 入口(紧贴官方 **「模板」** 上方),
点一下就会像官方「模板」那样在**屏幕正中弹出大面板**:左侧分类、顶部搜索、缩略图卡片网格,
点卡片直接在工作流新标签页打开,`Esc` / 点遮罩 / 右上角 ✕ 收起。
不需要改任何 ComfyUI 前端文件,升级 ComfyUI 也不会失效。
工作流数据都在 **`jzl_workflows/`** 一个目录里,方便单独管理:
```text
XB_ToolBox/
jzl_workflows/
__init__.py 注册入口(根 __init__.py 只调用它)
routes.py 后端路由:清单 / 工作流 / 预览图 / 图标样式
data/
index.json 英文清单
index.zh.json 中文清单(结构同官方 comfyui-workflow-templates)
anima_t2i_basic.json 工作流本身
anima_t2i_basic-1.webp 同名预览图,缺省时显示占位卡片
web/
xb_workflow_gallery.css
xb_icon_dark.png / xb_icon_light.png
web/jzl_workflows/
xb_workflow_gallery.js 面板脚本(必须在 web/ 下 ComfyUI 才会加载)
```
新增工作流只要两步:把工作流 JSON 放进 `jzl_workflows/data/`(文件名用 ASCII),
再在 `data/index.json`、`data/index.zh.json` 里各加一条条目(`name` 与 JSON 文件名一致)。
预览图放同目录,命名 `<name>-1.webp`。
---
### 🙏 致谢
XB_ToolBox 站在 ComfyUI 生态的肩膀上,部分节点为移植 / 套壳 / 改编,感谢原作者:
+72 -64
View File
@@ -28,8 +28,8 @@ async def choose_folder(request):
return web.json_response({"path": "", "error": f"弹窗调用失败: {str(e)}"})
from .nodes_audio_slicer import handle_audio_waveform
from .nodes_model_loader_v1 import _filter_by_keyword
from .nodes.audio.nodes_audio_slicer import handle_audio_waveform
from .nodes.llm.nodes_model_loader_v1 import _filter_by_keyword
@PromptServer.instance.routes.post("/xb_toolbox/audio_waveform")
async def get_audio_waveform(request):
@@ -39,7 +39,7 @@ async def get_audio_waveform(request):
@PromptServer.instance.routes.get("/xb_toolbox/yue2_recipes")
async def yue2_recipes(request):
"""YuE2 风格配方表:前端「YuE2 风格提示」节点联动用,与 yue2_styles.RECIPES 同一数据源"""
from .yue2_styles import GENRES, RECIPES
from .libs.yue2_styles import GENRES, RECIPES
return web.json_response({"recipes": RECIPES, "genres": GENRES})
@@ -61,6 +61,14 @@ async def get_model_list(request):
})
# ── 「机智罗的工作流」模板数据服务(清单 + 工作流 + 预览图 + 图标样式)──
try:
from .jzl_workflows import register_routes as _register_jzl_workflow_routes
_register_jzl_workflow_routes()
except Exception as _xb_tpl_err:
print(f"[XB_ToolBox] 工作流模板服务注册失败: {_xb_tpl_err}")
def print_success(msg):
print(f"\033[92m{msg}\033[0m")
@@ -82,42 +90,42 @@ XB_LOGO = "\n".join([
NODE_CLASS_MAPPINGS = {}
NODE_DISPLAY_NAME_MAPPINGS = {}
WEB_DIRECTORY = os.path.join(os.path.dirname(os.path.abspath(__file__)), "js")
WEB_DIRECTORY = os.path.join(os.path.dirname(os.path.abspath(__file__)), "web")
try:
from .nodes_vis import XB_VRAM_Calculator, XB_ChunkVisualization
from .nodes_vram import XTX_Data_Radar
from .nodes_video import XB_VideoParamsMaster, XB_ImageParamsMaster, XB_ImageParamsMasterMini, XB_MasterParameter, XB_VideoLoader, XB_VideoCombine
from .nodes_blockswap import XB_UNetBlockSwap, XB_CheckpointBlockSwap
from .nodes_blockswap_sage import XB_Sage_BlockSwap
from .nodes_wiring import XB_UNetNameBroadcaster, XB_CLIPNameBroadcaster
from .nodes_memory_viz import XB_MemoryVisualization
from .nodes_wan_vae import (XB_WanImageToVideo, XB_WanFirstLastFrameToVideo, XB_WanSoundImageToVideo, XB_WanFunControlToVideo, XB_WanVaceToVideo, XB_Wan22FunControlToVideo, XB_WanInfiniteTalkToVideo, XB_WanInfiniteTalkToVideo_Single, XB_WanInfiniteTalkToVideo_Dual, XB_WanFunInpaintToVideo, XB_WanCameraImageToVideo, XB_WanPhantomSubjectToVideo, XB_WanHuMoImageToVideo, XB_Wan22ImageToVideoLatent, XB_WanSoundImageToVideoExtend, XB_WanSCAILToVideo, XB_WanSCAILToVideoPro, XB_WanDancerVideo, XB_WanDanceSwitcher, XB_WanDancerCombo, XB_BerniniConditioning)
from .nodes_batch import XB_BatchFolderLoader, XB_BatchFolderBatchLoader
from .nodes_pipeline import XB_Wan_ParamBus, XB_Wan_RelayNode, XB_Wan_InfiniteRelayNode, XB_Video_Merger, XB_StoryboardSlicer,XB_WanAnimate_ParamBus,XB_WanAnimate_RelayNode, XB_WanInfiniteTalk_ParamBus, XB_WanInfiniteTalk_RelayNode, XB_Wan_InfiniteRelayNode_New, XB_WanAnimate_RelayNode_New, XB_WanInfiniteTalk_RelayNode_New, XB_WanSCAIL_ParamBus_New, XB_WanSCAIL_RelayNode_New, XB_WanInfiniteTalk_RelayNode_MultiRef, XB_WanInfiniteTalk_RelayNode_AllInOne
from .nodes_ltx_streamer import XB_LTX23_InfiniteStreamer
from .nodes_sageatt import XB_SageAttentionAccelerator
from .nodes_wan import XB_WanCompileSettings, XB_WanAnimateToVideo
from .nodes.hardware.nodes_vis import XB_VRAM_Calculator, XB_ChunkVisualization
from .nodes.hardware.nodes_vram import XTX_Data_Radar
from .nodes.video.nodes_video import XB_VideoParamsMaster, XB_ImageParamsMaster, XB_ImageParamsMasterMini, XB_MasterParameter, XB_VideoLoader, XB_VideoCombine
from .nodes.core.nodes_blockswap import XB_UNetBlockSwap, XB_CheckpointBlockSwap
from .nodes.core.nodes_blockswap_sage import XB_Sage_BlockSwap
from .nodes.core.nodes_wiring import XB_UNetNameBroadcaster, XB_CLIPNameBroadcaster
from .nodes.hardware.nodes_memory_viz import XB_MemoryVisualization
from .nodes.video.nodes_wan_vae import (XB_WanImageToVideo, XB_WanFirstLastFrameToVideo, XB_WanSoundImageToVideo, XB_WanFunControlToVideo, XB_WanVaceToVideo, XB_Wan22FunControlToVideo, XB_WanInfiniteTalkToVideo, XB_WanInfiniteTalkToVideo_Single, XB_WanInfiniteTalkToVideo_Dual, XB_WanFunInpaintToVideo, XB_WanCameraImageToVideo, XB_WanPhantomSubjectToVideo, XB_WanHuMoImageToVideo, XB_Wan22ImageToVideoLatent, XB_WanSoundImageToVideoExtend, XB_WanSCAILToVideo, XB_WanSCAILToVideoPro, XB_WanDancerVideo, XB_WanDanceSwitcher, XB_WanDancerCombo, XB_BerniniConditioning)
from .nodes.core.nodes_batch import XB_BatchFolderLoader, XB_BatchFolderBatchLoader
from .nodes.video.nodes_pipeline import XB_Wan_ParamBus, XB_Wan_RelayNode, XB_Wan_InfiniteRelayNode, XB_Video_Merger, XB_StoryboardSlicer,XB_WanAnimate_ParamBus,XB_WanAnimate_RelayNode, XB_WanInfiniteTalk_ParamBus, XB_WanInfiniteTalk_RelayNode, XB_Wan_InfiniteRelayNode_New, XB_WanAnimate_RelayNode_New, XB_WanInfiniteTalk_RelayNode_New, XB_WanSCAIL_ParamBus_New, XB_WanSCAIL_RelayNode_New, XB_WanInfiniteTalk_RelayNode_MultiRef, XB_WanInfiniteTalk_RelayNode_AllInOne
from .nodes.video.nodes_ltx_streamer import XB_LTX23_InfiniteStreamer
from .nodes.core.nodes_sageatt import XB_SageAttentionAccelerator
from .nodes.video.nodes_wan import XB_WanCompileSettings, XB_WanAnimateToVideo
from .nodes_label import XB_CanvasLabel
from .nodes_audio_slicer import XB_AudioSlicer, XB_AudioSlicerV1, XB_AudioSlicerV2, XB_AudioSlicerV3
from .nodes_digital_human import XB_DigitalHumanParams_Single, XB_DigitalHumanParams_Dual
from .nodes_model_loader_gguf import XB_ModelLoaderV1_GGUF, XB_ModelLoaderV2_GGUF, XB_ModelLoaderV3_GGUF
from .nodes_status_switch import XB_NodeStatusSwitch
from .nodes_fast_toggle import XB_DisableGroups, XB_DisableNodes, XB_AnySwitch
from .nodes_flux_encode import XB_FluxMultiImageEncode
from .nodes_qwen_encode import XB_TextEncodeQwenImageEdit, XB_TextEncodeQwenImageEditPlus
from .nodes_image_scale import XB_ImageScale, XB_ImageScalePixels
from .nodes_shenbi import XB_ShenBi
from .nodes_hailuo_video import XB_HailuoH3VideoParams
from .nodes_list_dispatcher import XB_ListDispatcher
from .nodes_batch_images import XB_BatchImages
from .nodes_reference_any import XB_ReferenceAny
from .nodes_segmentation import XB_HumanSegModelLoader, XB_HumanSegmentation
from .nodes.core.nodes_label import XB_CanvasLabel
from .nodes.audio.nodes_audio_slicer import XB_AudioSlicer, XB_AudioSlicerV1, XB_AudioSlicerV2, XB_AudioSlicerV3
from .nodes.video.nodes_digital_human import XB_DigitalHumanParams_Single, XB_DigitalHumanParams_Dual
from .nodes.llm.nodes_model_loader_gguf import XB_ModelLoaderV1_GGUF, XB_ModelLoaderV2_GGUF, XB_ModelLoaderV3_GGUF
from .nodes.core.nodes_status_switch import XB_NodeStatusSwitch
from .nodes.core.nodes_fast_toggle import XB_DisableGroups, XB_DisableNodes, XB_AnySwitch
from .nodes.image.nodes_flux_encode import XB_FluxMultiImageEncode
from .nodes.image.nodes_qwen_encode import XB_TextEncodeQwenImageEdit, XB_TextEncodeQwenImageEditPlus
from .nodes.image.nodes_image_scale import XB_ImageScale, XB_ImageScalePixels
from .nodes.image.nodes_shenbi import XB_ShenBi
from .nodes.video.nodes_hailuo_video import XB_HailuoH3VideoParams
from .nodes.core.nodes_list_dispatcher import XB_ListDispatcher
from .nodes.core.nodes_batch_images import XB_BatchImages
from .nodes.core.nodes_reference_any import XB_ReferenceAny
from .nodes.image.nodes_segmentation import XB_HumanSegModelLoader, XB_HumanSegmentation
# 注册视频转码端点(预览窗口实时缩放依赖)
from . import xb_video_server
from .nodes_vanilla_wrappers import (XB_KSampler, XB_KSamplerAdvanced,
from .libs import xb_video_server
from .nodes.core.nodes_vanilla_wrappers import (XB_KSampler, XB_KSamplerAdvanced,
XB_SamplerCustom, XB_SamplerCustomAdvanced,
XB_VAEDecode, XB_VAEDecodeTiled,
XB_VAEDecodeTiledImage,
@@ -125,16 +133,16 @@ try:
XB_VAEEncodeTiledImage,
XB_VAEEncodeForInpaint,
_AliasVAEDecodeTemporal)
from .nodes_string_merge import XB_StringMerge
from .nodes_clip_text_edit import XB_CLIPTextEdit
from .nodes_lora_loader import XB_TripleLoraLoader, XB_TripleLoraLoaderModelOnly
from .nodes_msr import XB_MSR
from .nodes_comic import XB_ComicPromptParser, XB_ComicTextRenderer, XB_AutoBubbleTextRenderer
from .nodes_super import XB_BerniniPromptRouter, XB_K2StyleRouter
from .nodes.core.nodes_string_merge import XB_StringMerge
from .nodes.core.nodes_clip_text_edit import XB_CLIPTextEdit
from .nodes.core.nodes_lora_loader import XB_TripleLoraLoader, XB_TripleLoraLoaderModelOnly
from .nodes.core.nodes_msr import XB_MSR
from .nodes.image.nodes_comic import XB_ComicPromptParser, XB_ComicTextRenderer, XB_AutoBubbleTextRenderer
from .nodes.core.nodes_super import XB_BerniniPromptRouter, XB_K2StyleRouter
# ── Llama 节点 (LLM/VLM) ──
try:
from .nodes_llama import (
from .nodes.llm.nodes_llama import (
XB_llamaModelLoader, XB_llamaModelLoaderPro, XB_llamaInstruct, XB_llamaParameters,
XB_llamaUnloadModel, XB_llamaCleanStates,
XB_llamaBBoxes2BBox, XB_llamaUnpackCodeBlock, XB_llamaPromptEnhancer,
@@ -144,7 +152,7 @@ try:
XB_RoleSceneDispatcher,
)
# ── 提示词增强反推(模型加载器 + 指令推理 + 提示词增强预设 + 推理参数 + LLM-API 设置 五合一)──
from .nodes_llama_prompt_reverse import XB_llamaPromptReverse
from .nodes.llm.nodes_llama_prompt_reverse import XB_llamaPromptReverse
_LLAMA_AVAILABLE = True
except ImportError as e:
print_warning(f"[XB-llama] llama-cpp-python 未安装, Llama/MiniMax 节点不可用: {e}")
@@ -152,7 +160,7 @@ try:
# ── 🖼️ 生图提示词预设Pro(生图提示词预设 + 提示词增强反推 融合)──
try:
from .nodes_image_prompt_preset_pro import XB_ImagePromptPresetPro
from .nodes.image.nodes_image_prompt_preset_pro import XB_ImagePromptPresetPro
_PRO_AVAILABLE = True
except Exception as e:
print_warning(f"[XB-BOX] 生图提示词预设Pro 未加载:{e}")
@@ -160,7 +168,7 @@ try:
# ── 🖼️ Qwen2.1提示词预设(生图提示词预设 + 官方 Load CLIP + 官方 Generate Text)──
try:
from .nodes_qwen_prompt_preset import XB_QwenPromptPreset
from .nodes.image.nodes_qwen_prompt_preset import XB_QwenPromptPreset
_QWEN_AVAILABLE = True
except Exception as e:
print_warning(f"[XB-BOX] Qwen2.1提示词预设 未加载:{e}")
@@ -168,30 +176,30 @@ try:
# ── 🎵 YuE2 音乐节点(风格提示 / 歌词 / 乐谱工作台 / 翻唱准备 / 转谱导入)──
try:
from .nodes_yue2_style import XB_YuE2StylePrompt
from .nodes_yue2_lyrics import XB_YuE2Lyrics
from .nodes_yue2_score import XB_YuE2ScoreWorkbench, XB_YuE2CoverPrep
from .nodes_yue2_transcription import XB_YuE2Transcription
from .nodes.audio.nodes_yue2_style import XB_YuE2StylePrompt
from .nodes.audio.nodes_yue2_lyrics import XB_YuE2Lyrics
from .nodes.audio.nodes_yue2_score import XB_YuE2ScoreWorkbench, XB_YuE2CoverPrep
from .nodes.audio.nodes_yue2_transcription import XB_YuE2Transcription
_YUE2_AVAILABLE = True
except Exception as e:
print_warning(f"[XB-BOX] YuE2 音乐节点 未加载:{e}")
_YUE2_AVAILABLE = False
# ── CosyVoice3 音频节点 ──
from .cosyvoice3.nodes.model_loader import XB_CosyVoice3_ModelLoader
from .cosyvoice3.nodes.zero_shot import XB_CosyVoice3_ZeroShot
from .cosyvoice3.nodes.cross_lingual import XB_CosyVoice3_CrossLingual
from .cosyvoice3.nodes.voice_conversion import XB_CosyVoice3_VoiceConversion
from .cosyvoice3.nodes.audio_crop import XB_CosyVoice3_AudioCrop
from .cosyvoice3.nodes.dialog import XB_CosyVoice3_Dialog
from .cosyvoice3.nodes.instruct2 import XB_CosyVoice3_Instruct2
from .cosyvoice3.nodes.save_speaker import XB_CosyVoice3_SaveSpeaker
from .cosyvoice3.nodes.speaker_clone import XB_CosyVoice3_SpeakerClone
from .cosyvoice3.nodes.speaker_instruct2 import XB_CosyVoice3_SpeakerInstruct2
from .nodes.cosyvoice3.nodes.model_loader import XB_CosyVoice3_ModelLoader
from .nodes.cosyvoice3.nodes.zero_shot import XB_CosyVoice3_ZeroShot
from .nodes.cosyvoice3.nodes.cross_lingual import XB_CosyVoice3_CrossLingual
from .nodes.cosyvoice3.nodes.voice_conversion import XB_CosyVoice3_VoiceConversion
from .nodes.cosyvoice3.nodes.audio_crop import XB_CosyVoice3_AudioCrop
from .nodes.cosyvoice3.nodes.dialog import XB_CosyVoice3_Dialog
from .nodes.cosyvoice3.nodes.instruct2 import XB_CosyVoice3_Instruct2
from .nodes.cosyvoice3.nodes.save_speaker import XB_CosyVoice3_SaveSpeaker
from .nodes.cosyvoice3.nodes.speaker_clone import XB_CosyVoice3_SpeakerClone
from .nodes.cosyvoice3.nodes.speaker_instruct2 import XB_CosyVoice3_SpeakerInstruct2
# ── 🎬 保存视频(Create Video + Save Video 二合一,依赖新版 V3 API)──
try:
from .nodes_save_video import XB_SaveVideo
from .nodes.video.nodes_save_video import XB_SaveVideo
_SAVE_VIDEO_AVAILABLE = True
except Exception as e:
print_warning(f"[XB-BOX] 保存视频节点 未加载:{e}")
@@ -199,27 +207,27 @@ try:
# ── 🎬 加载视频(Load Video + Get Video Components 二合一,依赖新版 V3 API)──
try:
from .nodes_load_video import XB_LoadVideo
from .nodes.video.nodes_load_video import XB_LoadVideo
_LOAD_VIDEO_AVAILABLE = True
except Exception as e:
print_warning(f"[XB-BOX] 加载视频节点 未加载:{e}")
_LOAD_VIDEO_AVAILABLE = False
# ── 生图提示词预设(提示词 + 空latent 一体化)──
from .nodes_image_prompt_preset import XB_ImagePromptPreset
from .nodes.image.nodes_image_prompt_preset import XB_ImagePromptPreset
# ── 生图预设(生图提示词预设 + 姿态参考图 / 穿搭参考图)──
# 同时 import xb_gallery → 注册 /xb_toolbox/gallery/* 端点(图库绑定、素材清单、图片读取)
try:
from . import xb_gallery # noqa: F401
from .nodes_image_preset import XB_ImagePreset
from .libs import xb_gallery # noqa: F401
from .nodes.image.nodes_image_preset import XB_ImagePreset
_IMAGE_PRESET_AVAILABLE = True
except Exception as e:
print_warning(f"[XB-BOX] 生图预设节点 未加载:{e}")
_IMAGE_PRESET_AVAILABLE = False
# ── 播放声音(复刻 pysssss PlaySound,自带音频资源)──
from .nodes_play_sound import XB_PlaySound
from .nodes.audio.nodes_play_sound import XB_PlaySound
NODE_CLASS_MAPPINGS = {
"XB_VRAM_Calculator": XB_VRAM_Calculator,
+17
View File
@@ -0,0 +1,17 @@
"""「机智罗的工作流」— 侧栏入口 + 居中弹窗的模板库
全部实现集中在 ``jzl_workflows/`` 一个目录里:
jzl_workflows/
__init__.py 本文件:对外只暴露 register_routes()
routes.py 后端路由(清单 / 工作流 / 预览图 / 图标与样式)
data/ index*.json 清单 + 工作流 JSON + 预览图
web/ 面板样式与深浅两套图标
前端脚本只有 ``js/jzl_workflows/xb_workflow_gallery.js`` 留在 ``js/`` 下,
因为 ComfyUI 只加载 ``WEB_DIRECTORY`` 内的 js(该目录会递归扫描子文件夹)。
"""
from .routes import register_routes
__all__ = ["register_routes"]
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because it is too large Load Diff
+55
View File
@@ -0,0 +1,55 @@
[
{
"moduleName": "jzl",
"category": "JZL Workflows",
"title": "Anima",
"icon": "icon-[lucide--image]",
"type": "image",
"templates": [
{
"name": "anima_t2i_basic",
"title": "Anima Text to Image (Basic)",
"description": "Anima basic text-to-image: prompt preset + model loader + sampling/decoding in one graph.",
"tags": ["Anima", "Text to Image"],
"thumbnail": ["anima_t2i_basic-1.webp"],
"date": "2026-11-09",
"username": "JZL",
"openSource": true,
"minComfyUIVersion": "0.0.0"
},
{
"name": "anima_inpaint_basic",
"title": "Anima Inpainting (Basic)",
"description": "Anima inpainting: basic masked-region repaint flow.",
"tags": ["Anima", "Inpainting"],
"thumbnail": ["anima_inpaint_basic-1.webp"],
"date": "2026-11-09",
"username": "JZL",
"openSource": true,
"minComfyUIVersion": "0.0.0"
},
{
"name": "anima_t2i_upscale",
"title": "Anima Text to Image + Upscale",
"description": "Anima text-to-image followed by an upscale pass.",
"tags": ["Anima", "Text to Image", "Upscale"],
"thumbnail": ["anima_t2i_upscale-1.webp"],
"date": "2026-11-09",
"username": "JZL",
"openSource": true,
"minComfyUIVersion": "0.0.0"
},
{
"name": "anima_control_t2i",
"title": "Anima Control to Image (Basic)",
"description": "Anima control-guided generation: basic flow with control conditions.",
"tags": ["Anima", "Control"],
"thumbnail": ["anima_control_t2i-1.webp"],
"date": "2026-11-09",
"username": "JZL",
"openSource": true,
"minComfyUIVersion": "0.0.0"
}
]
}
]
+55
View File
@@ -0,0 +1,55 @@
[
{
"moduleName": "jzl",
"category": "机智罗的工作流",
"title": "Anima",
"icon": "icon-[lucide--image]",
"type": "image",
"templates": [
{
"name": "anima_t2i_basic",
"title": "Anima 文生图基础工作流",
"description": "Anima 基础文生图:提示词预设 + 模型加载 + 采样解码一体化。",
"tags": ["Anima", "文生图"],
"thumbnail": ["anima_t2i_basic-1.webp"],
"date": "2026-11-09",
"username": "机智罗",
"openSource": true,
"minComfyUIVersion": "0.0.0"
},
{
"name": "anima_inpaint_basic",
"title": "Anima 局部重绘基础工作流",
"description": "Anima 局部重绘:遮罩区域重绘的基础流程。",
"tags": ["Anima", "局部重绘"],
"thumbnail": ["anima_inpaint_basic-1.webp"],
"date": "2026-11-09",
"username": "机智罗",
"openSource": true,
"minComfyUIVersion": "0.0.0"
},
{
"name": "anima_t2i_upscale",
"title": "Anima 文生图基础+放大工作流",
"description": "Anima 文生图后接放大流程。",
"tags": ["Anima", "文生图", "放大"],
"thumbnail": ["anima_t2i_upscale-1.webp"],
"date": "2026-11-09",
"username": "机智罗",
"openSource": true,
"minComfyUIVersion": "0.0.0"
},
{
"name": "anima_control_t2i",
"title": "Anima 控制生图基础工作流",
"description": "Anima 控制生图:带控制条件的基础流程。",
"tags": ["Anima", "控制生图"],
"thumbnail": ["anima_control_t2i-1.webp"],
"date": "2026-11-09",
"username": "机智罗",
"openSource": true,
"minComfyUIVersion": "0.0.0"
}
]
}
]
+114
View File
@@ -0,0 +1,114 @@
"""XB_ToolBox — 「机智罗的工作流」模板数据服务
=========================================================
清单结构照抄官方 ``comfyui-workflow-templates`` 的 ``index.json`` / ``index.zh.json``:
``[{moduleName, category, title, icon, type, templates: [...]}]``。
前端「机智罗」侧栏面板只读这两个清单,按 ``category`` / ``title`` 分组渲染卡片,
点击卡片再取 ``<name>.json`` 加载工作流,预览图按官方 ``<name>-1.webp`` 约定托管。
安全:只放行清单里登记过的文件名,且解析后的路径必须仍在 ``xb_templates`` 目录内。
"""
from __future__ import annotations
import json
import re
from pathlib import Path
from aiohttp import web
ROOT = Path(__file__).resolve().parent / "data"
# 面板静态资源(图标 / 样式):由本模块单独托管,绕开 /extensions 的 24h 强缓存
WEB_DIR = Path(__file__).resolve().parent / "web"
WEB_ASSETS = (
"xb_icon_dark.png",
"xb_icon_light.png",
"xb_workflow_gallery.css",
)
# 清单文件名(英文 / 中文)
MANIFESTS = ("index.json", "index.zh.json")
# 放行文件名只允许 ASCII 安全字符,天然排除 "/" 与 ".."
_SAFE_NAME = re.compile(r"^[A-Za-z0-9._-]+$")
def _read_manifest(name: str) -> list:
path = ROOT / name
if not path.is_file():
return []
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError):
return []
return data if isinstance(data, list) else []
def _allowed_files() -> set[str]:
"""清单登记过的文件名白名单(含两个 index 清单本身)。"""
allowed = set(MANIFESTS)
for manifest_name in MANIFESTS:
for group in _read_manifest(manifest_name):
if not isinstance(group, dict):
continue
templates = group.get("templates")
if not isinstance(templates, list):
continue
for tpl in templates:
if not isinstance(tpl, dict):
continue
name = tpl.get("name")
if isinstance(name, str) and _SAFE_NAME.match(name):
allowed.add(f"{name}.json")
thumbnails = tpl.get("thumbnail")
if isinstance(thumbnails, list):
for thumb in thumbnails:
if not isinstance(thumb, str):
continue
base = thumb.replace("\\", "/").rsplit("/", 1)[-1]
if _SAFE_NAME.match(base):
allowed.add(base)
return allowed
def register_routes():
from server import PromptServer
routes = PromptServer.instance.routes
@routes.get("/xb_toolbox/templates/{filename}")
async def get_template_asset(request):
filename = request.match_info.get("filename", "")
if not _SAFE_NAME.match(filename) or filename not in _allowed_files():
raise web.HTTPNotFound()
target = (ROOT / filename).resolve()
if target.parent != ROOT or not target.is_file():
raise web.HTTPNotFound()
is_manifest = filename in MANIFESTS
headers = {
# 清单必须实时反映磁盘改动,图片只做协商缓存
"Cache-Control": "no-store" if is_manifest else "no-cache",
"X-Content-Type-Options": "nosniff",
}
return web.FileResponse(target, headers=headers)
@routes.get("/xb_toolbox/assets/{name}")
async def get_web_asset(request):
"""面板静态资源(图标 / 样式)单独托管。
/extensions/* 由 ComfyUI 静态托管并带 ``max-age=86400``,换图换样式后浏览器会一直用旧的;
这里用 no-cache,让浏览器每次协商,改完普通刷新即可生效。
"""
name = request.match_info.get("name", "")
if name not in WEB_ASSETS:
raise web.HTTPNotFound()
target = WEB_DIR / name
if not target.is_file():
raise web.HTTPNotFound()
return web.FileResponse(
target,
headers={"Cache-Control": "no-cache", "X-Content-Type-Options": "nosniff"},
)
Binary file not shown.

After

Width:  |  Height:  |  Size: 16 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 9.7 KiB

+404
View File
@@ -0,0 +1,404 @@
/* XB_ToolBox — 「机智罗的工作流」侧栏面板样式
* 全部走 ComfyUI 的 CSS 变量,跟随明/暗主题。 */
.xb-wf-root {
display: flex;
flex-direction: column;
gap: 8px;
height: 100%;
min-height: 0;
box-sizing: border-box;
padding: 10px;
overflow-y: auto;
font-size: 13px;
line-height: 1.4;
color: var(--input-text, #ddd);
background: var(--comfy-menu-bg, #1e1e1e);
}
.xb-wf-root.is-busy {
cursor: progress;
}
.xb-wf-head {
display: flex;
align-items: center;
gap: 8px;
flex: 0 0 auto;
}
.xb-wf-logo {
width: 22px;
height: 22px;
object-fit: contain;
flex: 0 0 auto;
}
.xb-wf-title {
flex: 1 1 auto;
min-width: 0;
font-weight: 600;
font-size: 13px;
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
}
.xb-wf-icon-btn {
flex: 0 0 auto;
width: 24px;
height: 24px;
display: inline-flex;
align-items: center;
justify-content: center;
padding: 0;
border: 1px solid var(--border-color, #444);
border-radius: 6px;
background: transparent;
color: inherit;
font-size: 13px;
cursor: pointer;
transition: background-color 0.15s ease;
}
.xb-wf-icon-btn:hover:not(:disabled) {
background: var(--comfy-input-bg, rgba(255, 255, 255, 0.08));
}
.xb-wf-icon-btn:disabled {
opacity: 0.5;
cursor: default;
}
.xb-wf-search {
flex: 0 0 auto;
}
.xb-wf-search input {
width: 100%;
box-sizing: border-box;
padding: 5px 8px;
border: 1px solid var(--border-color, #444);
border-radius: 6px;
background: var(--comfy-input-bg, #2a2a2a);
color: var(--input-text, #ddd);
font-size: 12px;
outline: none;
}
.xb-wf-search input:focus {
border-color: var(--p-primary-color, #4a9eff);
}
.xb-wf-cats {
display: flex;
flex-wrap: wrap;
gap: 6px;
flex: 0 0 auto;
}
.xb-wf-cats.is-hidden {
display: none;
}
.xb-wf-chip {
padding: 3px 9px;
border: 1px solid var(--border-color, #444);
border-radius: 999px;
background: transparent;
color: inherit;
font-size: 12px;
cursor: pointer;
white-space: nowrap;
transition: background-color 0.15s ease, border-color 0.15s ease;
}
.xb-wf-chip:hover {
background: var(--comfy-input-bg, rgba(255, 255, 255, 0.08));
}
.xb-wf-chip.is-active {
border-color: var(--p-primary-color, #4a9eff);
background: var(--p-primary-color, #4a9eff);
color: #fff;
}
.xb-wf-grid {
display: grid;
grid-template-columns: repeat(auto-fill, minmax(140px, 1fr));
gap: 8px;
flex: 1 1 auto;
align-content: start;
}
.xb-wf-card {
display: flex;
flex-direction: column;
gap: 4px;
min-width: 0;
padding: 6px;
border: 1px solid var(--border-color, #444);
border-radius: 8px;
background: var(--comfy-input-bg, #252525);
color: inherit;
font: inherit;
text-align: left;
cursor: pointer;
transition: border-color 0.15s ease, transform 0.15s ease;
}
.xb-wf-card:hover {
border-color: var(--p-primary-color, #4a9eff);
transform: translateY(-1px);
}
.xb-wf-card:focus-visible {
outline: 2px solid var(--p-primary-color, #4a9eff);
outline-offset: 1px;
}
.xb-wf-root.is-busy .xb-wf-card {
pointer-events: none;
opacity: 0.7;
}
.xb-wf-thumb {
position: relative;
width: 100%;
aspect-ratio: 4 / 3;
overflow: hidden;
border-radius: 5px;
background: var(--comfy-menu-bg, #1a1a1a);
}
.xb-wf-thumb img {
display: block;
width: 100%;
height: 100%;
object-fit: cover;
}
.xb-wf-thumb.is-empty {
display: flex;
align-items: center;
justify-content: center;
padding: 6px;
box-sizing: border-box;
background: linear-gradient(135deg, rgba(74, 158, 255, 0.22), rgba(180, 74, 255, 0.22));
}
.xb-wf-thumb-text {
font-size: 11px;
line-height: 1.3;
text-align: center;
color: var(--input-text, #ddd);
word-break: break-word;
display: -webkit-box;
-webkit-line-clamp: 3;
-webkit-box-orient: vertical;
overflow: hidden;
}
.xb-wf-name {
font-size: 12px;
font-weight: 600;
display: -webkit-box;
-webkit-line-clamp: 2;
-webkit-box-orient: vertical;
overflow: hidden;
}
.xb-wf-tags {
display: flex;
flex-wrap: wrap;
gap: 4px;
}
.xb-wf-tag {
padding: 1px 6px;
border-radius: 999px;
background: rgba(128, 128, 128, 0.2);
font-size: 10px;
opacity: 0.85;
}
.xb-wf-empty {
grid-column: 1 / -1;
padding: 24px 12px;
text-align: center;
opacity: 0.6;
font-size: 12px;
}
.xb-wf-status {
flex: 0 0 auto;
min-height: 0;
font-size: 11px;
opacity: 0.75;
}
.xb-wf-status:empty {
display: none;
}
.xb-wf-status.is-error {
opacity: 1;
color: var(--error-text, #ff6b6b);
}
/* ── 居中大弹窗(与官方「模板」一致的布局)────────────────────────── */
.xb-wf-backdrop {
position: fixed;
inset: 0;
z-index: 10000;
display: flex;
align-items: center;
justify-content: center;
padding: 3vh 3vw;
box-sizing: border-box;
background: rgba(0, 0, 0, 0.6);
backdrop-filter: blur(2px);
}
.xb-wf-backdrop.is-busy {
cursor: progress;
}
.xb-wf-modal {
display: flex;
width: min(1240px, 100%);
height: min(840px, 100%);
overflow: hidden;
border: 1px solid var(--border-color, #3a3a3a);
border-radius: 12px;
background: var(--comfy-menu-bg, #202020);
color: var(--input-text, #ddd);
font-size: 13px;
line-height: 1.4;
box-shadow: 0 24px 64px rgba(0, 0, 0, 0.55);
}
.xb-wf-side {
flex: 0 0 208px;
display: flex;
flex-direction: column;
gap: 10px;
padding: 14px 10px;
overflow-y: auto;
border-right: 1px solid var(--border-color, #333);
background: rgba(255, 255, 255, 0.02);
}
.xb-wf-side-head {
display: flex;
align-items: center;
gap: 8px;
padding: 0 6px 10px;
border-bottom: 1px solid var(--border-color, #333);
}
.xb-wf-side-title {
font-weight: 600;
font-size: 13px;
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
}
.xb-wf-side-list {
display: flex;
flex-direction: column;
gap: 2px;
}
.xb-wf-side-item {
display: block;
width: 100%;
padding: 6px 10px;
border: 0;
border-radius: 6px;
background: transparent;
color: inherit;
font: inherit;
text-align: left;
cursor: pointer;
opacity: 0.85;
transition: background-color 0.12s ease, opacity 0.12s ease;
}
.xb-wf-side-item:hover {
background: rgba(255, 255, 255, 0.07);
opacity: 1;
}
.xb-wf-side-item.is-active {
background: var(--p-primary-color, #4a9eff);
color: #fff;
opacity: 1;
}
.xb-wf-main {
flex: 1 1 auto;
display: flex;
flex-direction: column;
min-width: 0;
min-height: 0;
}
.xb-wf-main-head {
display: flex;
align-items: center;
gap: 8px;
flex: 0 0 auto;
padding: 12px 14px;
border-bottom: 1px solid var(--border-color, #333);
}
.xb-wf-search-input {
flex: 1 1 auto;
min-width: 0;
padding: 6px 10px;
border: 1px solid var(--border-color, #444);
border-radius: 8px;
background: var(--comfy-input-bg, #2a2a2a);
color: var(--input-text, #ddd);
font-size: 13px;
outline: none;
}
.xb-wf-search-input:focus {
border-color: var(--p-primary-color, #4a9eff);
}
.xb-wf-modal .xb-wf-icon-btn {
width: 28px;
height: 28px;
font-size: 14px;
}
.xb-wf-modal .xb-wf-grid {
gap: 12px;
padding: 14px;
overflow-y: auto;
grid-template-columns: repeat(auto-fill, minmax(190px, 1fr));
}
.xb-wf-modal .xb-wf-card {
padding: 8px;
gap: 6px;
}
.xb-wf-modal .xb-wf-name {
font-size: 13px;
}
.xb-wf-modal .xb-wf-status {
padding: 0 14px 10px;
}
.xb-wf-backdrop.is-busy .xb-wf-card {
pointer-events: none;
opacity: 0.7;
}
View File
+1 -1
View File
@@ -37,7 +37,7 @@ MIME_BY_EXT = {".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg",
ANGLE_ORDER = ("正面", "斜侧", "侧面", "俯视", "仰视")
THUMB_MAX = 384
FOLDER_TTL = 5.0 # 普通文件夹扫描缓存(秒)
PACKAGE_ROOT = Path(__file__).resolve().parent
PACKAGE_ROOT = Path(__file__).resolve().parent.parent
_cache_lock = threading.Lock()
_manifest_cache: dict = {} # {(root, mtime, size): catalogue}
View File
+1 -1
View File
@@ -19,7 +19,7 @@ import os
import re
import time
ROOT_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "xb_skill")
ROOT_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "xb_skill")
DEFAULT_NODE = "XB_ImagePromptPreset"
ZH = "ZH"
+1 -1
View File
@@ -20,7 +20,7 @@ import re
import threading
import time
PACKAGE_DIR = os.path.dirname(os.path.abspath(__file__))
PACKAGE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
# 技能文件统一放在 xb_skill/skills(与预设词 txt 同一个 xb_skill 根目录)
SKILL_DIR = os.path.join(PACKAGE_DIR, "xb_skill", "skills")
# 旧位置(xb_skill/skills):首次使用时自动把文件迁移过去,老部署无需手动搬
View File
@@ -1,485 +1,485 @@
"""
XB-ToolBox 音频切片节点
======================
完全独立的音频加载/切片节点,支持可视化波形截取。
使用 av (PyAV) 解码,兼容所有格式(同 VHS 用 ffmpeg 的效果)
"""
import os, asyncio
import folder_paths
import torch
import av
from aiohttp import web
import server
from .xb_av import open_av
def _load_audio_file(filepath: str):
"""用 av 解码音频,返回 (waveform_2d, sample_rate)。加载失败返回 None"""
try:
with open_av(filepath) as af:
if not af.streams.audio:
return None
stream = af.streams.audio[0]
sr = stream.codec_context.sample_rate
frames = []
for frame in af.decode(streams=stream.index):
buf = torch.from_numpy(frame.to_ndarray())
if buf.dim() == 1:
buf = buf.unsqueeze(0)
elif buf.shape[0] != stream.channels:
buf = buf.view(-1, stream.channels).t()
frames.append(buf)
if not frames:
return None
wav = torch.cat(frames, dim=1)
# 统一转为 float32
if not wav.dtype.is_floating_point:
if wav.dtype == torch.int16:
wav = wav.float() / (2 ** 15)
elif wav.dtype == torch.int32:
wav = wav.float() / (2 ** 31)
return wav, sr
except Exception:
return None
def _make_audio(waveform, sample_rate):
if waveform.dim() == 1:
waveform = waveform.unsqueeze(0).unsqueeze(0)
elif waveform.dim() == 2:
waveform = waveform.unsqueeze(0)
return {"waveform": waveform, "sample_rate": sample_rate}
@server.PromptServer.instance.routes.post("/xb_audio_waveform")
@server.PromptServer.instance.routes.post("/xb_toolbox/audio_waveform")
async def handle_audio_waveform(request):
"""返回音频文件的波形峰值数据,支持按时间范围切片重算"""
try:
data = await request.json()
except Exception:
return web.json_response({"error": "Invalid JSON body"})
filename = data.get("filename", "")
num_peaks = data.get("num_peaks", 400)
start_time = data.get("start_time", None)
end_time = data.get("end_time", None)
if not filename:
return web.json_response({"peaks": [], "duration": 0})
input_dir = folder_paths.get_input_directory()
audio_path = os.path.join(input_dir, filename)
if not os.path.exists(audio_path):
return web.json_response({"peaks": [], "duration": 0})
try:
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(None, _load_audio_file, audio_path)
if result is None:
return web.json_response({"peaks": [], "duration": 0})
waveform, sample_rate = result
if waveform.shape[0] > 1:
waveform = waveform.mean(dim=0, keepdim=True)
duration = waveform.shape[1] / sample_rate
if start_time is not None and end_time is not None and end_time > start_time:
s_idx = max(0, int(start_time * sample_rate))
e_idx = min(waveform.shape[1], int(end_time * sample_rate))
waveform = waveform[:, s_idx:e_idx]
total_samples = waveform.shape[1]
if total_samples == 0:
return web.json_response({"peaks": [], "duration": duration})
chunk_size = max(1, total_samples // num_peaks)
peaks = []
for i in range(num_peaks):
chunk = waveform[0, i * chunk_size : (i + 1) * chunk_size]
if chunk.numel() == 0:
peaks.append([0.0, 0.0])
continue
peaks.append([float(chunk.max().item()), float(chunk.min().item())])
return web.json_response({"peaks": peaks, "duration": duration})
except Exception as e:
return web.json_response({"error": str(e)})
def _snap_4n1(frames):
"""将帧数对齐到 4N+1(满足 InfiniteTalk 的硬性要求)"""
return max(1, ((frames + 2) // 4) * 4 + 1)
# ============================================================
# XB_AudioSlicer — 音频切片(基础版)
# ============================================================
class XB_AudioSlicer:
"""音频加载与切片节点 — 支持频谱可视化 + 拖拽分割线截取
输出帧数(非秒数),可直接对接 InfiniteTalk 接力点的 segment_length。
额外输出 fps,方便连接到总线统一帧率。"""
@classmethod
def INPUT_TYPES(cls):
input_dir = folder_paths.get_input_directory()
audio_files = ["none"]
if os.path.exists(input_dir):
for f in sorted(os.listdir(input_dir)):
if f.lower().endswith(('.mp3', '.wav', '.flac', '.ogg', '.m4a', '.aac', '.opus', '.wma', '.webm')):
audio_files.append(f)
return {
"required": {
"audio": (audio_files,),
"fps": ("FLOAT", {"default": 25.0, "min": 1.0, "max": 120.0, "step": 1.0}),
"start_time": ("FLOAT", {"default": 0.00, "min": 0.00, "max": 99999.00, "step": 0.01, "precision": 2, "round": 0.01}),
"end_time": ("FLOAT", {"default": 10.00, "min": 0.00, "max": 99999.00, "step": 0.01, "precision": 2, "round": 0.01}),
"duration_display": ("STRING", {"default": "0 帧", "multiline": False}),
}
}
RETURN_TYPES = ("AUDIO", "INT", "FLOAT")
RETURN_NAMES = ("audio", "frame_count", "fps")
FUNCTION = "slice_audio"
CATEGORY = "XB_ToolBox/Audio"
@classmethod
def VALIDATE_INPUTS(cls, audio, **kwargs):
return True
def slice_audio(self, audio, fps, start_time, end_time, duration_display):
if audio == "none":
return (_make_audio(torch.zeros((1, 1), dtype=torch.float32), 44100), 0, fps)
result = _load_audio_file(os.path.join(folder_paths.get_input_directory(), audio))
if result is None:
return (_make_audio(torch.zeros((1, 1), dtype=torch.float32), 44100), 0, fps)
waveform, sample_rate = result
if waveform.shape[0] > 1:
waveform = waveform.mean(dim=0, keepdim=True)
total_duration = waveform.shape[1] / sample_rate
frame_duration = 1.0 / fps
start_time = max(0.0, min(start_time, total_duration))
end_time = max(start_time + frame_duration, min(end_time, total_duration))
start_sample = int(start_time * sample_rate)
end_sample = int(end_time * sample_rate)
if end_sample <= start_sample:
end_sample = min(start_sample + int(frame_duration * 4 * sample_rate), waveform.shape[1])
sliced = waveform[:, start_sample:end_sample]
duration_sec = (end_sample - start_sample) / sample_rate
raw_frames = int(round(duration_sec * fps))
frame_count = _snap_4n1(raw_frames)
# 🛡️ 音频物理长度对齐 4N+1 帧数,防止音视频时间轴错位
target_samples = int((frame_count / fps) * sample_rate)
if sliced.shape[1] < target_samples:
pad = torch.zeros((sliced.shape[0], target_samples - sliced.shape[1]), dtype=sliced.dtype, device=sliced.device)
sliced = torch.cat([sliced, pad], dim=1)
elif sliced.shape[1] > target_samples:
sliced = sliced[:, :target_samples]
if frame_count != raw_frames:
print(f"🔧 [音频切片] 帧数 {raw_frames} → {frame_count} (对齐 4N+1)")
return (_make_audio(sliced, sample_rate), frame_count, fps)
# ============================================================
# XB_AudioSlicerV1 — 可视化音频切片(波形+拖拽分割线)
# ============================================================
class XB_AudioSlicerV1:
"""音频切片节点 V1 — 可视化波形窗口 + 拖拽分割线截取
输出帧数(非秒数),可直接对接 InfiniteTalk 接力点的 segment_length。
额外输出 fps,方便连接到总线统一帧率。"""
@classmethod
def INPUT_TYPES(cls):
input_dir = folder_paths.get_input_directory()
audio_files = ["none"]
if os.path.exists(input_dir):
for f in sorted(os.listdir(input_dir)):
if f.lower().endswith(('.mp3', '.wav', '.flac', '.ogg', '.m4a', '.aac', '.opus', '.wma', '.webm')):
audio_files.append(f)
return {
"required": {
"audio": (audio_files,),
"fps": ("FLOAT", {"default": 25.0, "min": 1.0, "max": 120.0, "step": 1.0}),
"start_time": ("FLOAT", {"default": 0.00, "min": 0.00, "max": 99999.00, "step": 0.01}),
"end_time": ("FLOAT", {"default": 10.00, "min": 0.00, "max": 99999.00, "step": 0.01}),
"duration_display": ("STRING", {"default": "0 帧", "multiline": False}),
}
}
RETURN_TYPES = ("AUDIO", "INT", "FLOAT")
RETURN_NAMES = ("audio", "frame_count", "fps")
FUNCTION = "slice_audio"
CATEGORY = "XB_ToolBox/Audio"
@classmethod
def VALIDATE_INPUTS(cls, audio, **kwargs):
return True
def slice_audio(self, audio, fps, start_time, end_time, duration_display):
if audio == "none":
return (_make_audio(torch.zeros((1, 1), dtype=torch.float32), 44100), 0, fps)
result = _load_audio_file(os.path.join(folder_paths.get_input_directory(), audio))
if result is None:
return (_make_audio(torch.zeros((1, 1), dtype=torch.float32), 44100), 0, fps)
waveform, sample_rate = result
if waveform.shape[0] > 1:
waveform = waveform.mean(dim=0, keepdim=True)
total_duration = waveform.shape[1] / sample_rate
frame_duration = 1.0 / fps
start_time = max(0.0, min(start_time, total_duration))
end_time = max(start_time + frame_duration, min(end_time, total_duration))
start_sample = int(start_time * sample_rate)
end_sample = int(end_time * sample_rate)
if end_sample <= start_sample:
end_sample = min(start_sample + int(frame_duration * 4 * sample_rate), waveform.shape[1])
sliced = waveform[:, start_sample:end_sample]
duration_sec = (end_sample - start_sample) / sample_rate
raw_frames = int(round(duration_sec * fps))
frame_count = _snap_4n1(raw_frames)
if frame_count != raw_frames:
print(f"🔧 [音频切片V1] 帧数 {raw_frames} → {frame_count} (对齐 4N+1)")
return (_make_audio(sliced, sample_rate), frame_count, fps)
# ============================================================
# XB_AudioSlicerV2 — 双人音频切片
# ============================================================
class XB_AudioSlicerV2:
"""双人音频切片 — 音频1 + 间隔(帧) + 音频2,独立可视化
输出帧数(非秒数),可直接对接 InfiniteTalk 接力点的 segment_length。"""
@classmethod
def INPUT_TYPES(cls):
input_dir = folder_paths.get_input_directory()
audio_files = ["none"]
if os.path.exists(input_dir):
for f in sorted(os.listdir(input_dir)):
if f.lower().endswith(('.mp3', '.wav', '.flac', '.ogg', '.m4a', '.aac', '.opus', '.wma', '.webm')):
audio_files.append(f)
return {
"required": {
"audio1": (audio_files,),
"fps": ("FLOAT", {"default": 25.0, "min": 1.0, "max": 120.0, "step": 1.0}),
"start1": ("FLOAT", {"default": 0.00, "min": 0.00, "max": 99999.00, "step": 0.01}),
"end1": ("FLOAT", {"default": 10.00, "min": 0.00, "max": 99999.00, "step": 0.01}),
"audio2": (audio_files,),
"start2": ("FLOAT", {"default": 0.00, "min": 0.00, "max": 99999.00, "step": 0.01}),
"end2": ("FLOAT", {"default": 10.00, "min": 0.00, "max": 99999.00, "step": 0.01}),
"gap_frames": ("INT", {"default": 25, "min": 0, "max": 99999, "step": 1}),
"total_display": ("STRING", {"default": "0 帧", "multiline": False}),
}
}
RETURN_TYPES = ("AUDIO", "INT", "FLOAT", "AUDIO", "INT", "AUDIO", "INT")
RETURN_NAMES = ("combined_audio", "total_frames", "fps", "audio1", "frames1", "audio2", "frames2")
FUNCTION = "slice_dual"
CATEGORY = "XB_ToolBox/Audio"
@classmethod
def VALIDATE_INPUTS(cls, audio1, audio2, **kwargs):
return True
def _load(self, name):
if name == "none": return None, 0
result = _load_audio_file(os.path.join(folder_paths.get_input_directory(), name))
if result is None: return None, 0
wf, sr = result
if wf.shape[0] > 1: wf = wf.mean(dim=0, keepdim=True)
return {"waveform": wf, "sample_rate": sr}, wf.shape[1] / sr
def _trim(self, audio, td, st, et, sr, fps):
fd = 1.0 / fps
st = max(0.0, min(st, td)); et = max(st + fd, min(et, td))
ss, es = int(st * sr), int(et * sr)
if es <= ss: es = min(ss + int(fd * 4 * sr), audio.shape[1])
dur = (es - ss) / sr
raw = int(round(dur * fps))
fc = _snap_4n1(raw)
# 🛡️ 音频物理长度对齐 4N+1 帧数
sliced = audio[:, ss:es]
target = int((fc / fps) * sr)
if sliced.shape[1] < target:
pad = torch.zeros((sliced.shape[0], target - sliced.shape[1]), dtype=sliced.dtype, device=sliced.device)
sliced = torch.cat([sliced, pad], dim=1)
elif sliced.shape[1] > target:
sliced = sliced[:, :target]
return sliced, fc
def slice_dual(self, audio1, fps, start1, end1, audio2, start2, end2, gap_frames, total_display):
a1, d1f = self._load(audio1); a2, d2f = self._load(audio2)
sr = 44100
if a1: sr = a1["sample_rate"]
elif a2: sr = a2["sample_rate"]
# 各自用原生采样率裁剪
t1 = torch.zeros((1, int(0.04 * sr)), dtype=torch.float32); f1 = 0
t2 = torch.zeros((1, int(0.04 * sr)), dtype=torch.float32); f2 = 0
if a1 and d1f > 0: t1, f1 = self._trim(a1["waveform"], d1f, start1, end1, a1["sample_rate"], fps)
if a2 and d2f > 0: t2, f2 = self._trim(a2["waveform"], d2f, start2, end2, a2["sample_rate"], fps)
# 统一采样率
import torchaudio
if a2 and a2["sample_rate"] != sr:
t2 = torchaudio.functional.resample(t2, a2["sample_rate"], sr)
if a1 and a1["sample_rate"] != sr:
t1 = torchaudio.functional.resample(t1, a1["sample_rate"], sr)
gap_s = int(max(0, gap_frames) / fps * sr)
gap_wf = torch.zeros((1, gap_s), dtype=torch.float32)
combined = torch.cat([t1, gap_wf, t2], dim=1)
total_f = _snap_4n1(f1 + max(0, gap_frames) + f2)
if total_f != f1 + max(0, gap_frames) + f2:
print(f"🔧 [音频切片V2] 合并帧数 {f1 + max(0, gap_frames) + f2} → {total_f} (对齐 4N+1)")
return (_make_audio(combined, sr), total_f, fps,
_make_audio(t1, sr), f1, _make_audio(t2, sr), f2)
# ============================================================
# XB_AudioSlicerV3 — 高级双轨混音与静音切片 (视窗缩放版)
# ============================================================
class XB_AudioSlicerV3:
@classmethod
def INPUT_TYPES(cls):
input_dir = folder_paths.get_input_directory()
audio_files = ["none"]
if os.path.exists(input_dir):
for f in sorted(os.listdir(input_dir)):
if f.lower().endswith(('.mp3', '.wav', '.flac', '.ogg', '.m4a', '.aac', '.opus', '.wma', '.webm')):
audio_files.append(f)
return {
"required": {
"audio1": (audio_files,),
"fps": ("FLOAT", {"default": 24.0, "min": 1.0, "max": 120.0, "step": 1.0}),
"start1": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 3600.0, "step": 0.01}),
"end1": ("FLOAT", {"default": 10.0, "min": 0.0, "max": 3600.0, "step": 0.01}),
"mute_count1": ("INT", {"default": 0, "min": 0, "max": 20, "step": 1}),
"mutes1_data": ("STRING", {"default": ""}),
"audio2": (audio_files,),
"start2": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 3600.0, "step": 0.01}),
"end2": ("FLOAT", {"default": 10.0, "min": 0.0, "max": 3600.0, "step": 0.01}),
"mute_count2": ("INT", {"default": 0, "min": 0, "max": 20, "step": 1}),
"mutes2_data": ("STRING", {"default": ""}),
"merge_mode": (["接力", "重叠"], {"default": "接力"}),
"total_display": ("STRING", {"default": "0 / 0帧", "multiline": False}),
}
}
RETURN_TYPES = ("AUDIO", "INT", "FLOAT", "AUDIO", "INT", "AUDIO", "INT")
RETURN_NAMES = ("combined_audio", "total_frames", "fps", "audio1", "frames1", "audio2", "frames2")
FUNCTION = "slice_v3"
CATEGORY = "XB_ToolBox/Audio"
@classmethod
def VALIDATE_INPUTS(cls, audio1, audio2, **kwargs):
return True
def _load(self, name):
if name == "none": return None, 0
result = _load_audio_file(os.path.join(folder_paths.get_input_directory(), name))
if result is None: return None, 0
wf, sr = result
if wf.shape[0] > 1: wf = wf.mean(dim=0, keepdim=True)
return {"waveform": wf, "sample_rate": sr}, wf.shape[1] / sr
def _trim(self, audio, td, st, et, sr, fps):
fd = 1.0 / fps
st = max(0.0, min(st, td)); et = max(st + fd, min(et, td))
ss, es = int(st * sr), int(et * sr)
if es <= ss: es = min(ss + int(fd * 4 * sr), audio.shape[1])
dur = (es - ss) / sr
raw = int(round(dur * fps))
fc = _snap_4n1(raw)
sliced = audio[:, ss:es]
target = int((fc / fps) * sr)
if sliced.shape[1] < target:
pad = torch.zeros((sliced.shape[0], target - sliced.shape[1]), dtype=sliced.dtype, device=sliced.device)
sliced = torch.cat([sliced, pad], dim=1)
elif sliced.shape[1] > target:
sliced = sliced[:, :target]
return sliced, fc
def _apply_mutes_v3(self, waveform, sample_rate, mute_data_str, start_offset):
"""静音消音:必须减去 start_offset,将绝对时间转换为切片内的相对时间!"""
if not mute_data_str: return waveform
wav = waveform.clone()
for r in mute_data_str.split(';'):
if not r.strip(): continue
try:
s_str, e_str = r.split(',')
# 减去截取的起始时间,定位到当前波形张量中的真实物理位置
rel_s = float(s_str) - start_offset
rel_e = float(e_str) - start_offset
s_idx = max(0, int(rel_s * sample_rate))
e_idx = min(wav.shape[1], int(rel_e * sample_rate))
if e_idx > s_idx:
wav[:, s_idx:e_idx] = 0.0
except: pass
return wav
def slice_v3(self, audio1, fps, start1, end1, mute_count1, mutes1_data,
audio2, start2, end2, mute_count2, mutes2_data,
merge_mode, total_display):
a1, d1f = self._load(audio1)
a2, d2f = self._load(audio2)
sr = 44100
if a1: sr = a1["sample_rate"]
elif a2: sr = a2["sample_rate"]
t1 = torch.zeros((1, int(0.04 * sr)), dtype=torch.float32); f1 = 0
t2 = torch.zeros((1, int(0.04 * sr)), dtype=torch.float32); f2 = 0
if a1 and d1f > 0:
t1, f1 = self._trim(a1["waveform"], d1f, start1, end1, a1["sample_rate"], fps)
t1 = self._apply_mutes_v3(t1, a1["sample_rate"], mutes1_data, start1)
if a2 and d2f > 0:
t2, f2 = self._trim(a2["waveform"], d2f, start2, end2, a2["sample_rate"], fps)
t2 = self._apply_mutes_v3(t2, a2["sample_rate"], mutes2_data, start2)
# 采样率对齐
import torchaudio
if a2 and a2["sample_rate"] != sr:
t2 = torchaudio.functional.resample(t2, a2["sample_rate"], sr)
if a1 and a1["sample_rate"] != sr:
t1 = torchaudio.functional.resample(t1, a1["sample_rate"], sr)
# 核心混音引擎
if merge_mode == "接力":
combined = torch.cat([t1, t2], dim=1)
total_f = _snap_4n1(f1 + f2)
else: # 重叠混音模式
len1, len2 = t1.shape[1], t2.shape[1]
max_len = max(len1, len2)
if len1 < max_len:
t1 = torch.cat([t1, torch.zeros((t1.shape[0], max_len - len1), dtype=t1.dtype, device=t1.device)], dim=1)
if len2 < max_len:
t2 = torch.cat([t2, torch.zeros((t2.shape[0], max_len - len2), dtype=t2.dtype, device=t2.device)], dim=1)
# 硬件级压限(Hard Limiter),彻底斩断由于叠加造成的爆音
combined = torch.clamp(t1 + t2, min=-1.0, max=1.0)
total_f = _snap_4n1(max(f1, f2))
# 最终长度保护
target_samples = int((total_f / fps) * sr)
if combined.shape[1] < target_samples:
combined = torch.cat([combined, torch.zeros((combined.shape[0], target_samples - combined.shape[1]), dtype=combined.dtype, device=combined.device)], dim=1)
elif combined.shape[1] > target_samples:
combined = combined[:, :target_samples]
return (_make_audio(combined, sr), total_f, fps,
_make_audio(t1, sr), f1, _make_audio(t2, sr), f2)
"""
XB-ToolBox 音频切片节点
======================
完全独立的音频加载/切片节点,支持可视化波形截取。
使用 av (PyAV) 解码,兼容所有格式(同 VHS 用 ffmpeg 的效果)
"""
import os, asyncio
import folder_paths
import torch
import av
from aiohttp import web
import server
from ...libs.xb_av import open_av
def _load_audio_file(filepath: str):
"""用 av 解码音频,返回 (waveform_2d, sample_rate)。加载失败返回 None"""
try:
with open_av(filepath) as af:
if not af.streams.audio:
return None
stream = af.streams.audio[0]
sr = stream.codec_context.sample_rate
frames = []
for frame in af.decode(streams=stream.index):
buf = torch.from_numpy(frame.to_ndarray())
if buf.dim() == 1:
buf = buf.unsqueeze(0)
elif buf.shape[0] != stream.channels:
buf = buf.view(-1, stream.channels).t()
frames.append(buf)
if not frames:
return None
wav = torch.cat(frames, dim=1)
# 统一转为 float32
if not wav.dtype.is_floating_point:
if wav.dtype == torch.int16:
wav = wav.float() / (2 ** 15)
elif wav.dtype == torch.int32:
wav = wav.float() / (2 ** 31)
return wav, sr
except Exception:
return None
def _make_audio(waveform, sample_rate):
if waveform.dim() == 1:
waveform = waveform.unsqueeze(0).unsqueeze(0)
elif waveform.dim() == 2:
waveform = waveform.unsqueeze(0)
return {"waveform": waveform, "sample_rate": sample_rate}
@server.PromptServer.instance.routes.post("/xb_audio_waveform")
@server.PromptServer.instance.routes.post("/xb_toolbox/audio_waveform")
async def handle_audio_waveform(request):
"""返回音频文件的波形峰值数据,支持按时间范围切片重算"""
try:
data = await request.json()
except Exception:
return web.json_response({"error": "Invalid JSON body"})
filename = data.get("filename", "")
num_peaks = data.get("num_peaks", 400)
start_time = data.get("start_time", None)
end_time = data.get("end_time", None)
if not filename:
return web.json_response({"peaks": [], "duration": 0})
input_dir = folder_paths.get_input_directory()
audio_path = os.path.join(input_dir, filename)
if not os.path.exists(audio_path):
return web.json_response({"peaks": [], "duration": 0})
try:
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(None, _load_audio_file, audio_path)
if result is None:
return web.json_response({"peaks": [], "duration": 0})
waveform, sample_rate = result
if waveform.shape[0] > 1:
waveform = waveform.mean(dim=0, keepdim=True)
duration = waveform.shape[1] / sample_rate
if start_time is not None and end_time is not None and end_time > start_time:
s_idx = max(0, int(start_time * sample_rate))
e_idx = min(waveform.shape[1], int(end_time * sample_rate))
waveform = waveform[:, s_idx:e_idx]
total_samples = waveform.shape[1]
if total_samples == 0:
return web.json_response({"peaks": [], "duration": duration})
chunk_size = max(1, total_samples // num_peaks)
peaks = []
for i in range(num_peaks):
chunk = waveform[0, i * chunk_size : (i + 1) * chunk_size]
if chunk.numel() == 0:
peaks.append([0.0, 0.0])
continue
peaks.append([float(chunk.max().item()), float(chunk.min().item())])
return web.json_response({"peaks": peaks, "duration": duration})
except Exception as e:
return web.json_response({"error": str(e)})
def _snap_4n1(frames):
"""将帧数对齐到 4N+1(满足 InfiniteTalk 的硬性要求)"""
return max(1, ((frames + 2) // 4) * 4 + 1)
# ============================================================
# XB_AudioSlicer — 音频切片(基础版)
# ============================================================
class XB_AudioSlicer:
"""音频加载与切片节点 — 支持频谱可视化 + 拖拽分割线截取
输出帧数(非秒数),可直接对接 InfiniteTalk 接力点的 segment_length。
额外输出 fps,方便连接到总线统一帧率。"""
@classmethod
def INPUT_TYPES(cls):
input_dir = folder_paths.get_input_directory()
audio_files = ["none"]
if os.path.exists(input_dir):
for f in sorted(os.listdir(input_dir)):
if f.lower().endswith(('.mp3', '.wav', '.flac', '.ogg', '.m4a', '.aac', '.opus', '.wma', '.webm')):
audio_files.append(f)
return {
"required": {
"audio": (audio_files,),
"fps": ("FLOAT", {"default": 25.0, "min": 1.0, "max": 120.0, "step": 1.0}),
"start_time": ("FLOAT", {"default": 0.00, "min": 0.00, "max": 99999.00, "step": 0.01, "precision": 2, "round": 0.01}),
"end_time": ("FLOAT", {"default": 10.00, "min": 0.00, "max": 99999.00, "step": 0.01, "precision": 2, "round": 0.01}),
"duration_display": ("STRING", {"default": "0 帧", "multiline": False}),
}
}
RETURN_TYPES = ("AUDIO", "INT", "FLOAT")
RETURN_NAMES = ("audio", "frame_count", "fps")
FUNCTION = "slice_audio"
CATEGORY = "XB_ToolBox/Audio"
@classmethod
def VALIDATE_INPUTS(cls, audio, **kwargs):
return True
def slice_audio(self, audio, fps, start_time, end_time, duration_display):
if audio == "none":
return (_make_audio(torch.zeros((1, 1), dtype=torch.float32), 44100), 0, fps)
result = _load_audio_file(os.path.join(folder_paths.get_input_directory(), audio))
if result is None:
return (_make_audio(torch.zeros((1, 1), dtype=torch.float32), 44100), 0, fps)
waveform, sample_rate = result
if waveform.shape[0] > 1:
waveform = waveform.mean(dim=0, keepdim=True)
total_duration = waveform.shape[1] / sample_rate
frame_duration = 1.0 / fps
start_time = max(0.0, min(start_time, total_duration))
end_time = max(start_time + frame_duration, min(end_time, total_duration))
start_sample = int(start_time * sample_rate)
end_sample = int(end_time * sample_rate)
if end_sample <= start_sample:
end_sample = min(start_sample + int(frame_duration * 4 * sample_rate), waveform.shape[1])
sliced = waveform[:, start_sample:end_sample]
duration_sec = (end_sample - start_sample) / sample_rate
raw_frames = int(round(duration_sec * fps))
frame_count = _snap_4n1(raw_frames)
# 🛡️ 音频物理长度对齐 4N+1 帧数,防止音视频时间轴错位
target_samples = int((frame_count / fps) * sample_rate)
if sliced.shape[1] < target_samples:
pad = torch.zeros((sliced.shape[0], target_samples - sliced.shape[1]), dtype=sliced.dtype, device=sliced.device)
sliced = torch.cat([sliced, pad], dim=1)
elif sliced.shape[1] > target_samples:
sliced = sliced[:, :target_samples]
if frame_count != raw_frames:
print(f"🔧 [音频切片] 帧数 {raw_frames} → {frame_count} (对齐 4N+1)")
return (_make_audio(sliced, sample_rate), frame_count, fps)
# ============================================================
# XB_AudioSlicerV1 — 可视化音频切片(波形+拖拽分割线)
# ============================================================
class XB_AudioSlicerV1:
"""音频切片节点 V1 — 可视化波形窗口 + 拖拽分割线截取
输出帧数(非秒数),可直接对接 InfiniteTalk 接力点的 segment_length。
额外输出 fps,方便连接到总线统一帧率。"""
@classmethod
def INPUT_TYPES(cls):
input_dir = folder_paths.get_input_directory()
audio_files = ["none"]
if os.path.exists(input_dir):
for f in sorted(os.listdir(input_dir)):
if f.lower().endswith(('.mp3', '.wav', '.flac', '.ogg', '.m4a', '.aac', '.opus', '.wma', '.webm')):
audio_files.append(f)
return {
"required": {
"audio": (audio_files,),
"fps": ("FLOAT", {"default": 25.0, "min": 1.0, "max": 120.0, "step": 1.0}),
"start_time": ("FLOAT", {"default": 0.00, "min": 0.00, "max": 99999.00, "step": 0.01}),
"end_time": ("FLOAT", {"default": 10.00, "min": 0.00, "max": 99999.00, "step": 0.01}),
"duration_display": ("STRING", {"default": "0 帧", "multiline": False}),
}
}
RETURN_TYPES = ("AUDIO", "INT", "FLOAT")
RETURN_NAMES = ("audio", "frame_count", "fps")
FUNCTION = "slice_audio"
CATEGORY = "XB_ToolBox/Audio"
@classmethod
def VALIDATE_INPUTS(cls, audio, **kwargs):
return True
def slice_audio(self, audio, fps, start_time, end_time, duration_display):
if audio == "none":
return (_make_audio(torch.zeros((1, 1), dtype=torch.float32), 44100), 0, fps)
result = _load_audio_file(os.path.join(folder_paths.get_input_directory(), audio))
if result is None:
return (_make_audio(torch.zeros((1, 1), dtype=torch.float32), 44100), 0, fps)
waveform, sample_rate = result
if waveform.shape[0] > 1:
waveform = waveform.mean(dim=0, keepdim=True)
total_duration = waveform.shape[1] / sample_rate
frame_duration = 1.0 / fps
start_time = max(0.0, min(start_time, total_duration))
end_time = max(start_time + frame_duration, min(end_time, total_duration))
start_sample = int(start_time * sample_rate)
end_sample = int(end_time * sample_rate)
if end_sample <= start_sample:
end_sample = min(start_sample + int(frame_duration * 4 * sample_rate), waveform.shape[1])
sliced = waveform[:, start_sample:end_sample]
duration_sec = (end_sample - start_sample) / sample_rate
raw_frames = int(round(duration_sec * fps))
frame_count = _snap_4n1(raw_frames)
if frame_count != raw_frames:
print(f"🔧 [音频切片V1] 帧数 {raw_frames} → {frame_count} (对齐 4N+1)")
return (_make_audio(sliced, sample_rate), frame_count, fps)
# ============================================================
# XB_AudioSlicerV2 — 双人音频切片
# ============================================================
class XB_AudioSlicerV2:
"""双人音频切片 — 音频1 + 间隔(帧) + 音频2,独立可视化
输出帧数(非秒数),可直接对接 InfiniteTalk 接力点的 segment_length。"""
@classmethod
def INPUT_TYPES(cls):
input_dir = folder_paths.get_input_directory()
audio_files = ["none"]
if os.path.exists(input_dir):
for f in sorted(os.listdir(input_dir)):
if f.lower().endswith(('.mp3', '.wav', '.flac', '.ogg', '.m4a', '.aac', '.opus', '.wma', '.webm')):
audio_files.append(f)
return {
"required": {
"audio1": (audio_files,),
"fps": ("FLOAT", {"default": 25.0, "min": 1.0, "max": 120.0, "step": 1.0}),
"start1": ("FLOAT", {"default": 0.00, "min": 0.00, "max": 99999.00, "step": 0.01}),
"end1": ("FLOAT", {"default": 10.00, "min": 0.00, "max": 99999.00, "step": 0.01}),
"audio2": (audio_files,),
"start2": ("FLOAT", {"default": 0.00, "min": 0.00, "max": 99999.00, "step": 0.01}),
"end2": ("FLOAT", {"default": 10.00, "min": 0.00, "max": 99999.00, "step": 0.01}),
"gap_frames": ("INT", {"default": 25, "min": 0, "max": 99999, "step": 1}),
"total_display": ("STRING", {"default": "0 帧", "multiline": False}),
}
}
RETURN_TYPES = ("AUDIO", "INT", "FLOAT", "AUDIO", "INT", "AUDIO", "INT")
RETURN_NAMES = ("combined_audio", "total_frames", "fps", "audio1", "frames1", "audio2", "frames2")
FUNCTION = "slice_dual"
CATEGORY = "XB_ToolBox/Audio"
@classmethod
def VALIDATE_INPUTS(cls, audio1, audio2, **kwargs):
return True
def _load(self, name):
if name == "none": return None, 0
result = _load_audio_file(os.path.join(folder_paths.get_input_directory(), name))
if result is None: return None, 0
wf, sr = result
if wf.shape[0] > 1: wf = wf.mean(dim=0, keepdim=True)
return {"waveform": wf, "sample_rate": sr}, wf.shape[1] / sr
def _trim(self, audio, td, st, et, sr, fps):
fd = 1.0 / fps
st = max(0.0, min(st, td)); et = max(st + fd, min(et, td))
ss, es = int(st * sr), int(et * sr)
if es <= ss: es = min(ss + int(fd * 4 * sr), audio.shape[1])
dur = (es - ss) / sr
raw = int(round(dur * fps))
fc = _snap_4n1(raw)
# 🛡️ 音频物理长度对齐 4N+1 帧数
sliced = audio[:, ss:es]
target = int((fc / fps) * sr)
if sliced.shape[1] < target:
pad = torch.zeros((sliced.shape[0], target - sliced.shape[1]), dtype=sliced.dtype, device=sliced.device)
sliced = torch.cat([sliced, pad], dim=1)
elif sliced.shape[1] > target:
sliced = sliced[:, :target]
return sliced, fc
def slice_dual(self, audio1, fps, start1, end1, audio2, start2, end2, gap_frames, total_display):
a1, d1f = self._load(audio1); a2, d2f = self._load(audio2)
sr = 44100
if a1: sr = a1["sample_rate"]
elif a2: sr = a2["sample_rate"]
# 各自用原生采样率裁剪
t1 = torch.zeros((1, int(0.04 * sr)), dtype=torch.float32); f1 = 0
t2 = torch.zeros((1, int(0.04 * sr)), dtype=torch.float32); f2 = 0
if a1 and d1f > 0: t1, f1 = self._trim(a1["waveform"], d1f, start1, end1, a1["sample_rate"], fps)
if a2 and d2f > 0: t2, f2 = self._trim(a2["waveform"], d2f, start2, end2, a2["sample_rate"], fps)
# 统一采样率
import torchaudio
if a2 and a2["sample_rate"] != sr:
t2 = torchaudio.functional.resample(t2, a2["sample_rate"], sr)
if a1 and a1["sample_rate"] != sr:
t1 = torchaudio.functional.resample(t1, a1["sample_rate"], sr)
gap_s = int(max(0, gap_frames) / fps * sr)
gap_wf = torch.zeros((1, gap_s), dtype=torch.float32)
combined = torch.cat([t1, gap_wf, t2], dim=1)
total_f = _snap_4n1(f1 + max(0, gap_frames) + f2)
if total_f != f1 + max(0, gap_frames) + f2:
print(f"🔧 [音频切片V2] 合并帧数 {f1 + max(0, gap_frames) + f2} → {total_f} (对齐 4N+1)")
return (_make_audio(combined, sr), total_f, fps,
_make_audio(t1, sr), f1, _make_audio(t2, sr), f2)
# ============================================================
# XB_AudioSlicerV3 — 高级双轨混音与静音切片 (视窗缩放版)
# ============================================================
class XB_AudioSlicerV3:
@classmethod
def INPUT_TYPES(cls):
input_dir = folder_paths.get_input_directory()
audio_files = ["none"]
if os.path.exists(input_dir):
for f in sorted(os.listdir(input_dir)):
if f.lower().endswith(('.mp3', '.wav', '.flac', '.ogg', '.m4a', '.aac', '.opus', '.wma', '.webm')):
audio_files.append(f)
return {
"required": {
"audio1": (audio_files,),
"fps": ("FLOAT", {"default": 24.0, "min": 1.0, "max": 120.0, "step": 1.0}),
"start1": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 3600.0, "step": 0.01}),
"end1": ("FLOAT", {"default": 10.0, "min": 0.0, "max": 3600.0, "step": 0.01}),
"mute_count1": ("INT", {"default": 0, "min": 0, "max": 20, "step": 1}),
"mutes1_data": ("STRING", {"default": ""}),
"audio2": (audio_files,),
"start2": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 3600.0, "step": 0.01}),
"end2": ("FLOAT", {"default": 10.0, "min": 0.0, "max": 3600.0, "step": 0.01}),
"mute_count2": ("INT", {"default": 0, "min": 0, "max": 20, "step": 1}),
"mutes2_data": ("STRING", {"default": ""}),
"merge_mode": (["接力", "重叠"], {"default": "接力"}),
"total_display": ("STRING", {"default": "0 / 0帧", "multiline": False}),
}
}
RETURN_TYPES = ("AUDIO", "INT", "FLOAT", "AUDIO", "INT", "AUDIO", "INT")
RETURN_NAMES = ("combined_audio", "total_frames", "fps", "audio1", "frames1", "audio2", "frames2")
FUNCTION = "slice_v3"
CATEGORY = "XB_ToolBox/Audio"
@classmethod
def VALIDATE_INPUTS(cls, audio1, audio2, **kwargs):
return True
def _load(self, name):
if name == "none": return None, 0
result = _load_audio_file(os.path.join(folder_paths.get_input_directory(), name))
if result is None: return None, 0
wf, sr = result
if wf.shape[0] > 1: wf = wf.mean(dim=0, keepdim=True)
return {"waveform": wf, "sample_rate": sr}, wf.shape[1] / sr
def _trim(self, audio, td, st, et, sr, fps):
fd = 1.0 / fps
st = max(0.0, min(st, td)); et = max(st + fd, min(et, td))
ss, es = int(st * sr), int(et * sr)
if es <= ss: es = min(ss + int(fd * 4 * sr), audio.shape[1])
dur = (es - ss) / sr
raw = int(round(dur * fps))
fc = _snap_4n1(raw)
sliced = audio[:, ss:es]
target = int((fc / fps) * sr)
if sliced.shape[1] < target:
pad = torch.zeros((sliced.shape[0], target - sliced.shape[1]), dtype=sliced.dtype, device=sliced.device)
sliced = torch.cat([sliced, pad], dim=1)
elif sliced.shape[1] > target:
sliced = sliced[:, :target]
return sliced, fc
def _apply_mutes_v3(self, waveform, sample_rate, mute_data_str, start_offset):
"""静音消音:必须减去 start_offset,将绝对时间转换为切片内的相对时间!"""
if not mute_data_str: return waveform
wav = waveform.clone()
for r in mute_data_str.split(';'):
if not r.strip(): continue
try:
s_str, e_str = r.split(',')
# 减去截取的起始时间,定位到当前波形张量中的真实物理位置
rel_s = float(s_str) - start_offset
rel_e = float(e_str) - start_offset
s_idx = max(0, int(rel_s * sample_rate))
e_idx = min(wav.shape[1], int(rel_e * sample_rate))
if e_idx > s_idx:
wav[:, s_idx:e_idx] = 0.0
except: pass
return wav
def slice_v3(self, audio1, fps, start1, end1, mute_count1, mutes1_data,
audio2, start2, end2, mute_count2, mutes2_data,
merge_mode, total_display):
a1, d1f = self._load(audio1)
a2, d2f = self._load(audio2)
sr = 44100
if a1: sr = a1["sample_rate"]
elif a2: sr = a2["sample_rate"]
t1 = torch.zeros((1, int(0.04 * sr)), dtype=torch.float32); f1 = 0
t2 = torch.zeros((1, int(0.04 * sr)), dtype=torch.float32); f2 = 0
if a1 and d1f > 0:
t1, f1 = self._trim(a1["waveform"], d1f, start1, end1, a1["sample_rate"], fps)
t1 = self._apply_mutes_v3(t1, a1["sample_rate"], mutes1_data, start1)
if a2 and d2f > 0:
t2, f2 = self._trim(a2["waveform"], d2f, start2, end2, a2["sample_rate"], fps)
t2 = self._apply_mutes_v3(t2, a2["sample_rate"], mutes2_data, start2)
# 采样率对齐
import torchaudio
if a2 and a2["sample_rate"] != sr:
t2 = torchaudio.functional.resample(t2, a2["sample_rate"], sr)
if a1 and a1["sample_rate"] != sr:
t1 = torchaudio.functional.resample(t1, a1["sample_rate"], sr)
# 核心混音引擎
if merge_mode == "接力":
combined = torch.cat([t1, t2], dim=1)
total_f = _snap_4n1(f1 + f2)
else: # 重叠混音模式
len1, len2 = t1.shape[1], t2.shape[1]
max_len = max(len1, len2)
if len1 < max_len:
t1 = torch.cat([t1, torch.zeros((t1.shape[0], max_len - len1), dtype=t1.dtype, device=t1.device)], dim=1)
if len2 < max_len:
t2 = torch.cat([t2, torch.zeros((t2.shape[0], max_len - len2), dtype=t2.dtype, device=t2.device)], dim=1)
# 硬件级压限(Hard Limiter),彻底斩断由于叠加造成的爆音
combined = torch.clamp(t1 + t2, min=-1.0, max=1.0)
total_f = _snap_4n1(max(f1, f2))
# 最终长度保护
target_samples = int((total_f / fps) * sr)
if combined.shape[1] < target_samples:
combined = torch.cat([combined, torch.zeros((combined.shape[0], target_samples - combined.shape[1]), dtype=combined.dtype, device=combined.device)], dim=1)
elif combined.shape[1] > target_samples:
combined = combined[:, :target_samples]
return (_make_audio(combined, sr), total_f, fps,
_make_audio(t1, sr), f1, _make_audio(t2, sr), f2)
@@ -22,7 +22,7 @@ CATEGORY: JZL/YuE2
import re
try:
from .yue2_abc import OFFICIAL_SECTION_TAGS
from ...libs.yue2_abc import OFFICIAL_SECTION_TAGS
except ImportError: # 以文件方式直接加载时的兜底
from yue2_abc import OFFICIAL_SECTION_TAGS
@@ -26,7 +26,7 @@ CATEGORY: JZL/YuE2
import json
try:
from .yue2_abc import (
from ...libs.yue2_abc import (
AbcError, TOKEN, VOICES, compare_text, compile_events, convert_score,
diff_lines, mode_suggestion, parse_abc, sha_line, strip_chords,
summary_text, validate_instrumental,
@@ -20,7 +20,7 @@ CATEGORY: JZL/YuE2
import re
try:
from .yue2_styles import (
from ...libs.yue2_styles import (
BPM_HINTS, DENSITY_HINTS, GENRES, INSTRUMENT_HINTS, LANGUAGE_HINTS,
MOOD_HINTS, RECIPES, TECHNIQUE_HINTS, TEXTURE_HINTS, TIMBRE_HINTS,
VOCAL_FX_HINTS, VOCAL_HINTS, FOLLOW, CUSTOM, genre_options,
@@ -22,7 +22,7 @@ import os
import pathlib
try:
from .yue2_abc import (
from ...libs.yue2_abc import (
AbcError, compile_events, diff_lines, mode_suggestion, normalize_abc_text,
parse_abc, sha_line, summary_text,
)
+127 -127
View File
@@ -1,128 +1,128 @@
import os
import torch
import numpy as np
import re
from PIL import Image, ImageOps
import folder_paths
import datetime
from .nodes_batch_images import batch_images
VALID_EXTENSIONS = {'.png', '.jpg', '.jpeg', '.webp', '.bmp', '.jfif'}
def _natural_sort_key(s):
return [int(text) if text.isdigit() else text.lower() for text in re.split('([0-9]+)', s)]
def _list_image_files(directory):
if not os.path.isdir(directory):
raise FileNotFoundError(f"Directory not found: {directory}")
files = [f for f in os.listdir(directory) if os.path.splitext(f)[1].lower() in VALID_EXTENSIONS]
files.sort(key=_natural_sort_key)
return files
def _load_image_tensor(file_path):
"""读取单张图片 → (1, H, W, C) float32 张量。"""
img = Image.open(file_path)
img = ImageOps.exif_transpose(img)
image = img.convert("RGB")
image = np.array(image).astype(np.float32) / 255.0
return torch.from_numpy(image)[None,]
# ============================================================
# XB_BatchFolderLoader — 图片序列加载(单张步进)
# ============================================================
class XB_BatchFolderLoader:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"directory": ("STRING", {"default": ""}),
"seed": ("INT", {"default": 0, "min": 0, "max": 9999999, "step": 1, "tooltip": "Built-in auto-stepper using the seed mechanism"}),
}
}
RETURN_TYPES = ("IMAGE", "MASK", "STRING", "STRING")
RETURN_NAMES = ("IMAGE", "MASK", "FILE_NAME", "TOTAL_COUNT")
FUNCTION = "load_images"
CATEGORY = "XB_ToolBox/Batch_Process"
def load_images(self, directory, seed):
if not os.path.isdir(directory):
raise FileNotFoundError(f"Directory not found: {directory}")
valid_extensions = {'.png', '.jpg', '.jpeg', '.webp', '.bmp', '.jfif'}
files = [f for f in os.listdir(directory) if os.path.splitext(f)[1].lower() in valid_extensions]
def natural_sort_key(s):
return [int(text) if text.isdigit() else text.lower() for text in re.split('([0-9]+)', s)]
files.sort(key=natural_sort_key)
if not files:
raise Exception(f"No valid images found in directory: {directory}")
total_count = len(files)
current_file_idx = seed % total_count
file_path = os.path.join(directory, files[current_file_idx])
raw_name = os.path.splitext(files[current_file_idx])[0]
today_str = datetime.datetime.now().strftime("%Y-%m-%d")
formatted_file_name = f"Images/Batch/{today_str}/{raw_name}"
total_count_display = f"Total images detected: {total_count}"
img = Image.open(file_path)
img = ImageOps.exif_transpose(img)
image = img.convert("RGB")
image = np.array(image).astype(np.float32) / 255.0
image = torch.from_numpy(image)[None,]
if 'A' in img.getbands():
mask = np.array(img.getchannel('A')).astype(np.float32) / 255.0
mask = 1. - torch.from_numpy(mask)
mask = mask.unsqueeze(0) # (H, W) → (1, H, W)
else:
mask = torch.zeros((1, image.shape[1], image.shape[2]), dtype=torch.float32)
return (image, mask, formatted_file_name, total_count_display)
# ============================================================
# XB_BatchFolderBatchLoader — 图片批量加载(一次读取多张)
# ============================================================
class XB_BatchFolderBatchLoader:
"""选择文件夹后一次性批量加载多张图片,合并成一个图像批次传给下游(等价于批量图像节点接多张图)。"""
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"directory": ("STRING", {"default": ""}),
"count": ("INT", {"default": 4, "min": 1, "max": 500, "step": 1,
"tooltip": "本次批量加载的图片数量"}),
"skip": ("INT", {"default": 0, "min": 0, "max": 999999, "step": 1,
"tooltip": "跳过前 N 张,便于多次分批处理"}),
}
}
RETURN_TYPES = ("IMAGE",)
RETURN_NAMES = ("IMAGES",)
FUNCTION = "load_batch"
CATEGORY = "XB_ToolBox/Batch_Process"
def load_batch(self, directory, count, skip):
files = _list_image_files(directory)
if not files:
raise Exception(f"No valid images found in directory: {directory}")
selected = files[skip:skip + count]
if not selected:
raise Exception(
f"No images left: skipped {skip} of {len(files)} images in {directory}"
)
images = [_load_image_tensor(os.path.join(directory, f)) for f in selected]
import os
import torch
import numpy as np
import re
from PIL import Image, ImageOps
import folder_paths
import datetime
from .nodes_batch_images import batch_images
VALID_EXTENSIONS = {'.png', '.jpg', '.jpeg', '.webp', '.bmp', '.jfif'}
def _natural_sort_key(s):
return [int(text) if text.isdigit() else text.lower() for text in re.split('([0-9]+)', s)]
def _list_image_files(directory):
if not os.path.isdir(directory):
raise FileNotFoundError(f"Directory not found: {directory}")
files = [f for f in os.listdir(directory) if os.path.splitext(f)[1].lower() in VALID_EXTENSIONS]
files.sort(key=_natural_sort_key)
return files
def _load_image_tensor(file_path):
"""读取单张图片 → (1, H, W, C) float32 张量。"""
img = Image.open(file_path)
img = ImageOps.exif_transpose(img)
image = img.convert("RGB")
image = np.array(image).astype(np.float32) / 255.0
return torch.from_numpy(image)[None,]
# ============================================================
# XB_BatchFolderLoader — 图片序列加载(单张步进)
# ============================================================
class XB_BatchFolderLoader:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"directory": ("STRING", {"default": ""}),
"seed": ("INT", {"default": 0, "min": 0, "max": 9999999, "step": 1, "tooltip": "Built-in auto-stepper using the seed mechanism"}),
}
}
RETURN_TYPES = ("IMAGE", "MASK", "STRING", "STRING")
RETURN_NAMES = ("IMAGE", "MASK", "FILE_NAME", "TOTAL_COUNT")
FUNCTION = "load_images"
CATEGORY = "XB_ToolBox/Batch_Process"
def load_images(self, directory, seed):
if not os.path.isdir(directory):
raise FileNotFoundError(f"Directory not found: {directory}")
valid_extensions = {'.png', '.jpg', '.jpeg', '.webp', '.bmp', '.jfif'}
files = [f for f in os.listdir(directory) if os.path.splitext(f)[1].lower() in valid_extensions]
def natural_sort_key(s):
return [int(text) if text.isdigit() else text.lower() for text in re.split('([0-9]+)', s)]
files.sort(key=natural_sort_key)
if not files:
raise Exception(f"No valid images found in directory: {directory}")
total_count = len(files)
current_file_idx = seed % total_count
file_path = os.path.join(directory, files[current_file_idx])
raw_name = os.path.splitext(files[current_file_idx])[0]
today_str = datetime.datetime.now().strftime("%Y-%m-%d")
formatted_file_name = f"Images/Batch/{today_str}/{raw_name}"
total_count_display = f"Total images detected: {total_count}"
img = Image.open(file_path)
img = ImageOps.exif_transpose(img)
image = img.convert("RGB")
image = np.array(image).astype(np.float32) / 255.0
image = torch.from_numpy(image)[None,]
if 'A' in img.getbands():
mask = np.array(img.getchannel('A')).astype(np.float32) / 255.0
mask = 1. - torch.from_numpy(mask)
mask = mask.unsqueeze(0) # (H, W) → (1, H, W)
else:
mask = torch.zeros((1, image.shape[1], image.shape[2]), dtype=torch.float32)
return (image, mask, formatted_file_name, total_count_display)
# ============================================================
# XB_BatchFolderBatchLoader — 图片批量加载(一次读取多张)
# ============================================================
class XB_BatchFolderBatchLoader:
"""选择文件夹后一次性批量加载多张图片,合并成一个图像批次传给下游(等价于批量图像节点接多张图)。"""
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"directory": ("STRING", {"default": ""}),
"count": ("INT", {"default": 4, "min": 1, "max": 500, "step": 1,
"tooltip": "本次批量加载的图片数量"}),
"skip": ("INT", {"default": 0, "min": 0, "max": 999999, "step": 1,
"tooltip": "跳过前 N 张,便于多次分批处理"}),
}
}
RETURN_TYPES = ("IMAGE",)
RETURN_NAMES = ("IMAGES",)
FUNCTION = "load_batch"
CATEGORY = "XB_ToolBox/Batch_Process"
def load_batch(self, directory, count, skip):
files = _list_image_files(directory)
if not files:
raise Exception(f"No valid images found in directory: {directory}")
selected = files[skip:skip + count]
if not selected:
raise Exception(
f"No images left: skipped {skip} of {len(files)} images in {directory}"
)
images = [_load_image_tensor(os.path.join(directory, f)) for f in selected]
return (batch_images(images),)
@@ -1,232 +1,232 @@
import comfy.model_management as mm
import gc
import torch
import sys
from comfy.patcher_extension import CallbacksMP
from comfy.model_patcher import ModelPatcher
try:
from tqdm import tqdm
except ImportError:
def tqdm(iterable, **kwargs):
return iterable
class AnyType(str):
def __ne__(self, __value: object) -> bool:
return False
anyType = AnyType("*")
def is_dynamic_vram_active():
"""仅当用户显式传参 --enable-dynamic-vram 时视为动态显存激活,分块节点休眠。"""
if "--enable-dynamic-vram" in sys.argv:
print("\n\033[93m[XB Block Swap]\033[0m: ⚠️ 检测到 --enable-dynamic-vram,动态显存已启用。分块节点自动休眠。")
return True
return False
def is_unsupported_model(diffusion_model):
"""检测已知不支持物理分块的模型架构(仅黑名单,不检测量化格式)。
Qwen / FLUX 等模型的 GGUF 版本虽然参数类型特殊,但 module.to() 仍可正常工作,
不应被拦截。只有当模型的 C++ 量化引擎(如 comfy_kitchen)确实会因 module.to()
而崩溃时才需要拦截——但这类崩溃无法通过参数类型静态判断,由用户自行取舍。
"""
model_type = type(diffusion_model).__name__
blacklist = ["Lumina", "Lumina2", "ZImage", "HunyuanDiT", "ErnieImageModel"]
for b in blacklist:
if b in model_type:
return model_type
return None
# ============================================================
# XB_UNetBlockSwap — UNet 分块交换 (显存优化)
# ============================================================
class XB_UNetBlockSwap:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"unet_model": (anyType, {"tooltip": "Accepts pure UNet/DiT model input only"}),
"blocks_to_swap": ("INT", {
"default": 10,
"min": 0,
"max": 200,
"step": 1,
"tooltip": "Number of core blocks to offload to system RAM. Higher values save VRAM but slow down generation."
}),
},
}
RETURN_TYPES = (anyType,)
RETURN_NAMES = ("unet_model",)
CATEGORY = "XB_ToolBox/VRAM_Hacks"
FUNCTION = "set_callback"
def set_callback(self, unet_model: ModelPatcher, blocks_to_swap):
if not isinstance(unet_model, ModelPatcher):
return (unet_model,)
# ── 0 = 关闭分块功能:直接穿透,不做任何检测、不输出动态显存休眠警告 ──
if blocks_to_swap == 0:
print("\033[96m[XB UNet Block Swap]\033[0m: 分块数量 = 0,分块功能已关闭(节点穿透)")
return (unet_model,)
if is_dynamic_vram_active():
return (unet_model,)
def swap_blocks(model_patcher: ModelPatcher, device_to, lowvram_model_memory, force_patch_weights, full_load):
base_model = model_patcher.model
# 🛡️ 动态获取 ComfyUI 分配的 GPU 设备,绝不写死 'cuda:0'
main_device = model_patcher.load_device
diffusion_model = getattr(base_model, 'diffusion_model', None)
if not diffusion_model:
return
unsupported_name = is_unsupported_model(diffusion_model)
if unsupported_name:
print(f"\033[93m[XB-BOX 拦截盾]\033[0m: 模型 ({unsupported_name}) 属于高耦合架构,不支持物理分块。已自动跳过!")
return
all_blocks = []
block_paths = [
'transformer_blocks', 'blocks', 'down_blocks', 'up_blocks', 'mid_block',
'layers', 'attention_blocks', 'input_blocks', 'middle_block', 'output_blocks',
'double_blocks', 'single_blocks', 'joint_blocks'
]
for path in block_paths:
if hasattr(diffusion_model, path):
attr = getattr(diffusion_model, path)
if isinstance(attr, (list, torch.nn.ModuleList)):
for item in attr:
all_blocks.append(item)
elif isinstance(attr, torch.nn.Module) and path in ['mid_block', 'middle_block']:
all_blocks.append(attr)
if not all_blocks:
return
total_blocks = len(all_blocks)
# 🛡️ 防呆:用户输入不可超过模型总块数
if blocks_to_swap > total_blocks:
effective = total_blocks
tag = "匹配最大可分块参数"
else:
effective = blocks_to_swap
tag = "匹配用户设置的参数"
print(f"\033[96m[XB UNet Block Swap]\033[0m: 静态物理分块交换已激活!已找到 {total_blocks} 个引擎模块,分割 {effective} 个分块({tag})。")
for b, block in tqdm(enumerate(all_blocks), total=total_blocks, desc="Slicing UNet pipeline"):
if b > effective:
block.to(main_device)
else:
block.to(model_patcher.offload_device)
mm.soft_empty_cache()
gc.collect()
unet_model = unet_model.clone()
unet_model.add_callback(CallbacksMP.ON_LOAD, swap_blocks)
return (unet_model, )
# ============================================================
# XB_CheckpointBlockSwap — Checkpoint 分块交换 (显存优化)
# ============================================================
class XB_CheckpointBlockSwap:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"checkpoint_model": (anyType, {"tooltip": "Accepts any model input (including CLIP/VAE)."}),
"blocks_to_swap": ("INT", {"default": 15, "min": 0, "max": 1000, "step": 1}),
"offload_img_emb": ("BOOLEAN", {"default": True}),
"offload_txt_emb": ("BOOLEAN", {"default": True}),
"use_non_blocking": ("BOOLEAN", {"default": False}),
},
}
RETURN_TYPES = (anyType,)
RETURN_NAMES = ("checkpoint_model",)
CATEGORY = "XB_ToolBox/VRAM_Hacks"
FUNCTION = "set_callback"
def set_callback(self, checkpoint_model: ModelPatcher, blocks_to_swap, offload_txt_emb, offload_img_emb, use_non_blocking):
if not isinstance(checkpoint_model, ModelPatcher):
return (checkpoint_model,)
# ── 0 = 关闭分块功能:直接穿透,不做任何检测、不输出动态显存休眠警告 ──
if blocks_to_swap == 0:
print("\033[96m[XB Checkpoint Block Swap]\033[0m: 分块数量 = 0,分块功能已关闭(节点穿透)")
return (checkpoint_model,)
if is_dynamic_vram_active():
return (checkpoint_model,)
def swap_blocks(model_patcher: ModelPatcher, device_to, lowvram_model_memory, force_patch_weights, full_load):
base_model = model_patcher.model
# 🛡️ 动态获取 ComfyUI 分配的 GPU 设备
main_device = model_patcher.load_device
diffusion_model = getattr(base_model, 'diffusion_model', None)
if not diffusion_model:
return
unsupported_name = is_unsupported_model(diffusion_model)
if unsupported_name:
print(f"\033[93m[XB-BOX 拦截盾]\033[0m: 模型 ({unsupported_name}) 属于高耦合架构,不支持物理分块。已自动跳过!")
return
all_blocks = []
block_paths = [
'transformer_blocks', 'blocks', 'down_blocks', 'up_blocks', 'mid_block',
'layers', 'attention_blocks', 'input_blocks', 'middle_block', 'output_blocks',
'double_blocks', 'single_blocks', 'joint_blocks'
]
for path in block_paths:
if hasattr(diffusion_model, path):
attr = getattr(diffusion_model, path)
if isinstance(attr, (list, torch.nn.ModuleList)):
for item in attr:
all_blocks.append(item)
elif isinstance(attr, torch.nn.Module) and path in ['mid_block', 'middle_block']:
all_blocks.append(attr)
if all_blocks:
total_blocks = len(all_blocks)
# 🛡️ 防呆:用户输入不可超过模型总块数
if blocks_to_swap > total_blocks:
effective = total_blocks
tag = "匹配最大可分块参数"
else:
effective = blocks_to_swap
tag = "匹配用户设置的参数"
print(f"\033[96m[XB Checkpoint Block Swap]\033[0m: 静态物理分块交换已激活!已找到 {total_blocks} 个引擎模块,分割 {effective} 个分块({tag})。")
for b, block in tqdm(enumerate(all_blocks), total=total_blocks, desc="Slicing Checkpoint pipeline"):
if b > effective:
block.to(main_device)
else:
block.to(model_patcher.offload_device)
if offload_txt_emb:
for path in ['text_embedding', 'caption_encoder', 'text_encoder']:
if hasattr(diffusion_model, path) and getattr(diffusion_model, path) is not None:
getattr(diffusion_model, path).to(model_patcher.offload_device, non_blocking=use_non_blocking)
break
if offload_img_emb:
for path in ['img_emb', 'image_encoder', 'visual_encoder']:
if hasattr(diffusion_model, path) and getattr(diffusion_model, path) is not None:
getattr(diffusion_model, path).to(model_patcher.offload_device, non_blocking=use_non_blocking)
break
mm.soft_empty_cache()
gc.collect()
checkpoint_model = checkpoint_model.clone()
checkpoint_model.add_callback(CallbacksMP.ON_LOAD, swap_blocks)
import comfy.model_management as mm
import gc
import torch
import sys
from comfy.patcher_extension import CallbacksMP
from comfy.model_patcher import ModelPatcher
try:
from tqdm import tqdm
except ImportError:
def tqdm(iterable, **kwargs):
return iterable
class AnyType(str):
def __ne__(self, __value: object) -> bool:
return False
anyType = AnyType("*")
def is_dynamic_vram_active():
"""仅当用户显式传参 --enable-dynamic-vram 时视为动态显存激活,分块节点休眠。"""
if "--enable-dynamic-vram" in sys.argv:
print("\n\033[93m[XB Block Swap]\033[0m: ⚠️ 检测到 --enable-dynamic-vram,动态显存已启用。分块节点自动休眠。")
return True
return False
def is_unsupported_model(diffusion_model):
"""检测已知不支持物理分块的模型架构(仅黑名单,不检测量化格式)。
Qwen / FLUX 等模型的 GGUF 版本虽然参数类型特殊,但 module.to() 仍可正常工作,
不应被拦截。只有当模型的 C++ 量化引擎(如 comfy_kitchen)确实会因 module.to()
而崩溃时才需要拦截——但这类崩溃无法通过参数类型静态判断,由用户自行取舍。
"""
model_type = type(diffusion_model).__name__
blacklist = ["Lumina", "Lumina2", "ZImage", "HunyuanDiT", "ErnieImageModel"]
for b in blacklist:
if b in model_type:
return model_type
return None
# ============================================================
# XB_UNetBlockSwap — UNet 分块交换 (显存优化)
# ============================================================
class XB_UNetBlockSwap:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"unet_model": (anyType, {"tooltip": "Accepts pure UNet/DiT model input only"}),
"blocks_to_swap": ("INT", {
"default": 10,
"min": 0,
"max": 200,
"step": 1,
"tooltip": "Number of core blocks to offload to system RAM. Higher values save VRAM but slow down generation."
}),
},
}
RETURN_TYPES = (anyType,)
RETURN_NAMES = ("unet_model",)
CATEGORY = "XB_ToolBox/VRAM_Hacks"
FUNCTION = "set_callback"
def set_callback(self, unet_model: ModelPatcher, blocks_to_swap):
if not isinstance(unet_model, ModelPatcher):
return (unet_model,)
# ── 0 = 关闭分块功能:直接穿透,不做任何检测、不输出动态显存休眠警告 ──
if blocks_to_swap == 0:
print("\033[96m[XB UNet Block Swap]\033[0m: 分块数量 = 0,分块功能已关闭(节点穿透)")
return (unet_model,)
if is_dynamic_vram_active():
return (unet_model,)
def swap_blocks(model_patcher: ModelPatcher, device_to, lowvram_model_memory, force_patch_weights, full_load):
base_model = model_patcher.model
# 🛡️ 动态获取 ComfyUI 分配的 GPU 设备,绝不写死 'cuda:0'
main_device = model_patcher.load_device
diffusion_model = getattr(base_model, 'diffusion_model', None)
if not diffusion_model:
return
unsupported_name = is_unsupported_model(diffusion_model)
if unsupported_name:
print(f"\033[93m[XB-BOX 拦截盾]\033[0m: 模型 ({unsupported_name}) 属于高耦合架构,不支持物理分块。已自动跳过!")
return
all_blocks = []
block_paths = [
'transformer_blocks', 'blocks', 'down_blocks', 'up_blocks', 'mid_block',
'layers', 'attention_blocks', 'input_blocks', 'middle_block', 'output_blocks',
'double_blocks', 'single_blocks', 'joint_blocks'
]
for path in block_paths:
if hasattr(diffusion_model, path):
attr = getattr(diffusion_model, path)
if isinstance(attr, (list, torch.nn.ModuleList)):
for item in attr:
all_blocks.append(item)
elif isinstance(attr, torch.nn.Module) and path in ['mid_block', 'middle_block']:
all_blocks.append(attr)
if not all_blocks:
return
total_blocks = len(all_blocks)
# 🛡️ 防呆:用户输入不可超过模型总块数
if blocks_to_swap > total_blocks:
effective = total_blocks
tag = "匹配最大可分块参数"
else:
effective = blocks_to_swap
tag = "匹配用户设置的参数"
print(f"\033[96m[XB UNet Block Swap]\033[0m: 静态物理分块交换已激活!已找到 {total_blocks} 个引擎模块,分割 {effective} 个分块({tag})。")
for b, block in tqdm(enumerate(all_blocks), total=total_blocks, desc="Slicing UNet pipeline"):
if b > effective:
block.to(main_device)
else:
block.to(model_patcher.offload_device)
mm.soft_empty_cache()
gc.collect()
unet_model = unet_model.clone()
unet_model.add_callback(CallbacksMP.ON_LOAD, swap_blocks)
return (unet_model, )
# ============================================================
# XB_CheckpointBlockSwap — Checkpoint 分块交换 (显存优化)
# ============================================================
class XB_CheckpointBlockSwap:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"checkpoint_model": (anyType, {"tooltip": "Accepts any model input (including CLIP/VAE)."}),
"blocks_to_swap": ("INT", {"default": 15, "min": 0, "max": 1000, "step": 1}),
"offload_img_emb": ("BOOLEAN", {"default": True}),
"offload_txt_emb": ("BOOLEAN", {"default": True}),
"use_non_blocking": ("BOOLEAN", {"default": False}),
},
}
RETURN_TYPES = (anyType,)
RETURN_NAMES = ("checkpoint_model",)
CATEGORY = "XB_ToolBox/VRAM_Hacks"
FUNCTION = "set_callback"
def set_callback(self, checkpoint_model: ModelPatcher, blocks_to_swap, offload_txt_emb, offload_img_emb, use_non_blocking):
if not isinstance(checkpoint_model, ModelPatcher):
return (checkpoint_model,)
# ── 0 = 关闭分块功能:直接穿透,不做任何检测、不输出动态显存休眠警告 ──
if blocks_to_swap == 0:
print("\033[96m[XB Checkpoint Block Swap]\033[0m: 分块数量 = 0,分块功能已关闭(节点穿透)")
return (checkpoint_model,)
if is_dynamic_vram_active():
return (checkpoint_model,)
def swap_blocks(model_patcher: ModelPatcher, device_to, lowvram_model_memory, force_patch_weights, full_load):
base_model = model_patcher.model
# 🛡️ 动态获取 ComfyUI 分配的 GPU 设备
main_device = model_patcher.load_device
diffusion_model = getattr(base_model, 'diffusion_model', None)
if not diffusion_model:
return
unsupported_name = is_unsupported_model(diffusion_model)
if unsupported_name:
print(f"\033[93m[XB-BOX 拦截盾]\033[0m: 模型 ({unsupported_name}) 属于高耦合架构,不支持物理分块。已自动跳过!")
return
all_blocks = []
block_paths = [
'transformer_blocks', 'blocks', 'down_blocks', 'up_blocks', 'mid_block',
'layers', 'attention_blocks', 'input_blocks', 'middle_block', 'output_blocks',
'double_blocks', 'single_blocks', 'joint_blocks'
]
for path in block_paths:
if hasattr(diffusion_model, path):
attr = getattr(diffusion_model, path)
if isinstance(attr, (list, torch.nn.ModuleList)):
for item in attr:
all_blocks.append(item)
elif isinstance(attr, torch.nn.Module) and path in ['mid_block', 'middle_block']:
all_blocks.append(attr)
if all_blocks:
total_blocks = len(all_blocks)
# 🛡️ 防呆:用户输入不可超过模型总块数
if blocks_to_swap > total_blocks:
effective = total_blocks
tag = "匹配最大可分块参数"
else:
effective = blocks_to_swap
tag = "匹配用户设置的参数"
print(f"\033[96m[XB Checkpoint Block Swap]\033[0m: 静态物理分块交换已激活!已找到 {total_blocks} 个引擎模块,分割 {effective} 个分块({tag})。")
for b, block in tqdm(enumerate(all_blocks), total=total_blocks, desc="Slicing Checkpoint pipeline"):
if b > effective:
block.to(main_device)
else:
block.to(model_patcher.offload_device)
if offload_txt_emb:
for path in ['text_embedding', 'caption_encoder', 'text_encoder']:
if hasattr(diffusion_model, path) and getattr(diffusion_model, path) is not None:
getattr(diffusion_model, path).to(model_patcher.offload_device, non_blocking=use_non_blocking)
break
if offload_img_emb:
for path in ['img_emb', 'image_encoder', 'visual_encoder']:
if hasattr(diffusion_model, path) and getattr(diffusion_model, path) is not None:
getattr(diffusion_model, path).to(model_patcher.offload_device, non_blocking=use_non_blocking)
break
mm.soft_empty_cache()
gc.collect()
checkpoint_model = checkpoint_model.clone()
checkpoint_model.add_callback(CallbacksMP.ON_LOAD, swap_blocks)
return (checkpoint_model, )
+24 -24
View File
@@ -1,24 +1,24 @@
"""
XB-BOX Canvas Label — 画布文字标签节点
======================================
纯展示型节点,用于在工作流画布上添加说明文字。
所有渲染逻辑由前端 JS (xb_label.js) 处理。
"""
# ============================================================
# XB_CanvasLabel — 画布文字标签
# ============================================================
class XB_CanvasLabel:
"""纯显示标签节点 - 在画布上展示自定义文字,兼容所有 ComfyUI 版本"""
@classmethod
def INPUT_TYPES(cls):
return {"required": {}}
RETURN_TYPES = ()
FUNCTION = "noop"
CATEGORY = "XB_ToolBox/Display"
def noop(self, **kwargs):
return ()
"""
XB-BOX Canvas Label — 画布文字标签节点
======================================
纯展示型节点,用于在工作流画布上添加说明文字。
所有渲染逻辑由前端 JS (xb_label.js) 处理。
"""
# ============================================================
# XB_CanvasLabel — 画布文字标签
# ============================================================
class XB_CanvasLabel:
"""纯显示标签节点 - 在画布上展示自定义文字,兼容所有 ComfyUI 版本"""
@classmethod
def INPUT_TYPES(cls):
return {"required": {}}
RETURN_TYPES = ()
FUNCTION = "noop"
CATEGORY = "XB_ToolBox/Display"
def noop(self, **kwargs):
return ()
+321 -321
View File
@@ -1,321 +1,321 @@
"""
XB-SageAttention Accelerator
=============================
保持原版 KJNodes 设计, 仅添加 GPU 信息日志。
工作模式:
- 关闭: 直接透传模型
- 自动: 委托 KJNodes auto 逻辑 (sageattn 默认参数)
- pytorch attention: 融合官方 ModelAttentionBackend — 原生 SDPA
- comfy kitchen attention: 融合官方 ModelAttentionBackend — INT8 量化注意力
- Nvidia-Sage: KJNodes auto 逻辑 (N卡)
- AMD-RDNA4-Sage 2.2: KJNodes auto 逻辑 (RDNA4)
- 内置模式 A~D: 自定义 sage_config (M/N/GROUP/WAVE/WARP/NSTAGES)
- 自定模式 A~C: 自定义 sage_config (配合外部调参软件)
A卡兼容:
- RDNA3: Sage 1.0.6 不支持 sage_config, 选中内置/自定时会 fallback 到 SDPA
- RDNA4: Sage 2.2.0 完全兼容
- N卡: Sage 2.x 完全兼容
"""
import torch
from comfy.ldm.modules.attention import wrap_attn
# ── GPU 信息 (仅日志用途) ──
def _is_rocm():
try:
return torch.cuda.is_available() and hasattr(torch.version, "hip") and torch.version.hip is not None
except Exception:
return False
def _gpu_label():
try:
if not torch.cuda.is_available():
return "CPU"
if _is_rocm():
arch = ""
try:
raw = torch.cuda.get_device_properties(0).gcnArchName
arch = " " + raw.split(":")[0] if raw else ""
except Exception:
pass
return f"ROCm{arch}"
return "CUDA"
except Exception:
return "Unknown"
_GPU = _gpu_label()
print(f"[XB-SageAttn] GPU: {_GPU}, Sage节点已就绪")
# ══════════════════════════════════════════════════════════
# XB_SageAttentionAccelerator (保持原版设计)
# ══════════════════════════════════════════════════════════
class XB_SageAttentionAccelerator:
"""SageAttention 加速补丁节点。
选项说明:
- 关闭: 节点完全不生效,直接透传模型
- 自动: 一字不差使用 KJNodes 的 auto 逻辑 (sageattn 默认参数)
- pytorch attention: 融合官方 ModelAttentionBackend,使用原生 SDPA
- comfy kitchen attention: 融合官方 ModelAttentionBackend,使用 INT8 量化注意力
- 内置模式 A~D: 使用指定 sage_config 参数 (M/N/GROUP/WAVE/WARP/NSTAGES)
- 自定模式 A~C: 自定义 sage_config 参数
"""
# 融合官方 ModelAttentionBackend 的两个后端
_OFFICIAL_BACKENDS = ("pytorch attention", "comfy kitchen attention")
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"model": ("MODEL",),
"preset": ([
"关闭",
"自动",
"pytorch attention",
"comfy kitchen attention",
"Nvidia-Sage",
"AMD-RDNA4-Sage 2.2",
"内置模式 A (128x128x32)",
"内置模式 B (128x64x96)",
"内置模式 C (128x16x16)",
"内置模式 D (64x64x16)",
"自定模式 A (机智启动器)",
"自定模式 B (机智启动器)",
"自定模式 C (机智启动器)",
], {"default": "关闭"}),
}
}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "XB_ToolBox/VRAM_Hacks"
def patch(self, model, preset):
if preset == "关闭":
print(f"\033[96m[XB-SageAttn]\033[0m: 已关闭, 模型透传 (GPU={_GPU})")
return (model,)
try:
# 融合官方 ModelAttentionBackend
if preset in self._OFFICIAL_BACKENDS:
return (self._apply_official_backend(model, preset),)
return self._apply_patch(model, preset)
except Exception as e:
import traceback
print(f"\n\033[93m[XB-SageAttn 警告] 节点异常,自动跳过!工作流继续运行。\033[0m")
print(f"\033[93m[XB-SageAttn 错误] GPU={_GPU}, preset={preset}\033[0m")
print(f"\033[93m[XB-SageAttn 详情]\033[0m {e}")
traceback.print_exc()
return (model,)
def _apply_official_backend(self, model, preset):
"""融合官方 ModelAttentionBackend 节点 (pytorch / comfy kitchen attention)。"""
import comfy.ldm.modules.attention as attn_mod
backend_name = "comfy_kitchen_int8" if preset == "comfy kitchen attention" else "pytorch"
attention_function = attn_mod.get_attention_function(backend_name, None)
if attention_function is None:
print(f"\033[93m[XB-SageAttn]\033[0m: '{preset}' 当前不可用 "
f"(需 Nvidia/AMD GPU 且安装 comfy_kitchen),已回退 pytorch attention")
attention_function = attn_mod.get_attention_function("pytorch")
preset = "pytorch attention (回退)"
m = model.clone()
if "transformer_options" not in m.model_options:
m.model_options["transformer_options"] = {}
# 必须用官方 setter: 它会正确传递 container_function (INT8 kernel 依赖)
m.set_model_optimized_attention(attention_function)
print(f"\033[96m[XB-SageAttn]\033[0m: 引擎 → \033[92m{preset}\033[0m "
f"(官方 ModelAttentionBackend, GPU={_GPU})")
return m
def _apply_patch(self, model, preset):
# ── 自动:一字不差使用 KJNodes 的 auto 逻辑 ──
if preset in ("自动", "Nvidia-Sage", "AMD-RDNA4-Sage 2.2"):
from sageattention import sageattn
def sage_func(q, k, v, is_causal=False, attn_mask=None, tensor_layout="NHD"):
return sageattn(q, k, v, is_causal=is_causal, attn_mask=attn_mask, tensor_layout=tensor_layout)
@wrap_attn
def attention_sage(q, k, v, heads, mask=None, attn_precision=None,
skip_reshape=False, skip_output_reshape=False, **kwargs):
if kwargs.get("low_precision_attention", True) is False:
from comfy.ldm.modules.attention import attention_pytorch
return attention_pytorch(q, k, v, heads, mask=mask,
skip_reshape=skip_reshape,
skip_output_reshape=skip_output_reshape, **kwargs)
in_dtype = v.dtype
if q.dtype == torch.float32 or k.dtype == torch.float32 or v.dtype == torch.float32:
q, k, v = q.to(torch.float16), k.to(torch.float16), v.to(torch.float16)
if skip_reshape:
b, _, _, dim_head = q.shape
tensor_layout = "HND"
else:
b, _, dim_head = q.shape
dim_head //= heads
q, k, v = map(
lambda t: t.view(b, -1, heads, dim_head),
(q, k, v),
)
tensor_layout = "NHD"
if mask is not None:
if mask.ndim == 2:
mask = mask.unsqueeze(0)
if mask.ndim == 3:
mask = mask.unsqueeze(1)
out = sage_func(q, k, v, attn_mask=mask, is_causal=False,
tensor_layout=tensor_layout).to(in_dtype)
if tensor_layout == "HND":
if not skip_output_reshape:
out = out.transpose(1, 2).reshape(b, -1, heads * dim_head)
else:
if skip_output_reshape:
out = out.transpose(1, 2)
else:
out = out.reshape(b, -1, heads * dim_head)
return out
m = model.clone()
if "transformer_options" not in m.model_options:
m.model_options["transformer_options"] = {}
m.model_options["transformer_options"]["optimized_attention_override"] = \
lambda func, *args, **kwargs: attention_sage.__wrapped__(*args, **kwargs)
print(f"\033[96m[XB-SageAttn]\033[0m: 引擎 → \033[92m自动 (KJNodes auto)\033[0m (GPU={_GPU})")
return (m,)
# ── 内置/自定模式:使用指定 sage_config 参数 ──
configs = {
"内置模式 A (128x128x32)": {'M': 128, 'N': 128, 'GROUP': 32, 'WAVE': 2, 'WARP': 8, 'NSTAGES': 1},
"内置模式 B (128x64x96)": {'M': 128, 'N': 64, 'GROUP': 96, 'WAVE': 3, 'WARP': 8, 'NSTAGES': 2},
"内置模式 C (128x16x16)": {'M': 128, 'N': 16, 'GROUP': 16, 'WAVE': 2, 'WARP': 4, 'NSTAGES': 2},
"内置模式 D (64x64x16)": {'M': 64, 'N': 64, 'GROUP': 16, 'WAVE': 4, 'WARP': 4, 'NSTAGES': 2},
"自定模式 A (机智启动器)": {'M': 128, 'N': 128, 'GROUP': 16, 'WAVE': 4, 'WARP': 8, 'NSTAGES': 1},
"自定模式 B (机智启动器)": {'M': 128, 'N': 16, 'GROUP': 8, 'WAVE': 2, 'WARP': 8, 'NSTAGES': 2},
"自定模式 C (机智启动器)": {'M': 64, 'N': 128, 'GROUP': 96, 'WAVE': 2, 'WARP': 2, 'NSTAGES': 1},
}
selected_cfg = configs[preset]
from sageattention import sageattn
_warned = {}
def attention_override_sage(func, q, k, v, heads, mask=None, attn_precision=None,
skip_reshape=False, skip_output_reshape=False, **kwargs):
in_dtype = v.dtype
# 3D/4D 维度校验
if q.ndim != 3 and not skip_reshape:
return func(q, k, v, heads, mask=mask, attn_precision=attn_precision,
skip_reshape=skip_reshape, skip_output_reshape=skip_output_reshape, **kwargs)
if skip_reshape:
b, _, _, dim_head = q.shape
tensor_layout = "HND"
else:
b, _, d = q.shape
dim_head = d // heads
tensor_layout = "NHD"
if mask is not None:
if "mask" not in _warned:
print("\033[93m[SageAttn]\033[0m: Attention mask detected — falling back to SDPA.")
_warned["mask"] = True
return func(q, k, v, heads, mask=mask, attn_precision=attn_precision,
skip_reshape=skip_reshape, skip_output_reshape=skip_output_reshape, **kwargs)
# 动态补零策略
_STANDARD_DIMS = (16, 32, 64, 128, 256)
_orig_dim_head = dim_head
_pad_target = None
if dim_head not in _STANDARD_DIMS:
for _sd in _STANDARD_DIMS:
if _sd >= dim_head:
_pad_target = _sd
break
if _pad_target is not None:
if "pad" not in _warned:
print("\033[96m[SageAttn]\033[0m: Padding dim_head {} → {} for SageAttention compatibility."
.format(dim_head, _pad_target))
_warned["pad"] = True
_scale = (float(_pad_target) / float(_orig_dim_head)) ** 0.25
else:
if "large_dim" not in _warned:
print("\033[93m[SageAttn]\033[0m: dim_head ({}) exceeds max supported — falling back to SDPA."
.format(_orig_dim_head))
_warned["large_dim"] = True
return func(q, k, v, heads, mask=mask, attn_precision=attn_precision,
skip_reshape=skip_reshape, skip_output_reshape=skip_output_reshape, **kwargs)
if q.dtype == torch.float32 or k.dtype == torch.float32 or v.dtype == torch.float32:
q, k, v = q.to(torch.float16), k.to(torch.float16), v.to(torch.float16)
if not skip_reshape:
q = q.view(b, -1, heads, _orig_dim_head)
k = k.view(b, -1, heads, _orig_dim_head)
v = v.view(b, -1, heads, _orig_dim_head)
_orig_q, _orig_k, _orig_v = q, k, v
if _pad_target is not None:
_pad_len = _pad_target - _orig_dim_head
q = torch.nn.functional.pad(q, (0, _pad_len))
k = torch.nn.functional.pad(k, (0, _pad_len))
v = torch.nn.functional.pad(v, (0, _pad_len))
q = q * _scale
k = k * _scale
dim_head = _pad_target
is_causal = kwargs.get("is_causal", False)
try:
out = sageattn(q, k, v, is_causal=is_causal, tensor_layout=tensor_layout,
sage_config=selected_cfg).to(in_dtype)
except Exception as e:
if "sage_fallback" not in _warned:
print("\033[93m[SageAttn]\033[0m: SageAttention kernel failed (dim_head={}) — falling back to SDPA. "
"Error: {}".format(_orig_dim_head, str(e)[:120]))
_warned["sage_fallback"] = True
if skip_reshape:
return func(_orig_q, _orig_k, _orig_v, heads, mask=mask, attn_precision=attn_precision,
skip_reshape=True, skip_output_reshape=skip_output_reshape, **kwargs)
else:
q_3d = _orig_q.reshape(b, -1, heads * _orig_dim_head)
k_3d = _orig_k.reshape(b, -1, heads * _orig_dim_head)
v_3d = _orig_v.reshape(b, -1, heads * _orig_dim_head)
return func(q_3d, k_3d, v_3d, heads, mask=mask, attn_precision=attn_precision,
skip_reshape=False, skip_output_reshape=False, **kwargs)
if _pad_target is not None:
out = out[..., :_orig_dim_head]
dim_head = _orig_dim_head
if tensor_layout == "HND":
if not skip_output_reshape:
out = out.transpose(1, 2).reshape(b, -1, heads * dim_head)
else:
if skip_output_reshape:
out = out.transpose(1, 2)
else:
out = out.reshape(b, -1, heads * dim_head)
return out
m = model.clone()
if "transformer_options" not in m.model_options:
m.model_options["transformer_options"] = {}
m.model_options["transformer_options"]["optimized_attention_override"] = attention_override_sage
print(f"\033[96m[XB-SageAttn]\033[0m: 引擎 → \033[92m{preset}\033[0m (GPU={_GPU})")
return (m,)
"""
XB-SageAttention Accelerator
=============================
保持原版 KJNodes 设计, 仅添加 GPU 信息日志。
工作模式:
- 关闭: 直接透传模型
- 自动: 委托 KJNodes auto 逻辑 (sageattn 默认参数)
- pytorch attention: 融合官方 ModelAttentionBackend — 原生 SDPA
- comfy kitchen attention: 融合官方 ModelAttentionBackend — INT8 量化注意力
- Nvidia-Sage: KJNodes auto 逻辑 (N卡)
- AMD-RDNA4-Sage 2.2: KJNodes auto 逻辑 (RDNA4)
- 内置模式 A~D: 自定义 sage_config (M/N/GROUP/WAVE/WARP/NSTAGES)
- 自定模式 A~C: 自定义 sage_config (配合外部调参软件)
A卡兼容:
- RDNA3: Sage 1.0.6 不支持 sage_config, 选中内置/自定时会 fallback 到 SDPA
- RDNA4: Sage 2.2.0 完全兼容
- N卡: Sage 2.x 完全兼容
"""
import torch
from comfy.ldm.modules.attention import wrap_attn
# ── GPU 信息 (仅日志用途) ──
def _is_rocm():
try:
return torch.cuda.is_available() and hasattr(torch.version, "hip") and torch.version.hip is not None
except Exception:
return False
def _gpu_label():
try:
if not torch.cuda.is_available():
return "CPU"
if _is_rocm():
arch = ""
try:
raw = torch.cuda.get_device_properties(0).gcnArchName
arch = " " + raw.split(":")[0] if raw else ""
except Exception:
pass
return f"ROCm{arch}"
return "CUDA"
except Exception:
return "Unknown"
_GPU = _gpu_label()
print(f"[XB-SageAttn] GPU: {_GPU}, Sage节点已就绪")
# ══════════════════════════════════════════════════════════
# XB_SageAttentionAccelerator (保持原版设计)
# ══════════════════════════════════════════════════════════
class XB_SageAttentionAccelerator:
"""SageAttention 加速补丁节点。
选项说明:
- 关闭: 节点完全不生效,直接透传模型
- 自动: 一字不差使用 KJNodes 的 auto 逻辑 (sageattn 默认参数)
- pytorch attention: 融合官方 ModelAttentionBackend,使用原生 SDPA
- comfy kitchen attention: 融合官方 ModelAttentionBackend,使用 INT8 量化注意力
- 内置模式 A~D: 使用指定 sage_config 参数 (M/N/GROUP/WAVE/WARP/NSTAGES)
- 自定模式 A~C: 自定义 sage_config 参数
"""
# 融合官方 ModelAttentionBackend 的两个后端
_OFFICIAL_BACKENDS = ("pytorch attention", "comfy kitchen attention")
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"model": ("MODEL",),
"preset": ([
"关闭",
"自动",
"pytorch attention",
"comfy kitchen attention",
"Nvidia-Sage",
"AMD-RDNA4-Sage 2.2",
"内置模式 A (128x128x32)",
"内置模式 B (128x64x96)",
"内置模式 C (128x16x16)",
"内置模式 D (64x64x16)",
"自定模式 A (机智启动器)",
"自定模式 B (机智启动器)",
"自定模式 C (机智启动器)",
], {"default": "关闭"}),
}
}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "XB_ToolBox/VRAM_Hacks"
def patch(self, model, preset):
if preset == "关闭":
print(f"\033[96m[XB-SageAttn]\033[0m: 已关闭, 模型透传 (GPU={_GPU})")
return (model,)
try:
# 融合官方 ModelAttentionBackend
if preset in self._OFFICIAL_BACKENDS:
return (self._apply_official_backend(model, preset),)
return self._apply_patch(model, preset)
except Exception as e:
import traceback
print(f"\n\033[93m[XB-SageAttn 警告] 节点异常,自动跳过!工作流继续运行。\033[0m")
print(f"\033[93m[XB-SageAttn 错误] GPU={_GPU}, preset={preset}\033[0m")
print(f"\033[93m[XB-SageAttn 详情]\033[0m {e}")
traceback.print_exc()
return (model,)
def _apply_official_backend(self, model, preset):
"""融合官方 ModelAttentionBackend 节点 (pytorch / comfy kitchen attention)。"""
import comfy.ldm.modules.attention as attn_mod
backend_name = "comfy_kitchen_int8" if preset == "comfy kitchen attention" else "pytorch"
attention_function = attn_mod.get_attention_function(backend_name, None)
if attention_function is None:
print(f"\033[93m[XB-SageAttn]\033[0m: '{preset}' 当前不可用 "
f"(需 Nvidia/AMD GPU 且安装 comfy_kitchen),已回退 pytorch attention")
attention_function = attn_mod.get_attention_function("pytorch")
preset = "pytorch attention (回退)"
m = model.clone()
if "transformer_options" not in m.model_options:
m.model_options["transformer_options"] = {}
# 必须用官方 setter: 它会正确传递 container_function (INT8 kernel 依赖)
m.set_model_optimized_attention(attention_function)
print(f"\033[96m[XB-SageAttn]\033[0m: 引擎 → \033[92m{preset}\033[0m "
f"(官方 ModelAttentionBackend, GPU={_GPU})")
return m
def _apply_patch(self, model, preset):
# ── 自动:一字不差使用 KJNodes 的 auto 逻辑 ──
if preset in ("自动", "Nvidia-Sage", "AMD-RDNA4-Sage 2.2"):
from sageattention import sageattn
def sage_func(q, k, v, is_causal=False, attn_mask=None, tensor_layout="NHD"):
return sageattn(q, k, v, is_causal=is_causal, attn_mask=attn_mask, tensor_layout=tensor_layout)
@wrap_attn
def attention_sage(q, k, v, heads, mask=None, attn_precision=None,
skip_reshape=False, skip_output_reshape=False, **kwargs):
if kwargs.get("low_precision_attention", True) is False:
from comfy.ldm.modules.attention import attention_pytorch
return attention_pytorch(q, k, v, heads, mask=mask,
skip_reshape=skip_reshape,
skip_output_reshape=skip_output_reshape, **kwargs)
in_dtype = v.dtype
if q.dtype == torch.float32 or k.dtype == torch.float32 or v.dtype == torch.float32:
q, k, v = q.to(torch.float16), k.to(torch.float16), v.to(torch.float16)
if skip_reshape:
b, _, _, dim_head = q.shape
tensor_layout = "HND"
else:
b, _, dim_head = q.shape
dim_head //= heads
q, k, v = map(
lambda t: t.view(b, -1, heads, dim_head),
(q, k, v),
)
tensor_layout = "NHD"
if mask is not None:
if mask.ndim == 2:
mask = mask.unsqueeze(0)
if mask.ndim == 3:
mask = mask.unsqueeze(1)
out = sage_func(q, k, v, attn_mask=mask, is_causal=False,
tensor_layout=tensor_layout).to(in_dtype)
if tensor_layout == "HND":
if not skip_output_reshape:
out = out.transpose(1, 2).reshape(b, -1, heads * dim_head)
else:
if skip_output_reshape:
out = out.transpose(1, 2)
else:
out = out.reshape(b, -1, heads * dim_head)
return out
m = model.clone()
if "transformer_options" not in m.model_options:
m.model_options["transformer_options"] = {}
m.model_options["transformer_options"]["optimized_attention_override"] = \
lambda func, *args, **kwargs: attention_sage.__wrapped__(*args, **kwargs)
print(f"\033[96m[XB-SageAttn]\033[0m: 引擎 → \033[92m自动 (KJNodes auto)\033[0m (GPU={_GPU})")
return (m,)
# ── 内置/自定模式:使用指定 sage_config 参数 ──
configs = {
"内置模式 A (128x128x32)": {'M': 128, 'N': 128, 'GROUP': 32, 'WAVE': 2, 'WARP': 8, 'NSTAGES': 1},
"内置模式 B (128x64x96)": {'M': 128, 'N': 64, 'GROUP': 96, 'WAVE': 3, 'WARP': 8, 'NSTAGES': 2},
"内置模式 C (128x16x16)": {'M': 128, 'N': 16, 'GROUP': 16, 'WAVE': 2, 'WARP': 4, 'NSTAGES': 2},
"内置模式 D (64x64x16)": {'M': 64, 'N': 64, 'GROUP': 16, 'WAVE': 4, 'WARP': 4, 'NSTAGES': 2},
"自定模式 A (机智启动器)": {'M': 128, 'N': 128, 'GROUP': 16, 'WAVE': 4, 'WARP': 8, 'NSTAGES': 1},
"自定模式 B (机智启动器)": {'M': 128, 'N': 16, 'GROUP': 8, 'WAVE': 2, 'WARP': 8, 'NSTAGES': 2},
"自定模式 C (机智启动器)": {'M': 64, 'N': 128, 'GROUP': 96, 'WAVE': 2, 'WARP': 2, 'NSTAGES': 1},
}
selected_cfg = configs[preset]
from sageattention import sageattn
_warned = {}
def attention_override_sage(func, q, k, v, heads, mask=None, attn_precision=None,
skip_reshape=False, skip_output_reshape=False, **kwargs):
in_dtype = v.dtype
# 3D/4D 维度校验
if q.ndim != 3 and not skip_reshape:
return func(q, k, v, heads, mask=mask, attn_precision=attn_precision,
skip_reshape=skip_reshape, skip_output_reshape=skip_output_reshape, **kwargs)
if skip_reshape:
b, _, _, dim_head = q.shape
tensor_layout = "HND"
else:
b, _, d = q.shape
dim_head = d // heads
tensor_layout = "NHD"
if mask is not None:
if "mask" not in _warned:
print("\033[93m[SageAttn]\033[0m: Attention mask detected — falling back to SDPA.")
_warned["mask"] = True
return func(q, k, v, heads, mask=mask, attn_precision=attn_precision,
skip_reshape=skip_reshape, skip_output_reshape=skip_output_reshape, **kwargs)
# 动态补零策略
_STANDARD_DIMS = (16, 32, 64, 128, 256)
_orig_dim_head = dim_head
_pad_target = None
if dim_head not in _STANDARD_DIMS:
for _sd in _STANDARD_DIMS:
if _sd >= dim_head:
_pad_target = _sd
break
if _pad_target is not None:
if "pad" not in _warned:
print("\033[96m[SageAttn]\033[0m: Padding dim_head {} → {} for SageAttention compatibility."
.format(dim_head, _pad_target))
_warned["pad"] = True
_scale = (float(_pad_target) / float(_orig_dim_head)) ** 0.25
else:
if "large_dim" not in _warned:
print("\033[93m[SageAttn]\033[0m: dim_head ({}) exceeds max supported — falling back to SDPA."
.format(_orig_dim_head))
_warned["large_dim"] = True
return func(q, k, v, heads, mask=mask, attn_precision=attn_precision,
skip_reshape=skip_reshape, skip_output_reshape=skip_output_reshape, **kwargs)
if q.dtype == torch.float32 or k.dtype == torch.float32 or v.dtype == torch.float32:
q, k, v = q.to(torch.float16), k.to(torch.float16), v.to(torch.float16)
if not skip_reshape:
q = q.view(b, -1, heads, _orig_dim_head)
k = k.view(b, -1, heads, _orig_dim_head)
v = v.view(b, -1, heads, _orig_dim_head)
_orig_q, _orig_k, _orig_v = q, k, v
if _pad_target is not None:
_pad_len = _pad_target - _orig_dim_head
q = torch.nn.functional.pad(q, (0, _pad_len))
k = torch.nn.functional.pad(k, (0, _pad_len))
v = torch.nn.functional.pad(v, (0, _pad_len))
q = q * _scale
k = k * _scale
dim_head = _pad_target
is_causal = kwargs.get("is_causal", False)
try:
out = sageattn(q, k, v, is_causal=is_causal, tensor_layout=tensor_layout,
sage_config=selected_cfg).to(in_dtype)
except Exception as e:
if "sage_fallback" not in _warned:
print("\033[93m[SageAttn]\033[0m: SageAttention kernel failed (dim_head={}) — falling back to SDPA. "
"Error: {}".format(_orig_dim_head, str(e)[:120]))
_warned["sage_fallback"] = True
if skip_reshape:
return func(_orig_q, _orig_k, _orig_v, heads, mask=mask, attn_precision=attn_precision,
skip_reshape=True, skip_output_reshape=skip_output_reshape, **kwargs)
else:
q_3d = _orig_q.reshape(b, -1, heads * _orig_dim_head)
k_3d = _orig_k.reshape(b, -1, heads * _orig_dim_head)
v_3d = _orig_v.reshape(b, -1, heads * _orig_dim_head)
return func(q_3d, k_3d, v_3d, heads, mask=mask, attn_precision=attn_precision,
skip_reshape=False, skip_output_reshape=False, **kwargs)
if _pad_target is not None:
out = out[..., :_orig_dim_head]
dim_head = _orig_dim_head
if tensor_layout == "HND":
if not skip_output_reshape:
out = out.transpose(1, 2).reshape(b, -1, heads * dim_head)
else:
if skip_output_reshape:
out = out.transpose(1, 2)
else:
out = out.reshape(b, -1, heads * dim_head)
return out
m = model.clone()
if "transformer_options" not in m.model_options:
m.model_options["transformer_options"] = {}
m.model_options["transformer_options"]["optimized_attention_override"] = attention_override_sage
print(f"\033[96m[XB-SageAttn]\033[0m: 引擎 → \033[92m{preset}\033[0m (GPU={_GPU})")
return (m,)
+45 -45
View File
@@ -1,46 +1,46 @@
import folder_paths
class AnyType(str):
def __ne__(self, __value: object) -> bool:
return False
anyType = AnyType("*")
# ============================================================
# XB_UNetNameBroadcaster — UNet 名称广播器
# ============================================================
class XB_UNetNameBroadcaster:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"unet_name": (folder_paths.get_filename_list("unet"), ),
}
}
RETURN_TYPES = (anyType,)
RETURN_NAMES = ("UNet_Name",)
FUNCTION = "broadcast"
CATEGORY = "XB_ToolBox/Wiring"
def broadcast(self, unet_name):
return (unet_name,)
# ============================================================
# XB_CLIPNameBroadcaster — CLIP 名称广播器
# ============================================================
class XB_CLIPNameBroadcaster:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"clip_name": (folder_paths.get_filename_list("clip"), ),
}
}
RETURN_TYPES = (anyType,)
RETURN_NAMES = ("CLIP_Name",)
FUNCTION = "broadcast"
CATEGORY = "XB_ToolBox/Wiring"
def broadcast(self, clip_name):
import folder_paths
class AnyType(str):
def __ne__(self, __value: object) -> bool:
return False
anyType = AnyType("*")
# ============================================================
# XB_UNetNameBroadcaster — UNet 名称广播器
# ============================================================
class XB_UNetNameBroadcaster:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"unet_name": (folder_paths.get_filename_list("unet"), ),
}
}
RETURN_TYPES = (anyType,)
RETURN_NAMES = ("UNet_Name",)
FUNCTION = "broadcast"
CATEGORY = "XB_ToolBox/Wiring"
def broadcast(self, unet_name):
return (unet_name,)
# ============================================================
# XB_CLIPNameBroadcaster — CLIP 名称广播器
# ============================================================
class XB_CLIPNameBroadcaster:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"clip_name": (folder_paths.get_filename_list("clip"), ),
}
}
RETURN_TYPES = (anyType,)
RETURN_NAMES = ("CLIP_Name",)
FUNCTION = "broadcast"
CATEGORY = "XB_ToolBox/Wiring"
def broadcast(self, clip_name):
return (clip_name,)

Before

Width:  |  Height:  |  Size: 632 KiB

After

Width:  |  Height:  |  Size: 632 KiB

Some files were not shown because too many files have changed in this diff Show More